import os import re import json import time import asyncio from pathlib import Path from typing import Optional from urllib.parse import urlparse from dotenv import load_dotenv from telegram import Update from telegram.ext import ( Application, CommandHandler, MessageHandler, filters as tg_filters, ContextTypes, ) from pyrogram import Client as PyroClient load_dotenv(override=True) API_ID = int(os.getenv("API_ID", "0")) API_HASH = os.getenv("API_HASH", "").strip() BOT_TOKEN = os.getenv("BOT_TOKEN", "").strip() BASE_DIR = Path(__file__).resolve().parent DOWNLOAD_DIR = BASE_DIR / "downloads" QUEUE_DIR = BASE_DIR / "queue" QUEUE_FILE = QUEUE_DIR / "tasks.jsonl" STATUS_FILE = QUEUE_DIR / "status.jsonl" SETTINGS_FILE = QUEUE_DIR / "settings.json" DELETED_FILE = QUEUE_DIR / "deleted.jsonl" CANCEL_FILE = QUEUE_DIR / "cancelled.jsonl" DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True) QUEUE_DIR.mkdir(parents=True, exist_ok=True) if not API_ID or not API_HASH or not BOT_TOKEN: raise RuntimeError("Set API_ID, API_HASH, BOT_TOKEN in .env") # Pyrogram used only for MTProto file downloads (bypasses 20 MB Bot API limit) pyro = PyroClient("tel2rub", api_id=API_ID, api_hash=API_HASH, bot_token=BOT_TOKEN) waiting_for_zip_password = False # ── helpers ──────────────────────────────────────────────────────────────── def safe_filename(name: Optional[str]) -> str: name = (name or "file.bin").strip() name = re.sub(r'[<>:"/\\|?*\x00-\x1F]', "_", name) return name.rstrip(". ")[:200] or "file.bin" def split_name(filename: str): p = Path(filename) return p.stem, p.suffix def progress_bar(percent: float, length: int = 12) -> str: filled = int(length * percent / 100) return "█" * filled + "░" * (length - filled) def pretty_size(size) -> str: size = float(size or 0) units = ["B", "KB", "MB", "GB"] i = 0 while size >= 1024 and i < len(units) - 1: size /= 1024 i += 1 return f"{size:.2f} {units[i]}" def eta_text(seconds) -> str: if not seconds or seconds <= 0: return "نامشخص" seconds = int(seconds) h, m, s = seconds // 3600, (seconds % 3600) // 60, seconds % 60 if h: return f"{h}h {m}m" if m: return f"{m}m {s}s" return f"{s}s" def is_direct_url(text: str) -> bool: url = extract_first_url(text) if not url: return False try: p = urlparse(url) return p.scheme in ("http", "https") and bool(p.netloc) except Exception: return False def extract_first_url(text: str) -> Optional[str]: if not text: return None m = re.search(r"https?://\S+", text) return m.group(0) if m else None def load_settings() -> dict: try: if SETTINGS_FILE.exists(): return json.loads(SETTINGS_FILE.read_text(encoding="utf-8")) except Exception: pass return {"safe_mode": False, "zip_password": ""} def save_settings(data: dict): SETTINGS_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def cancel_job(job_id: str): with open(CANCEL_FILE, "a", encoding="utf-8") as f: f.write(json.dumps({"job_id": str(job_id)}, ensure_ascii=False) + "\n") def mark_deleted(task: dict): with open(DELETED_FILE, "a", encoding="utf-8") as f: f.write(json.dumps(task, ensure_ascii=False) + "\n") def was_deleted(job_id=None, message_id=None) -> bool: if not DELETED_FILE.exists(): return False with open(DELETED_FILE, "r", encoding="utf-8") as f: for line in f: if not line.strip(): continue item = json.loads(line) if job_id and str(item.get("job_id")) == str(job_id): return True if message_id and int(item.get("status_message_id", 0)) == message_id: return True return False # ── Queue ────────────────────────────────────────────────────────────────── class QueueManager: def __init__(self): self._cache = None self._mtime = 0 def all(self): mtime = QUEUE_FILE.stat().st_mtime if QUEUE_FILE.exists() else 0 if mtime == self._mtime and self._cache is not None: return self._cache self._cache = [] if QUEUE_FILE.exists(): with open(QUEUE_FILE, "r", encoding="utf-8") as f: self._cache = [json.loads(l) for l in f if l.strip()] self._mtime = mtime return self._cache def push(self, task): task.setdefault("job_id", str(int(time.time() * 1000))) with open(QUEUE_FILE, "a", encoding="utf-8") as f: f.write(json.dumps(task, ensure_ascii=False) + "\n") self._cache = None def remove(self, job_id=None, message_id=None): tasks = self.all() kept, removed = [], None for t in tasks: if (job_id and str(t.get("job_id")) == str(job_id)) or \ (message_id and int(t.get("status_message_id", 0)) == message_id): removed = t else: kept.append(t) if removed: with open(QUEUE_FILE, "w", encoding="utf-8") as f: f.writelines(json.dumps(t, ensure_ascii=False) + "\n" for t in kept) self._cache = None return removed queue = QueueManager() # ── Status watcher ───────────────────────────────────────────────────────── async def status_watcher(bot): pos = 0 while True: await asyncio.sleep(1) if not STATUS_FILE.exists(): continue try: with open(STATUS_FILE, "r", encoding="utf-8") as f: f.seek(pos) lines = f.readlines() pos = f.tell() for line in lines: if not line.strip(): continue data = json.loads(line) chat_id = data.get("chat_id") msg_id = data.get("message_id") text = data.get("text", "") percent = data.get("percent") if not chat_id or not msg_id: continue if percent is not None: text += f"\n\n`{progress_bar(float(percent))}` `{float(percent):.1f}%`" try: await bot.edit_message_text( chat_id=chat_id, message_id=msg_id, text=text, parse_mode="Markdown", ) except Exception: pass except Exception: pass # ── Command handlers ─────────────────────────────────────────────────────── async def start_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE): await update.message.reply_text( "سلام به ربات tele2rub خوش اومدی💙\n\n" "برای انتقال فایل از تلگرام به بله، کافیه فایل رو برام فوروارد کنی.\n" "اگه لینک مستقیم دانلود هم بدی، فایل رو برات دانلود می‌کنم و توی بله می‌فرستم.\n" "⚠️لطفا فایل‌ها رو حداکثر ۱۰ تا ۱۰ تا ارسال کن تا از سمت بله به مشکل نخوره.\n\n" "برای دانلود از یوتیوب، اینستاگرام و... از این ربات استفاده کن: @Gozilla_bot\n" "بعد فایل رو اینجا بفرست تا توی بله برات ارسال کنم.\n\n" "📌 راهنمای ربات:\n\n" "-حذف از صف:\n" "هر فایل وقتی تو صف قرار می‌گیره یه دستور لغو داره که با اون دستور می‌تونی حذفش کنی.\n" "⚠️اگر فایل در حال آپلود باشه، لغو بعد از پایان تلاش فعلی انجام میشه.\n\n" "-پاکسازی کل صف:\n" "/delall\n\n" "-حالت Safe Mode:\n" "همه فایل‌ها با رمز دلخواهت به صورت ZIP رمزدار ارسال میشن.\n\n" "برای فعال/غیرفعال کردن:\n" " /safemode on\n" " /safemode off\n\n" "⚠️برای فایل‌های حجیم و ویدیوها بهتره Safe Mode خاموش باشه.\n\n" "@caffeinexz" ) async def safemode_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE): global waiting_for_zip_password args = context.args if not args: await update.message.reply_text("از `/safemode on` یا `/safemode off` استفاده کن.") return action = args[0].strip().lower() settings = load_settings() if action == "on": settings["safe_mode"] = True save_settings(settings) waiting_for_zip_password = True await update.message.reply_text( "Safe Mode فعال شد.\n\n" "لطفا رمزی که می‌خواهید روی فایل‌های ZIP قرار بگیرد را ارسال کنید.\n" "از این به بعد فایل‌ها قبل از ارسال به بله با همین رمز ZIP می‌شوند." ) elif action == "off": settings["safe_mode"] = False settings["zip_password"] = "" save_settings(settings) waiting_for_zip_password = False await update.message.reply_text( "Safe Mode غیرفعال شد.\n\nاز این به بعد فایل‌ها به‌صورت عادی ارسال می‌شوند." ) else: await update.message.reply_text("از `/safemode on` یا `/safemode off` استفاده کن.") async def delall_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE): tasks = queue.all() if not tasks: await update.message.reply_text("صف خالی است.") return for task in tasks: mark_deleted(task) old_path = task.get("path") if old_path: try: p = Path(old_path) if p.exists(): p.unlink() except Exception: pass try: await context.bot.edit_message_text( chat_id=task["chat_id"], message_id=task["status_message_id"], text="این مورد از صف حذف شد.", ) except Exception: pass with open(QUEUE_FILE, "w", encoding="utf-8") as f: pass queue._cache = None queue._mtime = 0 await update.message.reply_text("تمام موارد در صف پاک شد.") async def del_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE): job_id = context.args[0].strip() if context.args else None reply_id = update.message.reply_to_message.message_id if update.message.reply_to_message else None tasks = queue.all() if not tasks: if job_id and was_deleted(job_id=job_id): await update.message.reply_text("این مورد قبلاً از صف حذف شده است.") return if job_id: cancel_job(job_id) await update.message.reply_text("لغو ثبت شد.") return await update.message.reply_text("موردی برای حذف در صف پیدا نشد.") return removed = queue.remove(job_id=job_id, message_id=reply_id) if removed: mark_deleted(removed) old_path = removed.get("path") if old_path: try: p = Path(old_path) if p.exists(): p.unlink() except Exception: pass try: await context.bot.edit_message_text( chat_id=removed["chat_id"], message_id=removed["status_message_id"], text="این مورد از صف حذف شد.", ) except Exception: pass await update.message.reply_text("از صف حذف شد.") return if job_id and was_deleted(job_id=job_id): await update.message.reply_text("این مورد قبلاً از صف حذف شده است.") return if job_id: cancel_job(job_id) await update.message.reply_text("دستور لغو ثبت شد.") # ── Message handlers ─────────────────────────────────────────────────────── async def text_msg(update: Update, context: ContextTypes.DEFAULT_TYPE): global waiting_for_zip_password text = update.message.text or "" if waiting_for_zip_password: password = text.strip() if not password: await update.message.reply_text("رمز نمی‌تواند خالی باشد. لطفاً یک رمز معتبر ارسال کنید.") return settings = load_settings() settings["safe_mode"] = True settings["zip_password"] = password save_settings(settings) waiting_for_zip_password = False await update.message.reply_text( "رمز ذخیره شد.\n\nاز این به بعد فایل‌ها قبل از ارسال به بله به‌صورت ZIP رمزدار آماده می‌شوند." ) return url = extract_first_url(text) if not url or not is_direct_url(url): return settings = load_settings() status_msg = await update.message.reply_text("لینک دریافت شد.\n\nوضعیت: در صف دانلود قرار گرفت.") task = { "type": "direct_url", "url": url, "chat_id": update.message.chat_id, "status_message_id": status_msg.message_id, "safe_mode": settings.get("safe_mode", False), "zip_password": settings.get("zip_password", ""), } queue.push(task) await status_msg.edit_text( f"لینک در صف قرار گرفت.\n\nشناسه: `{task['job_id']}`\nبرای حذف: `/del {task['job_id']}`", parse_mode="Markdown", ) async def _pyro_download_progress(current, total, status_msg, file_name, started_at, state): now = time.time() if now - state.get("last_update", 0) < 3 and current < total: return state["last_update"] = now percent = current * 100 / total if total else 0 elapsed = max(now - started_at, 1) speed = current / elapsed eta = (total - current) / speed if speed else None text = ( f"📥 در حال دریافت فایل از تلگرام\n\n" f"فایل: `{file_name}`\n" f"حجم: `{pretty_size(total)}`\n" f"پیشرفت: `{percent:.1f}%`\n" f"`{progress_bar(percent)}`\n" f"سرعت: `{pretty_size(speed)}/s`\n" f"زمان باقی‌مانده: `{eta_text(eta)}`" ) try: await status_msg.edit_text(text, parse_mode="Markdown") except Exception: pass async def media_msg(update: Update, context: ContextTypes.DEFAULT_TYPE): msg = update.message if msg.document: media = msg.document original_name = media.file_name or f"{media.file_unique_id}.bin" elif msg.video: media = msg.video original_name = getattr(media, "file_name", None) or f"{media.file_unique_id}.mp4" elif msg.audio: media = msg.audio original_name = getattr(media, "file_name", None) or f"{media.file_unique_id}.mp3" elif msg.voice: media = msg.voice original_name = f"{media.file_unique_id}.ogg" elif msg.photo: media = msg.photo[-1] original_name = f"{media.file_unique_id}.jpg" elif msg.animation: media = msg.animation original_name = getattr(media, "file_name", None) or f"{media.file_unique_id}.mp4" elif msg.video_note: media = msg.video_note original_name = f"{media.file_unique_id}.mp4" elif msg.sticker: media = msg.sticker original_name = f"{media.file_unique_id}.webp" else: await msg.reply_text("فایل قابل پردازش نیست.") return original_name = safe_filename(original_name) stem, suffix = split_name(original_name) download_name = safe_filename(f"{stem}_{msg.message_id}{suffix or '.bin'}") download_path = DOWNLOAD_DIR / download_name status_msg = await msg.reply_text("فایل دریافت شد.\n\nوضعیت: آماده‌سازی برای دانلود از تلگرام...") try: started_at = time.time() progress_state = {"last_update": 0} # Fetch the message via Pyrogram MTProto so large files (>20 MB) work pyro_msg = await pyro.get_messages(msg.chat_id, msg.message_id) downloaded = await pyro.download_media( pyro_msg, file_name=str(download_path), progress=_pyro_download_progress, progress_args=(status_msg, download_name, started_at, progress_state), ) if not downloaded: raise RuntimeError("Download failed.") downloaded_path = Path(downloaded) if not downloaded_path.exists(): raise RuntimeError("Downloaded file not found.") file_size = downloaded_path.stat().st_size settings = load_settings() task = { "type": "local_file", "path": str(downloaded_path), "caption": msg.caption or "", "chat_id": msg.chat_id, "status_message_id": status_msg.message_id, "file_name": download_name, "file_size": file_size, "safe_mode": settings.get("safe_mode", False), "zip_password": settings.get("zip_password", ""), } queue.push(task) await status_msg.edit_text( f"در صف قرار گرفت.\n\n" f"فایل: `{download_name}`\n" f"حجم: `{pretty_size(file_size)}`\n" f"شناسه: `{task['job_id']}`\n\n" f"برای حذف این مورد از صف:\n`/del {task['job_id']}`", parse_mode="Markdown", ) except Exception as e: await status_msg.edit_text(f"خطا: {str(e)}") # ── App lifecycle ────────────────────────────────────────────────────────── async def on_startup(app: Application) -> None: await pyro.start() asyncio.create_task(status_watcher(app.bot)) try: if STATUS_FILE.exists(): STATUS_FILE.unlink() except Exception: pass async def on_shutdown(app: Application) -> None: try: await pyro.stop() except Exception: pass def main(): application = ( Application.builder() .token(BOT_TOKEN) .post_init(on_startup) .post_shutdown(on_shutdown) .build() ) private = tg_filters.ChatType.PRIVATE application.add_handler(CommandHandler("start", start_cmd, filters=private)) application.add_handler(CommandHandler("safemode", safemode_cmd, filters=private)) application.add_handler(CommandHandler("delall", delall_cmd, filters=private)) application.add_handler(CommandHandler("del", del_cmd, filters=private)) application.add_handler(MessageHandler( private & tg_filters.TEXT & ~tg_filters.COMMAND, text_msg, )) application.add_handler(MessageHandler( private & ( tg_filters.Document.ALL | tg_filters.VIDEO | tg_filters.AUDIO | tg_filters.VOICE | tg_filters.PHOTO | tg_filters.ANIMATION | tg_filters.VIDEO_NOTE | tg_filters.Sticker.ALL ), media_msg, )) application.run_polling(allowed_updates=Update.ALL_TYPES) if __name__ == "__main__": main()