diff --git a/bot.py b/bot.py new file mode 100644 index 0000000..8db8634 --- /dev/null +++ b/bot.py @@ -0,0 +1,1367 @@ +import os +import re +import sys +import json +import asyncio +import logging +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 +from pyrogram.types import Message + +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) + +# ── 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_file_id(file_name: str) -> str | None: + try: + 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() + files = json.loads(out.decode()) + for f in files: + if f["Name"] == file_name: + return f["ID"] + return None + except Exception as e: + logger.warning(f"lsjson: {e}") + return None + +async def send_links(message: Message, file_name: str): + file_id = await get_drive_file_id(file_name) + if file_id: + 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**\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, + ) + +# ── 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\n" + f"🖥 RAM: {fmt_size(vm.available)} free · 📡 MTProto + rclone\n\n" + "━━━━━━━━━━━━━━━━━━━━\n\n" + "**Send a file:**\n" + "📄 Documents 🖼 Photos 🎬 Videos\n" + "🎵 Audio 🎤 Voice 📹 Video notes 😀 Stickers\n\n" + "**Send a URL:**\n" + "🌐 Paste any http:// or https:// link — I'll download and upload it for you.\n\n" + "**✏️ Rename before upload:**\n" + "Add a caption to any file to rename it on Drive.\n" + "_Example: send a video with caption_ `lecture_01.mp4`\n\n" + "**🔗 Direct download links:**\n" + "After every upload I send a wget-ready link so you can pull the file anywhere.\n\n" + "No file size limit — uses MTProto for large Telegram downloads.\n\n" + "━━━━━━━━━━━━━━━━━━━━\n" + "Made by @Coaxys" + ) + +# ── /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" + ) + +# ── /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}`") + +# ── /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\n" + "📊 **Info**\n" + "`/status` — Bot uptime, RAM, CPU, disk usage\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 · ✏️ Add caption to rename\n\n" + "Made by @Coaxys" + + auth_note + ) + +# ── URL handler ─────────────────────────────────────────────────────────────── + +@app.on_message(filters.text & ~filters.command( + ["start", "help", "list", "listall", "status", "search", "share", + "adduser", "removeuser", "logs", "restart", + "setquota", "clearquota", "quotas", "myquota"] +)) +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 + match = URL_RE.search(message.text or "") + if not match: + return + + url = match.group(0).rstrip(".,;)") + file_name = filename_from_url(url) + chunk = "512M" # URL downloads: always use max chunks + total_start = time.time() + + status = await message.reply_text( + f"🌐 **URL received**\n\n" + f"`{url}`\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", 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 + + # 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"🌐 **Downloading & uploading...**\n\n" + f"`{url}`\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"`{url}`\n\n" + f"💬 `{error}`" + ) + return + + global session_uploads + session_uploads += 1 + await status.edit_text( + f"✅ **Done!**\n\n" + f"📎 **{file_name}**\n" + f"🌐 Source: `{url[:60]}{'...' if len(url)>60 else ''}`\n\n" + f"🕐 Total: {fmt_dur(total_elapsed)} · 💾 Disk: **0 B**\n" + f"📁 → **Google Drive / {folder}**" + ) + + await send_links(message, file_name) + +# ── 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 + + 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) + +# ── 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) + +# ── 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) + logger.info("Bot starting — manual < 2 GB · stream ≥ 2 GB · URL download · caption rename") + app.run()