diff --git a/bot.py b/bot.py new file mode 100644 index 0000000..fca40eb --- /dev/null +++ b/bot.py @@ -0,0 +1,2810 @@ +import os +import re +import sys +import json +import asyncio +import logging +import secrets +import tempfile +import time +import collections +from datetime import datetime +from urllib.parse import urlparse, unquote +from pathlib import Path +from dotenv import load_dotenv +import psutil + +from pyrogram import Client, filters, idle +from pyrogram.types import ( + Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton, + BotCommand, +) + +load_dotenv() + +API_ID = int(os.getenv("API_ID", "0")) +API_HASH = os.getenv("API_HASH", "").strip() +BOT_TOKEN = os.getenv("BOT_TOKEN", "").strip() +RCLONE_REMOTE = "gdrive:" + +def classify_to_folder(file_name: str, mime_type: str = "") -> str: + ext = file_name.rsplit(".", 1)[-1].lower() if "." in file_name else "" + if mime_type.startswith("video/") or ext in ("mp4","mkv","avi","mov","webm","flv"): + return "Videos" + if mime_type.startswith("audio/") or ext in ("mp3","flac","wav","ogg","m4a","aac"): + return "Music" + if mime_type.startswith("image/") or ext in ("jpg","jpeg","png","gif","webp","bmp","heic"): + return "Photos" + if ext in ("exe","msi","deb","rpm","dmg","appimage"): + return "Software" + if ext in ("zip","rar","7z","tar","gz","bz2","xz","zst"): + return "Archives" + if ext == "gguf" or ext in ("safetensors","pt","ckpt","pth"): + return "Models" + if ext == "zim": + return "ZIM" + if ext in ("conf","ovpn","cfg","ini"): + return "Config" + if ext in ("pdf","doc","docx","txt","md","csv","xlsx","pptx"): + return "Docs" + if ext == "bin": + return "Bin" + return "Other" + +LOG_FILE = Path(__file__).parent / "bot.log" +USERS_FILE = Path(__file__).parent / "authorized_users.json" + +STREAM_THRESHOLD = 2 * (1 << 30) # 2 GB +QUOTAS_FILE = Path(__file__).parent / "quotas.json" + +# ── Persisted authorized users ──────────────────────────────────────────────── + +def _load_users() -> set: + try: + return set(json.loads(USERS_FILE.read_text())) + except Exception: + return {"coaxys"} + +def _save_users(users: set): + USERS_FILE.write_text(json.dumps(sorted(users))) + +AUTHORIZED_USERS: set = _load_users() + +# ── Quota system ────────────────────────────────────────────────────────────── +# quotas.json schema: +# { "quotas": {"username": }, "usage": {"username": {"YYYY-MM-DD": }} } + +def _load_quotas() -> dict: + try: + return json.loads(QUOTAS_FILE.read_text()) + except Exception: + return {"quotas": {}, "usage": {}} + +def _save_quotas(data: dict): + QUOTAS_FILE.write_text(json.dumps(data, indent=2)) + +def parse_size(s: str) -> int | None: + """Parse '5GB', '500MB', '2.5GB', '100KB' → bytes. Returns None on failure.""" + m = re.fullmatch(r'([\d.]+)\s*(B|KB|MB|GB|TB)', s.strip(), re.IGNORECASE) + if not m: + return None + n, unit = float(m.group(1)), m.group(2).upper() + mul = {"B": 1, "KB": 1<<10, "MB": 1<<20, "GB": 1<<30, "TB": 1<<40} + return int(n * mul[unit]) + +def today() -> str: + return datetime.utcnow().strftime("%Y-%m-%d") + +def get_quota(username: str) -> int | None: + """Daily quota in bytes, or None if not set.""" + return _load_quotas()["quotas"].get(username.lower()) + +def get_usage_today(username: str) -> int: + return _load_quotas()["usage"].get(username.lower(), {}).get(today(), 0) + +def quota_remaining(username: str) -> int | None: + """Returns remaining bytes today, or None if no quota set.""" + q = get_quota(username) + if q is None: + return None + return max(0, q - get_usage_today(username)) + +def record_upload(username: str, size_bytes: int): + """Add size_bytes to today's usage for username.""" + data = _load_quotas() + u = username.lower() + data["usage"].setdefault(u, {}) + data["usage"][u][today()] = data["usage"][u].get(today(), 0) + size_bytes + _save_quotas(data) + +def has_upload_access(message: Message) -> tuple[bool, str]: + """ + Returns (allowed, reason). + Auth users always pass. Others need a quota with remaining bytes. + """ + u = (message.from_user.username or "").lower() if message.from_user else "" + if u in AUTHORIZED_USERS: + return True, "auth" + q = get_quota(u) + if q is None: + return False, "no_access" + rem = quota_remaining(u) + if rem == 0: + return False, "quota_exceeded" + return True, "quota" + +BOT_START_TIME = time.time() +session_uploads = 0 + +logging.basicConfig( + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + level=logging.INFO +) +logger = logging.getLogger(__name__) + +app = Client( + "gdrive_bot", + api_id=API_ID, + api_hash=API_HASH, + bot_token=BOT_TOKEN, +) + +MEDIA_TYPES = ( + "document", "video", "audio", "voice", + "photo", "animation", "video_note", "sticker" +) + +URL_RE = re.compile(r'https?://\S+', re.IGNORECASE) + +GDRIVE_FILE_RE = re.compile(r'drive\.google\.com/file/d/([a-zA-Z0-9_-]+)') +GDRIVE_UC_RE = re.compile(r'drive\.google\.com/uc\?.*[?&]id=([a-zA-Z0-9_-]+)') +GDRIVE_OPEN_RE = re.compile(r'drive\.google\.com/open\?.*id=([a-zA-Z0-9_-]+)') + +def extract_gdrive_id(url: str) -> str | None: + for pat in (GDRIVE_FILE_RE, GDRIVE_UC_RE, GDRIVE_OPEN_RE): + m = pat.search(url) + if m: + return m.group(1) + return None + +# ── yt-dlp media sites ──────────────────────────────────────────────────────── +# URLs from these domains are handed to yt-dlp instead of rclone copyurl. +# Downloads land in gdrive:MEDIA-DLS// (NSFW sites → MEDIA-DLS/NSFW). + +MEDIA_DLS_DIR = "MEDIA-DLS" +COOKIES_FILE = Path(__file__).parent / "cookies.txt" + +# ── Torrents & Telegram delivery ────────────────────────────────────────────── +# /tmp is RAM-backed tmpfs here, so torrents (which must hit disk — BitTorrent +# delivers pieces out of order, no RAM-stream path) use a real-disk work dir. +TORRENTS_DIR = "Torrents" # Drive destination root +WORK_DIR = Path(__file__).parent / "dl_tmp" # on /dev/sda1, NOT tmpfs +TORRENT_STREAM_THRESHOLD = 8 * (1 << 30) # ≥ 8 GB → per-file mode +DISK_HEADROOM = 2 * (1 << 30) # keep 2 GB free +TG_SEND_LIMIT = 2000 * (1 << 20) # Telegram bot upload cap (~2 GB) + +MAGNET_RE = re.compile(r'magnet:\?xt=urn:[a-z0-9]+:[^\s]+', re.IGNORECASE) + +VIDEO_SEND_EXTS = (".mp4", ".mkv", ".webm", ".mov", ".avi", ".m4v") +AUDIO_SEND_EXTS = (".mp3", ".m4a", ".flac", ".ogg", ".opus", ".wav", ".aac") +IMAGE_SEND_EXTS = (".jpg", ".jpeg", ".png", ".gif", ".webp") + +YTDLP_DOMAINS = { + "youtube.com", "youtu.be", # videos, shorts, playlists, music + "instagram.com", # posts, reels, stories (needs cookies) + "tiktok.com", # videos, users, live + "twitter.com", "x.com", "t.co", # videos, spaces, broadcasts + "facebook.com", "fb.watch", # videos, reels + "reddit.com", "redd.it", # posts, v.redd.it + "pinterest.com", "pin.it", # video pins + "tumblr.com", + "linkedin.com", # posts, events + "aparat.com", +} + +NSFW_DOMAINS = { + "pornhub.com", "xvideos.com", "xhamster.com", + "youporn.com", "redtube.com", "spankbang.com", + "redgifs.com", +} + +# Spotify links are handled by spotdl (metadata from Spotify, audio fetched +# from YouTube Music — needs cookies.txt while this server's IP is YT-flagged). +SPOTIFY_DOMAINS = {"open.spotify.com", "play.spotify.com", "spotify.link"} + +def spotify_match(url: str) -> bool: + try: + host = urlparse(url).netloc.lower().split(":")[0] + except Exception: + return False + return any(host == d or host.endswith("." + d) for d in SPOTIFY_DOMAINS) + +def ytdlp_match(url: str) -> tuple[bool, bool]: + """Returns (handled_by_ytdlp, is_nsfw) based on the URL's domain.""" + try: + host = urlparse(url).netloc.lower().split(":")[0] + except Exception: + return False, False + for d in NSFW_DOMAINS: + if host == d or host.endswith("." + d): + return True, True + for d in YTDLP_DOMAINS: + if host == d or host.endswith("." + d): + return True, False + return False, False + +def media_dls_folder(file_name: str, nsfw: bool) -> str: + """Drive folder (relative to remote root) for a yt-dlp download.""" + if nsfw: + return f"{MEDIA_DLS_DIR}/NSFW" + return f"{MEDIA_DLS_DIR}/{classify_to_folder(file_name)}" + +# ── Utilities ───────────────────────────────────────────────────────────────── + +def get_media(message: Message): + for attr in MEDIA_TYPES: + m = getattr(message, attr, None) + if m: + return attr, m + return None, None + +def sanitize_filename(name: str) -> str: + """Clean up filenames from bots/channels: strip emoji, newlines, garbage chars.""" + # Normalise unicode (NFKC collapses weird lookalikes) + import unicodedata + name = unicodedata.normalize("NFKC", name) + # Control chars and newlines → space + name = re.sub(r'[\x00-\x1f\x7f]', ' ', name) + # Common garbage from forwarded bot messages + name = name.replace('→', '').replace('↵', '').replace('␊', '') + # Characters illegal in filenames on most filesystems + name = re.sub(r'[<>:"/\\|?*]', '_', name) + # Emoji: remove everything in supplementary Unicode planes (U+1F000+) and misc symbol blocks + name = re.sub( + r'[\U0001F000-\U0010FFFF' # supplementary planes — most emoji live here + r'☀-➿' # misc symbols, dingbats + r'⌀-⏿' # misc technical + r'︀-️' # variation selectors + r'​-‏' # zero-width / directional chars + r' -]', # general punctuation extras + '', name, flags=re.UNICODE + ) + # Collapse runs of whitespace/underscores to a single underscore + name = re.sub(r'[\s_]+', '_', name).strip('_. ') + # Truncate stem to 180 chars but keep extension intact + root, ext = os.path.splitext(name) + root = root[:180].rstrip('_. ') + name = (root + ext).strip() + # Final fallback + return name or f"file_{datetime.now().strftime('%Y%m%d_%H%M%S')}" + +def resolve_filename(media_type: str, media, caption: str | None) -> str: + # Caption overrides — rename before upload + if caption and caption.strip(): + name = caption.strip() + orig = getattr(media, "file_name", None) + if orig and "." in orig and "." not in name: + name += os.path.splitext(orig)[1] + return sanitize_filename(name) + name = getattr(media, "file_name", None) + if name: + return sanitize_filename(name) + ext = { + "photo": ".jpg", "voice": ".ogg", "video_note": ".mp4", + "sticker": ".webp", "animation": ".mp4", "video": ".mp4", "audio": ".mp3", + } + return f"file_{datetime.now().strftime('%Y%m%d_%H%M%S')}{ext.get(media_type, '.bin')}" + +def filename_from_url(url: str) -> str: + path = unquote(urlparse(url).path) + name = os.path.basename(path.rstrip("/")) + if not name or "." not in name: + name = f"download_{datetime.now().strftime('%Y%m%d_%H%M%S')}.bin" + return sanitize_filename(name) + +def fmt_size(b: int) -> str: + if b >= 1 << 30: return f"{b / (1<<30):.2f} GB" + if b >= 1 << 20: return f"{b / (1<<20):.1f} MB" + if b >= 1 << 10: return f"{b / (1<<10):.1f} KB" + return f"{b} B" + +def fmt_dur(s: float) -> str: + s = int(s) + if s < 60: return f"{s}s" + if s < 3600: return f"{s//60}m {s%60:02d}s" + return f"{s//3600}h {(s%3600)//60}m" + +def fmt_ram() -> str: + vm = psutil.virtual_memory() + return f"{fmt_size(vm.used)} / {fmt_size(vm.total)} ({vm.percent:.0f}%)" + +def spd(bps: float) -> str: + return f"{fmt_size(int(bps))}/s" if bps > 0 else "—" + +def choose_chunk(file_bytes: int) -> str: + if file_bytes == 0: return "256M" + if file_bytes < 10 * (1<<20): return "8M" + if file_bytes < 100 * (1<<20): return "64M" + if file_bytes < 500 * (1<<20): return "256M" + return "512M" + +async def get_drive_entry(file_name: str, folder: str | None = None) -> dict | None: + """lsjson metadata for a file — direct path lookup when the folder is + known (fast), recursive scan as fallback.""" + try: + if folder: + proc = await asyncio.create_subprocess_exec( + "rclone", "lsjson", f"{RCLONE_REMOTE}{folder}/{file_name}", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, _ = await proc.communicate() + if proc.returncode == 0: + files = json.loads(out.decode()) + if files: + entry = files[0] + entry["Path"] = f"{folder}/{file_name}" + return entry + proc = await asyncio.create_subprocess_exec( + "rclone", "lsjson", RCLONE_REMOTE, + "--recursive", "--files-only", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, _ = await proc.communicate() + for f in json.loads(out.decode()): + if f["Name"] == file_name: + return f + return None + except Exception as e: + logger.warning(f"lsjson: {e}") + return None + +async def get_drive_file_id(file_name: str) -> str | None: + entry = await get_drive_entry(file_name) + return entry["ID"] if entry else None + +# ── Inline file-action buttons ──────────────────────────────────────────────── +# Every uploaded file gets a short token; callbacks look up the token in +# FILE_ACTIONS (persisted, so buttons survive restarts). + +ACTIONS_FILE = Path(__file__).parent / "file_actions.json" + +def _load_actions() -> dict: + try: + return json.loads(ACTIONS_FILE.read_text()) + except Exception: + return {} + +FILE_ACTIONS: dict = _load_actions() + +def _save_actions(): + try: + while len(FILE_ACTIONS) > 500: # cap registry size + FILE_ACTIONS.pop(next(iter(FILE_ACTIONS))) + ACTIONS_FILE.write_text(json.dumps(FILE_ACTIONS)) + except Exception as e: + logger.warning(f"actions save: {e}") + +def register_file_action(name: str, folder: str, owner: str) -> str: + tok = secrets.token_hex(4) + FILE_ACTIONS[tok] = {"name": name, "folder": folder, "owner": owner} + _save_actions() + return tok + +MOVE_TARGETS = [ + "Videos", "Music", "Photos", "Docs", + "Archives", "Software", "Models", "Other", + f"{MEDIA_DLS_DIR}/Videos", f"{MEDIA_DLS_DIR}/Music", f"{MEDIA_DLS_DIR}/NSFW", +] + +def file_buttons(tok: str) -> InlineKeyboardMarkup: + return InlineKeyboardMarkup([ + [InlineKeyboardButton("🌍 Share", callback_data=f"share:{tok}"), + InlineKeyboardButton("ℹ️ Info", callback_data=f"info:{tok}")], + [InlineKeyboardButton("📂 Move", callback_data=f"mv:{tok}"), + InlineKeyboardButton("🗑 Delete", callback_data=f"del:{tok}")], + [InlineKeyboardButton("📥 Send to Telegram", callback_data=f"get:{tok}")], + ]) + +def move_keyboard(tok: str) -> InlineKeyboardMarkup: + rows, row = [], [] + for f in MOVE_TARGETS: + row.append(InlineKeyboardButton(f"📁 {f}", callback_data=f"mvto:{tok}:{f}")) + if len(row) == 2: + rows.append(row) + row = [] + if row: + rows.append(row) + rows.append([InlineKeyboardButton("↩️ Cancel", callback_data=f"back:{tok}")]) + return InlineKeyboardMarkup(rows) + +def del_confirm_keyboard(tok: str) -> InlineKeyboardMarkup: + return InlineKeyboardMarkup([[ + InlineKeyboardButton("✅ Yes, delete", callback_data=f"delc:{tok}"), + InlineKeyboardButton("↩️ Cancel", callback_data=f"back:{tok}"), + ]]) + +async def send_links(message: Message, file_name: str, + folder: str | None = None, owner: str = ""): + entry = await get_drive_entry(file_name, folder) + if not entry or not entry.get("ID"): + return + file_id = entry["ID"] + real_folder = folder or os.path.dirname(entry.get("Path", "")) or "Other" + tok = register_file_action(file_name, real_folder, owner) + dl = f"https://drive.google.com/uc?export=download&id={file_id}" + view = f"https://drive.google.com/file/d/{file_id}/view" + wget = f"wget -O \"{file_name}\" \"{dl}\"" + await message.reply_text( + f"🔗 **Links** — `{file_name}`\n\n" + f"📥 **Direct download:**\n`{dl}`\n\n" + f"👁 **View on Drive:**\n`{view}`\n\n" + f"🖥 **wget:**\n`{wget}`", + disable_web_page_preview=True, + reply_markup=file_buttons(tok), + ) + +# ── Auth ────────────────────────────────────────────────────────────────────── + +def is_authorized(message: Message) -> bool: + u = message.from_user + if not u: + return False + return (u.username or "").lower() in AUTHORIZED_USERS + +async def deny(message: Message): + await message.reply_text("⛔ You're not authorized to use this command.") + +# ── /start ──────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("start")) +async def cmd_start(client: Client, message: Message): + vm = psutil.virtual_memory() + await message.reply_text( + "**GDrive Upload Bot** ⚡\n" + f"🖥 RAM: {fmt_size(vm.available)} free · 📡 MTProto + rclone + yt-dlp + aria2\n" + "━━━━━━━━━━━━━━━━━━━━\n\n" + + USAGE_TUTORIAL + + "\n\n✏️ _Caption a file to rename it. After each upload you get a " + "download link plus buttons to Share / Move / Delete / send it back._\n\n" + "━━━━━━━━━━━━━━━━━━━━\n" + "Made by @Coaxys", + disable_web_page_preview=True, + ) + +# ── /folders ────────────────────────────────────────────────────────────────── + +FOLDER_LIST = ["Videos", "Music", "Photos", "Software", "Archives", "Models", + "ZIM", "Config", "Docs", "Bin", "Other"] + +@app.on_message(filters.command("folders")) +async def cmd_folders(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + desc = { + "Videos": "mp4, mkv, avi, mov, webm, flv", + "Music": "mp3, flac, wav, ogg, m4a, aac", + "Photos": "jpg, jpeg, png, gif, webp, bmp, heic", + "Software": "exe, msi, deb, rpm, dmg, appimage", + "Archives": "zip, rar, 7z, tar, gz, bz2, xz, zst", + "Models": "gguf, safetensors, pt, ckpt, pth", + "ZIM": "zim", + "Config": "conf, ovpn, cfg, ini", + "Docs": "pdf, doc, docx, txt, md, csv, xlsx, pptx", + "Bin": "bin", + "Other": "everything else", + } + lines = [f"📁 **{f}**\n `{desc.get(f, '')}`" for f in FOLDER_LIST] + await message.reply_text( + "📂 **Available Drive folders**\n\n" + + "\n\n".join(lines) + + "\n\n_↳ Files are auto-sorted by extension / MIME type._" + ) + +# ── /list ──────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("list")) +async def cmd_list(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + wait = await message.reply_text("📂 Fetching latest files from Drive...") + + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "lsjson", RCLONE_REMOTE, + "--recursive", "--files-only", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + raise RuntimeError(err.decode().strip()) + + files = json.loads(out.decode()) + files.sort(key=lambda x: x["ModTime"], reverse=True) + top = files[:10] + total = len(files) + + lines = [] + for i, f in enumerate(top, 1): + size = f["Size"] + if size >= 1<<30: s = f"{size/(1<<30):.2f} GB" + elif size >= 1<<20: s = f"{size/(1<<20):.1f} MB" + elif size >= 1<<10: s = f"{size/(1<<10):.1f} KB" + else: s = f"{size} B" + date = f["ModTime"][:10] + name = f["Name"] + if len(name) > 40: + name = name[:37] + "..." + fid = f["ID"] + folder = f["Path"].split("/")[0] if "/" in f["Path"] else "root" + link = f"https://drive.google.com/file/d/{fid}/view" + lines.append(f"`{i:2}.` [{name}]({link})\n 📁 {folder} · 📦 {s} · 📅 {date}") + + body = "\n\n".join(lines) + await wait.edit_text( + f"📂 **Latest {len(top)} of {total} files in Drive**\n\n{body}", + disable_web_page_preview=True, + ) + + except Exception as e: + logger.error(f"/list error: {e}") + await wait.edit_text(f"❌ Failed to fetch file list\n\n`{e}`") + +# ── /listall ───────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("listall")) +async def cmd_listall(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + wait = await message.reply_text("📂 Fetching all files from Drive...") + + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "lsjson", RCLONE_REMOTE, + "--recursive", "--files-only", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + raise RuntimeError(err.decode().strip()) + + files = json.loads(out.decode()) + files.sort(key=lambda x: x["ModTime"], reverse=True) + + if not files: + return await wait.edit_text("📂 Drive is empty.") + + total_size = sum(f["Size"] for f in files) + + def human(b): + if b >= 1<<30: return f"{b/(1<<30):.2f} GB" + if b >= 1<<20: return f"{b/(1<<20):.1f} MB" + if b >= 1<<10: return f"{b/(1<<10):.1f} KB" + return f"{b} B" + + # Build lines and split into ≤4000-char pages + header = ( + f"📂 **All files in Drive** ({len(files)} files · {human(total_size)} total)\n" + f"{'━'*22}\n\n" + ) + + pages = [] + current = header + for i, f in enumerate(files, 1): + name = f["Name"] + disp = name if len(name) <= 45 else name[:42] + "..." + link = f"https://drive.google.com/file/d/{f['ID']}/view" + line = f"`{i:3}.` [{disp}]({link})\n 📦 {human(f['Size'])} · 📅 {f['ModTime'][:10]}\n\n" + + if len(current) + len(line) > 4000: + pages.append(current) + current = line + else: + current += line + + if current.strip(): + pages.append(current) + + # Edit first message, send rest as new replies + await wait.edit_text( + pages[0] + (f"\n_Page 1 of {len(pages)}_" if len(pages) > 1 else ""), + disable_web_page_preview=True, + ) + for idx, page in enumerate(pages[1:], 2): + await message.reply_text( + page + f"\n_Page {idx} of {len(pages)}_", + disable_web_page_preview=True, + ) + + except Exception as e: + logger.error(f"/listall error: {e}") + await wait.edit_text(f"❌ Failed to fetch file list\n\n`{e}`") + +# ── /status ─────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("status")) +async def cmd_status(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + uptime = time.time() - BOT_START_TIME + vm = psutil.virtual_memory() + disk = psutil.disk_usage("/") + cpu = psutil.cpu_percent(interval=0.5) + + def bar(pct: float, width: int = 10) -> str: + filled = int(pct / 100 * width) + return "█" * filled + "░" * (width - filled) + + await message.reply_text( + f"📊 **Bot Status**\n\n" + f"⏱ **Uptime:** {fmt_dur(uptime)}\n" + f"📤 **Uploads this session:** {session_uploads}\n\n" + f"🖥 **CPU:** {cpu:.1f}% `[{bar(cpu)}]`\n\n" + f"💾 **RAM:**\n" + f" Used: {fmt_size(vm.used)} / {fmt_size(vm.total)}\n" + f" Free: {fmt_size(vm.available)}\n" + f" `[{bar(vm.percent)}]` {vm.percent:.1f}%\n\n" + f"🗄 **Disk (/):**\n" + f" Used: {fmt_size(disk.used)} / {fmt_size(disk.total)}\n" + f" Free: {fmt_size(disk.free)}\n" + f" `[{bar(disk.percent)}]` {disk.percent:.1f}%\n\n" + f"📡 **Method:** MTProto + rclone\n" + f"🔀 **Stream threshold:** ≥ 2 GB" + ) + +# ── /setquota ──────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("setquota")) +async def cmd_setquota(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + parts = message.command + if len(parts) < 3: + return await message.reply_text( + "Usage: `/setquota @username `\n" + "Examples: `/setquota @john 5GB` · `/setquota @john 500MB`" + ) + + username = parts[1].lstrip("@").lower() + size_bytes = parse_size(parts[2]) + if size_bytes is None: + return await message.reply_text( + "❌ Invalid size. Use formats like `5GB`, `500MB`, `2.5GB`, `100KB`." + ) + + data = _load_quotas() + data["quotas"][username] = size_bytes + _save_quotas(data) + logger.info(f"Quota set: {username} → {fmt_size(size_bytes)}/day") + await message.reply_text( + f"✅ Daily quota set for `@{username}`\n\n" + f"📦 **Limit:** {fmt_size(size_bytes)} / day\n" + f"🔄 Resets at midnight UTC" + ) + +# ── /clearquota ─────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("clearquota")) +async def cmd_clearquota(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + parts = message.command + if len(parts) < 2: + return await message.reply_text("Usage: `/clearquota @username`") + + username = parts[1].lstrip("@").lower() + data = _load_quotas() + + if username not in data["quotas"]: + return await message.reply_text(f"ℹ️ `@{username}` has no quota set.") + + del data["quotas"][username] + data["usage"].pop(username, None) + _save_quotas(data) + logger.info(f"Quota cleared: {username}") + await message.reply_text(f"✅ Quota and usage history cleared for `@{username}`.") + +# ── /quotas ─────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("quotas")) +async def cmd_quotas(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + data = _load_quotas() + if not data["quotas"]: + return await message.reply_text("📋 No quotas configured yet.\n\nUse `/setquota @user ` to add one.") + + lines = [] + for username, limit in sorted(data["quotas"].items()): + used = get_usage_today(username) + remaining = max(0, limit - used) + pct = used / limit * 100 if limit else 0 + bar_n = int(pct / 10) + bar = "█" * bar_n + "░" * (10 - bar_n) + lines.append( + f"👤 `@{username}`\n" + f" `[{bar}]` {pct:.1f}%\n" + f" 📦 Used: {fmt_size(used)} / {fmt_size(limit)}\n" + f" ✅ Left: {fmt_size(remaining)}" + ) + + await message.reply_text( + f"📋 **Quota Overview** — {today()} UTC\n\n" + + "\n\n".join(lines) + + "\n\n🔄 Resets at midnight UTC" + ) + +# ── /myquota ────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("myquota")) +async def cmd_myquota(client: Client, message: Message): + u = (message.from_user.username or "").lower() if message.from_user else "" + if not u: + return await message.reply_text("❌ Could not determine your username.") + + if u in AUTHORIZED_USERS: + return await message.reply_text("✅ You're an **authorized user** — no quota limits apply to you.") + + limit = get_quota(u) + if limit is None: + return await message.reply_text("🔒 You don't have a quota assigned. Contact an admin.") + + used = get_usage_today(u) + remaining = max(0, limit - used) + pct = used / limit * 100 if limit else 0 + bar_n = int(pct / 10) + bar = "█" * bar_n + "░" * (10 - bar_n) + + await message.reply_text( + f"📊 **Your Quota — {today()} UTC**\n\n" + f"`[{bar}]` {pct:.1f}%\n\n" + f"📦 Used: {fmt_size(used)}\n" + f"✅ Remaining: {fmt_size(remaining)}\n" + f"🔒 Daily limit: {fmt_size(limit)}\n\n" + f"🔄 Resets at midnight UTC" + ) + +# ── /get ────────────────────────────────────────────────────────────────────── + +async def _find_drive_file(query: str) -> tuple[dict | None, list[dict], str]: + """Look up a Drive file by exact-or-partial name. Returns (match, candidates, error).""" + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "lsjson", RCLONE_REMOTE, "--recursive", "--files-only", + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + return None, [], err.decode().strip() + files = json.loads(out.decode()) + exact = [f for f in files if f["Name"].lower() == query.lower()] + partial = [f for f in files if query.lower() in f["Name"].lower()] + matches = exact or partial + if not matches: + return None, [], "" + if len(matches) > 1 and not exact: + return None, matches, "" + return matches[0], [], "" + except Exception as e: + return None, [], str(e) + +@app.on_message(filters.command("get")) +async def cmd_get(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + query = " ".join(message.command[1:]).strip() + if not query: + return await message.reply_text( + "Usage: `/get `\n\n" + "Pulls a file **from** Drive and sends it back here.\n" + "Partial names work — e.g. `/get holiday` matches `holiday-2026.mp4`.\n" + f"Max size {fmt_size(TG_SEND_LIMIT)} (Telegram limit)." + ) + wait = await message.reply_text(f"🔍 Looking for `{query}` in Drive...") + match, candidates, err = await _find_drive_file(query) + if err: + return await wait.edit_text(f"❌ Drive lookup failed\n\n`{err}`") + if candidates: + names = "\n".join(f"• `{f['Name']}`" for f in candidates[:8]) + more = f"\n_…and {len(candidates)-8} more_" if len(candidates) > 8 else "" + return await wait.edit_text( + f"⚠️ **Multiple matches for** `{query}`\n\n{names}{more}\n\nBe more specific." + ) + if not match: + return await wait.edit_text(f"❌ No file matching `{query}` found in Drive.") + folder = os.path.dirname(match.get("Path", "")) or "Other" + await wait.delete() + await deliver_to_telegram(client, message, match["Name"], folder) + +# ── /share ─────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("share")) +async def cmd_share(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + query = " ".join(message.command[1:]).strip() + if not query: + return await message.reply_text( + "Usage: `/share `\n\n" + "Makes the file publicly accessible and returns a shareable link.\n" + "Tip: partial name works — e.g. `/share Mitm` matches `Mitm-Yt.mp4`" + ) + + wait = await message.reply_text(f"🔍 Looking for `{query}` in Drive...") + + # Find the file by exact or partial name match + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "lsjson", RCLONE_REMOTE, + "--recursive", "--files-only", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + raise RuntimeError(err.decode().strip()) + + files = json.loads(out.decode()) + # Prefer exact match, fall back to partial + exact = [f for f in files if f["Name"].lower() == query.lower()] + partial = [f for f in files if query.lower() in f["Name"].lower()] + matches = exact or partial + + if not matches: + return await wait.edit_text(f"❌ No file matching `{query}` found in Drive.") + + if len(matches) > 1 and not exact: + names = "\n".join(f"• `{f['Name']}`" for f in matches[:8]) + more = f"\n_…and {len(matches)-8} more_" if len(matches) > 8 else "" + return await wait.edit_text( + f"⚠️ **Multiple matches for** `{query}`\n\n{names}{more}\n\n" + f"Be more specific to share a single file." + ) + + target = matches[0] + name = target["Name"] + path = target["Path"] # includes subfolder, e.g. "Videos/filename.mp4" + size = target["Size"] + + except Exception as e: + logger.error(f"/share lookup error: {e}") + return await wait.edit_text(f"❌ Drive lookup failed\n\n`{e}`") + + await wait.edit_text(f"🔗 Making `{name}` public...") + + # Generate share link via rclone link + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "link", f"{RCLONE_REMOTE}{path}", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + raise RuntimeError(err.decode().strip() or out.decode().strip()) + + share_url = out.decode().strip() + file_id = share_url.split("id=")[-1] if "id=" in share_url else None + + except Exception as e: + logger.error(f"/share rclone link error: {e}") + return await wait.edit_text(f"❌ Failed to generate share link\n\n`{e}`") + + def human(b): + if b >= 1<<30: return f"{b/(1<<30):.2f} GB" + if b >= 1<<20: return f"{b/(1<<20):.1f} MB" + if b >= 1<<10: return f"{b/(1<<10):.1f} KB" + return f"{b} B" + + dl_link = f"https://drive.google.com/uc?export=download&id={file_id}" if file_id else None + wget_line = f"`wget -O \"{name}\" \"{dl_link}\"`" if dl_link else "" + + await wait.edit_text( + f"✅ **File is now public**\n\n" + f"📎 **{name}**\n" + f"📦 {human(size)}\n\n" + f"🔗 **Share link:**\n`{share_url}`\n\n" + + (f"📥 **Direct download:**\n`{dl_link}`\n\n" if dl_link else "") + + (f"🖥 **wget:**\n{wget_line}" if wget_line else ""), + disable_web_page_preview=True, + ) + +# ── /search ─────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("search")) +async def cmd_search(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + query = " ".join(message.command[1:]).strip().lower() + if not query: + return await message.reply_text("Usage: `/search `") + + wait = await message.reply_text(f"🔍 Searching for `{query}`...") + + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "lsjson", RCLONE_REMOTE, + "--recursive", "--files-only", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + raise RuntimeError(err.decode().strip()) + + files = json.loads(out.decode()) + hits = [f for f in files if query in f["Name"].lower()] + hits.sort(key=lambda x: x["ModTime"], reverse=True) + + if not hits: + return await wait.edit_text(f"🔍 No files matching `{query}` found in Drive.") + + lines = [] + for i, f in enumerate(hits[:15], 1): + size = f["Size"] + if size >= 1<<30: s = f"{size/(1<<30):.2f} GB" + elif size >= 1<<20: s = f"{size/(1<<20):.1f} MB" + elif size >= 1<<10: s = f"{size/(1<<10):.1f} KB" + else: s = f"{size} B" + name = f["Name"] + folder = f["Path"].split("/")[0] if "/" in f["Path"] else "root" + disp = name if len(name) <= 38 else name[:35] + "..." + link = f"https://drive.google.com/file/d/{f['ID']}/view" + lines.append( + f"`{i:2}.` [{disp}]({link})\n" + f" 📁 {folder} · 📦 {s} · 📅 {f['ModTime'][:10]}" + ) + + more = f"\n\n_…and {len(hits)-15} more_" if len(hits) > 15 else "" + await wait.edit_text( + f"🔍 **{len(hits)} result(s) for** `{query}`\n\n" + "\n\n".join(lines) + more, + disable_web_page_preview=True, + ) + except Exception as e: + logger.error(f"/search error: {e}") + await wait.edit_text(f"❌ Search failed\n\n`{e}`") + +# ── /stats ─────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("stats")) +async def cmd_stats(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + query = " ".join(message.command[1:]).strip() + if not query: + return await message.reply_text( + "Usage: `/stats `\n\n" + "Shows Drive file info: size, modified date, MIME type, folder path." + ) + + wait = await message.reply_text(f"🔍 Looking up `{query}`...") + + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "lsjson", RCLONE_REMOTE, + "--recursive", "--files-only", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + raise RuntimeError(err.decode().strip()) + + files = json.loads(out.decode()) + exact = [f for f in files if f["Name"].lower() == query.lower()] + partial = [f for f in files if query.lower() in f["Name"].lower()] + matches = exact or partial + + if not matches: + return await wait.edit_text(f"❌ No file matching `{query}` found in Drive.") + + if len(matches) > 1 and not exact: + names = "\n".join(f"• `{f['Name']}`" for f in matches[:8]) + more = f"\n_…and {len(matches)-8} more_" if len(matches) > 8 else "" + return await wait.edit_text( + f"⚠️ **Multiple matches for** `{query}`\n\n{names}{more}\n\n" + f"Be more specific." + ) + + f = matches[0] + size = f["Size"] + if size >= 1<<30: s = f"{size/(1<<30):.2f} GB" + elif size >= 1<<20: s = f"{size/(1<<20):.1f} MB" + elif size >= 1<<10: s = f"{size/(1<<10):.1f} KB" + else: s = f"{size} B" + + folder = f["Path"].split("/")[0] if "/" in f["Path"] else "root" + link = f"https://drive.google.com/file/d/{f['ID']}/view" + mtime = f["ModTime"][:19].replace("T", " ") + mime = f.get("MimeType", "—") + + await wait.edit_text( + f"📄 **File Info**\n\n" + f"**{f['Name']}**\n\n" + f"📦 Size: {s}\n" + f"📅 Modified: {mtime}\n" + f"🗂 MIME: `{mime}`\n" + f"📁 Drive path: `{folder}/`\n" + f"🔗 View: {link}", + disable_web_page_preview=True, + ) + + except Exception as e: + logger.error(f"/stats error: {e}") + await wait.edit_text(f"❌ Stats lookup failed\n\n`{e}`") + +# ── /delete ────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("delete")) +async def cmd_delete(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + query = " ".join(message.command[1:]).strip() + if not query: + return await message.reply_text( + "Usage: `/delete `\n\n" + "Permanently deletes a file from Google Drive. Use `/search` to find the exact name." + ) + + wait = await message.reply_text(f"🔍 Looking for `{query}` to delete...") + + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "lsjson", RCLONE_REMOTE, + "--recursive", "--files-only", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + raise RuntimeError(err.decode().strip()) + + files = json.loads(out.decode()) + exact = [f for f in files if f["Name"].lower() == query.lower()] + partial = [f for f in files if query.lower() in f["Name"].lower()] + matches = exact or partial + + if not matches: + return await wait.edit_text(f"❌ No file matching `{query}` found in Drive.") + + if len(matches) > 1 and not exact: + names = "\n".join(f"• `{f['Name']}`" for f in matches[:8]) + more = f"\n_…and {len(matches)-8} more_" if len(matches) > 8 else "" + return await wait.edit_text( + f"⚠️ **Multiple matches for** `{query}`\n\n{names}{more}\n\n" + f"Be more specific or use the full filename." + ) + + target = matches[0] + name = target["Name"] + path = target["Path"] + + except Exception as e: + logger.error(f"/delete lookup error: {e}") + return await wait.edit_text(f"❌ Drive lookup failed\n\n`{e}`") + + await wait.edit_text(f"🗑 Deleting `{name}`...") + + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "delete", f"{RCLONE_REMOTE}{path}", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + raise RuntimeError(err.decode().strip()) + + logger.info(f"Deleted: {path}") + await wait.edit_text(f"🗑 **Deleted**\n\n`{name}` has been removed from Google Drive.") + + except Exception as e: + logger.error(f"/delete error: {e}") + await wait.edit_text(f"❌ Delete failed\n\n`{e}`") + +# ── /df ────────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("df")) +async def cmd_df(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + wait = await message.reply_text("📊 Checking Drive storage...") + + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "about", RCLONE_REMOTE, "--json", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + raise RuntimeError(err.decode().strip()) + + info = json.loads(out.decode()) + total = info.get("total", 0) + used = info.get("used", 0) + free = info.get("free", 0) + # Trashed files are counted in "used". Also report "trashed" if available. + trashed = info.get("trashed", 0) + + pct = used / total * 100 if total else 0 + bar = "█" * int(pct / 5) + "░" * (20 - int(pct / 5)) + lines = [ + f"📦 **Drive Storage**\n", + f"`[{bar}]` {pct:.1f}%\n", + f"💾 Used: {fmt_size(used)}", + f"🟢 Free: {fmt_size(free)}", + f"📀 Total: {fmt_size(total)}", + ] + if trashed: + lines.append(f"🗑 Trashed: {fmt_size(trashed)}") + + await wait.edit_text("\n".join(lines)) + + except Exception as e: + logger.error(f"/df error: {e}") + # Fallback: rclone about on some backends returns nothing + await wait.edit_text(f"❌ Storage check failed\n\n`{e}`") + +# ── /adduser & /removeuser ──────────────────────────────────────────────────── + +@app.on_message(filters.command("adduser")) +async def cmd_adduser(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + parts = message.command + if len(parts) < 2: + return await message.reply_text("Usage: `/adduser `") + + username = parts[1].lstrip("@").lower() + if username in AUTHORIZED_USERS: + return await message.reply_text(f"ℹ️ `@{username}` is already authorized.") + + AUTHORIZED_USERS.add(username) + _save_users(AUTHORIZED_USERS) + logger.info(f"Added authorized user: {username}") + await message.reply_text(f"✅ `@{username}` added to authorized users.") + +@app.on_message(filters.command("removeuser")) +async def cmd_removeuser(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + parts = message.command + if len(parts) < 2: + return await message.reply_text("Usage: `/removeuser `") + + username = parts[1].lstrip("@").lower() + if username not in AUTHORIZED_USERS: + return await message.reply_text(f"ℹ️ `@{username}` is not in the authorized list.") + + AUTHORIZED_USERS.discard(username) + _save_users(AUTHORIZED_USERS) + logger.info(f"Removed authorized user: {username}") + await message.reply_text(f"✅ `@{username}` removed from authorized users.") + +# ── /logs ───────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("logs")) +async def cmd_logs(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + try: + lines = LOG_FILE.read_text(errors="replace").splitlines() + tail = "\n".join(lines[-50:]) + # Send as file if content is large, otherwise inline + if len(tail) > 3800: + with tempfile.NamedTemporaryFile("w", suffix=".log", + delete=False, prefix="botlog_") as f: + f.write(tail) + tmp = f.name + await message.reply_document(tmp, caption="📄 Last 50 log lines") + os.remove(tmp) + else: + await message.reply_text(f"📄 **Last 50 log lines**\n\n```\n{tail}\n```") + except Exception as e: + await message.reply_text(f"❌ Could not read log\n\n`{e}`") + +# ── /restart ────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("restart")) +async def cmd_restart(client: Client, message: Message): + if not is_authorized(message): + return await deny(message) + + await message.reply_text("♻️ Restarting bot... brb.") + logger.info("Restart requested via /restart") + + async def do_restart(): + await asyncio.sleep(1) + await client.stop() + os.execv(sys.executable, [sys.executable] + sys.argv) + + asyncio.create_task(do_restart()) + +# ── /help ───────────────────────────────────────────────────────────────────── + +@app.on_message(filters.command("help")) +async def cmd_help(client: Client, message: Message): + auth_note = ( + "\n✅ You are an **authorized user**." + if is_authorized(message) else + "\n🔒 Commands require authorization to run." + ) + await message.reply_text( + "**GDrive Bot — Commands**\n\n" + "━━━━━━━━━━━━━━━━━━━━\n" + "📂 **File & Transfer**\n" + "`/list` — Show the 10 most recent files in the Drive folder\n" + "`/listall` — List **every** file in the Drive folder (paginated)\n" + "`/search ` — Search files by name or keyword\n" + "`/share ` — Make a file public and get a shareable link\n" + "`/get ` — Pull a file **from** Drive back into this chat\n" + "`/stats ` — Show detailed Drive file info (size, date, MIME, path)\n" + "`/delete ` — Permanently delete a file from Drive\n\n" + "📁 **Folders**\n" + "`/folders` — List available Drive folders and which extensions go where\n\n" + "📊 **Info**\n" + "`/status` — Bot uptime, RAM, CPU, disk usage\n" + "`/df` — Show Google Drive storage usage (used, free, total)\n" + "`/logs` — Tail the last 50 lines of the bot log\n\n" + "👤 **User Management**\n" + "`/adduser <@username>` — Grant full (unlimited) upload access\n" + "`/removeuser <@username>` — Revoke full access\n\n" + "📊 **Quotas** _(for non-auth users)_\n" + "`/setquota @user ` — Set a daily upload quota (e.g. `5GB`, `500MB`)\n" + "`/clearquota @user` — Remove quota and reset usage\n" + "`/quotas` — Overview of all quotas and today's usage\n" + "`/myquota` — Check your own quota (any user)\n\n" + "⚙️ **System**\n" + "`/restart` — Gracefully restart the bot\n" + "`/help` — Show this message\n" + "`/start` — Show upload instructions\n\n" + "━━━━━━━━━━━━━━━━━━━━\n" + "**How to upload:**\n" + "📁 Send any file · 🌐 Paste a URL (or multiple!) · ✏️ Caption to rename\n" + "🔗 Google Drive shared links are auto-detected\n" + "🎬 Media links (YouTube, Instagram, TikTok, X, Reddit, Pinterest, " + "Tumblr, Facebook, LinkedIn, Aparat & more) are downloaded via yt-dlp " + f"into **{MEDIA_DLS_DIR}/**\n" + "🎵 Add the word `mp3` or `audio` after a media link to extract audio\n" + f"🎧 Spotify links (track/album/playlist) → spotdl → **{MEDIA_DLS_DIR}/Music**\n" + "🧲 Magnet links & `.torrent` files → aria2c → **Torrents/** (≥8 GB streams file-by-file)\n" + "🎛 Every upload gets buttons: 🌍 Share · ℹ️ Info · 📂 Move · 🗑 Delete · 📥 Send to TG\n\n" + "Made by @Coaxys" + + auth_note + ) + +# ── Button callbacks ────────────────────────────────────────────────────────── + +@app.on_callback_query() +async def handle_buttons(client: Client, cq: CallbackQuery): + parts = (cq.data or "").split(":", 2) + action = parts[0] + tok = parts[1] if len(parts) > 1 else "" + + entry = FILE_ACTIONS.get(tok) + if not entry: + return await cq.answer("⚠️ These buttons expired.", show_alert=True) + + user = (cq.from_user.username or "").lower() if cq.from_user else "" + if user not in AUTHORIZED_USERS and user != entry.get("owner", ""): + return await cq.answer("⛔ You can't manage this file.", show_alert=True) + + name = entry["name"] + folder = entry["folder"] + rpath = f"{RCLONE_REMOTE}{folder}/{name}" + base = cq.message.text.markdown if (cq.message and cq.message.text) else "🔗 **Links**" + + async def rclone(*args) -> tuple[bool, str]: + proc = await asyncio.create_subprocess_exec( + "rclone", *args, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + return proc.returncode == 0, (out.decode().strip() or err.decode().strip() + if proc.returncode == 0 + else err.decode().strip() or out.decode().strip()) + + try: + if action == "share": + await cq.answer("🌍 Making public...") + ok, res = await rclone("link", rpath) + if not ok: + return await cq.answer(f"❌ Share failed: {res[:150]}", show_alert=True) + await cq.message.edit_text( + base + f"\n\n🌍 **Public link:**\n`{res}`", + disable_web_page_preview=True, + reply_markup=file_buttons(tok), + ) + + elif action == "info": + e = await get_drive_entry(name, folder) + if not e: + return await cq.answer("❓ Not found in Drive — moved or deleted?", + show_alert=True) + await cq.answer( + f"📎 {name[:70]}\n" + f"📦 {fmt_size(e.get('Size', 0))}\n" + f"🗂 {e.get('MimeType', '?')}\n" + f"📁 {folder}/\n" + f"🕐 {e.get('ModTime', '')[:19].replace('T', ' ')}", + show_alert=True, + ) + + elif action == "mv": + await cq.message.edit_reply_markup(move_keyboard(tok)) + await cq.answer("📂 Pick a destination folder") + + elif action == "mvto": + dest = parts[2] if len(parts) > 2 else "" + if not dest: + return await cq.answer("❌ Bad destination.", show_alert=True) + if dest == folder: + await cq.message.edit_reply_markup(file_buttons(tok)) + return await cq.answer("Already there 😉") + await cq.answer(f"📂 Moving to {dest}/ ...") + ok, res = await rclone("moveto", rpath, f"{RCLONE_REMOTE}{dest}/{name}") + if not ok: + await cq.message.edit_reply_markup(file_buttons(tok)) + return await cq.answer(f"❌ Move failed: {res[:150]}", show_alert=True) + entry["folder"] = dest + _save_actions() + await cq.message.edit_text( + base + f"\n\n📂 Moved: **{folder}/** → **{dest}/**", + disable_web_page_preview=True, + reply_markup=file_buttons(tok), + ) + + elif action == "del": + await cq.message.edit_reply_markup(del_confirm_keyboard(tok)) + await cq.answer(f"⚠️ Delete “{name[:50]}” permanently?", show_alert=True) + + elif action == "delc": + await cq.answer("🗑 Deleting...") + ok, res = await rclone("deletefile", rpath) + if not ok: + await cq.message.edit_reply_markup(file_buttons(tok)) + return await cq.answer(f"❌ Delete failed: {res[:150]}", show_alert=True) + FILE_ACTIONS.pop(tok, None) + _save_actions() + await cq.message.edit_text( + base + "\n\n🗑 **Deleted from Drive** by " + f"@{cq.from_user.username or 'unknown'}.", + disable_web_page_preview=True, + reply_markup=None, + ) + + elif action == "get": + await cq.answer("📥 Fetching from Drive → Telegram...") + await deliver_to_telegram(client, cq.message, name, folder) + + elif action == "back": + await cq.message.edit_reply_markup(file_buttons(tok)) + await cq.answer() + + else: + await cq.answer() + + except Exception as e: + logger.error(f"button {action} failed: {e}") + try: + await cq.answer(f"❌ {str(e)[:150]}", show_alert=True) + except Exception: + pass + +# ── URL handler ─────────────────────────────────────────────────────────────── + +@app.on_message(filters.text & ~filters.command( + ["start", "help", "list", "listall", "status", "search", "share", + "adduser", "removeuser", "logs", "restart", + "setquota", "clearquota", "quotas", "myquota", + "folders", "delete", "stats", "df", "get"] +)) +async def handle_url(client: Client, message: Message): + allowed, reason = has_upload_access(message) + if not allowed: + if reason == "quota_exceeded": + u = (message.from_user.username or "").lower() + limit = get_quota(u) + await message.reply_text( + f"🚫 **Daily quota exceeded**\n\n" + f"You've used your full {fmt_size(limit)} for today.\n" + f"🔄 Resets at midnight UTC." + ) + return + + text = message.text or "" + + # Magnet links → torrent handler + magnets = MAGNET_RE.findall(text) + if magnets: + for i, mag in enumerate(magnets, 1): + await process_torrent(client, message, f"{i}/{len(magnets)}", magnet=mag) + return + + urls = [u.rstrip(".,;)") for u in URL_RE.findall(text)] + if not urls: + return + + # If multiple URLs found, process as batch + if len(urls) > 1: + batch_status = await message.reply_text( + f"📦 **Batch URL download**\n\n" + f"Found **{len(urls)} URLs**. Processing one by one..." + ) + success = 0 + fail = 0 + for i, url in enumerate(urls, 1): + ok = await _process_url(client, message, url, f"{i}/{len(urls)}") + if ok: + success += 1 + else: + fail += 1 + await batch_status.edit_text( + f"📦 **Batch complete** — {success} ✅ / {fail} ❌\n" + f"_Check individual results above._" + ) + return + + # Single URL + await _process_url(client, message, urls[0], "1/1") + + +async def _process_url(client: Client, message: Message, url: str, seq: str) -> bool: + """Process a single URL: media sites → yt-dlp, Google Drive links → direct + download, everything else → rclone copyurl. Returns True on success.""" + if spotify_match(url): + return await _spotify_download(client, message, url, seq) + + is_media, is_nsfw = ytdlp_match(url) + if is_media: + return await _ytdlp_download(client, message, url, seq, is_nsfw) + + chunk = "512M" + total_start = time.time() + + # Detect Google Drive shared links → use direct download URL + gdrive_id = extract_gdrive_id(url) + if gdrive_id: + dl_url = f"https://drive.google.com/uc?export=download&id={gdrive_id}" + # Try to get filename from content-disposition + file_name = await _fetch_gdrive_filename(gdrive_id) + if not file_name: + file_name = f"gdrive_{gdrive_id[:12]}" + # Try grabbing from the URL itself + from urllib.parse import unquote, urlparse + parsed = unquote(urlparse(url).path) + base = os.path.basename(parsed.rstrip("/")) + if base and "." in base: + file_name = sanitize_filename(base) + url_label = f"🔗 Google Drive file" + else: + dl_url = url + file_name = filename_from_url(url) + url_label = f"`{url[:80]}{'...' if len(url)>80 else ''}`" + + status = await message.reply_text( + f"🌐 **{seq}** — URL received\n\n" + f"{url_label}\n\n" + f"📎 Saving as: **{file_name}**\n" + f"🧩 Chunk: {chunk}\n\n" + f"⏳ Starting rclone stream (URL → RAM → Drive)..." + ) + + folder = classify_to_folder(file_name) + rclone_cmd = [ + "rclone", "copyurl", dl_url, + f"{RCLONE_REMOTE}{folder}/{file_name}", + "--drive-chunk-size", chunk, + "--buffer-size", "512M", + ] + + try: + proc = await asyncio.create_subprocess_exec( + *rclone_cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + except Exception as e: + await status.edit_text(f"❌ **rclone failed to start**\n\n`{e}`") + return False + + # Pulse-update while rclone works (copyurl gives no byte-level progress) + start_ts = time.time() + dots = ["⠋","⠙","⠹","⠸","⠼","⠴","⠦","⠧","⠇","⠏"] + dot_i = 0 + done_event = asyncio.Event() + + async def pulse(): + nonlocal dot_i + while not done_event.is_set(): + elapsed = time.time() - start_ts + try: + await status.edit_text( + f"🌐 **{seq}** — Downloading & uploading...\n\n" + f"📎 **{file_name}**\n" + f"{dots[dot_i % len(dots)]} Elapsed: {fmt_dur(elapsed)}\n\n" + f"🧩 Chunk: {chunk} · 💾 Disk: **0 B**\n" + f"🖥 RAM: {fmt_ram()}" + ) + except Exception: + pass + dot_i += 1 + await asyncio.sleep(4) + + pulse_task = asyncio.create_task(pulse()) + out, err = await proc.communicate() + done_event.set() + pulse_task.cancel() + try: + await pulse_task + except asyncio.CancelledError: + pass + + total_elapsed = time.time() - total_start + + if proc.returncode != 0: + error = err.decode().strip() or out.decode().strip() + logger.error(f"rclone copyurl failed: {error}") + await status.edit_text( + f"❌ **URL download failed**\n\n" + f"📎 **{file_name}**\n\n" + f"💬 `{error}`" + ) + return False + + global session_uploads + session_uploads += 1 + await status.edit_text( + f"✅ **Done!**\n\n" + f"📎 **{file_name}**\n" + f"📁 → **Google Drive / {folder}**\n\n" + f"🕐 Total: {fmt_dur(total_elapsed)} · 💾 Disk: **0 B**" + ) + + owner = (message.from_user.username or "").lower() if message.from_user else "" + await send_links(message, file_name, folder=folder, owner=owner) + return True + + +async def _fetch_gdrive_filename(file_id: str) -> str | None: + """Try to get the filename from a Google Drive link via a HEAD request.""" + try: + proc = await asyncio.create_subprocess_exec( + "curl", "-sI", "-L", + f"https://drive.google.com/uc?export=download&id={file_id}", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, _ = await proc.communicate(timeout=10) + for line in out.decode(errors="replace").splitlines(): + if line.lower().startswith("content-disposition:"): + m = re.search(r'filename\*?=(?:UTF-8\'\')?"?([^";\n]+)', line) + if m: + from urllib.parse import unquote + name = unquote(m.group(1).strip()) + if name and "." in name: + return sanitize_filename(name) + except Exception: + pass + return None + +# ── yt-dlp media downloader ─────────────────────────────────────────────────── + +YTDLP_PROGRESS_RE = re.compile( + r'\[download\]\s+(?P[\d.]+)%\s+of\s+~?\s*(?P[\d.]+\s*\S+?B)' + r'(?:\s+at\s+(?P\S+))?(?:\s+ETA\s+(?P\S+))?' +) +YTDLP_ITEM_RE = re.compile(r'\[download\] Downloading item (\d+) of (\d+)') +YTDLP_MAX_RUNTIME = 3 * 3600 # hard cap (livestreams could run forever) +YTDLP_PLAYLIST_CAP = "1:50" # max items pulled from a playlist/channel + +async def _ytdlp_download(client: Client, message: Message, url: str, + seq: str, nsfw: bool) -> bool: + """Download a media-site URL with yt-dlp to a temp dir, then rclone copy + every produced file into gdrive:MEDIA-DLS//. Returns success.""" + text = message.text or "" + wants_audio = bool(re.search(r'\b(mp3|audio)\b', text, re.IGNORECASE)) + username = (message.from_user.username or "").lower() if message.from_user else "" + site = urlparse(url).netloc.lower().removeprefix("www.") + temp_dir = tempfile.mkdtemp(prefix="ytdlp_") + total_start = time.time() + + status = await message.reply_text( + f"🎬 **{seq}** — Media link detected\n\n" + f"🌐 **{site}**{' · 🔞 NSFW' if nsfw else ''}\n" + f"🎚 Mode: {'🎵 audio (mp3)' if wants_audio else '🎞 best video+audio'}\n\n" + f"⏳ Starting yt-dlp..." + ) + + cmd = [ + sys.executable, "-m", "yt_dlp", + "--newline", "--no-warnings", "--color", "no_color", + "--no-playlist", "--playlist-items", YTDLP_PLAYLIST_CAP, + "--windows-filenames", "--trim-filenames", "180", + "--ffmpeg-location", "/usr/bin/ffmpeg", + "-o", os.path.join(temp_dir, "%(title)s [%(id)s].%(ext)s"), + ] + if wants_audio: + cmd += ["-x", "--audio-format", "mp3"] + else: + cmd += ["-f", "bv*+ba/b", "--merge-output-format", "mp4"] + if COOKIES_FILE.exists(): + cmd += ["--cookies", str(COOKIES_FILE)] + cmd.append(url) + + def _cleanup(): + for f in os.listdir(temp_dir): + try: os.remove(os.path.join(temp_dir, f)) + except Exception: pass + try: os.rmdir(temp_dir) + except Exception: pass + + try: + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + ) + except Exception as e: + await status.edit_text(f"❌ **yt-dlp failed to start**\n\n`{e}`") + _cleanup() + return False + + tail = collections.deque(maxlen=15) + last_edit = 0.0 + item_note = "" + + async def read_progress(): + nonlocal last_edit, item_note + while True: + line = await proc.stdout.readline() + if not line: + break + line = line.decode(errors="replace").rstrip() + if not line: + continue + tail.append(line) + m = YTDLP_ITEM_RE.search(line) + if m: + item_note = f" · 📚 Item {m.group(1)}/{m.group(2)}" + m = YTDLP_PROGRESS_RE.search(line) + now = time.time() + if m and now - last_edit >= 3: + last_edit = now + pct = min(float(m.group("pct")), 100.0) + bar = "█" * int(pct / 5) + "░" * (20 - int(pct / 5)) + try: + await status.edit_text( + f"🎬 **{seq}** — yt-dlp downloading{item_note}\n\n" + f"🌐 **{site}**\n" + f"`[{bar}]` **{pct:.1f}%** of {m.group('total')}\n" + f"⚡ {m.group('speed') or '—'} · ⏱ ETA {m.group('eta') or '—'}\n\n" + f"🕐 Elapsed: {fmt_dur(now - total_start)} · 🖥 RAM: {fmt_ram()}" + ) + except Exception: + pass + + try: + await asyncio.wait_for(read_progress(), timeout=YTDLP_MAX_RUNTIME) + await proc.wait() + except asyncio.TimeoutError: + try: proc.kill() + except Exception: pass + await status.edit_text( + f"❌ **yt-dlp timed out** after {fmt_dur(YTDLP_MAX_RUNTIME)}\n\n" + f"🌐 {site} — `{url[:80]}`\n" + f"_Livestreams are capped at {fmt_dur(YTDLP_MAX_RUNTIME)}._" + ) + _cleanup() + return False + + files = [f for f in sorted(os.listdir(temp_dir)) + if not f.endswith((".part", ".ytdl", ".temp", ".tmp"))] + + if not files: + err_tail = "\n".join(list(tail)[-5:]) or "no output" + logger.error(f"yt-dlp failed for {url}: {err_tail}") + hint = "" + if "cookies" in err_tail.lower() or "login" in err_tail.lower() \ + or "Sign in" in err_tail or "rate-limit" in err_tail.lower(): + hint = "\n\n💡 This site may need login cookies (`cookies.txt` in the bot folder)." + await status.edit_text( + f"❌ **yt-dlp failed** (exit {proc.returncode})\n\n" + f"🌐 {site}\n💬 `{err_tail[:600]}`{hint}" + ) + _cleanup() + return False + + return await _upload_media_files(message, status, seq, site, temp_dir, + files, nsfw, username, total_start) + +async def _upload_media_files(message, status, seq, site, temp_dir, files, + nsfw, username, total_start) -> bool: + """rclone-copy each downloaded file into MEDIA-DLS/, clean up + the temp dir, report the result and send Drive links. Returns success.""" + uploaded = [] + failed = [] + total_bytes = 0 + for i, fname in enumerate(files, 1): + fpath = os.path.join(temp_dir, fname) + fsize = os.path.getsize(fpath) + folder = media_dls_folder(fname, nsfw) + try: + await status.edit_text( + f"☁️ **{seq}** — Uploading {i}/{len(files)} to Drive\n\n" + f"📎 **{fname}**\n📦 {fmt_size(fsize)}\n" + f"📁 → **{folder}/**" + ) + except Exception: + pass + up = await asyncio.create_subprocess_exec( + "rclone", "copy", fpath, f"{RCLONE_REMOTE}{folder}/", + "--drive-chunk-size", choose_chunk(fsize), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await up.communicate() + if up.returncode == 0: + uploaded.append((fname, folder)) + total_bytes += fsize + try: os.remove(fpath) + except Exception: pass + else: + error = err.decode().strip() or out.decode().strip() + logger.error(f"rclone copy failed for {fname}: {error}") + failed.append(fname) + + for f in os.listdir(temp_dir): + try: os.remove(os.path.join(temp_dir, f)) + except Exception: pass + try: os.rmdir(temp_dir) + except Exception: pass + + if not uploaded: + await status.edit_text( + f"❌ **Drive upload failed** for all {len(files)} file(s)\n\n" + f"🌐 {site} — check `/logs` for rclone errors." + ) + return False + + global session_uploads + session_uploads += len(uploaded) + if username and username not in AUTHORIZED_USERS: + record_upload(username, total_bytes) + + total_elapsed = time.time() - total_start + file_lines = "\n".join(f"📎 `{n}` → **{f}/**" for n, f in uploaded[:10]) + if len(uploaded) > 10: + file_lines += f"\n_…and {len(uploaded) - 10} more_" + fail_note = f"\n⚠️ Failed to upload: {len(failed)} file(s)" if failed else "" + await status.edit_text( + f"✅ **Done!** ({seq})\n\n" + f"🌐 **{site}** · {len(uploaded)} file(s) · 📦 {fmt_size(total_bytes)}\n" + f"{file_lines}{fail_note}\n\n" + f"🕐 Total: {fmt_dur(total_elapsed)}" + ) + + if len(uploaded) <= 3: + for fname, ffolder in uploaded: + await send_links(message, fname, folder=ffolder, owner=username) + return True + +# ── Spotify downloader (spotdl) ─────────────────────────────────────────────── + +AUDIO_EXTS = (".mp3", ".m4a", ".ogg", ".opus", ".flac", ".wav") + +async def _spotify_download(client: Client, message: Message, url: str, + seq: str) -> bool: + """Download a Spotify track/album/playlist with spotdl (audio sourced from + YouTube Music), then upload to gdrive:MEDIA-DLS/Music. Returns success.""" + username = (message.from_user.username or "").lower() if message.from_user else "" + temp_dir = tempfile.mkdtemp(prefix="spotdl_") + total_start = time.time() + + status = await message.reply_text( + f"🎧 **{seq}** — Spotify link detected\n\n" + f"🔗 `{url[:80]}`\n\n" + f"⏳ Starting spotdl (Spotify metadata → YouTube Music audio)..." + ) + + cmd = [ + sys.executable, "-m", "spotdl", "download", url, + "--output", os.path.join(temp_dir, "{artists} - {title}.{output-ext}"), + ] + if COOKIES_FILE.exists(): + cmd += ["--cookie-file", str(COOKIES_FILE)] + + def _cleanup(): + for f in os.listdir(temp_dir): + try: os.remove(os.path.join(temp_dir, f)) + except Exception: pass + try: os.rmdir(temp_dir) + except Exception: pass + + try: + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + ) + except Exception as e: + await status.edit_text(f"❌ **spotdl failed to start**\n\n`{e}`") + _cleanup() + return False + + tail = collections.deque(maxlen=15) + last_edit = 0.0 + + async def read_output(): + nonlocal last_edit + while True: + line = await proc.stdout.readline() + if not line: + break + line = line.decode(errors="replace").strip() + if not line: + continue + tail.append(line) + now = time.time() + if now - last_edit >= 4: + last_edit = now + done = len([f for f in os.listdir(temp_dir) + if f.endswith(AUDIO_EXTS)]) + try: + await status.edit_text( + f"🎧 **{seq}** — spotdl working...\n\n" + f"🎵 Songs done: **{done}**\n" + f"💬 `{line[:120]}`\n\n" + f"🕐 Elapsed: {fmt_dur(now - total_start)} · 🖥 RAM: {fmt_ram()}" + ) + except Exception: + pass + + try: + await asyncio.wait_for(read_output(), timeout=YTDLP_MAX_RUNTIME) + await proc.wait() + except asyncio.TimeoutError: + try: proc.kill() + except Exception: pass + await status.edit_text( + f"❌ **spotdl timed out** after {fmt_dur(YTDLP_MAX_RUNTIME)}\n\n🔗 `{url[:80]}`" + ) + _cleanup() + return False + + files = [f for f in sorted(os.listdir(temp_dir)) if f.endswith(AUDIO_EXTS)] + + if not files: + err_tail = "\n".join(list(tail)[-5:]) or "no output" + logger.error(f"spotdl failed for {url}: {err_tail}") + hint = "" + if "AudioProviderError" in err_tail or "bot" in err_tail.lower(): + hint = ("\n\n💡 spotdl pulls audio from YouTube Music — this server's IP " + "is YT-flagged, so a `cookies.txt` in the bot folder is required.") + await status.edit_text( + f"❌ **spotdl failed** (exit {proc.returncode})\n\n" + f"🔗 `{url[:80]}`\n💬 `{err_tail[:600]}`{hint}" + ) + _cleanup() + return False + + return await _upload_media_files(message, status, seq, "spotify.com", + temp_dir, files, False, username, total_start) + +# ── Torrents (aria2c) ───────────────────────────────────────────────────────── + +def _bdecode(data: bytes, pos: int = 0): + """Minimal bencode decoder — enough to read a .torrent file list.""" + t = data[pos:pos + 1] + if t == b'i': + end = data.index(b'e', pos) + return int(data[pos + 1:end]), end + 1 + if t == b'l': + pos += 1; out = [] + while data[pos:pos + 1] != b'e': + v, pos = _bdecode(data, pos); out.append(v) + return out, pos + 1 + if t == b'd': + pos += 1; out = {} + while data[pos:pos + 1] != b'e': + k, pos = _bdecode(data, pos) + v, pos = _bdecode(data, pos) + out[k] = v + return out, pos + 1 + colon = data.index(b':', pos) + n = int(data[pos:colon]) + start = colon + 1 + return data[start:start + n], start + n + +def torrent_files(torrent_path: str) -> tuple[str, list[tuple[str, int]]]: + """Return (display_name, [(relative_path, size_bytes), ...]) from a .torrent. + relative_path includes the torrent's top folder for multi-file torrents.""" + # Names are used verbatim for both the on-disk path aria2 writes and the + # Drive destination (Google Drive accepts spaces/unicode), so they must NOT + # be altered here — sanitizing would break the local-file lookup. + data = Path(torrent_path).read_bytes() + meta, _ = _bdecode(data) + info = meta[b'info'] + name = info[b'name'].decode('utf-8', 'replace') + if b'files' in info: # multi-file + files = [] + for f in info[b'files']: + parts = [p.decode('utf-8', 'replace') for p in f[b'path']] + rel = os.path.join(name, *parts) + files.append((rel, int(f[b'length']))) + return name, files + return name, [(name, int(info[b'length']))] # single-file + +ARIA2_PROGRESS_RE = re.compile( + r'(?P[\d.]+\w*)/(?P[\d.]+\w*)\((?P\d+)%\)' + r'(?:\s+CN:(?P\d+))?(?:\s+SD:(?P\d+))?' + r'(?:\s+DL:(?P
\S+?))?(?:\s+ETA:(?P\S+?))?\]' +) + +ARIA2_COMMON = [ + "--seed-time=0", "--seed-ratio=0.0", "--bt-stop-timeout=300", + "--summary-interval=1", "--enable-color=false", "--console-log-level=warn", + "--max-connection-per-server=16", "--split=16", "--bt-max-peers=200", + "--allow-overwrite=true", "--auto-file-renaming=false", +] + +async def _aria2_fetch_metadata(magnet: str, work: str) -> str | None: + """Download only the .torrent metadata for a magnet; return its path.""" + proc = await asyncio.create_subprocess_exec( + "aria2c", magnet, "--dir", work, + "--bt-metadata-only=true", "--bt-save-metadata=true", + "--bt-stop-timeout=120", "--seed-time=0", "--console-log-level=warn", + stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, + ) + try: + await asyncio.wait_for(proc.wait(), timeout=180) + except asyncio.TimeoutError: + try: proc.kill() + except Exception: pass + return None + tfiles = [f for f in os.listdir(work) if f.endswith(".torrent")] + return os.path.join(work, tfiles[0]) if tfiles else None + +async def _aria2_run(source: str, work: str, status, header: str, + total_start: float, select_index: int | None = None) -> bool: + """Run aria2c on a magnet/.torrent (optionally one file via --select-file), + streaming progress to the status message. Returns True on success.""" + cmd = ["aria2c", source, "--dir", work] + ARIA2_COMMON + if select_index is not None: + cmd.append(f"--select-file={select_index}") + try: + proc = await asyncio.create_subprocess_exec( + *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, + ) + except Exception as e: + await status.edit_text(f"❌ **aria2c failed to start**\n\n`{e}`") + return False + + last_edit = 0.0 + async def pump(): + nonlocal last_edit + while True: + raw = await proc.stdout.readline() + if not raw: + break + for line in raw.decode(errors="replace").replace("\r", "\n").split("\n"): + m = ARIA2_PROGRESS_RE.search(line) + now = time.time() + if m and now - last_edit >= 3: + last_edit = now + pct = int(m.group("pct")) + bar = "█" * int(pct / 5) + "░" * (20 - int(pct / 5)) + try: + await status.edit_text( + f"{header}\n" + f"`[{bar}]` **{pct}%** ({m.group('done')}/{m.group('total')})\n" + f"⚡ {m.group('dl') or '—'}/s · 👥 {m.group('cn') or '0'} peers" + f" · ⏱ ETA {m.group('eta') or '—'}\n\n" + f"🕐 {fmt_dur(now - total_start)} · 🖥 RAM: {fmt_ram()}" + ) + except Exception: + pass + try: + await asyncio.wait_for(pump(), timeout=YTDLP_MAX_RUNTIME) + await proc.wait() + except asyncio.TimeoutError: + try: proc.kill() + except Exception: pass + return False + return proc.returncode == 0 + +async def _upload_one(local_path: str, rel_path: str, status, label: str) -> tuple[bool, int]: + """rclone-copy a single torrent file into Torrents//. Returns (ok, bytes).""" + if not os.path.exists(local_path): + return False, 0 + size = os.path.getsize(local_path) + rel_dir = os.path.dirname(rel_path) + dest = f"{RCLONE_REMOTE}{TORRENTS_DIR}/{rel_dir}/" if rel_dir else f"{RCLONE_REMOTE}{TORRENTS_DIR}/" + try: + await status.edit_text( + f"☁️ {label}\n\n📎 **{os.path.basename(rel_path)}**\n" + f"📦 {fmt_size(size)} · 📁 → **{TORRENTS_DIR}/{rel_dir}/**" + ) + except Exception: + pass + proc = await asyncio.create_subprocess_exec( + "rclone", "copy", local_path, dest, "--drive-chunk-size", choose_chunk(size), + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + if proc.returncode != 0: + logger.error(f"torrent upload failed {rel_path}: {err.decode().strip()}") + return False, 0 + return True, size + +async def process_torrent(client: Client, message: Message, seq: str, + magnet: str | None = None, torrent_path: str | None = None) -> bool: + """Download a magnet/.torrent via aria2c and upload to gdrive:Torrents/. + < 8 GB: whole download then upload. ≥ 8 GB: per-file (download→upload→delete) + to keep peak disk usage low. Returns success.""" + import shutil + username = (message.from_user.username or "").lower() if message.from_user else "" + total_start = time.time() + WORK_DIR.mkdir(exist_ok=True) + work = tempfile.mkdtemp(dir=str(WORK_DIR), prefix="t_") + + def _cleanup(): + try: shutil.rmtree(work, ignore_errors=True) + except Exception: pass + + status = await message.reply_text( + f"🧲 **{seq}** — Torrent received\n\n⏳ Fetching metadata (peers)..." + ) + + # Resolve to a local .torrent (parse file list before downloading data) + if magnet: + torrent_path = await _aria2_fetch_metadata(magnet, work) + if not torrent_path: + await status.edit_text( + "❌ **Couldn't fetch torrent metadata**\n\n" + "No peers responded within 3 min — the magnet may be dead." + ) + _cleanup() + return False + source = torrent_path + + try: + name, files = torrent_files(torrent_path) + except Exception as e: + logger.error(f"torrent parse failed: {e}") + await status.edit_text(f"❌ **Invalid .torrent file**\n\n`{e}`") + _cleanup() + return False + + total = sum(s for _, s in files) + per_file = total >= TORRENT_STREAM_THRESHOLD + biggest = max((s for _, s in files), default=0) + free = shutil.disk_usage(str(WORK_DIR)).free + need = (biggest if per_file else total) + DISK_HEADROOM + + if free < need: + await status.edit_text( + f"❌ **Not enough disk space**\n\n" + f"📦 Torrent: **{fmt_size(total)}** ({len(files)} files, biggest {fmt_size(biggest)})\n" + f"💾 Free disk: {fmt_size(free)} · needs ~{fmt_size(need)}\n\n" + f"_{'Per-file mode still needs the biggest file to fit.' if per_file else 'Try again when space frees up.'}_" + ) + _cleanup() + return False + + if username and username not in AUTHORIZED_USERS: + rem = quota_remaining(username) + if rem is not None and total > rem: + await status.edit_text( + f"🚫 **Not enough quota**\n\n📦 Torrent: {fmt_size(total)}\n" + f"✅ Remaining: {fmt_size(rem)}\n\n🔄 Resets at midnight UTC." + ) + _cleanup() + return False + + mode = f"🧩 Per-file (≥8 GB, low disk)" if per_file else "💾 Whole download" + await status.edit_text( + f"🧲 **{seq}** — {name}\n\n" + f"📦 {fmt_size(total)} · 📂 {len(files)} file(s)\n" + f"🔧 {mode} · 💾 Free disk: {fmt_size(free)}\n\n⏳ Starting aria2c..." + ) + + uploaded, failed, sent_bytes = [], [], 0 + + if not per_file: + ok = await _aria2_run(source, work, status, + f"🧲 **{seq}** — Downloading: {name}", total_start) + if not ok: + await status.edit_text(f"❌ **Torrent download failed** — {name}\n\nCheck `/logs`.") + _cleanup() + return False + for rel, _ in files: + up_ok, size = await _upload_one( + os.path.join(work, rel), rel, status, + f"**{seq}** — Uploading {len(uploaded)+1}/{len(files)}") + (uploaded if up_ok else failed).append(rel) + if up_ok: + sent_bytes += size + try: os.remove(os.path.join(work, rel)) + except Exception: pass + else: + for i, (rel, _) in enumerate(files, 1): # aria2 file index is 1-based + hdr = f"🧲 **{seq}** — File {i}/{len(files)}: {os.path.basename(rel)}\n📥 Downloading" + ok = await _aria2_run(source, work, status, hdr, total_start, select_index=i) + local = os.path.join(work, rel) + if not ok or not os.path.exists(local): + failed.append(rel) + continue + up_ok, size = await _upload_one( + local, rel, status, f"**{seq}** — Uploading file {i}/{len(files)}") + (uploaded if up_ok else failed).append(rel) + if up_ok: + sent_bytes += size + try: os.remove(local) # free disk before next file + except Exception: pass + + _cleanup() + + if not uploaded: + await status.edit_text(f"❌ **Torrent failed** — no files uploaded for {name}.\n\nCheck `/logs`.") + return False + + global session_uploads + session_uploads += len(uploaded) + if username and username not in AUTHORIZED_USERS: + record_upload(username, sent_bytes) + + fail_note = f"\n⚠️ Failed: {len(failed)} file(s)" if failed else "" + await status.edit_text( + f"✅ **Torrent done!** ({seq})\n\n" + f"🧲 **{name}**\n" + f"📂 {len(uploaded)} file(s) · 📦 {fmt_size(sent_bytes)}\n" + f"📁 → **Drive / {TORRENTS_DIR}/{name}/**{fail_note}\n\n" + f"🕐 Total: {fmt_dur(time.time() - total_start)}" + ) + if len(uploaded) <= 3: + for rel in uploaded: + rel_dir = os.path.dirname(rel) + await send_links(message, os.path.basename(rel), + folder=f"{TORRENTS_DIR}/{rel_dir}" if rel_dir else TORRENTS_DIR, + owner=username) + return True + +# ── Telegram delivery (Drive → Telegram) ────────────────────────────────────── + +async def deliver_to_telegram(client: Client, message: Message, + name: str, folder: str) -> bool: + """Download a file from Drive and send it back into Telegram. Returns success.""" + import shutil + entry = await get_drive_entry(name, folder) + if not entry or not entry.get("ID"): + await message.reply_text(f"❌ `{name}` not found in Drive (moved or deleted?).") + return False + size = entry.get("Size", 0) + if size > TG_SEND_LIMIT: + await message.reply_text( + f"⚠️ **Too big to send via Telegram**\n\n" + f"📎 **{name}** — {fmt_size(size)} (limit {fmt_size(TG_SEND_LIMIT)})\n\n" + f"Use 🌍 Share / `/share {name}` for a download link instead." + ) + return False + + status = await message.reply_text( + f"📤 **Fetching from Drive**\n\n📎 **{name}**\n📦 {fmt_size(size)}\n\n⏳ Downloading..." + ) + WORK_DIR.mkdir(exist_ok=True) + work = tempfile.mkdtemp(dir=str(WORK_DIR), prefix="get_") + rpath = f"{RCLONE_REMOTE}{folder}/{name}" + + proc = await asyncio.create_subprocess_exec( + "rclone", "copy", rpath, work, "--stats=1s", "--stats-one-line", + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, + ) + rstat_re = re.compile(r'Transferred:\s+\S+\s+\w+\s*/\s*\S+\s+\w+,\s+(\d+)%') + last = 0.0 + while True: + raw = await proc.stdout.readline() + if not raw: + break + m = rstat_re.search(raw.decode(errors="replace")) + now = time.time() + if m and now - last >= 3: + last = now + pct = int(m.group(1)); bar = "█"*int(pct/5)+"░"*(20-int(pct/5)) + try: await status.edit_text(f"📤 **Fetching from Drive**\n\n📎 **{name}**\n`[{bar}]` **{pct}%**") + except Exception: pass + await proc.wait() + + local = os.path.join(work, name) + if proc.returncode != 0 or not os.path.exists(local): + await status.edit_text(f"❌ **Drive download failed** for `{name}`.\n\nCheck `/logs`.") + shutil.rmtree(work, ignore_errors=True) + return False + + up_start = time.time() + last_up = 0.0 + async def up_prog(cur, tot): + nonlocal last_up + now = time.time() + if now - last_up < 3: return + last_up = now + pct = cur/tot*100 if tot else 0; bar = "█"*int(pct/5)+"░"*(20-int(pct/5)) + try: + await status.edit_text( + f"📲 **Sending to Telegram**\n\n📎 **{name}**\n`[{bar}]` **{pct:.0f}%**\n" + f"📦 {fmt_size(cur)} / {fmt_size(tot)}" + ) + except Exception: pass + + ext = os.path.splitext(name)[1].lower() + try: + if ext in VIDEO_SEND_EXTS: + await client.send_video(message.chat.id, local, caption=f"🎬 {name}", + progress=up_prog, reply_to_message_id=message.id, + supports_streaming=True) + elif ext in AUDIO_SEND_EXTS: + await client.send_audio(message.chat.id, local, caption=f"🎵 {name}", + progress=up_prog, reply_to_message_id=message.id) + else: + await client.send_document(message.chat.id, local, caption=f"📎 {name}", + progress=up_prog, reply_to_message_id=message.id, + force_document=(ext not in IMAGE_SEND_EXTS)) + await status.edit_text( + f"✅ **Sent!**\n\n📎 **{name}** · 📦 {fmt_size(size)}\n" + f"🕐 {fmt_dur(time.time() - up_start)}" + ) + ok = True + except Exception as e: + logger.error(f"deliver send failed for {name}: {e}") + await status.edit_text(f"❌ **Telegram send failed** for `{name}`\n\n`{str(e)[:200]}`") + ok = False + finally: + shutil.rmtree(work, ignore_errors=True) + return ok + +# ── File handler ────────────────────────────────────────────────────────────── + +@app.on_message(filters.media) +async def handle_file(client: Client, message: Message): + allowed, reason = has_upload_access(message) + if not allowed: + if reason == "quota_exceeded": + u = (message.from_user.username or "").lower() + limit = get_quota(u) + await message.reply_text( + f"🚫 **Daily quota exceeded**\n\n" + f"You've used your full {fmt_size(limit)} for today.\n" + f"🔄 Resets at midnight UTC." + ) + return + media_type, media = get_media(message) + if not media: + return + + # .torrent files → torrent handler (download the metadata file, then aria2c) + doc_name = getattr(media, "file_name", "") or "" + doc_mime = getattr(media, "mime_type", "") or "" + if media_type == "document" and (doc_name.lower().endswith(".torrent") + or "bittorrent" in doc_mime.lower()): + WORK_DIR.mkdir(exist_ok=True) + tpath = await client.download_media( + message, file_name=os.path.join(str(WORK_DIR), f"in_{message.id}.torrent")) + if tpath: + await process_torrent(client, message, "1/1", torrent_path=tpath) + try: os.remove(tpath) + except Exception: pass + return + + caption = (message.caption or "").strip() or None + file_name = resolve_filename(media_type, media, caption) + raw_size = getattr(media, "file_size", 0) or 0 + size_str = fmt_size(raw_size) if raw_size else "unknown" + mime = getattr(media, "mime_type", None) or media_type + dc_id = getattr(media, "dc_id", None) + dc_note = f"DC{dc_id}" if dc_id else "DC?" + chunk_label = choose_chunk(raw_size) + use_stream = raw_size >= STREAM_THRESHOLD + method = "🌊 RAM stream" if use_stream else "💾 Download → Upload" + total_start = time.time() + + username = (message.from_user.username or "").lower() if message.from_user else "" + rename_note = f"\n✏️ Renamed to: **{file_name}**" if caption else "" + + # Quota pre-check: reject if file would exceed remaining quota + if username not in AUTHORIZED_USERS and raw_size: + rem = quota_remaining(username) + if rem is not None and raw_size > rem: + return await message.reply_text( + f"🚫 **Not enough quota**\n\n" + f"📦 File size: {fmt_size(raw_size)}\n" + f"✅ Remaining: {fmt_size(rem)}\n\n" + f"🔄 Quota resets at midnight UTC." + ) + + status = await message.reply_text( + f"📥 **Received**\n\n" + f"**{file_name}**\n" + f"📦 {size_str} · 🗂 `{mime}` · 📡 {dc_note}{rename_note}\n" + f"🔧 Method: {method} · 🧩 Chunk: {chunk_label}\n\n" + f"⏳ Starting..." + ) + + folder = classify_to_folder(file_name, mime) + if use_stream: + await _stream_to_drive(client, message, status, file_name, raw_size, + size_str, mime, dc_note, chunk_label, total_start, + username=username, folder=folder) + else: + await _download_then_upload(client, message, status, file_name, raw_size, + size_str, mime, dc_note, chunk_label, total_start, + username=username, folder=folder) + +# ── Method A: download to disk → rclone copy (< 2 GB) ─────────────────────── + +async def _download_then_upload( + client, message, status, file_name, raw_size, + size_str, mime, dc_note, chunk_label, total_start, username="", folder="Other" +): + temp_dir = tempfile.mkdtemp() + local_path = os.path.join(temp_dir, file_name) + + dl_start = time.time() + last_edit = 0.0 + peak_dl = 0.0 + speed_win = collections.deque(maxlen=6) + prev_curr = 0 + + async def progress(current: int, total: int): + nonlocal last_edit, peak_dl, prev_curr + now = time.time() + if now - last_edit < 3: + return + last_edit = now + elapsed = now - dl_start + avg = current / elapsed if elapsed > 0 else 0 + speed_win.append(avg) + peak_dl = max(peak_dl, avg) + smooth = sum(speed_win) / len(speed_win) + pct = current / total * 100 if total else 0 + eta = (total - current) / smooth if smooth > 0 else 0 + bar = "█" * int(pct / 5) + "░" * (20 - int(pct / 5)) + try: + await status.edit_text( + f"⬇️ **Downloading** · {dc_note}\n\n" + f"**{file_name}**\n" + f"`[{bar}]` **{pct:.1f}%**\n" + f"📦 {fmt_size(current)} / {fmt_size(total)}\n\n" + f"⚡ Speed: {spd(smooth)}\n" + f"🏆 Peak: {spd(peak_dl)}\n" + f"⏱ ETA: {fmt_dur(eta)} · Elapsed: {fmt_dur(elapsed)}\n\n" + f"🧩 Chunk: {chunk_label} · 🖥 RAM: {fmt_ram()}" + ) + except Exception: + pass + + MAX_RETRIES = 3 + RETRY_DELAYS = [5, 15, 30] + last_error = None + + for attempt in range(1, MAX_RETRIES + 1): + try: + # Reset progress state for each attempt + dl_start = time.time() + last_edit = 0.0 + peak_dl = 0.0 + speed_win.clear() + + if attempt > 1: + try: + await status.edit_text( + f"🔄 **Retrying download** ({attempt}/{MAX_RETRIES}) · {dc_note}\n\n" + f"**{file_name}**\n" + f"📦 {size_str}\n\n" + f"⚠️ Previous error: `{last_error}`\n" + f"⏳ Starting attempt {attempt}..." + ) + except Exception: + pass + # Clean up partial file from last attempt + try: os.remove(local_path) + except Exception: pass + + downloaded = await client.download_media(message, file_name=local_path, progress=progress) + if not downloaded: + raise RuntimeError("download_media returned None") + break # success — exit retry loop + + except Exception as e: + last_error = str(e) + logger.warning(f"Download attempt {attempt}/{MAX_RETRIES} failed: {e}") + if attempt == MAX_RETRIES: + logger.error(f"All {MAX_RETRIES} download attempts failed for {file_name}") + await status.edit_text( + f"❌ **Download failed** after {MAX_RETRIES} attempts\n\n" + f"**{file_name}**\n\n" + f"💬 Last error: `{last_error}`" + ) + try: os.remove(local_path) + except Exception: pass + try: os.rmdir(temp_dir) + except Exception: pass + return + delay = RETRY_DELAYS[attempt - 1] + try: + await status.edit_text( + f"⚠️ **Connection dropped** (attempt {attempt}/{MAX_RETRIES}) · {dc_note}\n\n" + f"**{file_name}**\n\n" + f"💬 `{last_error}`\n\n" + f"⏳ Retrying in {delay}s..." + ) + except Exception: + pass + await asyncio.sleep(delay) + + dl_elapsed = time.time() - dl_start + avg_dl = raw_size / dl_elapsed if dl_elapsed > 0 and raw_size else 0 + + await status.edit_text( + f"☁️ **Uploading to Drive...**\n\n" + f"**{file_name}**\n" + f"📦 {size_str} downloaded in {fmt_dur(dl_elapsed)} ({spd(avg_dl)})\n\n" + f"⏳ rclone copying to Google Drive..." + ) + + up_start = time.time() + try: + proc = await asyncio.create_subprocess_exec( + "rclone", "copy", local_path, f"{RCLONE_REMOTE}{folder}/", + "--drive-chunk-size", chunk_label, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + up_elapsed = time.time() - up_start + total_elapsed = time.time() - total_start + avg_up = raw_size / up_elapsed if up_elapsed > 0 and raw_size else 0 + + if proc.returncode != 0: + error = err.decode().strip() or out.decode().strip() + logger.error(f"rclone copy failed: {error}") + await status.edit_text( + f"❌ **Upload failed**\n\n**{file_name}**\n\n💬 `{error}`\n\n" + f"💾 Local copy at:\n`{local_path}`" + ) + return + + global session_uploads + session_uploads += 1 + if username and username not in AUTHORIZED_USERS: + record_upload(username, raw_size or up_elapsed * 1) # track actual bytes + quota_note = "" + if username and username not in AUTHORIZED_USERS: + rem = quota_remaining(username) + if rem is not None: + quota_note = f"\n📊 Quota left today: {fmt_size(rem)}" + await status.edit_text( + f"✅ **Done!**\n\n" + f"**{file_name}**\n" + f"📦 {size_str} · 🗂 `{mime}`\n\n" + f"📊 **Stats**\n" + f"⬇️ Download: {fmt_dur(dl_elapsed)} @ {spd(avg_dl)}\n" + f"☁️ Upload: {fmt_dur(up_elapsed)} @ {spd(avg_up)}\n" + f"🕐 Total: {fmt_dur(total_elapsed)}\n\n" + f"🧩 Chunk: {chunk_label} · 📡 {dc_note}\n" + f"📁 → **Google Drive / {folder}**{quota_note}" + ) + except Exception as e: + logger.error(f"rclone copy error: {e}") + await status.edit_text(f"❌ **Upload error**\n\n**{file_name}**\n\n💬 `{e}`") + return + finally: + try: os.remove(local_path) + except Exception: pass + try: os.rmdir(temp_dir) + except Exception: pass + + await send_links(message, file_name, folder=folder, owner=username) + +# ── Method B: pure RAM stream via rclone rcat (≥ 2 GB) ────────────────────── + +async def _stream_to_drive( + client, message, status, file_name, raw_size, + size_str, mime, dc_note, chunk_label, total_start, username="", folder="Other" +): + rclone_cmd = [ + "rclone", "rcat", f"{RCLONE_REMOTE}{folder}/{file_name}", + "--drive-chunk-size", chunk_label, + ] + if raw_size: + rclone_cmd += ["--size", str(raw_size)] + + MAX_RETRIES = 3 + RETRY_DELAYS = [5, 15, 30] + last_error = None + bytes_sent = 0 + peak_spd = 0.0 + stream_start = time.time() + + for attempt in range(1, MAX_RETRIES + 1): + if attempt > 1: + delay = RETRY_DELAYS[attempt - 2] + try: + await status.edit_text( + f"⚠️ **Stream dropped** (attempt {attempt - 1}/{MAX_RETRIES}) · {dc_note}\n\n" + f"**{file_name}**\n" + f"📦 {fmt_size(bytes_sent)} received so far\n\n" + f"💬 `{last_error}`\n\n" + f"⏳ Retrying from 0% in {delay}s... (stream must restart)" + ) + except Exception: + pass + await asyncio.sleep(delay) + # Reset counters — stream restarts from beginning + bytes_sent = 0 + stream_start = time.time() + + # Launch a fresh rclone process for each attempt + try: + rclone = await asyncio.create_subprocess_exec( + *rclone_cmd, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + except Exception as e: + logger.error(f"rclone launch failed: {e}") + await status.edit_text(f"❌ **rclone failed to start**\n\n`{e}`") + return + + last_edit = 0.0 + speed_win = collections.deque(maxlen=6) + win_start = stream_start + win_bytes = 0 + stream_err = None + + try: + async for chunk in client.stream_media(message): + rclone.stdin.write(chunk) + await rclone.stdin.drain() + bytes_sent += len(chunk) + win_bytes += len(chunk) + + now = time.time() + win_age = now - win_start + if win_age >= 3: + inst = win_bytes / win_age + speed_win.append(inst) + peak_spd = max(peak_spd, inst) + win_bytes = 0 + win_start = now + + avg = sum(speed_win) / len(speed_win) if speed_win else 0 + + if now - last_edit >= 3: + last_edit = now + elapsed = now - stream_start + pct = bytes_sent / raw_size * 100 if raw_size else 0 + eta = (raw_size - bytes_sent) / avg if avg > 0 and raw_size else 0 + bar = "█" * int(pct / 5) + "░" * (20 - int(pct / 5)) + retry_note = f" · 🔁 attempt {attempt}/{MAX_RETRIES}" if attempt > 1 else "" + try: + await status.edit_text( + f"🌊 **Streaming** · {dc_note}{retry_note}\n\n" + f"**{file_name}**\n" + f"`[{bar}]` **{pct:.1f}%**\n" + f"📦 {fmt_size(bytes_sent)} / {size_str}\n\n" + f"⚡ Speed: {spd(avg)}\n" + f"🏆 Peak: {spd(peak_spd)}\n" + f"⏱ ETA: {fmt_dur(eta)} · Elapsed: {fmt_dur(elapsed)}\n\n" + f"🧩 Chunk: {chunk_label} · 💾 Disk: **0 B**\n" + f"🖥 RAM: {fmt_ram()}" + ) + except Exception: + pass + + rclone.stdin.close() + + except Exception as e: + stream_err = e + last_error = str(e) + logger.warning(f"Stream attempt {attempt}/{MAX_RETRIES} failed: {e}") + try: rclone.stdin.close() + except Exception: pass + try: rclone.kill() + except Exception: pass + + if stream_err is None: + break # stream completed cleanly + + if attempt == MAX_RETRIES: + await status.edit_text( + f"❌ **Stream failed** after {MAX_RETRIES} attempts\n\n" + f"**{file_name}**\n" + f"📦 {fmt_size(bytes_sent)} received before final failure\n\n" + f"💬 `{last_error}`" + ) + return + + stream_elapsed = time.time() - stream_start + try: + await status.edit_text( + f"☁️ **Flushing to Drive API...**\n\n" + f"**{file_name}**\n" + f"📦 {fmt_size(bytes_sent)} streamed in {fmt_dur(stream_elapsed)}\n" + f"⚡ Avg: {spd(bytes_sent / stream_elapsed if stream_elapsed else 0)}" + f" · 🏆 Peak: {spd(peak_spd)}\n\n" + f"⏳ rclone flushing last {chunk_label} chunk..." + ) + except Exception: + pass + + out, err = await rclone.communicate() + total_elapsed = time.time() - total_start + + if rclone.returncode != 0: + error = err.decode().strip() or out.decode().strip() + logger.error(f"rclone rcat exit {rclone.returncode}: {error}") + await status.edit_text( + f"❌ **Drive upload failed**\n\n**{file_name}**\n" + f"📦 {fmt_size(bytes_sent)} streamed — rclone exited {rclone.returncode}\n\n" + f"💬 `{error}`" + ) + return + + global session_uploads + session_uploads += 1 + if username and username not in AUTHORIZED_USERS: + record_upload(username, bytes_sent) + quota_note = "" + if username and username not in AUTHORIZED_USERS: + rem = quota_remaining(username) + if rem is not None: + quota_note = f"\n📊 Quota left today: {fmt_size(rem)}" + avg_final = bytes_sent / stream_elapsed if stream_elapsed > 0 else 0 + await status.edit_text( + f"✅ **Done!**\n\n" + f"**{file_name}**\n" + f"📦 {fmt_size(bytes_sent)} · 🗂 `{mime}`\n\n" + f"📊 **Stats**\n" + f"⚡ Avg speed: {spd(avg_final)}\n" + f"🏆 Peak speed: {spd(peak_spd)}\n" + f"⏱ Stream: {fmt_dur(stream_elapsed)}\n" + f"🕐 Total: {fmt_dur(total_elapsed)}\n\n" + f"🧩 Chunk: {chunk_label} · 📡 {dc_note} · 💾 Disk: **0 B**\n" + f"📁 → **Google Drive / {folder}**{quota_note}" + ) + + await send_links(message, file_name, folder=folder, owner=username) + +# ── yt-dlp auto-updater ─────────────────────────────────────────────────────── +# Extractors break whenever sites change; a daily upgrade keeps them fresh. +# Downloads spawn `python -m yt_dlp` subprocesses, so a new version is picked +# up immediately — no bot restart needed. + +YTDLP_UPDATE_INTERVAL = 24 * 3600 + +async def _ytdlp_version() -> str: + proc = await asyncio.create_subprocess_exec( + sys.executable, "-m", "yt_dlp", "--version", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + ) + out, _ = await proc.communicate() + return out.decode().strip() + +async def ytdlp_update_loop(): + while True: + try: + before = await _ytdlp_version() + proc = await asyncio.create_subprocess_exec( + sys.executable, "-m", "pip", "install", "-q", "--upgrade", "yt-dlp", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + ) + out, _ = await proc.communicate() + if proc.returncode != 0: + logger.warning(f"yt-dlp update failed (pip exit {proc.returncode}): " + f"{out.decode()[-300:]}") + else: + after = await _ytdlp_version() + if after != before: + logger.info(f"yt-dlp auto-updated: {before} → {after}") + else: + logger.info(f"yt-dlp up to date ({before})") + except Exception as e: + logger.warning(f"yt-dlp auto-update error: {e}") + await asyncio.sleep(YTDLP_UPDATE_INTERVAL) + +# ── Command menu (the "/" pop-up pane) ──────────────────────────────────────── +# Registered with Telegram on startup so typing "/" shows the list + hints. + +BOT_COMMANDS = [ + ("start", "👋 How to use the bot + system info"), + ("help", "📖 Full command reference"), + ("get", "📥 Send a Drive file back into this chat"), + ("search", "🔎 Find files in Drive by name"), + ("list", "🗂 Show the 10 most recent files"), + ("listall", "📚 List every file (paginated)"), + ("share", "🌍 Make a file public + get a link"), + ("stats", "ℹ️ Details for one Drive file"), + ("delete", "🗑 Permanently delete a Drive file"), + ("folders", "📁 Drive folders & where types go"), + ("df", "💽 Google Drive storage usage"), + ("status", "📊 Bot uptime, RAM, CPU, disk"), + ("logs", "📜 Tail the last 50 log lines"), + ("myquota", "📦 Check your own upload quota"), + ("quotas", "📈 All quotas & today's usage (admin)"), + ("setquota", "⚙️ Set a user's daily quota (admin)"), + ("clearquota", "♻️ Remove a user's quota (admin)"), + ("adduser", "➕ Grant a user full access (admin)"), + ("removeuser", "➖ Revoke a user's access (admin)"), + ("restart", "🔄 Gracefully restart the bot (admin)"), +] + +# How-to that doubles as the "/" pane intro — shown by /start and /help. +USAGE_TUTORIAL = ( + "**What I do:** send me anything and I save it to Google Drive.\n\n" + "📁 **Files** — forward or upload any file (caption = rename)\n" + "🌐 **Links** — paste a URL (or several); direct files stream straight to Drive\n" + "🎬 **Media sites** — YouTube, Instagram, TikTok, X, Reddit, Pinterest, " + "Tumblr, Facebook, Aparat, RedGifs & more → handled by yt-dlp\n" + "🎵 Add `mp3` after a media link to grab audio only\n" + "🎧 **Spotify** links → downloaded as MP3\n" + "🧲 **Torrents** — send a magnet link or a `.torrent` file\n" + "📥 **Get back** — `/get ` or the 📥 button pulls a file from Drive to Telegram\n\n" + "Type `/` to see every command." +) + +async def register_commands(): + try: + await app.set_bot_commands([BotCommand(c, d) for c, d in BOT_COMMANDS]) + logger.info(f"Registered {len(BOT_COMMANDS)} bot commands with Telegram") + except Exception as e: + logger.warning(f"set_bot_commands failed: {e}") + +# ── Entry ───────────────────────────────────────────────────────────────────── + +if __name__ == "__main__": + if not API_ID or not API_HASH or not BOT_TOKEN: + logger.error("API_ID, API_HASH, BOT_TOKEN must be set in .env") + raise SystemExit(1) + + # Rotate log on startup: keep last 1000 lines if it exceeds 1 MB + try: + if LOG_FILE.stat().st_size > 1024 * 1024: + lines = LOG_FILE.read_text(errors="replace").splitlines() + LOG_FILE.write_text("\n".join(lines[-1000:])) + logging.info("Log trimmed to last 1000 lines (exceeded 1 MB)") + except Exception: + pass + + logger.info("Bot starting — manual < 2 GB · stream ≥ 2 GB · URL download · yt-dlp media → MEDIA-DLS · caption rename") + + async def _main(): + await app.start() + await register_commands() + asyncio.create_task(ytdlp_update_loop()) + logger.info("yt-dlp auto-updater scheduled (every 24 h)") + await idle() + await app.stop() + + app.run(_main())