Create bot/handlers.py

bot/handlers.py

@@ -0,0 +1,699 @@
+from __future__ import annotations
+
+import os
+import json
+import asyncio
+from typing import Optional
+
+from hydrogram import Client, filters
+from hydrogram.types import (
+    Message,
+    InlineKeyboardMarkup,
+    InlineKeyboardButton,
+    CallbackQuery,
+)
+
+from app.state import STATE, UploadJob
+from app.settings import (
+    ROTATE_AFTER_PER_PROFILE,
+    DEFAULT_PRIVACY,
+    DEFAULT_TITLE_MODE,
+    YOUTUBE_CHUNK_SIZE,
+    PROGRESS_EDIT_EVERY_SEC,
+    MAX_CONCURRENT_UPLOADS,
+    TMP_DIR,
+)
+from app.progress import human_bytes, fmt_eta
+from app.youtube_api import youtube_resumable_upload, _clean_title
+from app.cf_api import CFClient
+from app.config import Config
+
+def _mkdir_tmp():
+    os.makedirs(TMP_DIR, exist_ok=True)
+
+def menu_kb(is_owner: bool):
+    rows = [
+        [InlineKeyboardButton("📦 Accounts", callback_data="menu:accounts"),
+         InlineKeyboardButton("➕ Add Account", callback_data="menu:addacc")],
+        [InlineKeyboardButton("⚙️ Auto Mode", callback_data="menu:automode")],
+    ]
+    if is_owner:
+        rows.append([InlineKeyboardButton("📊 Stats", callback_data="menu:stats")])
+        rows.append([InlineKeyboardButton("✅ Allow User", callback_data="menu:allow"),
+                     InlineKeyboardButton("⛔ Disallow User", callback_data="menu:disallow")])
+    return InlineKeyboardMarkup(rows)
+
+def upload_options_kb(u: UploadJob):
+    # Compact stable format
+    rows = [
+        [
+            InlineKeyboardButton(f"Privacy: {u.privacy}", callback_data=f"u:{u.upload_id}:privacy"),
+            InlineKeyboardButton("Change", callback_data=f"u:{u.upload_id}:privacy_menu"),
+        ],
+        [
+            InlineKeyboardButton(f"Title: {u.title_mode}", callback_data=f"u:{u.upload_id}:title"),
+            InlineKeyboardButton("Change", callback_data=f"u:{u.upload_id}:title_menu"),
+        ],
+        [
+            InlineKeyboardButton("▶️ Start Upload", callback_data=f"u:{u.upload_id}:start"),
+            InlineKeyboardButton("✖️ Cancel", callback_data=f"u:{u.upload_id}:cancel"),
+        ],
+    ]
+    return InlineKeyboardMarkup(rows)
+
+def privacy_kb(upload_id: str):
+    return InlineKeyboardMarkup([
+        [InlineKeyboardButton("private", callback_data=f"u:{upload_id}:set_priv:private"),
+         InlineKeyboardButton("unlisted", callback_data=f"u:{upload_id}:set_priv:unlisted"),
+         InlineKeyboardButton("public", callback_data=f"u:{upload_id}:set_priv:public")],
+        [InlineKeyboardButton("⬅ Back", callback_data=f"u:{upload_id}:back")],
+    ])
+
+def title_kb(upload_id: str):
+    return InlineKeyboardMarkup([
+        [InlineKeyboardButton("filename", callback_data=f"u:{upload_id}:set_title:filename"),
+         InlineKeyboardButton("caption", callback_data=f"u:{upload_id}:set_title:caption")],
+        [InlineKeyboardButton("custom", callback_data=f"u:{upload_id}:set_title:custom")],
+        [InlineKeyboardButton("⬅ Back", callback_data=f"u:{upload_id}:back")],
+    ])
+
+def accounts_kb(profiles: list, tg_id: str, default_profile_id: Optional[str], is_owner: bool):
+    rows = []
+    for p in profiles:
+        pid = p["profile_id"]
+        label = p.get("label") or "profile"
+        ch = p.get("channel_title") or p.get("channel_id") or "no-channel"
+        ok = "✅" if p.get("has_refresh") else "⚠️"
+        d = "⭐" if pid == default_profile_id else ""
+        rows.append([InlineKeyboardButton(f"{ok}{d} {label} — {ch}", callback_data=f"acc:info:{pid}")])
+    if profiles:
+        rows.append([InlineKeyboardButton("⭐ Set Default", callback_data="acc:setdef_menu")])
+        rows.append([InlineKeyboardButton("🗑 Remove Profile", callback_data="acc:remove_menu")])
+    rows.append([InlineKeyboardButton("⬅ Back", callback_data="menu:home")])
+    if is_owner:
+        rows.append([InlineKeyboardButton("List Allowed (if available)", callback_data="acc:list_allowed")])
+    return InlineKeyboardMarkup(rows)
+
+def pick_profile_kb(profiles: list, action: str):
+    # action: "setdef" or "remove"
+    rows = []
+    for p in profiles:
+        pid = p["profile_id"]
+        label = p.get("label") or "profile"
+        ch = p.get("channel_title") or p.get("channel_id") or "no-channel"
+        ok = "✅" if p.get("has_refresh") else "⚠️"
+        rows.append([InlineKeyboardButton(f"{ok} {label} — {ch}", callback_data=f"acc:{action}:{pid}")])
+    rows.append([InlineKeyboardButton("⬅ Back", callback_data="menu:accounts")])
+    return InlineKeyboardMarkup(rows)
+
+def is_owner(cfg: Config, user_id: int) -> bool:
+    return user_id == cfg.OWNER_ID
+
+def _safe_title_from(job: UploadJob) -> str:
+    if job.title_mode == "custom" and job.custom_title.strip():
+        return _clean_title(job.custom_title.strip())
+    if job.title_mode == "filename":
+        base = os.path.splitext(job.file_name)[0]
+        return _clean_title(base)
+    # caption default
+    if job.caption.strip():
+        # first line as title
+        first = job.caption.strip().splitlines()[0]
+        return _clean_title(first)
+    base = os.path.splitext(job.file_name)[0]
+    return _clean_title(base)
+
+async def register_handlers(app: Client, cfg: Config, cf: CFClient):
+
+    # semaphore size from settings
+    STATE.sem = asyncio.Semaphore(MAX_CONCURRENT_UPLOADS)
+
+    @app.on_message(filters.command("start"))
+    async def start_cmd(_, m: Message):
+        u = m.from_user.id
+        kb = menu_kb(is_owner(cfg, u))
+        await m.reply_text(
+            "✅ Bot online.\n\nSend a video to upload.\nUse buttons for accounts/settings.",
+            reply_markup=kb
+        )
+
+    @app.on_callback_query()
+    async def cb(_, q: CallbackQuery):
+        data = (q.data or "")
+        u = q.from_user.id
+        tg_id = str(u)
+
+        if data.startswith("menu:"):
+            act = data.split(":", 1)[1]
+
+            if act == "home":
+                await q.message.edit_text("Main menu:", reply_markup=menu_kb(is_owner(cfg, u)))
+                await q.answer()
+                return
+
+            if act == "accounts":
+                res = await cf.list_profiles_w2(tg_id)
+                profiles = res.get("profiles", [])
+                dpid = res.get("default_profile_id")
+                await q.message.edit_text(
+                    f"📦 Accounts for {tg_id}\nDefault: {dpid}\n\n✅ authorized | ⚠️ not authorized",
+                    reply_markup=accounts_kb(profiles, tg_id, dpid, is_owner(cfg, u))
+                )
+                await q.answer()
+                return
+
+            if act == "addacc":
+                STATE.waiting_client_id[u] = True
+                await q.message.edit_text(
+                    "➕ Add Account\n\nSend:\n1) Google OAuth JSON file (client_secret_*.json)\nOR\n2) client_id text (ends with apps.googleusercontent.com)",
+                    reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅ Back", callback_data="menu:home")]])
+                )
+                await q.answer()
+                return
+
+            if act == "automode":
+                cur = STATE.auto_mode.get(u, False)
+                STATE.auto_mode[u] = not cur
+                v = "ON ✅" if STATE.auto_mode[u] else "OFF ❌"
+                await q.answer(f"Auto Mode: {v}", show_alert=True)
+                # no edit required
+                return
+
+            if act == "stats":
+                if not is_owner(cfg, u):
+                    await q.answer("Owner only", show_alert=True)
+                    return
+                res = await cf.stats_today()
+                if not res.get("ok"):
+                    await q.answer("stats failed", show_alert=True)
+                    return
+                text = (
+                    f"📊 Stats (UTC day {res.get('day')})\n"
+                    f"Uploads today total: {res.get('uploads_today_total')}\n"
+                    f"Active users today: {res.get('active_users_today')}\n\n"
+                    f"Errors last20:\n"
+                )
+                errs = res.get("errors_last20") or []
+                if not errs:
+                    text += "— none —"
+                else:
+                    for e in errs[:20]:
+                        text += f"- {e.get('where')} | {e.get('tg_id')} | {e.get('err')}\n"
+                await q.message.edit_text(text, reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅ Back", callback_data="menu:home")]]))
+                await q.answer()
+                return
+
+            if act == "allow":
+                if not is_owner(cfg, u):
+                    await q.answer("Owner only", show_alert=True)
+                    return
+                await q.message.edit_text(
+                    "✅ Allow User\n\nForward a message from that user.\nIf forward privacy hides id, then send numeric user_id.",
+                    reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅ Back", callback_data="menu:home")]])
+                )
+                # mark mode using waiting_client_secret dict hack? better separate:
+                STATE.waiting_client_secret[u] = "__ALLOW__"
+                await q.answer()
+                return
+
+            if act == "disallow":
+                if not is_owner(cfg, u):
+                    await q.answer("Owner only", show_alert=True)
+                    return
+                await q.message.edit_text(
+                    "⛔ Disallow User\n\nForward a message from that user OR send numeric user_id.",
+                    reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅ Back", callback_data="menu:home")]])
+                )
+                STATE.waiting_client_secret[u] = "__DISALLOW__"
+                await q.answer()
+                return
+
+        # accounts submenus
+        if data == "acc:setdef_menu":
+            res = await cf.list_profiles_w2(tg_id)
+            profiles = res.get("profiles", [])
+            await q.message.edit_text("Pick profile to set default:", reply_markup=pick_profile_kb(profiles, "setdef"))
+            await q.answer()
+            return
+
+        if data == "acc:remove_menu":
+            res = await cf.list_profiles_w2(tg_id)
+            profiles = res.get("profiles", [])
+            await q.message.edit_text("Pick profile to remove:", reply_markup=pick_profile_kb(profiles, "remove"))
+            await q.answer()
+            return
+
+        if data.startswith("acc:setdef:"):
+            pid = data.split(":", 2)[2]
+            r = await cf.profile_set_default(tg_id, pid)
+            await q.answer("Default set ✅" if r.get("ok") else f"Failed: {r.get('err')}", show_alert=True)
+            # back to accounts
+            res = await cf.list_profiles_w2(tg_id)
+            profiles = res.get("profiles", [])
+            dpid = res.get("default_profile_id")
+            await q.message.edit_text("📦 Accounts", reply_markup=accounts_kb(profiles, tg_id, dpid, is_owner(cfg, u)))
+            return
+
+        if data.startswith("acc:remove:"):
+            pid = data.split(":", 2)[2]
+            r = await cf.profile_remove(tg_id, pid)
+            await q.answer("Removed ✅" if r.get("ok") else f"Failed: {r.get('err')}", show_alert=True)
+            res = await cf.list_profiles_w2(tg_id)
+            profiles = res.get("profiles", [])
+            dpid = res.get("default_profile_id")
+            await q.message.edit_text("📦 Accounts", reply_markup=accounts_kb(profiles, tg_id, dpid, is_owner(cfg, u)))
+            return
+
+        # upload callbacks
+        if data.startswith("u:"):
+            parts = data.split(":")
+            upload_id = parts[1]
+            act = parts[2] if len(parts) > 2 else ""
+            job = STATE.uploads.get(upload_id)
+            if not job:
+                await q.answer("This upload expired/restarted.", show_alert=True)
+                return
+            if job.user_id != u:
+                await q.answer("Not yours", show_alert=True)
+                return
+
+            if act == "privacy_menu":
+                await q.message.edit_text("Choose privacy:", reply_markup=privacy_kb(upload_id))
+                await q.answer()
+                return
+            if act == "title_menu":
+                await q.message.edit_text("Choose title mode:", reply_markup=title_kb(upload_id))
+                await q.answer()
+                return
+            if act == "set_priv":
+                val = parts[3]
+                job.privacy = val
+                await q.message.edit_text(
+                    f"📤 Ready\nFile: {job.file_name}\nPrivacy: {job.privacy}\nTitle mode: {job.title_mode}",
+                    reply_markup=upload_options_kb(job)
+                )
+                await q.answer("Saved ✅")
+                return
+            if act == "set_title":
+                val = parts[3]
+                job.title_mode = val
+                if val == "custom":
+                    STATE.waiting_custom_title[u] = upload_id
+                    await q.message.edit_text("Send custom title text now (next message).", reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("⬅ Back", callback_data=f"u:{upload_id}:back")]]))
+                    await q.answer()
+                    return
+                await q.message.edit_text(
+                    f"📤 Ready\nFile: {job.file_name}\nPrivacy: {job.privacy}\nTitle mode: {job.title_mode}",
+                    reply_markup=upload_options_kb(job)
+                )
+                await q.answer("Saved ✅")
+                return
+            if act == "back":
+                await q.message.edit_text(
+                    f"📤 Ready\nFile: {job.file_name}\nPrivacy: {job.privacy}\nTitle mode: {job.title_mode}",
+                    reply_markup=upload_options_kb(job)
+                )
+                await q.answer()
+                return
+            if act == "cancel":
+                STATE.uploads.pop(upload_id, None)
+                await q.message.edit_text("Cancelled ✅", reply_markup=menu_kb(is_owner(cfg, u)))
+                await q.answer()
+                return
+            if act == "start":
+                await q.answer("Starting upload…")
+                asyncio.create_task(run_upload(app, cfg, cf, job))
+                return
+
+        await q.answer()
+
+    @app.on_message(filters.text & filters.private)
+    async def text_in(_, m: Message):
+        u = m.from_user.id
+
+        # custom title input
+        if u in STATE.waiting_custom_title:
+            upload_id = STATE.waiting_custom_title.pop(u)
+            job = STATE.uploads.get(upload_id)
+            if job:
+                job.custom_title = m.text.strip()
+                await m.reply_text(
+                    f"✅ Custom title saved.\nNow press Start Upload for file: {job.file_name}",
+                    reply_markup=upload_options_kb(job)
+                )
+            return
+
+        # owner allow/disallow mode
+        if u in STATE.waiting_client_secret and STATE.waiting_client_secret[u] in ("__ALLOW__", "__DISALLOW__"):
+            mode = STATE.waiting_client_secret.pop(u)
+            target = (m.text or "").strip()
+
+            # if numeric in text
+            tg_id = None
+            if target.isdigit():
+                tg_id = target
+            else:
+                await m.reply_text("Send numeric user_id or forward a message from user (if forward shows id).")
+                return
+
+            if mode == "__ALLOW__":
+                r = await cf.allow_user(tg_id)
+                await m.reply_text("✅ Allowed" if r.get("ok") else f"❌ Failed: {r.get('err')}", reply_markup=menu_kb(True))
+                return
+            else:
+                r = await cf.disallow_user(tg_id)
+                await m.reply_text("⛔ Disallowed" if r.get("ok") else f"❌ Failed: {r.get('err')}", reply_markup=menu_kb(True))
+                return
+
+        # add account flow (client_id then client_secret)
+        if STATE.waiting_client_id.get(u):
+            t = (m.text or "").strip()
+            if "apps.googleusercontent.com" in t:
+                STATE.waiting_client_id.pop(u, None)
+                STATE.waiting_client_secret[u] = t
+                await m.reply_text("Now send CLIENT_SECRET (starts with GOCSPX-...)")
+                return
+            await m.reply_text("Send client_id (apps.googleusercontent.com) or upload Google JSON file.")
+            return
+
+        if u in STATE.waiting_client_secret and STATE.waiting_client_secret[u] and not STATE.waiting_client_secret[u].startswith("__"):
+            client_id = STATE.waiting_client_secret[u]
+            client_secret = (m.text or "").strip()
+            if not client_secret:
+                await m.reply_text("Empty secret. Send again.")
+                return
+            STATE.waiting_client_secret.pop(u, None)
+
+            tg_id = str(u)
+            # create profile + login link
+            r = await cf.profile_add(tg_id, client_id, client_secret, label="main", ttl_sec=600)
+            if not r.get("ok"):
+                await m.reply_text(f"❌ Failed to add profile: {r.get('err')}")
+                return
+            login_url = r.get("login_url")
+            pid = r.get("profile_id")
+            await m.reply_text(
+                f"✅ Profile created: {pid}\n\nOpen login link to authorize:\n(If Telegram preview consumes ticket, just add again for now.)",
+                reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("🔐 Login with Google", url=login_url)]])
+            )
+            return
+
+    @app.on_message(filters.document & filters.private)
+    async def doc_in(_, m: Message):
+        u = m.from_user.id
+        if not STATE.waiting_client_id.get(u):
+            return
+
+        doc = m.document
+        name = (doc.file_name or "").lower()
+        if not name.endswith(".json"):
+            await m.reply_text("Send Google OAuth JSON file (client_secret_*.json).")
+            return
+
+        _mkdir_tmp()
+        path = os.path.join(TMP_DIR, f"cred_{u}.json")
+        await m.download(file_name=path)
+
+        try:
+            with open(path, "r", encoding="utf-8") as f:
+                j = json.load(f)
+            block = j.get("installed") or j.get("web") or {}
+            client_id = block.get("client_id", "")
+            client_secret = block.get("client_secret", "")
+            if not client_id or not client_secret:
+                raise ValueError("missing client_id/client_secret in json")
+        except Exception as e:
+            await m.reply_text(f"❌ Bad JSON: {e}")
+            return
+        finally:
+            try:
+                os.remove(path)
+            except Exception:
+                pass
+
+        STATE.waiting_client_id.pop(u, None)
+        tg_id = str(u)
+        r = await cf.profile_add(tg_id, client_id, client_secret, label="main", ttl_sec=600)
+        if not r.get("ok"):
+            await m.reply_text(f"❌ Failed to add profile: {r.get('err')}")
+            return
+        await m.reply_text(
+            f"✅ Profile created: {r.get('profile_id')}\nAuthorize now:",
+            reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("🔐 Login with Google", url=r.get("login_url"))]])
+        )
+
+    @app.on_message((filters.video | filters.document) & filters.private)
+    async def video_in(_, m: Message):
+        u = m.from_user.id
+        tg_id = str(u)
+
+        # owner always allowed
+        if not is_owner(cfg, u):
+            ok = await cf.is_allowed(tg_id)
+            if not ok:
+                await m.reply_text("⛔ You are not allowed. Ask owner to allow you.")
+                return
+
+        # Determine file
+        file_id = None
+        file_name = "video.mp4"
+        file_type = "video"
+        caption = (m.caption or "")
+
+        if m.video:
+            file_id = m.video.file_id
+            file_name = m.video.file_name or "video.mp4"
+            file_type = "video"
+        elif m.document and (m.document.mime_type or "").startswith("video/"):
+            file_id = m.document.file_id
+            file_name = m.document.file_name or "video.mp4"
+            file_type = "document"
+        else:
+            return
+
+        # Create upload job
+        upload_id = STATE.new_upload_id()
+        job = UploadJob(
+            upload_id=upload_id,
+            user_id=u,
+            chat_id=m.chat.id,
+            src_msg_id=m.id,
+            file_type=file_type,
+            tg_file_id=file_id,
+            file_name=file_name,
+            caption=caption,
+            privacy=DEFAULT_PRIVACY,
+            title_mode=DEFAULT_TITLE_MODE,
+        )
+        STATE.uploads[upload_id] = job
+
+        # Auto mode: start immediately with defaults (bulk-friendly)
+        auto = STATE.auto_mode.get(u, False)
+        if auto:
+            status = await m.reply_text(
+                f"📤 Auto Mode ON\nUploading with defaults…\nFile: {job.file_name}\nPrivacy: {job.privacy}\nTitle mode: {job.title_mode}",
+            )
+            job.status_msg_id = status.id
+            asyncio.create_task(run_upload(app, cfg, cf, job))
+            return
+
+        # Manual mode: show options
+        msg = await m.reply_text(
+            f"📤 Ready\nFile: {job.file_name}\nPrivacy: {job.privacy}\nTitle mode: {job.title_mode}",
+            reply_markup=upload_options_kb(job)
+        )
+        job.status_msg_id = msg.id
+
+async def run_upload(app: Client, cfg: Config, cf: CFClient, job: UploadJob):
+    """
+    Upload pipeline:
+    - check default profile/channel
+    - pick rotated profile for same channel
+    - get access token
+    - download file
+    - resumable upload with progress (speed/ETA)
+    - record upload + cleanup
+    """
+    tg_id = str(job.user_id)
+
+    async with STATE.sem:
+        _mkdir_tmp()
+        local_path = os.path.join(TMP_DIR, f"{job.upload_id}_{job.file_name}")
+        status_msg = None
+        try:
+            status_msg = await app.get_messages(job.chat_id, job.status_msg_id) if job.status_msg_id else None
+        except Exception:
+            status_msg = None
+
+        def _edit(text: str):
+            async def _do():
+                try:
+                    if status_msg:
+                        await status_msg.edit_text(text)
+                except Exception:
+                    pass
+            return _do()
+
+        try:
+            # 1) list profiles to find default + channel
+            profs = await cf.list_profiles_w2(tg_id)
+            if not profs.get("ok"):
+                await _edit(f"❌ Worker2 list_profiles failed: {profs.get('err')}")
+                return
+
+            default_pid = profs.get("default_profile_id")
+            profiles = profs.get("profiles") or []
+            default = next((p for p in profiles if p.get("profile_id") == default_pid), None)
+
+            if not default_pid or not default:
+                await _edit("⚠️ No default profile.\nGo: Accounts → set default.")
+                return
+            if not default.get("has_refresh"):
+                await _edit("⚠️ Default profile not authorized.\nGo: Accounts → choose an authorized profile and set default.")
+                return
+            channel_id = default.get("channel_id")
+            if not channel_id:
+                await _edit("⚠️ Default profile has no channel_id.\nAuthorize again and then set default.")
+                return
+
+            # 2) pick rotated profile for SAME channel
+            pick = await cf.pick_profile(tg_id, channel_id, ROTATE_AFTER_PER_PROFILE)
+            if not pick.get("ok"):
+                await _edit(f"❌ pick_profile failed: {pick.get('err')}")
+                return
+            profile_id = pick.get("profile_id")
+
+            # 3) access token
+            tok = await cf.access_token(tg_id, profile_id)
+            if not tok.get("ok"):
+                await cf.log_error(tg_id, profile_id, "access_token", tok.get("err", "token_failed"))
+                await _edit(f"❌ access_token failed: {tok.get('err')}")
+                return
+            access_token = tok.get("access_token")
+
+            # 4) download file with download progress
+            await _edit("⬇️ Downloading from Telegram…")
+
+            # hydrogram download (no progress hook reliably in all forks),
+            # so we just download and then show upload progress.
+            msg = await app.get_messages(job.chat_id, job.src_msg_id)
+            await msg.download(file_name=local_path)
+
+            size = os.path.getsize(local_path)
+
+            # 5) upload to YouTube with progress edits
+            title = _safe_title_from(job)
+            desc = job.caption.strip()
+
+            last_update = 0.0
+
+            async def progress_upload(done: int, total: int, speed: float, eta: float):
+                nonlocal last_update
+                import time
+                now = time.time()
+                if now - last_update < PROGRESS_EDIT_EVERY_SEC:
+                    return
+                last_update = now
+                pct = (done / max(1, total)) * 100.0
+                text = (
+                    f"⬆️ Uploading to YouTube…\n"
+                    f"File: {job.file_name}\n"
+                    f"Title: {title}\n"
+                    f"Privacy: {job.privacy}\n\n"
+                    f"{pct:.1f}%  ({human_bytes(done)}/{human_bytes(total)})\n"
+                    f"Speed: {human_bytes(speed)}/s\n"
+                    f"ETA: {fmt_eta(eta)}"
+                )
+                await _edit(text)
+
+            await _edit(
+                f"⬆️ Starting YouTube upload…\n"
+                f"File: {job.file_name} ({human_bytes(size)})\n"
+                f"Title: {title}\n"
+                f"Privacy: {job.privacy}"
+            )
+
+            # First attempt
+            http = None
+            try:
+                import httpx
+                http = httpx.AsyncClient(timeout=None)
+                res = await youtube_resumable_upload(
+                    access_token=access_token,
+                    file_path=local_path,
+                    title=title,
+                    description=desc,
+                    privacy=job.privacy,
+                    chunk_size=YOUTUBE_CHUNK_SIZE,
+                    progress_cb=lambda d, t, s, e: asyncio.create_task(progress_upload(d, t, s, e)),
+                    http=http,
+                )
+            finally:
+                if http:
+                    await http.aclose()
+
+            # Retry once if token expired
+            if not res.get("ok") and "401" in str(res.get("err", "")):
+                tok2 = await cf.access_token(tg_id, profile_id)
+                if tok2.get("ok"):
+                    access_token = tok2.get("access_token")
+                    http = None
+                    try:
+                        import httpx
+                        http = httpx.AsyncClient(timeout=None)
+                        res = await youtube_resumable_upload(
+                            access_token=access_token,
+                            file_path=local_path,
+                            title=title,
+                            description=desc,
+                            privacy=job.privacy,
+                            chunk_size=YOUTUBE_CHUNK_SIZE,
+                            progress_cb=lambda d, t, s, e: asyncio.create_task(progress_upload(d, t, s, e)),
+                            http=http,
+                        )
+                    finally:
+                        if http:
+                            await http.aclose()
+
+            if not res.get("ok"):
+                err = res.get("err", "upload_failed")
+                await cf.log_error(tg_id, profile_id, "youtube_upload", err)
+                await _edit(f"❌ Upload failed: {err}")
+                return
+
+            video_id = res.get("video_id")
+            link = f"https://youtu.be/{video_id}" if video_id else "no_link"
+
+            # 6) record upload
+            await cf.record_upload(tg_id, profile_id)
+
+            # 7) final message (stable format)
+            final_text = (
+                f"✅ Uploaded\n"
+                f"Title: {title}\n"
+                f"Privacy: {job.privacy}\n"
+                f"Channel: {tok.get('channel_title') or tok.get('channel_id')}\n"
+                f"Profile: {profile_id}\n"
+                f"Link: {link}"
+            )
+            await _edit(final_text)
+
+        except Exception as e:
+            try:
+                await cf.log_error(tg_id, "", "run_upload_exception", str(e)[:200])
+            except Exception:
+                pass
+            if status_msg:
+                try:
+                    await status_msg.edit_text(f"❌ Crash: {e}")
+                except Exception:
+                    pass
+        finally:
+            # cleanup
+            try:
+                if os.path.exists(local_path):
+                    os.remove(local_path)
+            except Exception:
+                pass
+            # drop job
+            STATE.uploads.pop(job.upload_id, None)