import asyncio import logging import os import shutil import sqlite3 import subprocess import time import uuid import zipfile from pathlib import Path from typing import Optional, List, Tuple from dotenv import load_dotenv from PIL import Image, ImageOps, UnidentifiedImageError from telegram import Update, InputMediaPhoto from telegram.error import TimedOut, NetworkError, RetryAfter, BadRequest from telegram.ext import ( Application, ApplicationBuilder, CommandHandler, ContextTypes, MessageHandler, filters, ) load_dotenv() BOT_TOKEN = os.getenv("BOT_TOKEN", "").strip() FFMPEG_PATH = os.getenv("FFMPEG_PATH", "/usr/bin/ffmpeg").strip() MAX_MEDIA_PER_FINAL = int(os.getenv("MAX_MEDIA_PER_FINAL", "1000")) STATUS_EVERY = int(os.getenv("STATUS_EVERY", "25")) SEND_DELAY = float(os.getenv("SEND_DELAY", "1.0")) DOWNLOAD_DELAY = float(os.getenv("DOWNLOAD_DELAY", "0.05")) BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" DOWNLOAD_DIR = BASE_DIR / "downloads" OUTPUT_DIR = BASE_DIR / "output" DB_PATH = DATA_DIR / "queue.sqlite3" for directory in [DATA_DIR, DOWNLOAD_DIR, OUTPUT_DIR]: directory.mkdir(parents=True, exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", ) logger = logging.getLogger("media_to_photo_bot") def db() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.execute("PRAGMA journal_mode=WAL") conn.execute( """ CREATE TABLE IF NOT EXISTS inbox_media ( id INTEGER PRIMARY KEY AUTOINCREMENT, chat_id INTEGER NOT NULL, user_id INTEGER NOT NULL, message_id INTEGER NOT NULL, file_id TEXT NOT NULL, file_unique_id TEXT, media_type TEXT NOT NULL, file_name TEXT, mime_type TEXT, queued INTEGER NOT NULL DEFAULT 0, processed INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL, UNIQUE(chat_id, user_id, message_id, file_unique_id) ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS media_queue ( id INTEGER PRIMARY KEY AUTOINCREMENT, inbox_id INTEGER, chat_id INTEGER NOT NULL, user_id INTEGER NOT NULL, message_id INTEGER, file_id TEXT NOT NULL, file_unique_id TEXT, media_type TEXT NOT NULL, file_name TEXT, mime_type TEXT, created_at INTEGER NOT NULL, processed INTEGER NOT NULL DEFAULT 0, UNIQUE(chat_id, user_id, file_unique_id, message_id) ) """ ) # Migration for old DBs cols = [r[1] for r in conn.execute("PRAGMA table_info(media_queue)").fetchall()] if "processed" not in cols: conn.execute("ALTER TABLE media_queue ADD COLUMN processed INTEGER NOT NULL DEFAULT 0") if "inbox_id" not in cols: conn.execute("ALTER TABLE media_queue ADD COLUMN inbox_id INTEGER") if "message_id" not in cols: conn.execute("ALTER TABLE media_queue ADD COLUMN message_id INTEGER") if "file_unique_id" not in cols: conn.execute("ALTER TABLE media_queue ADD COLUMN file_unique_id TEXT") conn.commit() return conn def extract_media(update: Update): msg = update.effective_message if not msg: return None file_id = None file_unique_id = None media_type = None file_name = None mime_type = None if msg.photo: best_photo = msg.photo[-1] file_id = best_photo.file_id file_unique_id = best_photo.file_unique_id media_type = "image" file_name = f"telegram_photo_{file_unique_id}.jpg" mime_type = "image/jpeg" elif msg.video: file_id = msg.video.file_id file_unique_id = msg.video.file_unique_id media_type = "video" file_name = msg.video.file_name or f"telegram_video_{file_unique_id}.mp4" mime_type = msg.video.mime_type or "video/mp4" elif msg.animation: file_id = msg.animation.file_id file_unique_id = msg.animation.file_unique_id media_type = "video" file_name = msg.animation.file_name or f"telegram_animation_{file_unique_id}.mp4" mime_type = msg.animation.mime_type or "video/mp4" elif msg.document: doc = msg.document file_id = doc.file_id file_unique_id = doc.file_unique_id file_name = doc.file_name or f"telegram_document_{file_unique_id}" mime_type = doc.mime_type or "" if mime_type.startswith("image/"): media_type = "image" elif mime_type.startswith("video/"): media_type = "video" else: return None if not file_id or not media_type: return None return { "chat_id": update.effective_chat.id, "user_id": update.effective_user.id, "message_id": msg.message_id, "file_id": file_id, "file_unique_id": file_unique_id or file_id, "media_type": media_type, "file_name": file_name, "mime_type": mime_type, } def save_to_inbox(media: dict) -> Optional[int]: conn = db() cur = conn.cursor() try: cur.execute( """ INSERT OR IGNORE INTO inbox_media (chat_id, user_id, message_id, file_id, file_unique_id, media_type, file_name, mime_type, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( media["chat_id"], media["user_id"], media["message_id"], media["file_id"], media["file_unique_id"], media["media_type"], media["file_name"], media["mime_type"], int(time.time()), ), ) conn.commit() row = conn.execute( """ SELECT id FROM inbox_media WHERE chat_id = ? AND user_id = ? AND message_id = ? AND file_unique_id = ? """, ( media["chat_id"], media["user_id"], media["message_id"], media["file_unique_id"], ), ).fetchone() return row[0] if row else None finally: conn.close() def add_to_queue(media: dict, inbox_id: Optional[int] = None) -> bool: conn = db() try: cur = conn.cursor() cur.execute( """ INSERT OR IGNORE INTO media_queue (inbox_id, chat_id, user_id, message_id, file_id, file_unique_id, media_type, file_name, mime_type, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( inbox_id, media["chat_id"], media["user_id"], media.get("message_id"), media["file_id"], media["file_unique_id"], media["media_type"], media["file_name"], media["mime_type"], int(time.time()), ), ) inserted = cur.rowcount > 0 if inbox_id: conn.execute("UPDATE inbox_media SET queued = 1 WHERE id = ?", (inbox_id,)) conn.commit() return inserted finally: conn.close() def scan_inbox_to_queue(chat_id: int, user_id: int) -> int: conn = db() try: rows = conn.execute( """ SELECT id, chat_id, user_id, message_id, file_id, file_unique_id, media_type, file_name, mime_type FROM inbox_media WHERE chat_id = ? AND user_id = ? AND queued = 0 AND processed = 0 ORDER BY id ASC """ , (chat_id, user_id), ).fetchall() added = 0 for row in rows: inbox_id, r_chat_id, r_user_id, message_id, file_id, file_unique_id, media_type, file_name, mime_type = row cur = conn.execute( """ INSERT OR IGNORE INTO media_queue (inbox_id, chat_id, user_id, message_id, file_id, file_unique_id, media_type, file_name, mime_type, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( inbox_id, r_chat_id, r_user_id, message_id, file_id, file_unique_id, media_type, file_name, mime_type, int(time.time()), ), ) if cur.rowcount > 0: added += 1 conn.execute("UPDATE inbox_media SET queued = 1 WHERE id = ?", (inbox_id,)) conn.commit() return added finally: conn.close() def get_queue(chat_id: int, user_id: int) -> List[Tuple]: conn = db() try: rows = conn.execute( """ SELECT id, inbox_id, file_id, media_type, file_name, mime_type FROM media_queue WHERE chat_id = ? AND user_id = ? AND processed = 0 ORDER BY id ASC LIMIT ? """, (chat_id, user_id, MAX_MEDIA_PER_FINAL), ).fetchall() return rows finally: conn.close() def queue_count(chat_id: int, user_id: int) -> int: conn = db() try: return conn.execute( """ SELECT COUNT(*) FROM media_queue WHERE chat_id = ? AND user_id = ? AND processed = 0 """, (chat_id, user_id), ).fetchone()[0] finally: conn.close() def inbox_count(chat_id: int, user_id: int) -> int: conn = db() try: return conn.execute( """ SELECT COUNT(*) FROM inbox_media WHERE chat_id = ? AND user_id = ? AND queued = 0 AND processed = 0 """, (chat_id, user_id), ).fetchone()[0] finally: conn.close() def clear_queue(chat_id: int, user_id: int) -> int: conn = db() try: cur = conn.cursor() cur.execute( """ DELETE FROM media_queue WHERE chat_id = ? AND user_id = ? AND processed = 0 """, (chat_id, user_id), ) deleted = cur.rowcount conn.execute( """ UPDATE inbox_media SET queued = 0 WHERE chat_id = ? AND user_id = ? AND processed = 0 """, (chat_id, user_id), ) conn.commit() return deleted finally: conn.close() def mark_processed(queue_ids: List[int]) -> None: if not queue_ids: return conn = db() try: placeholders = ",".join("?" for _ in queue_ids) inbox_ids = [ r[0] for r in conn.execute( f"SELECT inbox_id FROM media_queue WHERE id IN ({placeholders}) AND inbox_id IS NOT NULL", queue_ids, ).fetchall() ] conn.execute( f"UPDATE media_queue SET processed = 1 WHERE id IN ({placeholders})", queue_ids, ) if inbox_ids: inbox_placeholders = ",".join("?" for _ in inbox_ids) conn.execute( f"UPDATE inbox_media SET processed = 1, queued = 1 WHERE id IN ({inbox_placeholders})", inbox_ids, ) conn.commit() finally: conn.close() def delete_processed(chat_id: int, user_id: int) -> None: conn = db() try: conn.execute( "DELETE FROM media_queue WHERE chat_id = ? AND user_id = ? AND processed = 1", (chat_id, user_id), ) conn.commit() finally: conn.close() def safe_suffix(file_name: Optional[str], fallback: str) -> str: if not file_name: return fallback suffix = Path(file_name).suffix.lower() if not suffix: return fallback if len(suffix) > 10: return fallback return suffix def ffprobe_path() -> str: ffmpeg = Path(FFMPEG_PATH) possible = ffmpeg.with_name("ffprobe") return str(possible) if possible.exists() else "ffprobe" def get_video_duration(path: Path) -> Optional[float]: try: result = subprocess.run( [ ffprobe_path(), "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", str(path), ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=30, ) if result.returncode != 0: return None return float(result.stdout.strip()) except Exception: return None def convert_image_to_jpg(input_path: Path, output_path: Path) -> bool: try: with Image.open(input_path) as img: img = ImageOps.exif_transpose(img) if img.mode in ("RGBA", "LA", "P"): img = img.convert("RGBA") background = Image.new("RGBA", img.size, (255, 255, 255, 255)) background.alpha_composite(img) img = background.convert("RGB") else: img = img.convert("RGB") img.save(output_path, "JPEG", quality=90, optimize=True) return True except UnidentifiedImageError: return False except Exception as e: logger.exception("Image conversion failed: %s", e) return False def convert_video_to_jpg(input_path: Path, output_path: Path) -> bool: duration = get_video_duration(input_path) seek_time = 1 if duration and duration > 3: seek_time = max(1, int(duration / 2)) try: result = subprocess.run( [ FFMPEG_PATH, "-y", "-ss", str(seek_time), "-i", str(input_path), "-frames:v", "1", "-q:v", "3", str(output_path), ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=180, ) return result.returncode == 0 and output_path.exists() and output_path.stat().st_size > 0 except Exception as e: logger.exception("Video conversion failed: %s", e) return False async def safe_reply(msg, text: str): try: return await msg.reply_text(text, read_timeout=60, write_timeout=60, connect_timeout=30, pool_timeout=30) except Exception as e: logger.warning("safe_reply failed: %s", e) return None async def safe_edit(message, text: str): if not message: return try: return await message.edit_text(text, read_timeout=60, write_timeout=60, connect_timeout=30, pool_timeout=30) except BadRequest as e: if "Message is not modified" not in str(e): logger.warning("safe_edit bad request: %s", e) except Exception as e: logger.warning("safe_edit failed: %s", e) async def retry_async(func, *args, retries: int = 5, **kwargs): delay = 2 last_error = None for attempt in range(1, retries + 1): try: return await func(*args, **kwargs) except RetryAfter as e: await asyncio.sleep(int(e.retry_after) + 2) last_error = e except (TimedOut, NetworkError) as e: last_error = e logger.warning("Telegram network error attempt %s/%s: %s", attempt, retries, e) await asyncio.sleep(delay) delay = min(delay * 2, 30) raise last_error async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: text = ( "بات آماده است.\n\n" "حالت جدید:\n" "- مدیاها ذخیره می‌شوند.\n" "- برای وارد کردن مدیاهای ذخیره‌شده به صف: /scan\n" "- برای تبدیل و ارسال: /final\n\n" "دستورها:\n" "/queue - وضعیت صف و inbox\n" "/scan - اضافه کردن مدیاهای دیده‌شده قبلی به صف\n" "/final - دانلود، تبدیل به عکس، ارسال خروجی\n" "/clear - پاک کردن صف فعلی" ) await safe_reply(update.effective_message, text) async def queue_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: chat_id = update.effective_chat.id user_id = update.effective_user.id q = queue_count(chat_id, user_id) inbox = inbox_count(chat_id, user_id) await safe_reply( update.effective_message, f"صف فعلی: {q}\nمدیای ذخیره‌شده ولی صف‌نشده: {inbox}", ) async def scan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: chat_id = update.effective_chat.id user_id = update.effective_user.id added = scan_inbox_to_queue(chat_id, user_id) q = queue_count(chat_id, user_id) await safe_reply( update.effective_message, f"اسکن انجام شد.\nاضافه‌شده به صف: {added}\nصف فعلی: {q}\nبرای اجرا: /final", ) async def clear(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: chat_id = update.effective_chat.id user_id = update.effective_user.id deleted = clear_queue(chat_id, user_id) await safe_reply(update.effective_message, f"صف پاک شد. تعداد حذف‌شده: {deleted}") async def collect_media(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: media = extract_media(update) if not media: return inbox_id = save_to_inbox(media) # Auto queue is still enabled for new media. # If you want manual-only queue, comment the next line. inserted = add_to_queue(media, inbox_id=inbox_id) # Do not reply for every file. This was causing floods/timeouts. # Send a small confirmation only every STATUS_EVERY new files. if inserted: q = queue_count(media["chat_id"], media["user_id"]) if q == 1 or q % STATUS_EVERY == 0: await safe_reply( update.effective_message, f"ذخیره شد. صف فعلی: {q}\nبرای اجرا: /final", ) async def final(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: msg = update.effective_message chat_id = update.effective_chat.id user_id = update.effective_user.id rows = get_queue(chat_id, user_id) if not rows: await safe_reply(msg, "صف خالی است. اگر مدیاها قبلاً به بات رسیده‌اند، اول /scan را بزن.") return job_id = uuid.uuid4().hex[:12] job_download_dir = DOWNLOAD_DIR / f"{chat_id}_{user_id}_{job_id}" job_output_dir = OUTPUT_DIR / f"{chat_id}_{user_id}_{job_id}" job_download_dir.mkdir(parents=True, exist_ok=True) job_output_dir.mkdir(parents=True, exist_ok=True) status = await safe_reply(msg, f"شروع شد. تعداد فایل‌ها: {len(rows)}") output_images: List[Path] = [] processed_queue_ids: List[int] = [] failed = 0 skipped_big = 0 try: for index, row in enumerate(rows, start=1): queue_id, inbox_id, file_id, media_type, file_name, mime_type = row if index == 1 or index % STATUS_EVERY == 0 or index == len(rows): await safe_edit(status, f"درحال دانلود و تبدیل: {index}/{len(rows)}") suffix = safe_suffix(file_name, ".bin") input_path = job_download_dir / f"{queue_id}{suffix}" output_path = job_output_dir / f"{queue_id}.jpg" try: tg_file = await retry_async( context.bot.get_file, file_id, retries=5, read_timeout=120, write_timeout=120, connect_timeout=60, pool_timeout=60, ) await retry_async( tg_file.download_to_drive, custom_path=str(input_path), retries=5, read_timeout=300, write_timeout=300, connect_timeout=60, pool_timeout=60, ) ok = False if media_type == "image": ok = await asyncio.to_thread(convert_image_to_jpg, input_path, output_path) elif media_type == "video": ok = await asyncio.to_thread(convert_video_to_jpg, input_path, output_path) if ok: output_images.append(output_path) processed_queue_ids.append(queue_id) else: failed += 1 except BadRequest as e: error_text = str(e).lower() if "file is too big" in error_text: failed += 1 skipped_big += 1 processed_queue_ids.append(queue_id) logger.warning( "Skipped too-big file queue_id=%s file_id=%s error=%s", queue_id, file_id, e, ) else: failed += 1 logger.exception( "BadRequest item queue_id=%s file_id=%s error=%s", queue_id, file_id, e, ) except Exception as e: failed += 1 logger.exception("Failed item queue_id=%s file_id=%s error=%s", queue_id, file_id, e) await asyncio.sleep(DOWNLOAD_DELAY) if not output_images: await safe_edit(status, f"هیچ خروجی تصویری ساخته نشد. ناموفق: {failed}") return await safe_edit(status, f"تبدیل تمام شد. درحال ساخت ZIP از {len(output_images)} عکس...") zip_path = job_output_dir / f"ziper_output_{job_id}.zip" with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: for idx, image_path in enumerate(output_images, start=1): zf.write(image_path, arcname=f"{idx:04d}_{image_path.name}") await safe_edit(status, f"ZIP ساخته شد. درحال ارسال فایل...") sent = 0 try: with open(zip_path, "rb") as f: await retry_async( context.bot.send_document, chat_id=chat_id, document=f, filename=zip_path.name, caption=( f"ZIP آماده شد.\\n" f"تعداد عکس داخل ZIP: {len(output_images)}\\n" f"ناموفق: {failed}\\n" f"ردشده به‌خاطر حجم زیاد: {skipped_big}" ), reply_to_message_id=msg.message_id, retries=5, read_timeout=300, write_timeout=600, connect_timeout=60, pool_timeout=60, ) sent = len(output_images) except BadRequest as e: if "file is too big" in str(e).lower(): await safe_edit( status, "ZIP ساخته شد ولی برای ارسال با Bot API خیلی بزرگ است. " "باید حجم خروجی کمتر شود یا Local Bot API / Telethon استفاده شود." ) logger.exception("ZIP too big to send: %s", e) return raise mark_processed(processed_queue_ids) delete_processed(chat_id, user_id) remain = queue_count(chat_id, user_id) done_text = ( f"تمام شد.\n" f"عکس داخل ZIP: {sent}\n" f"پردازش‌شده و حذف‌شده از صف: {len(processed_queue_ids)}\n" f"ناموفق: {failed}\n" f"ردشده به‌خاطر حجم زیاد: {skipped_big}\n" f"باقی‌مانده در صف: {remain}" ) await safe_edit(status, done_text) except Exception as e: logger.exception("Final command failed: %s", e) await safe_edit(status, f"خطا هنگام پردازش: {e}") finally: shutil.rmtree(job_download_dir, ignore_errors=True) shutil.rmtree(job_output_dir, ignore_errors=True) async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE) -> None: logger.exception("Unhandled error: %s", context.error) def main() -> None: if not BOT_TOKEN: raise RuntimeError("BOT_TOKEN is missing in .env") application: Application = ( ApplicationBuilder() .token(BOT_TOKEN) .connect_timeout(60) .read_timeout(120) .write_timeout(120) .pool_timeout(60) .media_write_timeout(300) .get_updates_connect_timeout(60) .get_updates_read_timeout(60) .get_updates_write_timeout(60) .get_updates_pool_timeout(60) .concurrent_updates(False) .build() ) application.add_handler(CommandHandler("start", start)) application.add_handler(CommandHandler("queue", queue_status)) application.add_handler(CommandHandler("scan", scan)) application.add_handler(CommandHandler("clear", clear)) application.add_handler(CommandHandler("final", final)) media_filter = ( filters.PHOTO | filters.VIDEO | filters.ANIMATION | filters.Document.IMAGE | filters.Document.VIDEO ) application.add_handler(MessageHandler(media_filter, collect_media)) application.add_error_handler(error_handler) logger.info("Bot started.") application.run_polling( allowed_updates=Update.ALL_TYPES, drop_pending_updates=False, ) if __name__ == "__main__": main()