Update bot/handlers.py

bot/handlers.py

@@ -1,37 +1,43 @@
 # PATH: bot/handlers.py
+from __future__ import annotations
+
 import asyncio
 import json
 import os
-import re
 import time
-from typing import Dict
+from typing import Dict, Optional, Any
 
 from hydrogram import Client, filters
 from hydrogram.types import Message, CallbackQuery
 
-from bot.config import Workers
+from bot.config import Telegram, Workers
 from bot.ui import texts
-from bot.ui.keyboards import profiles_keyboard, auth_menu_keyboard, main_menu_keyboard
+from bot.ui.keyboards import (
+    main_menu_keyboard,
+    auth_menu_keyboard,
+    profiles_keyboard,
+    filename_keyboard,
+    upload_confirm_keyboard,
+)
 from bot.ui.callbacks import (
     parse_cb,
-    AUTH_JSON, AUTH_CI, CANCEL,
-    MENU_HELP, MENU_AUTH, MENU_PROFILES, MENU_SPEEDTEST
+    AUTH_JSON, AUTH_CI, CANCEL, BACK,
+    MENU_HELP, MENU_AUTH, MENU_PROFILES, MENU_SPEEDTEST,
+    NAME_ORIGINAL, NAME_CAPTION, NAME_CUSTOM,
+    UP_GO, UP_EDIT, UP_PRIV, UP_CANCEL,
 )
 
-from bot.core.auth import is_owner_id, require_allowed
+from bot.core.auth import is_owner_id, is_admin_id, require_allowed
+from bot.core.link_parser import parse_link
 from bot.core.progress import SpeedETA, human_bytes, human_eta
-from bot.core.tasks import create_task, set_task
-
 from bot.core.speedtest import (
     ping_ms,
     net_download_test,
     net_upload_test,
     disk_total_free,
-    bytes_to_mb,
     bytes_per_sec_to_mb_s,
 )
-
-from bot.integrations.cf_worker1 import profile_add, profile_list, profile_set_default
+from bot.integrations.cf_worker1 import profile_add
 from bot.integrations.cf_worker2 import (
     allow_user,
     disallow_user,
@@ -39,24 +45,51 @@ from bot.integrations.cf_worker2 import (
     access_token,
     record_upload,
     stats_today,
+    profile_list,
+    profile_set_default,
 )
+from bot.integrations.diag_extra import dns_check
+from bot.integrations.http import fetch_status
 
 from bot.telegram.media import download_to_temp
 from bot.telegram.parse import extract_title_description
 from bot.telegram.replies import safe_reply, safe_edit
 from bot.temp.files import cleanup_file
-
 from bot.youtube.uploader import upload_video
 
-from bot.integrations.diag_extra import dns_check
-from bot.integrations.http import fetch_status
-
-_AWAIT_AUTH: Dict[int, str] = {}
 
+# -----------------------
+# GLOBAL STATE (in-memory)
+# -----------------------
+_AWAIT_AUTH: Dict[int, str] = {}         # {uid: "json"|"ci"}
+_AWAIT_CUSTOM: Dict[int, bool] = {}      # {uid: True}
+_AWAIT_EDIT: Dict[int, bool] = {}        # {uid: True}
+_SPEED_COOLDOWN: Dict[int, float] = {}   # {uid: last_ts}
+
+# pending job state per-user:
+# {
+#   "mode": "direct"|"link",
+#   "src_msg": Message,
+#   "status_msg": Message,
+#   "downloader": Client,
+#   "original_name": str,
+#   "caption_name": str,
+#   "filename": str (chosen),
+#   "title": str,
+#   "description": str,
+#   "privacy": str,
+#   "link": str,
+# }
+_PENDING: Dict[int, Dict[str, Any]] = {}
+
+# batch gate (one batch at a time)
+BATCH_SEM = asyncio.Semaphore(1)
 _STARTED_AT = time.time()
-_SPEED_COOLDOWN: dict[int, float] = {}
 
 
+# -----------------------
+# HELPERS
+# -----------------------
 def _pick_login_url(j: dict) -> str:
     if not isinstance(j, dict):
         return ""
@@ -81,11 +114,9 @@ def _pick_login_url(j: dict) -> str:
 async def _render_speedtest(mb: int) -> str:
     mb = max(1, min(200, int(mb)))
     dl_bytes = mb * 1024 * 1024
-    up_mb = min(20, max(3, mb // 5))  # 3..20MB
+    up_mb = min(20, max(3, mb // 5))
     up_bytes = up_mb * 1024 * 1024
-
     up_sec = int(max(0, time.time() - _STARTED_AT))
-    up_txt = human_eta(up_sec)
 
     d = disk_total_free("/")
     total_gb = d["total"] / (1024.0**3)
@@ -98,27 +129,194 @@ async def _render_speedtest(mb: int) -> str:
     dl, up, p = await asyncio.gather(dl_task, up_task, p_task)
 
     ping_line = f"{p:.0f} ms" if p is not None else "N/A"
-    dl_mb_s = bytes_per_sec_to_mb_s(dl["bps"])
-    up_mb_s = bytes_per_sec_to_mb_s(up["bps"])
-
     return (
         "⚡ **Server Speed Test**\n\n"
-        "🕒 Uptime\n"
-        f"└ {up_txt}\n\n"
+        f"🕒 Uptime: {human_eta(up_sec)}\n\n"
         "📡 Network\n"
         f"├ 🟢 Ping: {ping_line}\n"
-        f"├ ⬇️ Download: {dl_mb_s:.2f} MB/s\n"
-        f"│   └ {bytes_to_mb(dl['bytes']):.2f} MB in {dl['seconds']:.2f}s\n"
-        f"└ ⬆️ Upload: {up_mb_s:.2f} MB/s\n"
-        f"    └ {bytes_to_mb(up['bytes']):.2f} MB in {up['seconds']:.2f}s\n\n"
+        f"├ ⬇️ DL: {bytes_per_sec_to_mb_s(dl['bps']):.2f} MB/s\n"
+        f"└ ⬆️ UL: {bytes_per_sec_to_mb_s(up['bps']):.2f} MB/s\n\n"
         "💾 Storage\n"
         f"├ Total: {total_gb:.1f} GB\n"
-        f"└ Free : {free_gb:.1f} GB\n\n"
-        "🧪 Measured from server (real MB/s, not Mbps)"
+        f"└ Free : {free_gb:.1f} GB"
     )
 
 
-def setup_handlers(app: Client) -> None:
+def _sanitize_filename(name: str, default_ext: str = ".mp4") -> str:
+    name = (name or "").strip()
+    if not name:
+        name = "video"
+    # remove path-ish
+    name = name.replace("\\", "_").replace("/", "_").replace("..", "_")
+    # force extension if missing
+    if "." not in name:
+        name += default_ext
+    return name
+
+
+def _toggle_privacy(p: str) -> str:
+    p = (p or "private").lower().strip()
+    if p == "private":
+        return "unlisted"
+    if p == "unlisted":
+        return "public"
+    return "private"
+
+
+def _require_admin_for_links(uid: int) -> bool:
+    # admin/owner-only for link/range uploads
+    return is_admin_id(uid)
+
+
+async def _ensure_profile_and_token(uid: int):
+    """
+    Returns (profile_id, access_token_string)
+    """
+    pl = await profile_list(uid)
+    if not pl.get("ok"):
+        raise RuntimeError(texts.NEED_AUTH)
+
+    default_id = pl.get("default_profile_id")
+    profiles = pl.get("profiles", []) or []
+    dp = next((p for p in profiles if p.get("profile_id") == default_id), None)
+    if not dp:
+        raise RuntimeError("❌ No active profile. Use /profiles and set default.")
+
+    pick = await pick_profile(uid, dp["channel_id"])
+    if not pick or not pick.get("profile_id"):
+        raise RuntimeError(texts.PICK_FAIL.format(pick))
+
+    tok = await access_token(uid, pick["profile_id"])
+    if not tok.get("ok") or not tok.get("access_token"):
+        raise RuntimeError(texts.TOKEN_FAIL.format(tok))
+
+    return pick["profile_id"], tok["access_token"]
+
+
+def _make_throttled_progress_editor(
+    status_msg: Message, prefix: str, min_interval_s: float = 2.5
+):
+    last_ts = 0.0
+
+    async def cb(done: int, total: int):
+        nonlocal last_ts
+        now = time.time()
+        if now - last_ts < min_interval_s:
+            return
+        last_ts = now
+        if total > 0:
+            pct = done * 100.0 / total
+            await safe_edit(
+                status_msg,
+                f"{prefix}\n{human_bytes(done)}/{human_bytes(total)} ({pct:.1f}%)",
+            )
+        else:
+            await safe_edit(status_msg, f"{prefix}\n{human_bytes(done)}")
+
+    return cb
+
+
+async def _preview_and_confirm(uid: int, status: Message):
+    """
+    Shows preview UI for the current pending item, using upload_confirm_keyboard.
+    """
+    d = _PENDING.get(uid)
+    if not d:
+        return
+    title = d.get("title") or "Untitled"
+    desc = d.get("description") or ""
+    privacy = d.get("privacy") or "private"
+    filename = d.get("filename") or d.get("original_name") or "video.mp4"
+
+    # keep preview short
+    desc_short = desc.strip()
+    if len(desc_short) > 400:
+        desc_short = desc_short[:400] + "…"
+
+    src = d.get("link") or "Direct upload"
+    await safe_edit(
+        status,
+        "📦 **Ready to Upload**\n\n"
+        f"🗂 **Source:** `{src}`\n"
+        f"📄 **File:** `{filename}`\n"
+        f"🏷 **Title:** `{title}`\n"
+        f"🔒 **Privacy:** `{privacy}`\n\n"
+        f"📝 **Description (preview):**\n{desc_short or '_empty_'}\n\n"
+        "👇 Choose action:",
+        reply_markup=upload_confirm_keyboard(privacy),
+    )
+
+
+async def _do_upload(uid: int):
+    """
+    Performs: download (if needed) -> upload to YouTube -> cleanup -> final message.
+    """
+    d = _PENDING.get(uid)
+    if not d:
+        return
+
+    status: Message = d["status_msg"]
+    msg: Message = d["src_msg"]
+    downloader: Client = d["downloader"]
+    filename: str = d.get("filename") or d.get("original_name") or "video.mp4"
+    filename = _sanitize_filename(filename)
+
+    # auth
+    profile_id, access = await _ensure_profile_and_token(uid)
+
+    # download progress
+    await safe_edit(status, f"⬇️ **Downloading:** `{filename}` …")
+    dl_cb = _make_throttled_progress_editor(status, "⬇️ **Downloading…**")
+
+    path = ""
+    try:
+        path, _, _ = await download_to_temp(downloader, msg, progress_cb=dl_cb)
+        if not path:
+            raise RuntimeError("❌ Download failed.")
+
+        # upload progress
+        await safe_edit(status, "⬆️ **Uploading to YouTube…**")
+        se = SpeedETA()
+        last_upd = 0.0
+
+        async def up_cb(done: int, total: int):
+            nonlocal last_upd
+            now = time.time()
+            if now - last_upd < 3.0:
+                return
+            last_upd = now
+            snap = se.update(done, total)
+            await safe_edit(
+                status,
+                "⬆️ **Uploading…**\n"
+                f"{human_bytes(done)}/{human_bytes(total)}\n"
+                f"⚡ {human_bytes(snap['speed_bps'])}/s",
+            )
+
+        yt_url = await upload_video(
+            access_token=access,
+            file_path=path,
+            title=d.get("title") or filename.rsplit(".", 1)[0],
+            description=d.get("description") or "",
+            privacy=(d.get("privacy") or "private"),
+            progress_cb=up_cb,
+        )
+
+        await record_upload(uid, profile_id)
+        await safe_edit(status, f"✅ **Uploaded!**\n\n📺 `{filename}`\n🔗 {yt_url}")
+
+    finally:
+        if path:
+            cleanup_file(path)
+
+
+# -----------------------
+# HANDLERS SETUP
+# -----------------------
+def setup_handlers(app: Client, user_app: Client = None):
+    # -------------------
+    # STANDARD COMMANDS
+    # -------------------
     @app.on_message(filters.command(["start"]) & filters.private)
     async def start_handler(_: Client, m: Message):
         await safe_reply(m, texts.START_TEXT, reply_markup=main_menu_keyboard())
@@ -133,362 +331,477 @@ def setup_handlers(app: Client) -> None:
 
     @app.on_message(filters.command(["me"]) & filters.private)
     async def me_handler(_: Client, m: Message):
-        uid = m.from_user.id if m.from_user else None
-        await safe_reply(m, f"👤 Your ID: `{uid}`")
+        await safe_reply(m, f"👤 Your ID: `{m.from_user.id}`")
+
+    @app.on_message(filters.command(["cancel"]) & filters.private)
+    async def cancel_cmd(_: Client, m: Message):
+        uid = m.from_user.id
+        _AWAIT_AUTH.pop(uid, None)
+        _AWAIT_CUSTOM.pop(uid, None)
+        _AWAIT_EDIT.pop(uid, None)
+        _PENDING.pop(uid, None)
+        await safe_reply(m, texts.CANCELLED, reply_markup=main_menu_keyboard())
 
-    # -------------------- SPEEDTEST (any user) --------------------
+    # -------------------
+    # SPEEDTEST
+    # -------------------
     @app.on_message(filters.command(["speedtest"]) & filters.private)
     async def speedtest_cmd(_: Client, m: Message):
-        uid = m.from_user.id if m.from_user else 0
-
+        uid = m.from_user.id
         now = time.time()
-        last = _SPEED_COOLDOWN.get(uid, 0.0)
-        if now - last < 30:
-            return await safe_reply(m, "⏳ Wait 30s then try again.")
+        if now - _SPEED_COOLDOWN.get(uid, 0.0) < 30:
+            return await safe_reply(m, "⏳ Wait 30s...")
         _SPEED_COOLDOWN[uid] = now
 
-        parts = (m.text or "").split()
-        try:
-            mb = int(parts[1]) if len(parts) > 1 else 8
-        except Exception:
-            mb = 8
-
         msg = await safe_reply(m, "⚡ Running speedtest…")
         try:
-            txt = await _render_speedtest(mb)
+            txt = await _render_speedtest(8)
             await safe_edit(msg, txt)
         except Exception as e:
-            await safe_edit(msg, f"❌ Speedtest failed: `{type(e).__name__}: {e}`")
-
-    # -------- OWNER allow/disallow --------
-    @app.on_message(filters.command(["allow"]) & filters.private)
-    async def allow_cmd(_: Client, m: Message):
-        if not is_owner_id(m.from_user.id if m.from_user else None):
-            return await safe_reply(m, texts.OWNER_ONLY)
-
-        target = None
-        if m.reply_to_message and m.reply_to_message.from_user:
-            target = m.reply_to_message.from_user.id
-        elif m.forward_from:
-            target = m.forward_from.id
-        else:
-            parts = (m.text or "").split()
-            if len(parts) >= 2 and parts[1].isdigit():
-                target = int(parts[1])
-
-        if not target:
-            return await safe_reply(m, "Usage:\n• Reply/forward user msg then /allow\n• or /allow <user_id>")
-
-        j = await allow_user(target)
-        await safe_reply(m, f"✅ allowed: `{target}`\n`{j}`")
-
-    @app.on_message(filters.command(["disallow"]) & filters.private)
-    async def disallow_cmd(_: Client, m: Message):
-        if not is_owner_id(m.from_user.id if m.from_user else None):
+            await safe_edit(msg, f"❌ Failed: {e}")
+
+    # -------------------
+    # OWNER COMMANDS
+    # -------------------
+    @app.on_message(filters.command(["allow", "disallow", "stats", "diag"]) & filters.private)
+    async def owner_cmds(_: Client, m: Message):
+        if not is_owner_id(m.from_user.id):
             return await safe_reply(m, texts.OWNER_ONLY)
 
-        parts = (m.text or "").split()
-        if len(parts) < 2 or not parts[1].isdigit():
-            return await safe_reply(m, "Usage: /disallow <user_id>")
-        target = int(parts[1])
-        j = await disallow_user(target)
-        await safe_reply(m, f"✅ disallowed: `{target}`\n`{j}`")
+        cmd = m.command[0]
+        if cmd == "stats":
+            return await safe_reply(m, f"📊 Stats:\n`{await stats_today()}`")
 
-    @app.on_message(filters.command(["stats"]) & filters.private)
-    async def stats_cmd(_: Client, m: Message):
-        if not is_owner_id(m.from_user.id if m.from_user else None):
-            return await safe_reply(m, texts.OWNER_ONLY)
-        j = await stats_today()
-        await safe_reply(m, f"📊 Stats:\n`{j}`")
+        if cmd == "diag":
+            r1 = await fetch_status(Workers.WORKER1_URL)
+            return await safe_reply(m, f"🔎 Diag:\nW1: {r1}\nDNS: {dns_check(Workers.WORKER1_URL)}")
 
-    @app.on_message(filters.command(["diag"]) & filters.private)
-    async def diag_cmd(_: Client, m: Message):
-        if not is_owner_id(m.from_user.id if m.from_user else None):
-            return await safe_reply(m, texts.OWNER_ONLY)
+        # allow/disallow
+        if len(m.command) < 2:
+            return await safe_reply(m, f"Usage: /{cmd} <id>")
+        target = int(m.command[1])
 
-        r1 = await fetch_status((Workers.WORKER1_URL or "") + "/")
-        r2 = await fetch_status((Workers.WORKER2_URL or "") + "/")
-
-        await safe_reply(
-            m,
-            "🔎 DIAG\n"
-            f"W1={Workers.WORKER1_URL} -> {r1}\n"
-            f"W2={Workers.WORKER2_URL} -> {r2}\n\n"
-            f"DNS W1 -> {dns_check(Workers.WORKER1_URL)}\n"
-            f"DNS W2 -> {dns_check(Workers.WORKER2_URL)}\n"
-            f"DNS google -> {dns_check('https://www.google.com')}\n",
-        )
+        if cmd == "allow":
+            res = await allow_user(target)
+            await safe_reply(m, f"✅ Allowed {target}: {res}")
+        else:
+            res = await disallow_user(target)
+            await safe_reply(m, f"🚫 Disallowed {target}: {res}")
 
-    # -------- AUTH --------
+    # -------------------
+    # AUTH + PROFILES
+    # -------------------
     @app.on_message(filters.command(["auth"]) & filters.private)
     async def auth_cmd(_: Client, m: Message):
-        uid = m.from_user.id if m.from_user else 0
+        uid = m.from_user.id
         if not await require_allowed(uid):
             return await safe_reply(m, texts.NOT_ALLOWED)
-
         _AWAIT_AUTH.pop(uid, None)
-        await safe_reply(m, "🔐 Add YouTube Profile\n\nChoose method:", reply_markup=auth_menu_keyboard())
+        await safe_reply(m, "🔐 Add Profile:", reply_markup=auth_menu_keyboard())
 
-    @app.on_message(filters.command(["cancel"]) & filters.private)
-    async def cancel_cmd(_: Client, m: Message):
-        _AWAIT_AUTH.pop(m.from_user.id, None)
-        await safe_reply(m, texts.CANCELLED, reply_markup=main_menu_keyboard())
+    @app.on_message(filters.command(["profiles"]) & filters.private)
+    async def profiles_cmd(_: Client, m: Message):
+        uid = m.from_user.id
+        if not await require_allowed(uid):
+            return await safe_reply(m, texts.NOT_ALLOWED)
 
-    @app.on_callback_query()
-    async def cb(_: Client, q: CallbackQuery):
-        uid = q.from_user.id if q.from_user else 0
-        action, value = parse_cb(q.data or "")
+        j = await profile_list(uid)
+        if not j.get("ok"):
+            return await safe_reply(m, texts.PROFILE_LIST_FAIL.format(j))
 
-        # main menu
-        if action == MENU_HELP:
-            try:
-                await q.message.edit_text(texts.HELP_TEXT, reply_markup=main_menu_keyboard())
-            except Exception:
-                pass
-            return await q.answer("Help")
+        profiles = j.get("profiles") or []
+        kb = profiles_keyboard(profiles)
+        await safe_reply(m, f"👤 **Profiles ({len(profiles)})**", reply_markup=kb)
+
+    # -------------------
+    # LINK ARCHIVE (/archive, /yt, /dl)  admin/owner only
+    # -------------------
+    @app.on_message(filters.command(["archive", "yt", "dl"]) & filters.private)
+    async def archive_link_handler(c: Client, m: Message):
+        uid = m.from_user.id
+        if not await require_allowed(uid):
+            return await safe_reply(m, texts.NOT_ALLOWED)
 
-        if action == MENU_SPEEDTEST:
-            await q.answer("Use /speedtest", show_alert=False)
-            return
+        if not _require_admin_for_links(uid):
+            return await safe_reply(m, "❌ Link-archive is admin/owner only.")
 
-        if action == MENU_AUTH:
-            if not await require_allowed(uid):
-                return await q.answer("Not allowed", show_alert=True)
-            _AWAIT_AUTH.pop(uid, None)
-            try:
-                await q.message.edit_text("🔐 Add YouTube Profile\n\nChoose method:", reply_markup=auth_menu_keyboard())
-            except Exception:
-                pass
-            return await q.answer("Auth")
+        if not user_app:
+            return await safe_reply(m, "❌ `USER_SESSION_STRING` missing. Link-archive disabled.")
 
-        if action == MENU_PROFILES:
-            await q.answer("Use /profiles", show_alert=False)
-            return
+        if len(m.command) < 2:
+            return await safe_reply(m, "Usage: `/archive <t.me message link>`")
 
-        # auth menu
-        if action in (AUTH_JSON, AUTH_CI, CANCEL):
-            if action == CANCEL:
-                _AWAIT_AUTH.pop(uid, None)
-                return await q.answer("Cancelled")
-            if action == AUTH_JSON:
-                _AWAIT_AUTH[uid] = "json"
-                await q.message.edit_text("📄 Paste your client JSON here.\n\n(You can /cancel anytime)")
-                return await q.answer("Paste JSON")
-            if action == AUTH_CI:
-                _AWAIT_AUTH[uid] = "ci"
-                await q.message.edit_text("🔑 Send Client ID and Client Secret.\n\nFormat:\n<id> | <secret>\n(or 2 lines)\n\n(/cancel anytime)")
-                return await q.answer("Send ID & Secret")
-            return
+        link = m.command[1]
+        ids = parse_link(link)
+        if not ids:
+            return await safe_reply(m, "❌ Invalid link.")
+        chat_id, msg_id = ids
 
-        if action == "setdef" and value:
-            if not await require_allowed(uid):
-                return await q.answer("Not allowed", show_alert=True)
+        status = await safe_reply(m, "🔍 **Fetching message…**")
 
-            j = await profile_set_default(uid, value)
-            if not j.get("ok"):
-                return await q.answer("Failed", show_alert=True)
+        try:
+            target_msg = await user_app.get_messages(chat_id, msg_id)
+            if not target_msg or target_msg.empty:
+                return await safe_edit(status, "❌ Message not found or inaccessible.")
+
+            media = target_msg.video or target_msg.document
+            if not media:
+                return await safe_edit(status, "❌ No video/document found in that message.")
+
+            original_name = getattr(media, "file_name", None) or "video.mp4"
+            file_size = int(getattr(media, "file_size", 0) or 0)
+            caption_base = (target_msg.caption or "video").replace("\n", " ").strip()[:60] or "video"
+            caption_name = f"{caption_base}.mp4"
+
+            _PENDING[uid] = {
+                "mode": "link",
+                "src_msg": target_msg,
+                "status_msg": status,
+                "downloader": user_app,
+                "original_name": original_name,
+                "caption_name": caption_name,
+                "privacy": "private",
+                "link": link,
+            }
+
+            size_mb = f"{file_size / 1024 / 1024:.2f} MB"
+            await safe_edit(
+                status,
+                "📂 **File Found!**\n"
+                f"📄 **Original:** `{original_name}`\n"
+                f"📦 **Size:** `{size_mb}`\n\n"
+                "👇 **Select filename:**",
+                reply_markup=filename_keyboard(),
+            )
 
-            await q.answer("Current set ✅", show_alert=False)
-            try:
-                await q.message.edit_text(f"🟢 Current profile set:\n`{value}`", reply_markup=main_menu_keyboard())
-            except Exception:
-                pass
-            return
+        except Exception as e:
+            await safe_edit(status, f"❌ Error: {e}")
+
+    # -------------------
+    # BATCH (admin/owner only)
+    # -------------------
+    @app.on_message(filters.command(["batch"]) & filters.private)
+    async def batch_handler(c: Client, m: Message):
+        uid = m.from_user.id
+        if not await require_allowed(uid):
+            return await safe_reply(m, texts.NOT_ALLOWED)
 
-        await q.answer("OK")
+        if not _require_admin_for_links(uid):
+            return await safe_reply(m, "❌ Batch is admin/owner only.")
 
-    # auth sink
-    @app.on_message(
-        filters.private
-        & filters.text
-        & ~filters.command(["start", "help", "ping", "me", "auth", "cancel", "profiles", "speedtest", "stats", "allow", "disallow", "diag"])
-    )
-    async def auth_sink(_: Client, m: Message):
-        uid = m.from_user.id if m.from_user else 0
-        mode = _AWAIT_AUTH.get(uid)
-        if not mode:
-            return
+        if not user_app:
+            return await safe_reply(m, "❌ `USER_SESSION_STRING` missing. Batch disabled.")
 
-        try:
-            if mode == "json":
-                data = json.loads(m.text)
-                root = data.get("installed") or data.get("web") or {}
-                client_id = root.get("client_id") or data.get("client_id")
-                client_secret = root.get("client_secret") or data.get("client_secret")
-                if not client_id or not client_secret:
-                    return await safe_reply(m, texts.PARSE_FAIL)
-            else:
-                txt = m.text.strip()
-                parts = [p.strip() for p in re.split(r"[|\n]+", txt) if p.strip()]
-                if len(parts) < 2:
-                    return await safe_reply(m, texts.PARSE_FAIL)
-                client_id, client_secret = parts[0], parts[1]
-
-            j = await profile_add(uid, client_id, client_secret, label="main", ttl_sec=600)
-            if not isinstance(j, dict):
-                return await safe_reply(m, f"❌ profile_add bad response: `{j}`")
-
-            if not j.get("ok") and not (isinstance(j.get("data"), dict) and j["data"].get("ok")):
-                return await safe_reply(m, texts.PROFILE_ADD_FAIL.format(j))
-
-            login_url = _pick_login_url(j)
-            if not login_url:
-                return await safe_reply(m, f"❌ profile_add OK but login link missing.\nBackend response:\n`{j}`")
+        if len(m.command) < 3:
+            return await safe_reply(m, "Usage: `/batch <start_link> <end_link>`")
 
-            _AWAIT_AUTH.pop(uid, None)
-            await safe_reply(m, texts.SENT_AUTH_LINK + login_url, reply_markup=main_menu_keyboard())
+        s_ids = parse_link(m.command[1])
+        e_ids = parse_link(m.command[2])
+        if not s_ids or not e_ids:
+            return await safe_reply(m, "❌ Invalid start/end link.")
 
-        except Exception as e:
-            await safe_reply(m, f"{texts.PARSE_FAIL}\n`{type(e).__name__}: {e}`")
+        chat_id1, start_id = s_ids
+        chat_id2, end_id = e_ids
+        if chat_id1 != chat_id2:
+            return await safe_reply(m, "❌ Start/end must be from same chat.")
 
-    @app.on_message(filters.command(["profiles"]) & filters.private)
-    async def profiles_cmd(_: Client, m: Message):
-        uid = m.from_user.id if m.from_user else 0
-        if not await require_allowed(uid):
-            return await safe_reply(m, texts.NOT_ALLOWED)
+        if start_id > end_id:
+            start_id, end_id = end_id, start_id
 
-        j = await profile_list(uid)
-        if not j.get("ok"):
-            return await safe_reply(m, texts.PROFILE_LIST_FAIL.format(j))
+        status = await safe_reply(m, f"🔄 Batching {start_id} → {end_id}…")
+        uploaded = 0
 
-        profiles = j.get("profiles") or []
-        default_id = j.get("default_profile_id")
-
-        txt = "👤 **Profiles**\n\n"
-        txt += f"🟢 Current: `{default_id}`\n\n"
-
-        for i, p in enumerate(profiles, start=1):
-            pid = p.get("profile_id")
-            ch = (p.get("channel_title") or "—").strip()
-            cid = p.get("client_id_hint") or "—"
-            sec = p.get("client_secret_hint") or "****…****"
-            connected = "✅ Connected" if (p.get("has_refresh") and p.get("channel_id")) else "⏳ Pending"
-            cur = "🟢 Current" if pid == default_id else ""
-            txt += f"[{i}] {ch}  {cur}\n"
-            txt += f"    id: {cid}\n"
-            txt += f"    secret: {sec}\n"
-            txt += f"    {connected}\n\n"
-
-        kb = profiles_keyboard(profiles)
-        await safe_reply(m, txt, reply_markup=kb)
+        async with BATCH_SEM:
+            try:
+                profile_id, access = await _ensure_profile_and_token(uid)
+            except Exception as e:
+                return await safe_edit(status, f"❌ Auth error: {e}")
 
-    # -------- Upload --------
+            for mid in range(start_id, end_id + 1):
+                try:
+                    await safe_edit(status, f"🔄 Processing message `{mid}`…")
+                    msg = await user_app.get_messages(chat_id1, mid)
+                    if not msg or msg.empty:
+                        continue
+                    if not (msg.video or msg.document):
+                        continue
+
+                    # derive basic meta
+                    t, dsc = extract_title_description(msg, default_title=f"TG {mid}")
+                    filename = getattr((msg.video or msg.document), "file_name", None) or f"{mid}.mp4"
+
+                    # download
+                    dl_status = _make_throttled_progress_editor(status, f"⬇️ Downloading `{filename}`…", 2.0)
+                    path, _, _ = await download_to_temp(user_app, msg, progress_cb=dl_status)
+
+                    # upload (no spam)
+                    await safe_edit(status, f"⬆️ Uploading `{filename}`…")
+                    await upload_video(
+                        access_token=access,
+                        file_path=path,
+                        title=t,
+                        description=dsc,
+                        privacy="private",
+                        progress_cb=None,  # keep batch quiet
+                    )
+                    cleanup_file(path)
+                    uploaded += 1
+                    await asyncio.sleep(2.0)
+                except Exception:
+                    # best-effort batch
+                    try:
+                        if "path" in locals() and path:
+                            cleanup_file(path)
+                    except Exception:
+                        pass
+                    continue
+
+            await record_upload(uid, profile_id)
+            await safe_edit(status, f"✅ Batch done. Uploaded: **{uploaded}**")
+
+    # -------------------
+    # DIRECT UPLOAD (user sends video/document in private)
+    # -------------------
     @app.on_message(filters.private & (filters.video | filters.document))
-    async def upload_handler(app_: Client, m: Message):
-        uid = m.from_user.id if m.from_user else 0
+    async def direct_upload_handler(c: Client, m: Message):
+        uid = m.from_user.id
         if not await require_allowed(uid):
-            return await safe_reply(m, texts.NOT_ALLOWED)
+            return
 
+        # prepare preview from message itself
         media = m.video or m.document
-        if not media:
-            return await safe_reply(m, "Send a video/document.")
+        fname = getattr(media, "file_name", None) or "video.mp4"
+        fname = _sanitize_filename(fname)
+
+        status = await safe_reply(m, "📦 Preparing preview…")
+        title, desc = extract_title_description(m, default_title=fname.rsplit(".", 1)[0])
+
+        _PENDING[uid] = {
+            "mode": "direct",
+            "src_msg": m,
+            "status_msg": status,
+            "downloader": c,  # bot can download private user DM
+            "original_name": fname,
+            "caption_name": fname,
+            "filename": fname,
+            "title": title,
+            "description": desc,
+            "privacy": "private",
+            "link": "Direct upload",
+        }
+
+        await _preview_and_confirm(uid, status)
+
+    # -------------------
+    # CALLBACKS (all buttons)
+    # -------------------
+    @app.on_callback_query()
+    async def global_cb(c: Client, q: CallbackQuery):
+        uid = q.from_user.id
+        action, val = parse_cb(q.data)
+
+        # ---- filename selection (link-archive) ----
+        if action in (NAME_ORIGINAL, NAME_CAPTION, NAME_CUSTOM):
+            d = _PENDING.get(uid)
+            if not d:
+                return await q.answer("❌ Expired", show_alert=True)
+
+            if action == NAME_CUSTOM:
+                _AWAIT_CUSTOM[uid] = True
+                await safe_edit(q.message, "✍️ **Send custom filename now:**\nExample: `myvideo.mp4`")
+                return await q.answer()
+
+            # choose filename
+            chosen = d["original_name"] if action == NAME_ORIGINAL else d["caption_name"]
+            chosen = _sanitize_filename(chosen)
+            d["filename"] = chosen
+
+            # build preview meta now that name chosen
+            title, desc = extract_title_description(d["src_msg"], default_title=chosen.rsplit(".", 1)[0])
+            d["title"] = title
+            d["description"] = desc
+            d["privacy"] = d.get("privacy") or "private"
+
+            await q.answer()
+            await _preview_and_confirm(uid, d["status_msg"])
+            return
 
-        pl = await profile_list(uid)
-        if not pl.get("ok"):
-            return await safe_reply(m, texts.PROFILE_LIST_FAIL.format(pl))
+        # ---- main menu callbacks ----
+        if action == MENU_AUTH:
+            if not await require_allowed(uid):
+                return await q.answer("Not allowed", show_alert=True)
+            _AWAIT_AUTH.pop(uid, None)
+            await safe_edit(q.message, "🔐 Auth Menu", reply_markup=auth_menu_keyboard())
+            return await q.answer()
 
-        default_id = pl.get("default_profile_id")
-        profiles = pl.get("profiles") or []
-        dp = next((p for p in profiles if p.get("profile_id") == default_id), None)
-        if not dp or not dp.get("has_refresh"):
-            return await safe_reply(m, texts.NEED_AUTH)
+        if action == MENU_PROFILES:
+            # reuse profiles command
+            fake = q.message
+            await q.answer()
+            return await profiles_cmd(c, fake)
 
-        channel_id = dp.get("channel_id")
-        if not channel_id:
-            return await safe_reply(m, "❌ channel_id missing. Re-auth profile.")
+        if action == MENU_SPEEDTEST:
+            await q.answer("Running…")
+            txt = await _render_speedtest(8)
+            await safe_edit(q.message, txt)
+            return
 
-        pick = await pick_profile(uid, channel_id)
-        if not pick.get("ok"):
-            return await safe_reply(m, texts.PICK_FAIL.format(pick))
-        use_profile_id = pick["profile_id"]
+        if action == MENU_HELP:
+            await q.answer()
+            await safe_edit(q.message, texts.HELP_TEXT, reply_markup=main_menu_keyboard())
+            return
 
-        tok = await access_token(uid, use_profile_id)
-        if not tok.get("ok"):
-            return await safe_reply(m, texts.TOKEN_FAIL.format(tok))
-        access_tok = tok["access_token"]
+        # ---- auth buttons ----
+        if action == AUTH_JSON:
+            _AWAIT_AUTH[uid] = "json"
+            await q.answer()
+            await safe_edit(q.message, texts.ASK_JSON)
+            return
 
-        status = await safe_reply(m, "⬇️ Downloading…")
-        task_id = f"{uid}:{m.id}"
-        create_task(task_id, uid)
+        if action == AUTH_CI:
+            _AWAIT_AUTH[uid] = "ci"
+            await q.answer()
+            await safe_edit(q.message, texts.ASK_ID_SECRET)
+            return
 
-        file_path = ""
-        file_size = 0
-        file_name = "video.bin"
+        if action == "setdef":
+            try:
+                await profile_set_default(uid, val)
+                await q.answer("Updated!")
+                await safe_edit(q.message, f"✅ Profile set as default: `{val}`", reply_markup=main_menu_keyboard())
+            except Exception as e:
+                await q.answer("Failed", show_alert=True)
+                await safe_edit(q.message, f"❌ Failed: {e}", reply_markup=main_menu_keyboard())
+            return
 
-        try:
-            # ---- Download ----
-            set_task(task_id, "downloading", "")
-            dl_t0 = time.time()
-            file_path, file_size, file_name = await download_to_temp(app_, m)
-            dl_sec = max(0.001, time.time() - dl_t0)
+        # ---- upload confirm UI ----
+        if action == UP_EDIT:
+            _AWAIT_EDIT[uid] = True
+            await q.answer()
+            await safe_edit(q.message, texts.EDIT_PROMPT)
+            return
 
-            if (not file_size) and file_path and os.path.exists(file_path):
-                try:
-                    file_size = os.path.getsize(file_path)
-                except Exception:
-                    file_size = 0
+        if action == UP_PRIV:
+            d = _PENDING.get(uid)
+            if not d:
+                return await q.answer("Expired", show_alert=True)
+            d["privacy"] = _toggle_privacy(d.get("privacy"))
+            await q.answer("Privacy updated")
+            await _preview_and_confirm(uid, d["status_msg"])
+            return
 
-            dl_avg_bps = int((file_size or 0) / dl_sec) if file_size else 0
+        if action == UP_GO:
+            await q.answer("Uploading…")
+            try:
+                await _do_upload(uid)
+            except Exception as e:
+                d = _PENDING.get(uid)
+                if d:
+                    await safe_edit(d["status_msg"], f"❌ Error: {e}")
+            finally:
+                _PENDING.pop(uid, None)
+                _AWAIT_CUSTOM.pop(uid, None)
+                _AWAIT_EDIT.pop(uid, None)
+            return
 
-            await safe_edit(
-                status,
-                "✅ Downloaded\n"
-                f"• File: `{file_name}`\n"
-                f"• Size: {human_bytes(file_size)}\n"
-                f"• Time: {human_eta(int(dl_sec))}\n"
-                f"• Avg: {human_bytes(dl_avg_bps)}/s\n\n"
-                "⬆️ Starting upload…",
-            )
+        if action in (UP_CANCEL, CANCEL):
+            _AWAIT_AUTH.pop(uid, None)
+            _AWAIT_CUSTOM.pop(uid, None)
+            _AWAIT_EDIT.pop(uid, None)
+            _PENDING.pop(uid, None)
+            await q.answer("Cancelled")
+            await safe_edit(q.message, texts.CANCELLED, reply_markup=main_menu_keyboard())
+            return
 
-            # ---- Upload ----
-            set_task(task_id, "uploading", "")
-            up_t0 = time.time()
-
-            se = SpeedETA()
-            last_edit = 0.0
-
-            async def prog(done: int, total: int):
-                nonlocal last_edit
-                snap = se.update(done, total)
-                if time.time() - last_edit < 2.0 and done < total:
-                    return
-                last_edit = time.time()
-                txt = (
-                    "⬆️ Uploading to YouTube…\n"
-                    f"{human_bytes(done)}/{human_bytes(total)}\n"
-                    f"speed: {human_bytes(snap['speed_bps'])}/s | eta: {human_eta(snap['eta_sec'])}"
-                )
-                await safe_edit(status, txt)
-
-            title, desc = extract_title_description(m, file_name)
-
-            yt_url = await upload_video(
-                access_tok,
-                file_path,
-                title,
-                desc,
-                privacy="private",
-                progress_cb=prog,
-            )
+        if action == BACK:
+            await q.answer()
+            await safe_edit(q.message, texts.START_TEXT, reply_markup=main_menu_keyboard())
+            return
 
-            up_sec = max(0.001, time.time() - up_t0)
-            up_avg_bps = int((file_size or 0) / up_sec) if file_size else 0
+        await q.answer()
+
+    # -------------------
+    # TEXT INPUT handler (auth/custom/edit)
+    # -------------------
+    @app.on_message(filters.text & filters.private)
+    async def text_handler(c: Client, m: Message):
+        uid = m.from_user.id
+        txt = (m.text or "").strip()
+
+        # A) Custom filename input
+        if _AWAIT_CUSTOM.get(uid):
+            d = _PENDING.get(uid)
+            if d:
+                name = _sanitize_filename(txt)
+                d["filename"] = name
+
+                title, desc = extract_title_description(d["src_msg"], default_title=name.rsplit(".", 1)[0])
+                d["title"] = title
+                d["description"] = desc
+                d["privacy"] = d.get("privacy") or "private"
+
+                _AWAIT_CUSTOM.pop(uid, None)
+                try:
+                    await m.delete()
+                except Exception:
+                    pass
+                await _preview_and_confirm(uid, d["status_msg"])
+            return
 
-            await record_upload(uid, use_profile_id)
-            set_task(task_id, "done", "", yt_url=yt_url)
+        # B) Edit title/desc input
+        if _AWAIT_EDIT.get(uid):
+            d = _PENDING.get(uid)
+            if not d:
+                _AWAIT_EDIT.pop(uid, None)
+                return
 
-            await safe_edit(
-                status,
-                "✅ **Uploaded!**\n\n"
-                f"🎬 {title}\n"
-                f"🔗 {yt_url}\n\n"
-                f"⬇️ Download: {human_eta(int(dl_sec))} | avg {human_bytes(dl_avg_bps)}/s\n"
-                f"⬆️ Upload: {human_eta(int(up_sec))} | avg {human_bytes(up_avg_bps)}/s",
-            )
+            lines = [ln.strip() for ln in (txt.splitlines() if txt else []) if ln.strip()]
+            if not lines:
+                return await safe_reply(m, texts.PARSE_FAIL)
 
-        except Exception as e:
-            set_task(task_id, "error", str(e))
-            await safe_edit(status, f"❌ Upload failed:\n`{type(e).__name__}: {e}`")
-        finally:
-            if file_path:
-                cleanup_file(file_path)
+            new_title = lines[0]
+            new_desc = "\n".join(lines[1:]) if len(lines) > 1 else (d.get("description") or "")
+
+            d["title"] = new_title
+            d["description"] = new_desc
+            _AWAIT_EDIT.pop(uid, None)
+            try:
+                await m.delete()
+            except Exception:
+                pass
+            await _preview_and_confirm(uid, d["status_msg"])
+            return
+
+        # C) Auth input
+        mode = _AWAIT_AUTH.get(uid)
+        if mode:
+            try:
+                cid = sec = ""
+                if mode == "json":
+                    data = json.loads(txt)
+                    cid = (data.get("installed", {}) or {}).get("client_id") or (data.get("web", {}) or {}).get("client_id") or ""
+                    sec = (data.get("installed", {}) or {}).get("client_secret") or (data.get("web", {}) or {}).get("client_secret") or ""
+                else:
+                    # allow: "id | secret" OR "id secret"
+                    if "|" in txt:
+                        parts = [p.strip() for p in txt.split("|", 1)]
+                    else:
+                        parts = [p.strip() for p in txt.split() if p.strip()]
+                    if len(parts) >= 2:
+                        cid, sec = parts[0], parts[1]
+
+                if cid and sec:
+                    res = await profile_add(uid, cid, sec, "main")
+                    link = _pick_login_url(res)
+                    _AWAIT_AUTH.pop(uid, None)
+                    if not link:
+                        return await safe_reply(m, f"✅ Saved. (No login url returned)\n`{res}`", reply_markup=main_menu_keyboard())
+                    return await safe_reply(m, texts.SENT_AUTH_LINK + link, reply_markup=main_menu_keyboard())
+                else:
+                    return await safe_reply(m, "❌ Invalid format.", reply_markup=main_menu_keyboard())
+
+            except Exception as e:
+                return await safe_reply(m, f"❌ Error: {e}", reply_markup=main_menu_keyboard())
+
+        # if none of the above, ignore
+        return