Update bot/handlers.py

bot/handlers.py

@@ -1,807 +1,725 @@
-# PATH: bot/handlers.py
+# bot/handlers.py
 from __future__ import annotations
 
-import asyncio
-import json
 import os
+import json
 import time
-from typing import Dict, Optional, Any
+import asyncio
+from dataclasses import dataclass
+from typing import Any, Dict, Optional, Tuple, Union
 
 from hydrogram import Client, filters
-from hydrogram.types import Message, CallbackQuery
-
-from bot.config import Telegram, Workers
-from bot.ui import texts
-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, BACK,
-    MENU_HELP, MENU_AUTH, MENU_PROFILES, MENU_SPEEDTEST,
-    NAME_ORIGINAL, NAME_CAPTION, NAME_CUSTOM,
-    UP_GO, UP_EDIT, UP_PRIV, UP_CANCEL,
+from hydrogram.types import (
+    Message,
+    CallbackQuery,
+    InlineKeyboardMarkup,
+    InlineKeyboardButton,
 )
 
-from bot.core.auth import is_owner_id, is_admin_id, require_allowed
-from bot.core.link_parser import parse_link
+from bot.core.auth import require_allowed, is_owner_id
 from bot.core.progress import SpeedETA, human_bytes, human_eta
-from bot.core.speedtest import (
-    ping_ms,
-    net_download_test,
-    net_upload_test,
-    disk_total_free,
-    bytes_per_sec_to_mb_s,
+
+# ✅ profile_* are ADMIN routes => cf_worker1
+from bot.integrations.cf_worker1 import (
+    profile_add,
+    profile_list,
+    profile_set_default,
 )
-from bot.integrations.cf_worker1 import profile_add
+
+# ✅ HF/private routes => cf_worker2
 from bot.integrations.cf_worker2 import (
-    allow_user,
-    disallow_user,
     pick_profile,
     access_token,
     record_upload,
+    allow_user,
+    disallow_user,
     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.telegram.parse import extract_title_description
+from bot.telegram.media import download_to_temp
 from bot.temp.files import cleanup_file
-from bot.youtube.uploader import upload_video
-
-
-# -----------------------
-# 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()
-
-
-# -----------------------
-# HELPERS
-# -----------------------
-def _pick_login_url(j: dict) -> str:
-    if not isinstance(j, dict):
-        return ""
-    data = j.get("data") if isinstance(j.get("data"), dict) else {}
-
-    def _get(d: dict, k: str) -> str:
-        v = d.get(k)
-        return (v or "").strip() if isinstance(v, str) else ""
 
-    return (
-        _get(j, "login_url")
-        or _get(j, "loginUrl")
-        or _get(j, "auth_url")
-        or _get(j, "url")
-        or _get(data, "login_url")
-        or _get(data, "loginUrl")
-        or _get(data, "auth_url")
-        or _get(data, "url")
-    )
-
-
-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))
-    up_bytes = up_mb * 1024 * 1024
-    up_sec = int(max(0, time.time() - _STARTED_AT))
+from bot.ui.keyboards import (
+    main_menu_keyboard,
+    auth_menu_keyboard,
+    profiles_keyboard,
+    upload_confirm_keyboard,
+)
+from bot.ui.callbacks import (
+    parse_cb,
+    make,
+    MENU_AUTH,
+    MENU_PROFILES,
+    MENU_HELP,
+    MENU_SPEEDTEST,
+    BACK,
+    AUTH_JSON,
+    AUTH_CI,
+    UP_GO,
+    UP_EDIT,
+    UP_PRIV,
+    UP_CANCEL,
+)
+from bot.ui.texts import (
+    START_TEXT,
+    HELP_TEXT,
+    OWNER_ONLY,
+    NOT_ALLOWED,
+    CANCELLED,
+    NEED_AUTH,
+)
 
-    d = disk_total_free("/")
-    total_gb = d["total"] / (1024.0**3)
-    free_gb = d["free"] / (1024.0**3)
 
-    dl_task = asyncio.create_task(net_download_test(bytes_target=dl_bytes, timeout=40.0))
-    up_task = asyncio.create_task(net_upload_test(bytes_target=up_bytes, timeout=40.0))
-    p_task = asyncio.create_task(ping_ms())
+# =========================
+# In-memory state (per-user)
+# =========================
+
+@dataclass
+class PendingUpload:
+    src_msg: Message
+    downloader: Client
+    file_name: str
+    title: str
+    description: str
+    privacy: str  # private | unlisted | public
+    status_msg: Message
+    via_link: bool = False
+
+
+_AWAIT_AUTH_MODE: Dict[int, str] = {}           # uid -> "json"|"ci"
+_AWAIT_EDIT: Dict[int, bool] = {}               # uid -> True when waiting for title/desc text
+_PENDING_UPLOAD: Dict[int, PendingUpload] = {}  # uid -> pending upload
+_IN_PROGRESS: Dict[int, bool] = {}              # uid -> upload is running
+_SPEED_COOLDOWN_UNTIL: Dict[int, float] = {}    # uid -> unix time
+
+
+# =========================
+# Helpers
+# =========================
+
+def _admin_ids_from_env() -> set[int]:
+    raw = (os.getenv("ADMIN_IDS") or "").strip()
+    out: set[int] = set()
+    for part in raw.split(","):
+        part = part.strip()
+        if not part:
+            continue
+        try:
+            out.add(int(part))
+        except ValueError:
+            continue
+    return out
 
-    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"
-    return (
-        "⚡ **Server Speed Test**\n\n"
-        f"🕒 Uptime: {human_eta(up_sec)}\n\n"
-        "📡 Network\n"
-        f"├ 🟢 Ping: {ping_line}\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"
-    )
+_ADMIN_IDS = _admin_ids_from_env()
 
 
-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 _is_admin_or_owner(uid: int) -> bool:
+    return is_owner_id(uid) or (uid in _ADMIN_IDS)
 
 
-def _toggle_privacy(p: str) -> str:
-    p = (p or "private").lower().strip()
-    if p == "private":
+def _next_privacy(cur: str) -> str:
+    cur = (cur or "").lower()
+    if cur == "private":
         return "unlisted"
-    if p == "unlisted":
+    if cur == "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)
+def _media_and_filename(m: Message) -> Tuple[Optional[Any], str, int]:
+    if getattr(m, "video", None):
+        v = m.video
+        name = (getattr(v, "file_name", None) or "video.mp4").strip() or "video.mp4"
+        size = int(getattr(v, "file_size", 0) or 0)
+        return v, name, size
 
+    if getattr(m, "document", None):
+        d = m.document
+        name = (getattr(d, "file_name", None) or "file.bin").strip() or "file.bin"
+        size = int(getattr(d, "file_size", 0) or 0)
+        return d, name, size
 
-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)
+    return None, "file.bin", 0
 
-    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))
+def _render_preview(p: PendingUpload) -> str:
+    size = 0
+    if getattr(p.src_msg, "video", None):
+        size = int(getattr(p.src_msg.video, "file_size", 0) or 0)
+    elif getattr(p.src_msg, "document", None):
+        size = int(getattr(p.src_msg.document, "file_size", 0) or 0)
 
-    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"]
+    desc = p.description or ""
+    if len(desc) > 500:
+        desc = desc[:500] + "…"
 
+    return (
+        "✅ *Ready to upload*\n\n"
+        f"**File:** `{p.file_name}`\n"
+        f"**Size:** {human_bytes(size) if size else '—'}\n"
+        f"**Privacy:** `{p.privacy}`\n\n"
+        f"**Title:** {p.title or '—'}\n"
+        f"**Description:** {desc or '—'}"
+    )
 
-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)}")
+def _pick_login_url(resp: Any) -> Optional[str]:
+    if not isinstance(resp, dict):
+        return None
+    if isinstance(resp.get("login_url"), str):
+        return resp["login_url"]
+    data = resp.get("data")
+    if isinstance(data, dict) and isinstance(data.get("login_url"), str):
+        return data["login_url"]
+    return None
+
+
+def parse_telegram_link(link: str) -> Tuple[Union[int, str], int]:
+    s = (link or "").strip()
+    if not s:
+        raise ValueError("empty link")
+
+    s = s.replace("https://", "").replace("http://", "")
+    if s.startswith("t.me/"):
+        s = s[len("t.me/") :]
+
+    s = s.split("?", 1)[0].split("#", 1)[0]
+    s = s.strip("/")
+
+    parts = s.split("/")
+    if len(parts) < 2:
+        raise ValueError("link must include message id")
+
+    if parts[0] == "c":
+        if len(parts) < 3:
+            raise ValueError("invalid /c/ link")
+        internal_id = parts[1]
+        msg_id = parts[2]
+        if not internal_id.isdigit() or not msg_id.isdigit():
+            raise ValueError("invalid numeric ids in /c/ link")
+        chat_id = int(f"-100{internal_id}")
+        return chat_id, int(msg_id)
+
+    username = parts[0]
+    msg_id = parts[1]
+    if not msg_id.isdigit():
+        raise ValueError("invalid message id")
+    return username, int(msg_id)
+
+
+async def _ensure_allowed(m: Message) -> bool:
+    uid = m.from_user.id if m.from_user else 0
+    if not uid:
+        await safe_reply(m, NOT_ALLOWED)
+        return False
+    if not await require_allowed(uid):
+        await safe_reply(m, NOT_ALLOWED)
+        return False
+    return True
+
+
+async def _start_pending_upload(
+    *,
+    chat_msg: Message,
+    src_msg: Message,
+    downloader: Client,
+    privacy: str = "private",
+    via_link: bool = False,
+) -> None:
+    uid = chat_msg.from_user.id
+    _, file_name, _ = _media_and_filename(src_msg)
+    title, desc = extract_title_description(src_msg, file_name)
+
+    preview = PendingUpload(
+        src_msg=src_msg,
+        downloader=downloader,
+        file_name=file_name,
+        title=title,
+        description=desc,
+        privacy=privacy,
+        status_msg=chat_msg,
+        via_link=via_link,
+    )
+    status = await safe_reply(chat_msg, _render_preview(preview), reply_markup=upload_confirm_keyboard(privacy))
+    if not status:
+        return
 
-    return cb
+    _PENDING_UPLOAD[uid] = PendingUpload(
+        src_msg=src_msg,
+        downloader=downloader,
+        file_name=file_name,
+        title=title,
+        description=desc,
+        privacy=privacy,
+        status_msg=status,
+        via_link=via_link,
+    )
 
 
-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:
+async def _run_upload(uid: int) -> None:
+    if _IN_PROGRESS.get(uid):
+        return
+    pending = _PENDING_UPLOAD.get(uid)
+    if not pending:
         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),
-    )
 
+    _IN_PROGRESS[uid] = True
+    st = pending.status_msg
+    file_path = None
 
-async def _do_upload(uid: int):
-    """
-    Performs: download (if needed) -> upload to YouTube -> cleanup -> final message.
-    """
-    d = _PENDING.get(uid)
-    if not d:
-        return
+    try:
+        pl = await profile_list(uid)
+        if not (isinstance(pl, dict) and pl.get("ok")):
+            await safe_edit(st, "❌ Failed to fetch profiles. Try /profiles again.")
+            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)
+        profiles = pl.get("profiles") or []
+        if not profiles:
+            await safe_edit(st, NEED_AUTH)
+            return
 
-    # auth
-    profile_id, access = await _ensure_profile_and_token(uid)
+        default_id = pl.get("default_profile_id") or profiles[0].get("profile_id")
+        default_profile = next((p for p in profiles if p.get("profile_id") == default_id), profiles[0])
+        channel_id = default_profile.get("channel_id")
+        if not channel_id:
+            await safe_edit(st, "❌ Default profile is not connected. Use /profiles → set a connected one as default.")
+            return
 
-    # download progress
-    await safe_edit(status, f"⬇️ **Downloading:** `{filename}` …")
-    dl_cb = _make_throttled_progress_editor(status, "⬇️ **Downloading…**")
+        pick = await pick_profile(uid, channel_id)
+        if not (isinstance(pick, dict) and pick.get("ok")):
+            err = pick.get("err", "unknown") if isinstance(pick, dict) else "unknown"
+            await safe_edit(st, f"❌ Profile pick failed: `{err}`")
+            return
 
-    path = ""
-    try:
-        path, _, _ = await download_to_temp(downloader, msg, progress_cb=dl_cb)
-        if not path:
-            raise RuntimeError("❌ Download failed.")
+        prof_id = pick.get("profile_id")
+        if not prof_id:
+            await safe_edit(st, "❌ pick_profile returned no profile_id.")
+            return
 
-        # upload progress
-        await safe_edit(status, "⬆️ **Uploading to YouTube…**")
-        se = SpeedETA()
-        last_upd = 0.0
+        tok = await access_token(uid, prof_id)
+        if not (isinstance(tok, dict) and tok.get("ok") and tok.get("access_token")):
+            err = tok.get("err", "token_failed") if isinstance(tok, dict) else "token_failed"
+            await safe_edit(st, f"❌ Token failed: `{err}`")
+            return
+
+        access = tok["access_token"]
+
+        await safe_edit(st, "⬇️ Downloading media…")
+        file_path, _, _ = await download_to_temp(pending.downloader, pending.src_msg)
 
-        async def up_cb(done: int, total: int):
-            nonlocal last_upd
+        from bot.youtube.uploader import upload_video  # local import
+        speed = SpeedETA()
+        last_ui = 0.0
+        start_t = time.time()
+
+        async def progress_cb(sent: int, total: int) -> None:
+            nonlocal last_ui
             now = time.time()
-            if now - last_upd < 3.0:
+            if now - last_ui < 0.8:
                 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",
+            last_ui = now
+            rate = speed.update(sent, total)
+            txt = (
+                "⬆️ Uploading…\n\n"
+                f"{human_bytes(sent)} / {human_bytes(total)}\n"
+                f"Speed: {human_bytes(rate)}/s\n"
+                f"ETA: {human_eta(speed.eta_seconds)}"
             )
+            await safe_edit(st, txt)
 
-        yt_url = await upload_video(
+        up = 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,
+            file_path=file_path,
+            title=pending.title,
+            description=pending.description,
+            privacy=pending.privacy,
+            progress_cb=progress_cb,
         )
 
-        await record_upload(uid, profile_id)
-        await safe_edit(status, f"✅ **Uploaded!**\n\n📺 `{filename}`\n🔗 {yt_url}")
+        if not (isinstance(up, dict) and up.get("ok")):
+            err = up.get("err", "upload_failed") if isinstance(up, dict) else "upload_failed"
+            await safe_edit(st, f"❌ Upload failed: `{err}`")
+            return
 
-    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())
-
-    @app.on_message(filters.command(["help"]) & filters.private)
-    async def help_handler(_: Client, m: Message):
-        await safe_reply(m, texts.HELP_TEXT, reply_markup=main_menu_keyboard())
-
-    @app.on_message(filters.command(["ping"]) & filters.private)
-    async def ping_handler(_: Client, m: Message):
-        await safe_reply(m, "🏓 Pong!")
-
-    @app.on_message(filters.command(["me"]) & filters.private)
-    async def me_handler(_: Client, m: Message):
-        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
-    # -------------------
-    @app.on_message(filters.command(["speedtest"]) & filters.private)
-    async def speedtest_cmd(_: Client, m: Message):
-        uid = m.from_user.id
-        now = time.time()
-        if now - _SPEED_COOLDOWN.get(uid, 0.0) < 30:
-            return await safe_reply(m, "⏳ Wait 30s...")
-        _SPEED_COOLDOWN[uid] = now
+        url = up.get("url") or up.get("video_url") or up.get("watch_url") or ""
+        await record_upload(uid, prof_id)
 
-        msg = await safe_reply(m, "⚡ Running speedtest…")
-        try:
-            txt = await _render_speedtest(8)
-            await safe_edit(msg, txt)
-        except Exception as e:
-            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)
-
-        cmd = m.command[0]
-        if cmd == "stats":
-            return await safe_reply(m, f"📊 Stats:\n`{await stats_today()}`")
-
-        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)}")
-
-        # allow/disallow
-        if len(m.command) < 2:
-            return await safe_reply(m, f"Usage: /{cmd} <id>")
-        target = int(m.command[1])
-
-        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 + PROFILES
-    # -------------------
-    @app.on_message(filters.command(["auth"]) & filters.private)
-    async def auth_cmd(_: Client, m: Message):
-        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 Profile:", reply_markup=auth_menu_keyboard())
+        dur = max(1.0, time.time() - start_t)
+        done = "✅ Uploaded!"
+        if url:
+            done += f"\n\n{url}"
+        done += f"\n\nTime: {dur:.1f}s"
+        await safe_edit(st, done)
 
-    @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)
-
-        j = await profile_list(uid)
-        if not j.get("ok"):
-            return await safe_reply(m, texts.PROFILE_LIST_FAIL.format(j))
-
-        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)
+        _PENDING_UPLOAD.pop(uid, None)
+        _AWAIT_EDIT.pop(uid, None)
 
-        if not _require_admin_for_links(uid):
-            return await safe_reply(m, "❌ Link-archive is admin/owner only.")
+    except Exception as e:
+        await safe_edit(st, f"❌ Error: `{str(e)[:180]}`")
+    finally:
+        if file_path:
+            try:
+                cleanup_file(file_path)
+            except Exception:
+                pass
+        _IN_PROGRESS.pop(uid, None)
 
-        if not user_app:
-            return await safe_reply(m, "❌ `USER_SESSION_STRING` missing. Link-archive disabled.")
 
-        if len(m.command) < 2:
-            return await safe_reply(m, "Usage: `/archive <t.me message link>`")
+# =========================
+# Handlers
+# =========================
 
-        link = m.command[1]
-        ids = parse_link(link)
-        if not ids:
-            return await safe_reply(m, "❌ Invalid link.")
-        chat_id, msg_id = ids
+def setup_handlers(app: Client, user_app: Optional[Client] = None) -> None:
+    @app.on_message(filters.command("start") & filters.private)
+    async def start_cmd(_, m: Message) -> None:
+        await safe_reply(m, START_TEXT, reply_markup=main_menu_keyboard())
 
-        status = await safe_reply(m, "🔍 **Fetching message…**")
+    @app.on_message(filters.command("help") & filters.private)
+    async def help_cmd(_, m: Message) -> None:
+        await safe_reply(m, HELP_TEXT, reply_markup=main_menu_keyboard())
 
-        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(),
-            )
+    @app.on_message(filters.command("cancel") & filters.private)
+    async def cancel_cmd(_, m: Message) -> None:
+        uid = m.from_user.id
+        _AWAIT_AUTH_MODE.pop(uid, None)
+        _AWAIT_EDIT.pop(uid, None)
+        _PENDING_UPLOAD.pop(uid, None)
+        await safe_reply(m, CANCELLED, reply_markup=main_menu_keyboard())
 
-        except Exception as e:
-            await safe_edit(status, f"❌ Error: {e}")
+    # owner/admin
+    @app.on_message(filters.command("allow") & filters.private)
+    async def allow_cmd(_, m: Message) -> None:
+        uid = m.from_user.id
+        if not _is_admin_or_owner(uid):
+            await safe_reply(m, OWNER_ONLY)
+            return
+        parts = (m.text or "").split()
+        if len(parts) < 2 or not parts[1].isdigit():
+            await safe_reply(m, "Usage: /allow <tg_id>")
+            return
+        r = await allow_user(int(parts[1]))
+        await safe_reply(m, json.dumps(r, indent=2)[:3900])
 
-    # -------------------
-    # BATCH (admin/owner only)
-    # -------------------
-    @app.on_message(filters.command(["batch"]) & filters.private)
-    async def batch_handler(c: Client, m: Message):
+    @app.on_message(filters.command("disallow") & filters.private)
+    async def disallow_cmd(_, m: Message) -> None:
         uid = m.from_user.id
-        if not await require_allowed(uid):
-            return await safe_reply(m, texts.NOT_ALLOWED)
+        if not _is_admin_or_owner(uid):
+            await safe_reply(m, OWNER_ONLY)
+            return
+        parts = (m.text or "").split()
+        if len(parts) < 2 or not parts[1].isdigit():
+            await safe_reply(m, "Usage: /disallow <tg_id>")
+            return
+        r = await disallow_user(int(parts[1]))
+        await safe_reply(m, json.dumps(r, indent=2)[:3900])
 
-        if not _require_admin_for_links(uid):
-            return await safe_reply(m, "❌ Batch is admin/owner only.")
+    @app.on_message(filters.command("stats") & filters.private)
+    async def stats_cmd(_, m: Message) -> None:
+        uid = m.from_user.id
+        if not _is_admin_or_owner(uid):
+            await safe_reply(m, OWNER_ONLY)
+            return
+        r = await stats_today()
+        await safe_reply(m, json.dumps(r, indent=2)[:3900])
 
-        if not user_app:
-            return await safe_reply(m, "❌ `USER_SESSION_STRING` missing. Batch disabled.")
+    # profiles/auth
+    @app.on_message(filters.command("auth") & filters.private)
+    async def auth_cmd(_, m: Message) -> None:
+        if not await _ensure_allowed(m):
+            return
+        await safe_reply(m, "Choose auth mode:", reply_markup=auth_menu_keyboard())
 
-        if len(m.command) < 3:
-            return await safe_reply(m, "Usage: `/batch <start_link> <end_link>`")
+    @app.on_message(filters.command("profiles") & filters.private)
+    async def profiles_cmd(_, m: Message) -> None:
+        if not await _ensure_allowed(m):
+            return
+        uid = m.from_user.id
+        pl = await profile_list(uid)
+        if not (isinstance(pl, dict) and pl.get("ok")):
+            err = pl.get("err", "unknown") if isinstance(pl, dict) else "unknown"
+            await safe_reply(m, f"❌ profile_list failed: `{err}`")
+            return
+        await safe_reply(
+            m,
+            f"Profiles for `{uid}`\nDefault: `{pl.get('default_profile_id')}`\n\nPick default:",
+            reply_markup=profiles_keyboard(pl),
+        )
 
-        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.")
+    # incoming media
+    @app.on_message((filters.video | filters.document) & filters.private)
+    async def incoming_media(_, m: Message) -> None:
+        if not await _ensure_allowed(m):
+            return
+        uid = m.from_user.id
+        if _IN_PROGRESS.get(uid):
+            await safe_reply(m, "⏳ Upload already running. Use /cancel to reset.")
+            return
+        media, _, _ = _media_and_filename(m)
+        if not media:
+            await safe_reply(m, "Send a video or a document file.")
+            return
+        await _start_pending_upload(chat_msg=m, src_msg=m, downloader=app, via_link=False)
 
-        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.")
+    # link/archive mode (owner/admin only)
+    @app.on_message(filters.command(["yt", "dl", "archive"]) & filters.private)
+    async def archive_cmd(_, m: Message) -> None:
+        if not await _ensure_allowed(m):
+            return
+        uid = m.from_user.id
+        if not _is_admin_or_owner(uid):
+            await safe_reply(m, OWNER_ONLY)
+            return
+        args = (m.text or "").split(maxsplit=1)
+        if len(args) < 2:
+            await safe_reply(m, "Send: /archive <t.me message link>")
+            return
+        if user_app is None:
+            await safe_reply(m, "❌ Link mode is not configured (user session missing).")
+            return
 
-        if start_id > end_id:
-            start_id, end_id = end_id, start_id
+        try:
+            chat_ref, msg_id = parse_telegram_link(args[1].strip())
+        except Exception as e:
+            await safe_reply(m, f"❌ Bad link: {e}")
+            return
 
-        status = await safe_reply(m, f"🔄 Batching {start_id} → {end_id}…")
-        uploaded = 0
+        await safe_reply(m, "🔎 Fetching message…")
+        try:
+            src = await user_app.get_messages(chat_ref, msg_id)
+        except Exception as e:
+            await safe_reply(m, f"❌ Cannot fetch: `{str(e)[:180]}`")
+            return
 
-        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}")
-
-            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 direct_upload_handler(c: Client, m: Message):
-        uid = m.from_user.id
-        if not await require_allowed(uid):
-            return
-
-        # prepare preview from message itself
-        media = m.video or m.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):
+        media, _, _ = _media_and_filename(src)
+        if not media:
+            await safe_reply(m, "❌ That message has no video/document.")
+            return
+
+        await _start_pending_upload(chat_msg=m, src_msg=src, downloader=user_app, via_link=True)
+
+    # callbacks
+    @app.on_callback_query(filters.private)
+    async def cb_handler(_, q: CallbackQuery) -> None:
         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"
+        action, value = parse_cb(q.data or "")
 
+        try:
             await q.answer()
-            await _preview_and_confirm(uid, d["status_msg"])
+        except Exception:
+            pass
+
+        if action == BACK:
+            await safe_edit(q.message, START_TEXT, reply_markup=main_menu_keyboard())
+            return
+
+        if action == MENU_HELP:
+            await safe_edit(q.message, HELP_TEXT, reply_markup=main_menu_keyboard())
             return
 
-        # ---- 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()
+                await safe_edit(q.message, NOT_ALLOWED, reply_markup=main_menu_keyboard())
+                return
+            await safe_edit(q.message, "Choose auth mode:", reply_markup=auth_menu_keyboard())
+            return
 
         if action == MENU_PROFILES:
-            # reuse profiles command
-            fake = q.message
-            await q.answer()
-            return await profiles_cmd(c, fake)
+            if not await require_allowed(uid):
+                await safe_edit(q.message, NOT_ALLOWED, reply_markup=main_menu_keyboard())
+                return
+            pl = await profile_list(uid)
+            if not (isinstance(pl, dict) and pl.get("ok")):
+                await safe_edit(q.message, "❌ Failed to load profiles.", reply_markup=main_menu_keyboard())
+                return
+            await safe_edit(
+                q.message,
+                f"Profiles for `{uid}`\nDefault: `{pl.get('default_profile_id')}`\n\nPick default:",
+                reply_markup=profiles_keyboard(pl),
+            )
+            return
 
         if action == MENU_SPEEDTEST:
-            await q.answer("Running…")
-            txt = await _render_speedtest(8)
-            await safe_edit(q.message, txt)
-            return
+            now = time.time()
+            until = _SPEED_COOLDOWN_UNTIL.get(uid, 0.0)
+            if now < until:
+                await safe_edit(q.message, f"⏳ Try again in {int(until - now)}s.", reply_markup=main_menu_keyboard())
+                return
+            _SPEED_COOLDOWN_UNTIL[uid] = now + 60
 
-        if action == MENU_HELP:
-            await q.answer()
-            await safe_edit(q.message, texts.HELP_TEXT, reply_markup=main_menu_keyboard())
+            from bot.core.speedtest import ping_ms, net_download_test, net_upload_test, disk_total_free
+
+            await safe_edit(q.message, "🧪 Running speed test…")
+            try:
+                p = await ping_ms()
+                dl = await net_download_test()
+                ul = await net_upload_test()
+                disk = disk_total_free()
+
+                dl_mbps = (float(dl.get("bps", 0) or 0) * 8.0) / 1e6 if isinstance(dl, dict) else 0.0
+                ul_mbps = (float(ul.get("bps", 0) or 0) * 8.0) / 1e6 if isinstance(ul, dict) else 0.0
+                total = int(disk.get("total", 0) or 0) if isinstance(disk, dict) else 0
+                free = int(disk.get("free", 0) or 0) if isinstance(disk, dict) else 0
+
+                txt = (
+                    "📶 *Speed test*\n\n"
+                    f"Ping: `{int(p)} ms`\n"
+                    f"Download: `{dl_mbps:.2f} Mbps`\n"
+                    f"Upload: `{ul_mbps:.2f} Mbps`\n"
+                    f"Disk: `{human_bytes(free)} free / {human_bytes(total)} total`"
+                )
+            except Exception as e:
+                txt = f"❌ Speed test error: `{str(e)[:180]}`"
+            await safe_edit(q.message, txt, reply_markup=main_menu_keyboard())
             return
 
-        # ---- auth buttons ----
         if action == AUTH_JSON:
-            _AWAIT_AUTH[uid] = "json"
-            await q.answer()
-            await safe_edit(q.message, texts.ASK_JSON)
+            _AWAIT_AUTH_MODE[uid] = "json"
+            await safe_edit(
+                q.message,
+                "Paste credentials JSON:\n`{\"client_id\":\"...\",\"client_secret\":\"...\",\"label\":\"optional\"}`",
+                reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("Back", callback_data=make(BACK))]]),
+            )
             return
 
         if action == AUTH_CI:
-            _AWAIT_AUTH[uid] = "ci"
-            await q.answer()
-            await safe_edit(q.message, texts.ASK_ID_SECRET)
+            _AWAIT_AUTH_MODE[uid] = "ci"
+            await safe_edit(
+                q.message,
+                "Send in 2–3 lines:\n`client_id`\n`client_secret`\n(optional 3rd line = label)",
+                reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("Back", callback_data=make(BACK))]]),
+            )
             return
 
         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())
+            if not await require_allowed(uid):
+                await safe_edit(q.message, NOT_ALLOWED, reply_markup=main_menu_keyboard())
+                return
+            prof_id = (value or "").strip()
+            if not prof_id:
+                return
+            r = await profile_set_default(uid, prof_id)
+            if isinstance(r, dict) and r.get("ok"):
+                await safe_edit(q.message, "✅ Default profile updated.", reply_markup=main_menu_keyboard())
+            else:
+                err = r.get("err", "unknown") if isinstance(r, dict) else "unknown"
+                await safe_edit(q.message, f"❌ Failed: `{err}`", reply_markup=main_menu_keyboard())
             return
 
-        # ---- upload confirm UI ----
-        if action == UP_EDIT:
-            _AWAIT_EDIT[uid] = True
-            await q.answer()
-            await safe_edit(q.message, texts.EDIT_PROMPT)
-            return
+        if action in (UP_GO, UP_EDIT, UP_PRIV, UP_CANCEL):
+            pending = _PENDING_UPLOAD.get(uid)
+            if not pending:
+                await safe_edit(q.message, "No pending upload. Send a video/document first.", reply_markup=main_menu_keyboard())
+                return
 
-        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
+            pending.status_msg = q.message
+            _PENDING_UPLOAD[uid] = pending
 
-        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)
+            if action == UP_CANCEL:
+                _PENDING_UPLOAD.pop(uid, None)
                 _AWAIT_EDIT.pop(uid, None)
-            return
+                await safe_edit(q.message, CANCELLED, reply_markup=main_menu_keyboard())
+                return
 
-        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
+            if action == UP_PRIV:
+                pending.privacy = _next_privacy(pending.privacy)
+                _PENDING_UPLOAD[uid] = pending
+                await safe_edit(q.message, _render_preview(pending), reply_markup=upload_confirm_keyboard(pending.privacy))
+                return
 
-        if action == BACK:
-            await q.answer()
-            await safe_edit(q.message, texts.START_TEXT, reply_markup=main_menu_keyboard())
-            return
+            if action == UP_EDIT:
+                _AWAIT_EDIT[uid] = True
+                await safe_reply(
+                    q.message,
+                    "✏️ Send new title + optional description.\n\nFormats:\n1) `Title | Description`\n2) `Title` then blank line then description",
+                )
+                return
 
-        await q.answer()
+            if action == UP_GO:
+                await safe_edit(q.message, "⏳ Starting upload…")
+                asyncio.create_task(_run_upload(uid))
+                return
 
-    # -------------------
-    # TEXT INPUT handler (auth/custom/edit)
-    # -------------------
+    # text router (auth + 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()
+    async def text_router(_, m: Message) -> None:
+        uid = m.from_user.id if m.from_user else 0
+        if not uid:
+            return
 
-        # A) Custom filename input
-        if _AWAIT_CUSTOM.get(uid):
-            d = _PENDING.get(uid)
-            if d:
-                name = _sanitize_filename(txt)
-                d["filename"] = name
+        mode = _AWAIT_AUTH_MODE.get(uid)
+        if mode:
+            if not await _ensure_allowed(m):
+                _AWAIT_AUTH_MODE.pop(uid, None)
+                return
+
+            text = (m.text or "").strip()
+            try:
+                if mode == "json":
+                    obj = json.loads(text)
+                    client_id = str(obj.get("client_id", "")).strip()
+                    client_secret = str(obj.get("client_secret", "")).strip()
+                    label = str(obj.get("label", "")).strip()[:40] if obj.get("label") else ""
+                else:
+                    lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
+                    client_id = lines[0] if len(lines) >= 1 else ""
+                    client_secret = lines[1] if len(lines) >= 2 else ""
+                    label = lines[2][:40] if len(lines) >= 3 else ""
+
+                if not client_id or not client_secret:
+                    raise ValueError("missing client_id/client_secret")
 
-                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_AUTH_MODE.pop(uid, None)
 
-                _AWAIT_CUSTOM.pop(uid, None)
-                try:
-                    await m.delete()
-                except Exception:
-                    pass
-                await _preview_and_confirm(uid, d["status_msg"])
+                resp = await profile_add(uid, client_id, client_secret, label)
+                if not (isinstance(resp, dict) and resp.get("ok")):
+                    err = resp.get("err", "unknown") if isinstance(resp, dict) else "unknown"
+                    await safe_reply(m, f"❌ profile_add failed: `{err}`")
+                    return
+
+                login_url = _pick_login_url(resp)
+                if not login_url:
+                    await safe_reply(m, "✅ Profile added, but login link missing from response. Check your Pages backend.")
+                    return
+
+                await safe_reply(m, f"✅ Profile added.\n\nOpen this link to authorize:\n{login_url}")
+                await safe_reply(m, "After authorizing, send a video/document here to upload.", reply_markup=main_menu_keyboard())
+
+            except Exception as e:
+                await safe_reply(m, f"❌ Parse error: `{str(e)[:180]}`\n\nTry /auth again.")
             return
 
-        # B) Edit title/desc input
         if _AWAIT_EDIT.get(uid):
-            d = _PENDING.get(uid)
-            if not d:
+            pending = _PENDING_UPLOAD.get(uid)
+            if not pending:
                 _AWAIT_EDIT.pop(uid, None)
                 return
 
-            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)
+            raw = (m.text or "").strip()
+            if not raw or raw.startswith("/"):
+                return
 
-            new_title = lines[0]
-            new_desc = "\n".join(lines[1:]) if len(lines) > 1 else (d.get("description") or "")
+            title = raw
+            desc = pending.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
+            if "\n\n" in raw:
+                a, b = raw.split("\n\n", 1)
+                title = a.strip()
+                desc = b.strip()
+            elif "|" in raw:
+                a, b = raw.split("|", 1)
+                title = a.strip()
+                desc = b.strip()
 
-        # 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())
+            title = (title or pending.title or "video").strip()[:95]
+            desc = (desc or "").strip()[:4900]
 
-            except Exception as e:
-                return await safe_reply(m, f"❌ Error: {e}", reply_markup=main_menu_keyboard())
+            pending.title = title
+            pending.description = desc
+            _PENDING_UPLOAD[uid] = pending
+            _AWAIT_EDIT.pop(uid, None)
 
-        # if none of the above, ignore
-        return
+            await safe_edit(pending.status_msg, _render_preview(pending), reply_markup=upload_confirm_keyboard(pending.privacy))
+            return