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": <bytes>}, "usage": {"username": {"YYYY-MM-DD": <bytes>}} }

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/<category>/ (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 <size>`\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 <size>` 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 <filename or keyword>`\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 <filename or keyword>`\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 <filename or keyword>`")

    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 <filename or keyword>`\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 <filename>`\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>`")

    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>`")

    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 <query>` — Search files by name or keyword\n"
        "`/share <filename>` — Make a file public and get a shareable link\n"
        "`/get <filename>` — Pull a file **from** Drive back into this chat\n"
        "`/stats <filename>` — Show detailed Drive file info (size, date, MIME, path)\n"
        "`/delete <filename>` — 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 <size>` — 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<pct>[\d.]+)%\s+of\s+~?\s*(?P<total>[\d.]+\s*\S+?B)'
    r'(?:\s+at\s+(?P<speed>\S+))?(?:\s+ETA\s+(?P<eta>\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/<category>/. 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/<category>, 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<done>[\d.]+\w*)/(?P<total>[\d.]+\w*)\((?P<pct>\d+)%\)'
    r'(?:\s+CN:(?P<cn>\d+))?(?:\s+SD:(?P<sd>\d+))?'
    r'(?:\s+DL:(?P<dl>\S+?))?(?:\s+ETA:(?P<eta>\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/<reldir>/. 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 <name>` 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())