Update bot/handlers.py

bot/handlers.py

@@ -1,76 +1,43 @@
-# bot/handlers.py
+# FILE: bot/handlers.py
+# NOTE: Updated:
+# - Fix float(dict) crash by using SpeedETA that returns float and safe int conversions
+# - Show download speed/ETA + upload speed/ETA + total time summary
+# - Fix upload_video return-type mismatch (now dict)
+# - /cancel actually cancels running upload and batch tasks
+# - Add /batch (admin/owner only)
+
 from __future__ import annotations
 
+import asyncio
 import os
-import json
 import time
-import asyncio
 from dataclasses import dataclass
-from typing import Any, Dict, Optional, Tuple, Union
+from typing import Any, Dict, Optional, Tuple
 
-import httpx
 from hydrogram import Client, filters
-from hydrogram.types import (
-    Message,
-    CallbackQuery,
-    InlineKeyboardMarkup,
-    InlineKeyboardButton,
-)
-
-from bot.config import Workers
-from bot.core.auth import require_allowed, is_owner_id
+from hydrogram.types import CallbackQuery, InlineKeyboardMarkup, Message
+
+from bot.config import Auth, Workers
+from bot.integrations.auth import (allow_user, disallow_user, get_stats,
+                                   is_allowed)
+from bot.integrations.cf_worker1 import (profile_add, profile_check_auth,
+                                         profile_delete)
+from bot.integrations.cf_worker2 import (get_default_profile, list_profiles,
+                                         record_upload, set_default_profile)
+from bot.telegram.files import cleanup_file
+from bot.telegram.media import download_to_temp
+from bot.telegram.replies import safe_edit, safe_reply
+from bot.ui.callbacks import (MENU_AUTH, MENU_HELP, MENU_PROFILES, MENU_SPEED,
+                              UP_DEL, UP_EDIT, UP_GO, UP_PRIV)
+from bot.ui.keyboards import (auth_menu_keyboard, main_menu_keyboard,
+                              profiles_keyboard, upload_confirm_keyboard)
+from bot.ui.parse import parse_cb
+from bot.ui.texts import CANCELLED, HELP_TEXT, NEED_AUTH, NOT_ALLOWED, OWNER_ONLY
 from bot.core.progress import SpeedETA, human_bytes, human_eta
+from bot.core.settings import Settings
+from bot.core.uptime import uptime_text
+from bot.youtube.link_parser import parse_telegram_link
 
-# ✅ profile_add is ADMIN route => cf_worker1 (Pages, BOT_BACKEND_KEY)
-from bot.integrations.cf_worker1 import profile_add
-
-# ✅ HF/private routes => cf_worker2 (Pages, HF_API_KEY)
-from bot.integrations.cf_worker2 import (
-    pick_profile,
-    access_token,
-    record_upload,
-    allow_user,
-    disallow_user,
-    stats_today,
-)
-
-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.ui.keyboards import (
-    main_menu_keyboard,
-    auth_menu_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,
-)
-
-# =========================
-# In-memory state (per-user)
-# =========================
 
 @dataclass
 class PendingUpload:
@@ -79,345 +46,97 @@ class PendingUpload:
     file_name: str
     title: str
     description: str
-    privacy: str  # private | unlisted | public
-    status_msg: Message
+    privacy: str = "private"
+    status_msg: Optional[Message] = None
     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
-
-_PENDING_DELETE: Dict[int, str] = {}            # uid -> profile_id pending delete confirmation
-
-
-# =========================
-# Helpers
-# =========================
-
-def _admin_ids_from_env() -> set[int]:
-    raw = (os.getenv("ADMIN_IDS") or "").strip()
-    out: set[int] = set()
-    for part in raw.replace(";", ",").replace("|", ",").split(","):
-        part = part.strip()
-        if not part:
-            continue
-        try:
-            out.add(int(part))
-        except ValueError:
-            continue
-    return out
+@dataclass
+class EditState:
+    title: str
+    description: str
+    privacy: str
 
 
-_ADMIN_IDS = _admin_ids_from_env()
+# In-memory state
+_PENDING_UPLOAD: Dict[int, PendingUpload] = {}          # uid -> pending upload
+_AWAIT_EDIT: Dict[int, EditState] = {}                  # uid -> editing title/desc
+_AWAIT_AUTH_MODE: Dict[int, str] = {}                   # uid -> "oauth" | "token"
+_PENDING_DELETE: Dict[int, str] = {}                    # uid -> profile_id to delete
+_IN_PROGRESS: Dict[int, bool] = {}                      # uid -> upload is running
+_UPLOAD_TASK: Dict[int, asyncio.Task] = {}              # uid -> running upload task (so /cancel can stop it)
+_BATCH_TASK: Dict[int, asyncio.Task] = {}               # uid -> running batch task
 
 
 def _is_admin_or_owner(uid: int) -> bool:
-    return is_owner_id(uid) or (uid in _ADMIN_IDS)
-
-
-def _next_privacy(cur: str) -> str:
-    cur = (cur or "").lower()
-    if cur == "private":
-        return "unlisted"
-    if cur == "unlisted":
-        return "public"
-    return "private"
-
-
-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
-
-    return None, "file.bin", 0
-
-
-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)
-
-    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 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)
+    return uid in set(Auth.OWNERS) or uid in set(Auth.ADMINS)
 
 
 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
-
-    # ✅ owner/admin must never get blocked by allowlist
+    uid = m.from_user.id
     if _is_admin_or_owner(uid):
         return True
-
-    if not await require_allowed(uid):
+    ok = await is_allowed(uid)
+    if not ok:
         await safe_reply(m, NOT_ALLOWED)
-        return False
-    return True
-
-
-# =========================
-# Worker calls (no integration dependency)
-# =========================
-
-_HTTP_TIMEOUT = httpx.Timeout(20.0, connect=10.0)
-
-async def _post_json(url: str, bearer: str, payload: dict) -> dict:
-    headers = {"authorization": f"Bearer {bearer}"}
-    try:
-        async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT) as c:
-            r = await c.post(url, headers=headers, json=payload)
-        ct = (r.headers.get("content-type") or "").lower()
-        if "application/json" in ct:
-            j = r.json()
-            return j if isinstance(j, dict) else {"ok": False, "err": "bad_json"}
-        # Pages sometimes returns text (Unauthorized etc)
-        return {"ok": False, "err": f"http_{r.status_code}", "detail": (r.text or "")[:200]}
-    except Exception as e:
-        return {"ok": False, "err": "network_error", "detail": str(e)[:200]}
-
-
-async def _hf_list_profiles(uid: int, only_connected: bool = False) -> dict:
-    if not Workers.WORKER2_URL or not Workers.HF_API_KEY:
-        return {"ok": False, "err": "WORKER2_URL/HF_API_KEY missing"}
-    return await _post_json(
-        f"{Workers.WORKER2_URL}/api/list_profiles",
-        Workers.HF_API_KEY,
-        {"tg_id": str(uid), "only_connected": bool(only_connected)},
-    )
-
-
-async def _admin_profile_set_default(uid: int, profile_id: str) -> dict:
-    if not Workers.WORKER1_URL or not Workers.BOT_BACKEND_KEY:
-        return {"ok": False, "err": "WORKER1_URL/BOT_BACKEND_KEY missing"}
-    return await _post_json(
-        f"{Workers.WORKER1_URL}/api/profile/set_default",
-        Workers.BOT_BACKEND_KEY,
-        {"tg_id": str(uid), "profile_id": str(profile_id)},
-    )
-
-
-async def _admin_profile_remove(uid: int, profile_id: str) -> dict:
-    if not Workers.WORKER1_URL or not Workers.BOT_BACKEND_KEY:
-        return {"ok": False, "err": "WORKER1_URL/BOT_BACKEND_KEY missing"}
-    return await _post_json(
-        f"{Workers.WORKER1_URL}/api/profile/remove",
-        Workers.BOT_BACKEND_KEY,
-        {"tg_id": str(uid), "profile_id": str(profile_id)},
-    )
+    return ok
 
 
-async def _admin_profile_login_link(uid: int, profile_id: str, force: bool = False) -> dict:
-    if not Workers.WORKER1_URL or not Workers.BOT_BACKEND_KEY:
-        return {"ok": False, "err": "WORKER1_URL/BOT_BACKEND_KEY missing"}
-    return await _post_json(
-        f"{Workers.WORKER1_URL}/api/profile/login_link",
-        Workers.BOT_BACKEND_KEY,
-        {"tg_id": str(uid), "profile_id": str(profile_id), "ttl_sec": 600, "force": bool(force)},
+def _media_and_filename(m: Message) -> Tuple[Optional[Any], str, int]:
+    if m.video:
+        return m.video, (m.video.file_name or "video.mp4"), (m.video.file_size or 0)
+    if m.document:
+        return m.document, (m.document.file_name or "file.bin"), (m.document.file_size or 0)
+    return None, "", 0
+
+
+def extract_title_description(src: Message, file_name: str) -> Tuple[str, str]:
+    # Prefer caption text
+    caption = (src.caption or "").strip()
+    if caption:
+        # title = first line, desc = remaining lines
+        parts = caption.splitlines()
+        title = parts[0].strip()
+        desc = "\n".join([p.strip() for p in parts[1:]]).strip()
+        return title[:Settings.MAX_TITLE], desc[:Settings.MAX_DESC]
+    # fallback to filename
+    base = os.path.splitext(file_name)[0].strip()
+    title = (base or "Untitled")[:Settings.MAX_TITLE]
+    return title, ""
+
+
+def _render_preview(p: PendingUpload, size: int) -> str:
+    return (
+        "📦 *Ready to upload*\\n\\n"
+        f"*File:* `{p.file_name}`\\n"
+        f"*Size:* {human_bytes(size)}\\n"
+        f"*Privacy:* `{p.privacy}`\\n\\n"
+        f"*Title:* {p.title}\\n"
+        f"*Description:* {(p.description[:600] + '…') if len(p.description) > 600 else (p.description or '—')}"
     )
 
 
-# =========================
-# Profiles UI (fixed)
-# =========================
-
-def _is_connected_profile(p: dict) -> bool:
-    return bool(p.get("has_refresh")) and bool(p.get("channel_id"))
-
-
-def _profiles_text(uid: int, pl: dict) -> str:
-    default_id = pl.get("default_profile_id") or "—"
-    day = pl.get("day") or ""
-    profiles = pl.get("profiles") or []
-    if not profiles:
-        return (
-            f"👤 Profiles for `{uid}`\n\n"
-            "No profiles found.\n\n"
-            "Use **Add Profile** (or /auth) to add credentials."
-        )
-
-    lines = [
-        f"👤 Profiles for `{uid}`",
-        f"Default: `{default_id}`",
-    ]
-    if day:
-        lines.append(f"Day: `{day}`")
-    lines.append("")
-    lines.append("Legend: ✅ connected | ⚠️ not connected")
-    lines.append("")
-
-    for p in profiles:
-        pid = p.get("profile_id") or ""
-        label = (p.get("label") or pid[:6] or "profile").strip()
-        mark = "⭐" if pid == default_id else " "
-        conn = "✅" if _is_connected_profile(p) else "⚠️"
-        ch = p.get("channel_title") or p.get("channel_id") or "—"
-        used = p.get("used_today")
-        used_txt = f" • used:{used}" if isinstance(used, int) else ""
-        lines.append(f"{mark}{conn} **{label}** — `{pid}` — {ch}{used_txt}")
-
-    lines.append("")
-    lines.append("Tap buttons below to set default / login / delete.")
-    return "\n".join(lines)
-
-
-def _profiles_markup(pl: dict) -> InlineKeyboardMarkup:
-    profiles = pl.get("profiles") or []
-    default_id = pl.get("default_profile_id") or ""
-
-    rows: list[list[InlineKeyboardButton]] = []
-
-    for p in profiles:
-        pid = str(p.get("profile_id") or "")
-        if not pid:
-            continue
-        connected = _is_connected_profile(p)
-        is_def = (pid == default_id)
-
-        # Row: Default + Login/Reauth + Delete
-        btn_default = InlineKeyboardButton(
-            "✅ Default" if is_def else "⭐ Set default",
-            callback_data=make("pdef", pid) if not is_def else make("pref"),
-        )
-        btn_login = InlineKeyboardButton(
-            "🔁 Re-auth" if connected else "🔑 Login",
-            callback_data=make("preauth" if connected else "plog", pid),
-        )
-        btn_del = InlineKeyboardButton("🗑 Delete", callback_data=make("pdel", pid))
-
-        rows.append([btn_default, btn_login, btn_del])
-
-    # Footer
-    rows.append(
-        [
-            InlineKeyboardButton("🔄 Refresh", callback_data=make("pref")),
-            InlineKeyboardButton("➕ Add profile", callback_data=make(MENU_AUTH)),
-        ]
-    )
-    rows.append([InlineKeyboardButton("⬅️ Back", callback_data=make(BACK))])
-
-    return InlineKeyboardMarkup(rows)
-
-
-async def _show_profiles(msg: Message, uid: int, *, edit: bool) -> None:
-    pl = await _hf_list_profiles(uid, only_connected=False)
-    if not (isinstance(pl, dict) and pl.get("ok")):
-        err = pl.get("err", "unknown") if isinstance(pl, dict) else "unknown"
-        detail = pl.get("detail") if isinstance(pl, dict) else None
-        txt = f"❌ Failed to load profiles: `{err}`"
-        if detail:
-            txt += f"\n`{str(detail)[:180]}`"
-        if edit:
-            await safe_edit(msg, txt, reply_markup=main_menu_keyboard())
-        else:
-            await safe_reply(msg, txt, reply_markup=main_menu_keyboard())
+async def _start_pending_upload(uid: int, src: Message, downloader: Client, via_link: bool = False) -> None:
+    media, file_name, size = _media_and_filename(src)
+    if not media:
         return
 
-    txt = _profiles_text(uid, pl)
-    kb = _profiles_markup(pl)
-    if edit:
-        await safe_edit(msg, txt, reply_markup=kb)
-    else:
-        await safe_reply(msg, txt, reply_markup=kb)
-
-
-# =========================
-# Upload flow
-# =========================
-
-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
+    title, desc = extract_title_description(src, file_name)
 
-    _PENDING_UPLOAD[uid] = PendingUpload(
-        src_msg=src_msg,
+    p = PendingUpload(
+        src_msg=src,
         downloader=downloader,
         file_name=file_name,
         title=title,
         description=desc,
-        privacy=privacy,
-        status_msg=status,
+        privacy="private",
         via_link=via_link,
+        status_msg=None,
     )
+    _PENDING_UPLOAD[uid] = p
+
+    txt = _render_preview(p, size)
+    await safe_reply(src, txt, reply_markup=upload_confirm_keyboard(p.privacy))
 
 
 async def _run_upload(uid: int) -> None:
@@ -426,71 +145,28 @@ async def _run_upload(uid: int) -> None:
     pending = _PENDING_UPLOAD.get(uid)
     if not pending:
         return
+    if not pending.status_msg:
+        return
 
     _IN_PROGRESS[uid] = True
     st = pending.status_msg
-    file_path = None
+    file_path: Optional[str] = None
 
     try:
-        # Use HF list (shows default + connection state)
-        pl = await _hf_list_profiles(uid, only_connected=False)
-        if not (isinstance(pl, dict) and pl.get("ok")):
-            await safe_edit(st, "❌ Failed to fetch profiles. Try /profiles again.")
-            return
-
-        profiles = pl.get("profiles") or []
-        if not profiles:
-            await safe_edit(st, NEED_AUTH)
-            return
-
-        default_id = pl.get("default_profile_id") or ""
-        default_profile = next((p for p in profiles if p.get("profile_id") == default_id), None)
-
-        # Default must be connected; otherwise fallback to first connected and instruct user
-        chosen = default_profile if default_profile and _is_connected_profile(default_profile) else None
-        if not chosen:
-            chosen = next((p for p in profiles if _is_connected_profile(p)), None)
-
-        if not chosen:
-            await safe_edit(
-                st,
-                "❌ No connected YouTube profile.\n\nOpen /profiles → tap **Login** to authorize a profile.",
-            )
-            return
-
-        channel_id = chosen.get("channel_id")
-        if not channel_id:
-            await safe_edit(st, "❌ चुने हुए profile में channel_id missing है. /profiles में Re-auth करो.")
-            return
-
-        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
-
-        prof_id = pick.get("profile_id")
-        if not prof_id:
-            await safe_edit(st, "❌ pick_profile returned no profile_id.")
-            return
-
-        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}`")
+        # Ensure there is a default profile
+        prof = await get_default_profile(uid)
+        if not (isinstance(prof, dict) and prof.get("ok") and prof.get("profile_id") and prof.get("access_token")):
+            await safe_edit(st, NEED_AUTH, reply_markup=main_menu_keyboard())
             return
 
-        access = tok["access_token"]
+        prof_id = str(prof["profile_id"])
+        access_token = str(prof["access_token"])
 
-        await safe_edit(st, "⬇️ Downloading media…")
-        file_path, _, _ = await download_to_temp(pending.downloader, pending.src_msg)
-
-        from bot.youtube.uploader import upload_video  # local import
-        speed = SpeedETA()
-        last_ui = 0.0
-        start_t = time.time()
+        # Track total time (download + upload)
+        overall_start = time.time()
 
-        # ✅ Fix: uploader sometimes passes dict/str instead of int → prevent float(dict) crash
+        # Safety: Telegram/hydrogram callbacks sometimes pass dict/None.
+        # Convert to int to avoid int(dict)/float(dict) crashes.
         def _as_int(v: Any) -> int:
             if v is None:
                 return 0
@@ -504,40 +180,82 @@ async def _run_upload(uid: int) -> None:
                 except Exception:
                     return 0
             if isinstance(v, dict):
-                for k in ("sent", "uploaded", "current", "bytes_sent", "bytes", "done", "value", "progress"):
+                for k in ("current", "sent", "bytes", "done", "downloaded", "uploaded", "processed"):
                     if k in v:
-                        return _as_int(v.get(k))
-                if len(v) == 1:
-                    try:
-                        return _as_int(next(iter(v.values())))
-                    except Exception:
-                        return 0
-                return 0
+                        try:
+                            return int(float(v.get(k) or 0))
+                        except Exception:
+                            continue
+                # common pattern: {"current":..., "total":...}
+                try:
+                    return int(float(next(iter(v.values()))))
+                except Exception:
+                    return 0
             return 0
 
+        # ---- Download (TG -> temp) with live speed ----
+        await safe_edit(st, "⬇️ Downloading…")
+        dl_speed = SpeedETA()
+        dl_last_ui = 0.0
+        dl_start = time.time()
+
+        async def dl_progress(cur: Any, total: Any) -> None:
+            nonlocal dl_last_ui
+            now = time.time()
+            if now - dl_last_ui < 0.8:
+                return
+            dl_last_ui = now
+
+            cur_i = _as_int(cur)
+            total_i = _as_int(total)
+            if total_i <= 0:
+                total_i = max(cur_i, 1)
+
+            rate = dl_speed.update(cur_i, total_i)
+            txt = (
+                "⬇️ Downloading…\\n\\n"
+                f"{human_bytes(cur_i)} / {human_bytes(total_i)}\\n"
+                f"Speed: {human_bytes(rate)}/s\\n"
+                f"ETA: {human_eta(dl_speed.eta_seconds)}"
+            )
+            await safe_edit(st, txt)
+
+        file_path, _, _ = await download_to_temp(pending.downloader, pending.src_msg, progress_cb=dl_progress)
+        dl_dur = max(0.001, time.time() - dl_start)
+        try:
+            file_bytes = int(os.path.getsize(file_path))
+        except Exception:
+            file_bytes = 0
+
+        # ---- Upload (temp -> YouTube) with live speed ----
+        from bot.youtube.uploader import upload_video  # local import
+        ul_speed = SpeedETA()
+        ul_last_ui = 0.0
+        ul_start = time.time()
+
         async def progress_cb(sent: Any, total: Any) -> None:
-            nonlocal last_ui
+            nonlocal ul_last_ui
             now = time.time()
-            if now - last_ui < 0.8:
+            if now - ul_last_ui < 0.8:
                 return
-            last_ui = now
+            ul_last_ui = now
 
             sent_i = _as_int(sent)
             total_i = _as_int(total)
             if total_i <= 0:
                 total_i = max(sent_i, 1)
 
-            rate = speed.update(sent_i, total_i)
+            rate = ul_speed.update(sent_i, total_i)
             txt = (
-                "⬆️ Uploading…\n\n"
-                f"{human_bytes(sent_i)} / {human_bytes(total_i)}\n"
-                f"Speed: {human_bytes(rate)}/s\n"
-                f"ETA: {human_eta(speed.eta_seconds)}"
+                "⬆️ Uploading…\\n\\n"
+                f"{human_bytes(sent_i)} / {human_bytes(total_i)}\\n"
+                f"Speed: {human_bytes(rate)}/s\\n"
+                f"ETA: {human_eta(ul_speed.eta_seconds)}"
             )
             await safe_edit(st, txt)
 
         up = await upload_video(
-            access_token=access,
+            access_token=access_token,
             file_path=file_path,
             title=pending.title,
             description=pending.description,
@@ -547,46 +265,66 @@ async def _run_upload(uid: int) -> None:
 
         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}`")
+            detail = up.get("detail") if isinstance(up, dict) else None
+
+            msg = f"❌ Upload failed: `{err}`"
+            if detail:
+                d = str(detail)
+                msg += f"\n`{d[:280]}`"
+
+                if "uploadLimitExceeded" in d or "quotaExceeded" in d:
+                    msg += "\n\nℹ️ This looks like a YouTube daily upload/quota limit. Try another profile or wait and retry later."
+
+            await safe_edit(st, msg)
             return
 
-        url = up.get("url") or up.get("video_url") or up.get("watch_url") or ""
+        url = up.get("url") or ""
+
         await record_upload(uid, prof_id)
 
-        dur = max(1.0, time.time() - start_t)
+        ul_dur = max(0.001, time.time() - ul_start)
+        total_dur = max(0.001, time.time() - overall_start)
+
+        dl_avg = (human_bytes(file_bytes / dl_dur) + "/s") if file_bytes and dl_dur else "—"
+        ul_avg = (human_bytes(file_bytes / ul_dur) + "/s") if file_bytes and ul_dur else "—"
+
         done = "✅ Uploaded!"
         if url:
             done += f"\n\n{url}"
-        done += f"\n\nTime: {dur:.1f}s"
+        done += (
+            f"\n\nDownload: {dl_dur:.1f}s • avg {dl_avg}"
+            f"\nUpload: {ul_dur:.1f}s • avg {ul_avg}"
+            f"\nTotal: {total_dur:.1f}s"
+        )
         await safe_edit(st, done)
 
-        _PENDING_UPLOAD.pop(uid, None)
-        _AWAIT_EDIT.pop(uid, None)
-
+    except asyncio.CancelledError:
+        # /cancel pressed
+        try:
+            await safe_edit(st, CANCELLED, reply_markup=main_menu_keyboard())
+        except Exception:
+            pass
     except Exception as e:
         await safe_edit(st, f"❌ Error: `{str(e)[:180]}`")
     finally:
+        _PENDING_UPLOAD.pop(uid, None)
+        _AWAIT_EDIT.pop(uid, None)
         if file_path:
             try:
                 cleanup_file(file_path)
             except Exception:
                 pass
         _IN_PROGRESS.pop(uid, None)
+        _UPLOAD_TASK.pop(uid, None)
 
 
-# =========================
-# Handlers
-# =========================
-
-def setup_handlers(app: Client, user_app: Optional[Client] = None) -> None:
-    @app.on_message(filters.command("start") & filters.private)
+def setup_handlers(app: Client, user_app: Optional[Client]) -> None:
+    # ---- basic ----
+    @app.on_message(filters.command(["start", "help"]) & filters.private)
     async def start_cmd(_, m: Message) -> None:
+        from bot.ui.texts import START_TEXT
         await safe_reply(m, START_TEXT, reply_markup=main_menu_keyboard())
 
-    @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())
-
     @app.on_message(filters.command("cancel") & filters.private)
     async def cancel_cmd(_, m: Message) -> None:
         uid = m.from_user.id
@@ -594,38 +332,38 @@ def setup_handlers(app: Client, user_app: Optional[Client] = None) -> None:
         _AWAIT_EDIT.pop(uid, None)
         _PENDING_UPLOAD.pop(uid, None)
         _PENDING_DELETE.pop(uid, None)
+
+        # stop running upload task (if any)
+        task = _UPLOAD_TASK.pop(uid, None)
+        if task and not task.done():
+            task.cancel()
+
+        btask = _BATCH_TASK.pop(uid, None)
+        if btask and not btask.done():
+            btask.cancel()
+
+        _IN_PROGRESS.pop(uid, None)
+
         await safe_reply(m, CANCELLED, reply_markup=main_menu_keyboard())
 
-    # ✅ diag (owner/admin) - fixes “command not working” perception
-    @app.on_message(filters.command("diag") & filters.private)
-    async def diag_cmd(_, m: Message) -> None:
-        uid = m.from_user.id
-        if not _is_admin_or_owner(uid):
-            await safe_reply(m, OWNER_ONLY)
-            return
-        hf = await _hf_list_profiles(uid, only_connected=False)
-        txt = {
-            "ok": True,
-            "worker1_url": Workers.WORKER1_URL,
-            "worker2_url": Workers.WORKER2_URL,
-            "hf_list_profiles_ok": bool(isinstance(hf, dict) and hf.get("ok")),
-            "hf_err": hf.get("err") if isinstance(hf, dict) else "bad_resp",
-        }
-        await safe_reply(m, json.dumps(txt, indent=2)[:3900], reply_markup=main_menu_keyboard())
-
-    # owner/admin
+    # ---- admin allowlist ----
     @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():
+        args = (m.text or "").split(maxsplit=1)
+        if len(args) < 2:
             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])
+        try:
+            tid = int(args[1].strip())
+        except Exception:
+            await safe_reply(m, "Invalid tg_id")
+            return
+        await allow_user(tid)
+        await safe_reply(m, f"✅ Allowed `{tid}`")
 
     @app.on_message(filters.command("disallow") & filters.private)
     async def disallow_cmd(_, m: Message) -> None:
@@ -633,12 +371,17 @@ def setup_handlers(app: Client, user_app: Optional[Client] = None) -> None:
         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():
+        args = (m.text or "").split(maxsplit=1)
+        if len(args) < 2:
             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])
+        try:
+            tid = int(args[1].strip())
+        except Exception:
+            await safe_reply(m, "Invalid tg_id")
+            return
+        await disallow_user(tid)
+        await safe_reply(m, f"✅ Disallowed `{tid}`")
 
     @app.on_message(filters.command("stats") & filters.private)
     async def stats_cmd(_, m: Message) -> None:
@@ -646,347 +389,346 @@ def setup_handlers(app: Client, user_app: Optional[Client] = None) -> None:
         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])
+        st = await get_stats()
+        if not isinstance(st, dict):
+            await safe_reply(m, "❌ stats failed")
+            return
+        await safe_reply(
+            m,
+            "📊 *Today stats*\n\n"
+            f"Allowed users: `{st.get('allowed_users', 0)}`\n"
+            f"Profiles: `{st.get('profiles', 0)}`\n"
+            f"Uploads today: `{st.get('uploads_today', 0)}`",
+        )
+
+    @app.on_message(filters.command("diag") & filters.private)
+    async def diag_cmd(_, m: Message) -> None:
+        uid = m.from_user.id
+        if not _is_admin_or_owner(uid):
+            await safe_reply(m, OWNER_ONLY)
+            return
 
-    # profiles/auth
+        w1 = Workers.WORKER1_URL or "—"
+        w2 = Workers.WORKER2_URL or "—"
+
+        await safe_reply(
+            m,
+            "🧪 *Diag*\n\n"
+            f"Uptime: `{uptime_text()}`\n"
+            f"WORKER1_URL: `{w1}`\n"
+            f"WORKER2_URL: `{w2}`\n"
+            f"Owners: `{len(Auth.OWNERS)}` Admins: `{len(Auth.ADMINS)}`",
+        )
+
+    # ---- auth/profile flow ----
     @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())
+        uid = m.from_user.id
+        _AWAIT_AUTH_MODE[uid] = "oauth"
+        await safe_reply(m, "🔐 Add a profile:", reply_markup=auth_menu_keyboard())
 
     @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
-        await _show_profiles(m, uid, edit=False)
+        data = await list_profiles(uid, only_connected=False)
+        if not (isinstance(data, dict) and data.get("ok")):
+            await safe_reply(m, "❌ Failed to list profiles.")
+            return
+        profiles = data.get("profiles") or []
+        default_id = data.get("default_profile_id") or ""
+        await safe_reply(m, "👤 *Profiles*", reply_markup=profiles_keyboard(profiles, default_id))
 
-    # incoming media
-    @app.on_message((filters.video | filters.document) & filters.private)
-    async def incoming_media(_, m: Message) -> None:
+    # ---- upload from DM media ----
+    @app.on_message(filters.private & (filters.video | filters.document))
+    async def media_in_dm(_, 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.")
+            await safe_reply(m, "⏳ Upload already running. Use /cancel to stop.")
             return
-        await _start_pending_upload(chat_msg=m, src_msg=m, downloader=app, via_link=False)
+        await _start_pending_upload(uid, m, app, via_link=False)
 
-    # link/archive mode (owner/admin only)
+    # ---- upload from link (admin/owner) ----
     @app.on_message(filters.command(["yt", "dl", "archive"]) & filters.private)
     async def archive_cmd(_, m: Message) -> None:
+        uid = m.from_user.id
         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
+        if user_app is None:
+            await safe_reply(m, "❌ This mode requires user session (user_app missing).")
+            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).")
+            await safe_reply(m, "Usage: /archive <t.me message link>")
             return
-
         try:
             chat_ref, msg_id = parse_telegram_link(args[1].strip())
         except Exception as e:
-            await safe_reply(m, f"❌ Bad link: {e}")
+            await safe_reply(m, f"Bad link: {e}")
+            return
+        st = await safe_reply(m, "🔎 Fetching message…")
+        if not st:
             return
-
-        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]}`")
+            await safe_edit(st, f"❌ Fetch failed: `{str(e)[:180]}`")
             return
+        await safe_edit(st, "✅ Message fetched. Preparing preview…")
+        await _start_pending_upload(uid, src, user_app, via_link=True)
 
-        media, _, _ = _media_and_filename(src)
-        if not media:
-            await safe_reply(m, "❌ That message has no video/document.")
+    # batch mode (owner/admin only): /batch <t.me links...> (one per line)
+    @app.on_message(filters.command("batch") & filters.private)
+    async def batch_cmd(_, m: Message) -> None:
+        if not await _ensure_allowed(m):
             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, value = parse_cb(q.data or "")
-
-        try:
-            await q.answer()
-        except Exception:
-            pass
-
-        if action == BACK:
-            await safe_edit(q.message, START_TEXT, reply_markup=main_menu_keyboard())
+        uid = m.from_user.id
+        if not _is_admin_or_owner(uid):
+            await safe_reply(m, OWNER_ONLY)
+            return
+        if user_app is None:
+            await safe_reply(m, "❌ Batch mode is not configured (user session missing).")
             return
 
-        if action == MENU_HELP:
-            await safe_edit(q.message, HELP_TEXT, reply_markup=main_menu_keyboard())
+        args = (m.text or "").split(maxsplit=1)
+        if len(args) < 2:
+            await safe_reply(m, "Send: /batch <t.me message links> (one per line)")
             return
 
-        if action == MENU_AUTH:
-            if not await require_allowed(uid) and not _is_admin_or_owner(uid):
-                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())
+        links = [ln.strip() for ln in args[1].splitlines() if ln.strip()]
+        if not links:
+            await safe_reply(m, "No links found. Put one t.me link per line after /batch")
             return
 
-        if action == MENU_PROFILES:
-            if not await require_allowed(uid) and not _is_admin_or_owner(uid):
-                await safe_edit(q.message, NOT_ALLOWED, reply_markup=main_menu_keyboard())
-                return
-            await _show_profiles(q.message, uid, edit=True)
+        if _BATCH_TASK.get(uid) and not _BATCH_TASK[uid].done():
+            await safe_reply(m, "⏳ A batch is already running. Use /cancel to stop it.")
             return
 
-        if action == MENU_SPEEDTEST:
-            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
+        await safe_reply(m, f"🧾 Batch starting: {len(links)} item(s).")
+
+        async def runner() -> None:
+            batch_start = time.time()
+            for i, link in enumerate(links, 1):
+                try:
+                    chat_ref, msg_id = parse_telegram_link(link)
+                except Exception as e:
+                    await safe_reply(m, f"❌ {i}/{len(links)} bad link: {e}")
+                    continue
 
-            from bot.core.speedtest import ping_ms, net_download_test, net_upload_test, disk_total_free
+                st = await safe_reply(m, f"🔎 Batch {i}/{len(links)}: fetching…")
+                if not st:
+                    continue
 
-            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`"
+                try:
+                    src = await user_app.get_messages(chat_ref, msg_id)
+                except Exception as e:
+                    await safe_edit(st, f"❌ Batch {i}/{len(links)} fetch failed: `{str(e)[:180]}`")
+                    continue
+
+                media, file_name, _ = _media_and_filename(src)
+                if not media:
+                    await safe_edit(st, f"❌ Batch {i}/{len(links)}: no video/document in that message.")
+                    continue
+
+                title, desc = extract_title_description(src, file_name)
+
+                _PENDING_UPLOAD[uid] = PendingUpload(
+                    src_msg=src,
+                    downloader=user_app,
+                    file_name=file_name,
+                    title=title,
+                    description=desc,
+                    privacy="private",
+                    status_msg=st,
+                    via_link=True,
                 )
-            except Exception as e:
-                txt = f"❌ Speed test error: `{str(e)[:180]}`"
-            await safe_edit(q.message, txt, reply_markup=main_menu_keyboard())
-            return
 
-        # ===== Profiles actions (fixed) =====
-        if action == "pref":
-            await _show_profiles(q.message, uid, edit=True)
+                await safe_edit(st, f"⏳ Batch {i}/{len(links)}: starting upload…")
+                await _run_upload(uid)
+
+            batch_dur = max(0.001, time.time() - batch_start)
+            await safe_reply(m, f"✅ Batch done in {batch_dur:.1f}s.")
+
+        t = asyncio.create_task(runner())
+        _BATCH_TASK[uid] = t
+
+    # callbacks
+    @app.on_callback_query(filters.private)
+    async def cb_handler(_, q: CallbackQuery) -> None:
+        uid = q.from_user.id
+        action, value = parse_cb(q.data or "")
+        # Menus
+        if action == MENU_HELP:
+            await safe_edit(q.message, HELP_TEXT, reply_markup=main_menu_keyboard())
             return
 
-        if action == "pdef":
-            pid = (value or "").strip()
-            if not pid:
+        if action == MENU_SPEED:
+            if not await _ensure_allowed(q.message):
                 return
-            r = await _admin_profile_set_default(uid, pid)
-            if isinstance(r, dict) and r.get("ok"):
-                await _show_profiles(q.message, uid, edit=True)
-            else:
-                err = r.get("err", "unknown") if isinstance(r, dict) else "unknown"
-                await safe_edit(q.message, f"❌ Set default failed: `{err}`", reply_markup=main_menu_keyboard())
+            from bot.core.speedtest import net_download_test, net_upload_test
+            await safe_edit(q.message, "⏱ Running speed test…")
+            dl = await net_download_test()
+            ul = await net_upload_test()
+            try:
+                dl_bps = float(dl.get("bps", 0) or 0)
+            except Exception:
+                dl_bps = 0.0
+            try:
+                ul_bps = float(ul.get("bps", 0) or 0)
+            except Exception:
+                ul_bps = 0.0
+
+            await safe_edit(
+                q.message,
+                "📶 *Speed Test*\n\n"
+                f"Download: `{human_bytes(dl_bps)}/s`\n"
+                f"Upload: `{human_bytes(ul_bps)}/s`",
+                reply_markup=main_menu_keyboard(),
+            )
             return
 
-        if action == "pdel":
-            pid = (value or "").strip()
-            if not pid:
+        if action == MENU_AUTH:
+            if not await _ensure_allowed(q.message):
                 return
-            _PENDING_DELETE[uid] = pid
-            kb = InlineKeyboardMarkup(
-                [
-                    [
-                        InlineKeyboardButton("✅ Yes, delete", callback_data=make("pdelok", pid)),
-                        InlineKeyboardButton("❌ Cancel", callback_data=make("pref")),
-                    ]
-                ]
-            )
-            await safe_edit(q.message, f"⚠️ Delete profile `{pid}` ?", reply_markup=kb)
+            await safe_edit(q.message, "🔐 Add a profile:", reply_markup=auth_menu_keyboard())
             return
 
-        if action == "pdelok":
-            pid = (value or "").strip()
-            if not pid:
+        if action == MENU_PROFILES:
+            if not await _ensure_allowed(q.message):
                 return
-            if _PENDING_DELETE.get(uid) != pid:
-                await _show_profiles(q.message, uid, edit=True)
+            data = await list_profiles(uid, only_connected=False)
+            if not (isinstance(data, dict) and data.get("ok")):
+                await safe_edit(q.message, "❌ Failed to list profiles.", reply_markup=main_menu_keyboard())
                 return
-            _PENDING_DELETE.pop(uid, None)
-            r = await _admin_profile_remove(uid, pid)
-            if isinstance(r, dict) and r.get("ok"):
-                await _show_profiles(q.message, uid, edit=True)
-            else:
-                err = r.get("err", "unknown") if isinstance(r, dict) else "unknown"
-                await safe_edit(q.message, f"❌ Delete failed: `{err}`", reply_markup=main_menu_keyboard())
+            profiles = data.get("profiles") or []
+            default_id = data.get("default_profile_id") or ""
+            await safe_edit(q.message, "👤 *Profiles*", reply_markup=profiles_keyboard(profiles, default_id))
             return
 
-        if action in ("plog", "preauth"):
-            pid = (value or "").strip()
-            if not pid:
+        # Upload buttons
+        if action == UP_PRIV:
+            p = _PENDING_UPLOAD.get(uid)
+            if not p:
                 return
-            force = (action == "preauth")
-            r = await _admin_profile_login_link(uid, pid, force=force)
-            if not (isinstance(r, dict) and r.get("ok") and r.get("login_url")):
-                err = r.get("err", "unknown") if isinstance(r, dict) else "unknown"
-                await safe_edit(q.message, f"❌ Login link failed: `{err}`", reply_markup=main_menu_keyboard())
-                return
-            url = str(r.get("login_url"))
-            kb = InlineKeyboardMarkup(
-                [
-                    [InlineKeyboardButton("🌐 Open Google Login", url=url)],
-                    [InlineKeyboardButton("🔄 Back to profiles", callback_data=make(MENU_PROFILES))],
-                ]
-            )
-            msg = "🔑 Open this link to authorize your YouTube channel."
-            if force:
-                msg = "🔁 Re-auth link generated (fresh consent). Open this link:"
-            await safe_edit(q.message, msg, reply_markup=kb)
+            cycle = {"private": "unlisted", "unlisted": "public", "public": "private"}
+            p.privacy = cycle.get(p.privacy, "private")
+            _PENDING_UPLOAD[uid] = p
+            media, _, size = _media_and_filename(p.src_msg)
+            await safe_edit(q.message, _render_preview(p, size), reply_markup=upload_confirm_keyboard(p.privacy))
             return
 
-        # ===== Auth mode selection =====
-        if action == AUTH_JSON:
-            _AWAIT_AUTH_MODE[uid] = "json"
+        if action == UP_EDIT:
+            p = _PENDING_UPLOAD.get(uid)
+            if not p:
+                return
+            _AWAIT_EDIT[uid] = EditState(title=p.title, description=p.description, privacy=p.privacy)
             await safe_edit(
                 q.message,
-                "Paste credentials JSON:\n`{\"client_id\":\"...\",\"client_secret\":\"...\",\"label\":\"optional\"}`",
-                reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("Back", callback_data=make(BACK))]]),
+                "✍️ Send new title + description like this:\n\n"
+                "Title line\n"
+                "Description lines…\n\n"
+                "(Send only 1 line to change title only)",
             )
             return
 
-        if action == AUTH_CI:
-            _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))]]),
-            )
+        if action == UP_DEL:
+            _PENDING_UPLOAD.pop(uid, None)
+            await safe_edit(q.message, CANCELLED, reply_markup=main_menu_keyboard())
             return
 
-        # ===== Upload controls =====
-        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())
+        if action == UP_GO:
+            p = _PENDING_UPLOAD.get(uid)
+            if not p:
                 return
-
-            pending.status_msg = q.message
-            _PENDING_UPLOAD[uid] = pending
-
-            if action == UP_CANCEL:
-                _PENDING_UPLOAD.pop(uid, None)
-                _AWAIT_EDIT.pop(uid, None)
-                await safe_edit(q.message, CANCELLED, reply_markup=main_menu_keyboard())
+            if _IN_PROGRESS.get(uid):
+                await safe_edit(q.message, "⏳ Upload already running. Use /cancel to stop.", reply_markup=main_menu_keyboard())
                 return
+            p.status_msg = q.message
+            _PENDING_UPLOAD[uid] = p
+            await safe_edit(q.message, "⏳ Starting upload…")
+            t = asyncio.create_task(_run_upload(uid))
+            _UPLOAD_TASK[uid] = t
+            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))
+        # Profile callbacks (from profiles_keyboard)
+        if action == "pdel":
+            if not await _ensure_allowed(q.message):
                 return
+            pid = value
+            _PENDING_DELETE[uid] = pid
+            await safe_edit(q.message, f"❓ Delete profile `{pid}`? Reply YES to confirm.")
+            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",
-                )
+        if action == "pdef":
+            if not await _ensure_allowed(q.message):
                 return
+            pid = value
+            out = await set_default_profile(uid, pid)
+            if isinstance(out, dict) and out.get("ok"):
+                data = await list_profiles(uid, only_connected=False)
+                profiles = (data or {}).get("profiles") or []
+                default_id = (data or {}).get("default_profile_id") or ""
+                await safe_edit(q.message, "✅ Default set.", reply_markup=profiles_keyboard(profiles, default_id))
+            else:
+                await safe_edit(q.message, "❌ Failed to set default.", reply_markup=main_menu_keyboard())
+            return
 
-            if action == UP_GO:
-                await safe_edit(q.message, "⏳ Starting upload…")
-                asyncio.create_task(_run_upload(uid))
+        if action == "plog":
+            if not await _ensure_allowed(q.message):
                 return
-
-    # text router (auth + edit)
-    @app.on_message(filters.text & filters.private)
-    async def text_router(_, m: Message) -> None:
-        uid = m.from_user.id if m.from_user else 0
-        if not uid:
+            pid = value
+            chk = await profile_check_auth(uid, pid)
+            if isinstance(chk, dict) and chk.get("ok") and chk.get("login_url"):
+                await safe_edit(
+                    q.message,
+                    f"🔗 Login URL:\n{chk.get('login_url')}\n\nAfter login, reopen /profiles.",
+                    reply_markup=main_menu_keyboard(),
+                )
+            else:
+                await safe_edit(q.message, "❌ Failed to create login URL.", reply_markup=main_menu_keyboard())
             return
 
-        mode = _AWAIT_AUTH_MODE.get(uid)
-        if mode:
-            if not await _ensure_allowed(m):
-                _AWAIT_AUTH_MODE.pop(uid, None)
-                return
+    # ---- edit text handler ----
+    @app.on_message(filters.private & filters.text)
+    async def text_in_dm(_, m: Message) -> None:
+        uid = m.from_user.id
 
-            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 ""
+        # Confirm delete
+        if uid in _PENDING_DELETE:
+            if (m.text or "").strip().lower() == "yes":
+                pid = _PENDING_DELETE.pop(uid)
+                out = await profile_delete(uid, pid)
+                if isinstance(out, dict) and out.get("ok"):
+                    await safe_reply(m, "✅ Profile deleted.", reply_markup=main_menu_keyboard())
                 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")
-
-                _AWAIT_AUTH_MODE.pop(uid, None)
-
-                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 = resp.get("login_url")
-                if not login_url:
-                    # fallback: tell user to open Profiles and tap Login
-                    await safe_reply(m, "✅ Profile added. Now open /profiles and tap **Login** to authorize.")
-                    return
-
-                await safe_reply(m, f"✅ Profile added.\n\nOpen this link to authorize:\n{login_url}")
-                await safe_reply(m, "After authorizing, open /profiles and set default.", reply_markup=main_menu_keyboard())
-
-            except Exception as e:
-                await safe_reply(m, f"❌ Parse error: `{str(e)[:180]}`\n\nTry /auth again.")
+                    await safe_reply(m, "❌ Delete failed.", reply_markup=main_menu_keyboard())
+            else:
+                _PENDING_DELETE.pop(uid, None)
+                await safe_reply(m, "Cancelled.", reply_markup=main_menu_keyboard())
             return
 
-        if _AWAIT_EDIT.get(uid):
-            pending = _PENDING_UPLOAD.get(uid)
-            if not pending:
-                _AWAIT_EDIT.pop(uid, None)
+        # Edit title/desc
+        if uid in _AWAIT_EDIT:
+            st = _AWAIT_EDIT.pop(uid)
+            p = _PENDING_UPLOAD.get(uid)
+            if not p:
                 return
-
-            raw = (m.text or "").strip()
-            if not raw or raw.startswith("/"):
-                return
-
-            title = raw
-            desc = pending.description or ""
-
-            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()
-
-            title = (title or pending.title or "video").strip()[:95]
-            desc = (desc or "").strip()[:4900]
-
-            pending.title = title
-            pending.description = desc
-            _PENDING_UPLOAD[uid] = pending
-            _AWAIT_EDIT.pop(uid, None)
-
-            await safe_edit(pending.status_msg, _render_preview(pending), reply_markup=upload_confirm_keyboard(pending.privacy))
+            txt = (m.text or "").strip()
+            lines = txt.splitlines()
+            if len(lines) == 1:
+                p.title = lines[0].strip()[:Settings.MAX_TITLE]
+            else:
+                p.title = lines[0].strip()[:Settings.MAX_TITLE]
+                p.description = "\n".join(lines[1:]).strip()[:Settings.MAX_DESC]
+            _PENDING_UPLOAD[uid] = p
+            media, _, size = _media_and_filename(p.src_msg)
+            await safe_reply(m, _render_preview(p, size), reply_markup=upload_confirm_keyboard(p.privacy))
             return