Update bot/handlers.py

bot/handlers.py

@@ -1,31 +1,233 @@
 # PATH: bot/handlers.py
+import asyncio
 from hydrogram import Client, filters
-from hydrogram.types import Message
+from hydrogram.types import Message, CallbackQuery
+
 from bot.config import Telegram
+from bot.ui import texts
+from bot.ui.keyboards import profiles_keyboard
+from bot.ui.callbacks import parse_cb
+
+from bot.core.auth import is_owner_id, require_allowed
+from bot.core.progress import SpeedETA, human_bytes, human_eta
+from bot.core.tasks import create_task, set_task
+
+from bot.integrations.cf_worker1 import profile_add, profile_list, profile_set_default
+from bot.integrations.cf_worker2 import allow_user, disallow_user, pick_profile, access_token, record_upload, stats_today
+
+from bot.telegram.media import download_to_temp
+from bot.telegram.parse import extract_media_message, extract_title_description
+from bot.telegram.replies import safe_reply, safe_edit
+from bot.temp.files import cleanup_file
+
+from bot.youtube.uploader import upload_video
 
 def setup_handlers(app: Client) -> None:
-    @app.on_message(filters.command(["start"]))
+
+    @app.on_message(filters.command(["start"]) & filters.private)
     async def start_handler(_: Client, m: Message):
-        await m.reply_text(
-            "✅ Online.\n"
-            "Commands:\n"
-            "/ping\n"
-            "/me\n"
-        )
-
-    @app.on_message(filters.command(["ping"]))
+        await safe_reply(m, texts.START_TEXT)
+
+    @app.on_message(filters.command(["ping"]) & filters.private)
     async def ping_handler(_: Client, m: Message):
-        await m.reply_text("🏓 Pong!")
+        await safe_reply(m, "🏓 Pong!")
 
-    @app.on_message(filters.command(["me"]))
+    @app.on_message(filters.command(["me"]) & filters.private)
     async def me_handler(_: Client, m: Message):
         uid = m.from_user.id if m.from_user else None
-        await m.reply_text(f"👤 Your ID: `{uid}`", quote=True)
-
-    @app.on_message(filters.private & filters.text & ~filters.command(["start", "ping", "me"]))
-    async def echo_handler(_: Client, m: Message):
-        # Normal simple reply bot
-        txt = (m.text or "").strip()
-        if not txt:
-            return
-        await m.reply_text(f"🗣 You said: {txt}")
+        await safe_reply(m, f"👤 Your ID: `{uid}`")
+
+    # -------- OWNER allow/disallow via forward OR id --------
+    @app.on_message(filters.command(["allow"]) & filters.private)
+    async def allow_cmd(_: Client, m: Message):
+        if not is_owner_id(m.from_user.id if m.from_user else None):
+            return await safe_reply(m, texts.OWNER_ONLY)
+
+        # If forward present -> use forwarded user id
+        target = None
+        if m.reply_to_message and m.reply_to_message.from_user:
+            target = m.reply_to_message.from_user.id
+        elif m.forward_from:
+            target = m.forward_from.id
+        else:
+            parts = (m.text or "").split()
+            if len(parts) >= 2 and parts[1].isdigit():
+                target = int(parts[1])
+
+        if not target:
+            return await safe_reply(m, "Usage:\n• Reply/forward user msg then /allow\n• or /allow <user_id>")
+
+        j = await allow_user(target)
+        await safe_reply(m, f"✅ allowed: `{target}`\n{j}")
+
+    @app.on_message(filters.command(["disallow"]) & filters.private)
+    async def disallow_cmd(_: Client, m: Message):
+        if not is_owner_id(m.from_user.id if m.from_user else None):
+            return await safe_reply(m, texts.OWNER_ONLY)
+
+        parts = (m.text or "").split()
+        if len(parts) < 2 or not parts[1].isdigit():
+            return await safe_reply(m, "Usage: /disallow <user_id>")
+
+        target = int(parts[1])
+        j = await disallow_user(target)
+        await safe_reply(m, f"✅ disallowed: `{target}`\n{j}")
+
+    # -------- Owner stats --------
+    @app.on_message(filters.command(["stats"]) & filters.private)
+    async def stats_cmd(_: Client, m: Message):
+        if not is_owner_id(m.from_user.id if m.from_user else None):
+            return await safe_reply(m, texts.OWNER_ONLY)
+        j = await stats_today()
+        await safe_reply(m, f"📊 Stats:\n`{j}`")
+
+    # -------- Auth flow (Worker1) --------
+    @app.on_message(filters.command(["auth"]) & filters.private)
+    async def auth_cmd(_: Client, m: Message):
+        uid = m.from_user.id if m.from_user else 0
+        ok = await require_allowed(uid)
+        if not ok:
+            return await safe_reply(m, texts.NOT_ALLOWED)
+
+        # expects: /auth <client_id> <client_secret>
+        parts = (m.text or "").split(maxsplit=2)
+        if len(parts) < 3:
+            return await safe_reply(m, "Usage:\n/auth <CLIENT_ID> <CLIENT_SECRET>\n(baad me JSON method add kar denge)")
+
+        client_id = parts[1].strip()
+        client_secret = parts[2].strip()
+
+        j = await profile_add(uid, client_id, client_secret, label="main", ttl_sec=600)
+        if not j.get("ok"):
+            return await safe_reply(m, f"❌ profile_add failed: `{j}`")
+
+        await safe_reply(m, texts.SENT_AUTH_LINK + j["login_url"])
+
+    @app.on_message(filters.command(["profiles"]) & filters.private)
+    async def profiles_cmd(_: Client, m: Message):
+        uid = m.from_user.id if m.from_user else 0
+        ok = await require_allowed(uid)
+        if not ok:
+            return await safe_reply(m, texts.NOT_ALLOWED)
+
+        j = await profile_list(uid)
+        if not j.get("ok"):
+            return await safe_reply(m, f"❌ profile_list failed: `{j}`")
+
+        profiles = j.get("profiles") or []
+        txt = f"Default: `{j.get('default_profile_id')}`\n\n"
+        for p in profiles:
+            txt += f"- `{p['profile_id']}` | {p.get('label')} | refresh={p.get('has_refresh')} | ch={p.get('channel_title')}\n"
+
+        kb = profiles_keyboard(profiles)
+        await safe_reply(m, txt, reply_markup=kb)
+
+    @app.on_callback_query()
+    async def cb(_: Client, q: CallbackQuery):
+        uid = q.from_user.id if q.from_user else 0
+        if not await require_allowed(uid):
+            return await q.answer("Not allowed", show_alert=True)
+
+        action, value = parse_cb(q.data or "")
+        if action == "setdef" and value:
+            j = await profile_set_default(uid, value)
+            if not j.get("ok"):
+                return await q.answer("Failed", show_alert=True)
+            await q.answer("Default set ✅", show_alert=False)
+            try:
+                await q.message.edit_text(f"✅ Default set: `{value}`")
+            except Exception:
+                pass
+        else:
+            await q.answer("OK")
+
+    # -------- Upload handler (send video/document) --------
+    @app.on_message(filters.private & (filters.video | filters.document))
+    async def upload_handler(app_: Client, m: Message):
+        uid = m.from_user.id if m.from_user else 0
+        ok = await require_allowed(uid)
+        if not ok:
+            return await safe_reply(m, texts.NOT_ALLOWED)
+
+        media, kind = extract_media_message(m)
+        if not media:
+            return await safe_reply(m, "Send a video/document.")
+
+        # find default profile + channel_id from worker1 list
+        pl = await profile_list(uid)
+        if not pl.get("ok"):
+            return await safe_reply(m, f"❌ profile_list failed: `{pl}`")
+
+        default_id = pl.get("default_profile_id")
+        profiles = pl.get("profiles") or []
+        dp = next((p for p in profiles if p.get("profile_id") == default_id), None)
+        if not dp or not dp.get("has_refresh"):
+            return await safe_reply(m, texts.NEED_AUTH)
+
+        channel_id = dp.get("channel_id")
+        if not channel_id:
+            return await safe_reply(m, "❌ channel_id missing. Re-auth profile.")
+
+        # pick profile (rotation same channel only)
+        pick = await pick_profile(uid, channel_id)
+        if not pick.get("ok"):
+            return await safe_reply(m, f"❌ pick_profile failed: `{pick}`")
+        use_profile_id = pick["profile_id"]
+
+        tok = await access_token(uid, use_profile_id)
+        if not tok.get("ok"):
+            return await safe_reply(m, f"❌ access_token failed: `{tok}`")
+        access_tok = tok["access_token"]
+
+        # status message
+        status = await safe_reply(m, texts.UPLOAD_START)
+
+        task_id = f"{uid}:{m.id}"
+        create_task(task_id, uid)
+        set_task(task_id, "downloading", "")
+
+        file_path = ""
+        try:
+            # download
+            file_path, file_size, file_name = await download_to_temp(app_, m)
+
+            # metadata rules
+            title, desc = extract_title_description(m, file_name)
+
+            # upload progress updates
+            se = SpeedETA()
+            last_edit = 0.0
+
+            async def prog(done: int, total: int):
+                nonlocal last_edit
+                snap = se.update(done, total)
+                now = snap["done"]
+                if not status:
+                    return
+                # throttle edit
+                import time
+                if time.time() - last_edit < 2.0 and done < total:
+                    return
+                last_edit = time.time()
+                txt = (
+                    f"{texts.UPLOAD_TO_YT}\n"
+                    f"{human_bytes(now)}/{human_bytes(total)}\n"
+                    f"speed: {human_bytes(snap['speed_bps'])}/s | eta: {human_eta(snap['eta_sec'])}"
+                )
+                await safe_edit(status, txt)
+
+            set_task(task_id, "uploading", "")
+            yt_url = await upload_video(access_tok, file_path, title, desc, privacy="private", progress_cb=prog)
+
+            # record upload counter
+            await record_upload(uid, use_profile_id)
+
+            set_task(task_id, "done", "", yt_url=yt_url)
+            await safe_reply(m, f"{texts.DONE}\n\n🎬 {title}\n🔗 {yt_url}")
+
+        except Exception as e:
+            set_task(task_id, "error", str(e))
+            await safe_reply(m, f"❌ Upload failed: `{type(e).__name__}: {e}`")
+        finally:
+            if file_path:
+                cleanup_file(file_path)