Update bot/handlers.py

bot/handlers.py

@@ -8,6 +8,7 @@ import asyncio
 from dataclasses import dataclass
 from typing import Any, Dict, Optional, Tuple, Union
 
+import httpx
 from hydrogram import Client, filters
 from hydrogram.types import (
     Message,
@@ -16,17 +17,14 @@ from hydrogram.types import (
     InlineKeyboardButton,
 )
 
+from bot.config import Workers
 from bot.core.auth import require_allowed, is_owner_id
 from bot.core.progress import SpeedETA, human_bytes, human_eta
 
-# ✅ profile_* are ADMIN routes => cf_worker1
-from bot.integrations.cf_worker1 import (
-    profile_add,
-    profile_list,
-    profile_set_default,
-)
+# ✅ profile_add is ADMIN route => cf_worker1 (Pages, BOT_BACKEND_KEY)
+from bot.integrations.cf_worker1 import profile_add
 
-# ✅ HF/private routes => cf_worker2
+# ✅ HF/private routes => cf_worker2 (Pages, HF_API_KEY)
 from bot.integrations.cf_worker2 import (
     pick_profile,
     access_token,
@@ -44,7 +42,6 @@ from bot.temp.files import cleanup_file
 from bot.ui.keyboards import (
     main_menu_keyboard,
     auth_menu_keyboard,
-    profiles_keyboard,
     upload_confirm_keyboard,
 )
 from bot.ui.callbacks import (
@@ -71,7 +68,6 @@ from bot.ui.texts import (
     NEED_AUTH,
 )
 
-
 # =========================
 # In-memory state (per-user)
 # =========================
@@ -94,6 +90,8 @@ _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
@@ -102,7 +100,7 @@ _SPEED_COOLDOWN_UNTIL: Dict[int, float] = {}    # uid -> unix time
 def _admin_ids_from_env() -> set[int]:
     raw = (os.getenv("ADMIN_IDS") or "").strip()
     out: set[int] = set()
-    for part in raw.split(","):
+    for part in raw.replace(";", ",").replace("|", ",").split(","):
         part = part.strip()
         if not part:
             continue
@@ -166,17 +164,6 @@ def _render_preview(p: PendingUpload) -> str:
     )
 
 
-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:
@@ -215,12 +202,186 @@ async def _ensure_allowed(m: Message) -> bool:
     if not uid:
         await safe_reply(m, NOT_ALLOWED)
         return False
+
+    # ✅ owner/admin must never get blocked by allowlist
+    if _is_admin_or_owner(uid):
+        return True
+
     if not await require_allowed(uid):
         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)},
+    )
+
+
+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)},
+    )
+
+
+# =========================
+# 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())
+        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,
@@ -271,7 +432,8 @@ async def _run_upload(uid: int) -> None:
     file_path = None
 
     try:
-        pl = await profile_list(uid)
+        # 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
@@ -281,11 +443,24 @@ async def _run_upload(uid: int) -> None:
             await safe_edit(st, NEED_AUTH)
             return
 
-        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")
+        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, "❌ Default profile is not connected. Use /profiles → set a connected one as default.")
+            await safe_edit(st, "❌ चुने हुए profile में channel_id missing है. /profiles में Re-auth करो.")
             return
 
         pick = await pick_profile(uid, channel_id)
@@ -387,8 +562,26 @@ def setup_handlers(app: Client, user_app: Optional[Client] = None) -> None:
         _AWAIT_AUTH_MODE.pop(uid, None)
         _AWAIT_EDIT.pop(uid, None)
         _PENDING_UPLOAD.pop(uid, None)
+        _PENDING_DELETE.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
     @app.on_message(filters.command("allow") & filters.private)
     async def allow_cmd(_, m: Message) -> None:
@@ -437,16 +630,7 @@ def setup_handlers(app: Client, user_app: Optional[Client] = None) -> 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),
-        )
+        await _show_profiles(m, uid, edit=False)
 
     # incoming media
     @app.on_message((filters.video | filters.document) & filters.private)
@@ -520,25 +704,17 @@ def setup_handlers(app: Client, user_app: Optional[Client] = None) -> None:
             return
 
         if action == MENU_AUTH:
-            if not await require_allowed(uid):
+            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())
             return
 
         if action == MENU_PROFILES:
-            if not await require_allowed(uid):
+            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
-            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),
-            )
+            await _show_profiles(q.message, uid, edit=True)
             return
 
         if action == MENU_SPEEDTEST:
@@ -575,6 +751,79 @@ def setup_handlers(app: Client, user_app: Optional[Client] = None) -> None:
             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)
+            return
+
+        if action == "pdef":
+            pid = (value or "").strip()
+            if not pid:
+                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())
+            return
+
+        if action == "pdel":
+            pid = (value or "").strip()
+            if not pid:
+                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)
+            return
+
+        if action == "pdelok":
+            pid = (value or "").strip()
+            if not pid:
+                return
+            if _PENDING_DELETE.get(uid) != pid:
+                await _show_profiles(q.message, uid, edit=True)
+                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())
+            return
+
+        if action in ("plog", "preauth"):
+            pid = (value or "").strip()
+            if not pid:
+                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)
+            return
+
+        # ===== Auth mode selection =====
         if action == AUTH_JSON:
             _AWAIT_AUTH_MODE[uid] = "json"
             await safe_edit(
@@ -593,21 +842,7 @@ def setup_handlers(app: Client, user_app: Optional[Client] = None) -> None:
             )
             return
 
-        if action == "setdef":
-            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 controls =====
         if action in (UP_GO, UP_EDIT, UP_PRIV, UP_CANCEL):
             pending = _PENDING_UPLOAD.get(uid)
             if not pending:
@@ -679,13 +914,14 @@ def setup_handlers(app: Client, user_app: Optional[Client] = None) -> None:
                     await safe_reply(m, f"❌ profile_add failed: `{err}`")
                     return
 
-                login_url = _pick_login_url(resp)
+                login_url = resp.get("login_url")
                 if not login_url:
-                    await safe_reply(m, "✅ Profile added, but login link missing from response. Check your Pages backend.")
+                    # 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, send a video/document here to upload.", reply_markup=main_menu_keyboard())
+                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.")