app.py

import asyncio
import base64
import json
import subprocess
import threading
import time
from urllib.parse import urlparse

from flask import Flask, request, Response
from aiogram import Bot, Dispatcher, F
from aiogram.enums import ChatType
from aiogram.filters import Command, CommandStart
from aiogram.types import FSInputFile, Message
from aiogram.client.session.aiohttp import AiohttpSession
from aiogram.client.telegram import TelegramAPIServer

from agent1 import Config, DB, engine, scheduler, supabase_store

# ╔═══════════════════════════════════════════════════════════════════╗
# ║             🌉 THE CURL BRIDGE (HUGGING FACE BYPASS)             ║
# ╚═══════════════════════════════════════════════════════════════════╝

BRIDGE_PORT = 7860
# We pull your URLs directly from your Config class!
PROXY_TARGET = Config.PROXY_TARGET 
CLOUDFLARE_IP = Config.CLOUDFLARE_IP 

bridge_app = Flask(__name__)

def run_curl(method, url, data=None):
    parsed = urlparse(url)
    domain = parsed.hostname
    
    cmd = [
        "curl", "-X", method, url,
        "--resolve", f"{domain}:443:{CLOUDFLARE_IP}",
        # ⚡️ THE FIX: Full Windows Browser Spoof to bypass Cloudflare Bot Protection
        "-H", "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
        "-H", "Content-Type: application/json",
        "-k", "--max-time", "30"
    ]
    
    input_str = None
    if data:
        cmd.extend(["--data-binary", "@-"])
        input_str = json.dumps(data)

    try:
        # I removed '-s' so curl will actually tell us if it fails
        result = subprocess.run(cmd, input=input_str, capture_output=True, text=True, timeout=35)
        
        if not result.stdout:
            print(f"❌ [CURL FAILED] STDERR: {result.stderr}")
            return json.dumps({"ok": False, "description": f"Empty Curl Response. Error: {result.stderr}", "error_code": 500})
            
        return result.stdout
    except Exception as e:
        print(f"❌ [BRIDGE CRASH] {e}")
        return json.dumps({"ok": False, "description": str(e), "error_code": 500})

@bridge_app.route('/bot<token>/<method>', methods=['POST', 'GET'])
def proxy(token, method):
    real_url = f"{PROXY_TARGET}/bot{token}/{method}"
    data = request.get_json(force=True, silent=True)
    if not data:
        data = request.form.to_dict() or request.args.to_dict()
    response_text = run_curl(request.method, real_url, data)
    return Response(response_text, mimetype='application/json')

def start_bridge():
    print(f"🚀 Starting Bridge on 0.0.0.0:{BRIDGE_PORT}")
    bridge_app.run(host="0.0.0.0", port=BRIDGE_PORT, threaded=True)

# Start the bridge in the background IMMEDIATELY
threading.Thread(target=start_bridge, daemon=True).start()
time.sleep(3)

# ── KEEP YOUR dp = Dispatcher() AND ALL YOUR MESSAGE HANDLERS BELOW THIS LINE ──


dp = Dispatcher()


def is_owner_user(m: Message) -> bool:
    uid = m.from_user.id if m.from_user else 0
    username = ((m.from_user.username or "") if m.from_user else "").lower()
    return Config.is_admin(uid) or username in Config.OWNER_USERNAMES


def user_mode(m: Message) -> str:
    # Owner can force agent for everyone using /agent_on and /agent_off.
    row = DB.q("SELECT value FROM kv WHERE key='public_agent_mode'", fetchone=True)
    public_agent = (row["value"] == "1") if row else False

    if is_owner_user(m):
        return "agent"

    if m.chat.type == ChatType.PRIVATE:
        return "assistant" if not public_agent else "agent"

    # Group chat users can talk when mentioning/replying bot.
    return "assistant" if not public_agent else "agent"


async def collect_attachments(bot: Bot, m: Message):
    out = []

    if m.photo:
        ph = m.photo[-1]
        f = await bot.get_file(ph.file_id)
        data = await bot.download_file(f.file_path)
        raw = data.read()
        out.append({"type": "image", "meta": f"{ph.width}x{ph.height}", "b64": base64.b64encode(raw).decode("utf-8")})

    if m.document:
        f = await bot.get_file(m.document.file_id)
        data = await bot.download_file(f.file_path)
        raw = data.read()
        preview = ""
        if (m.document.file_name or "").lower().endswith((".txt", ".md", ".py", ".json", ".csv", ".log")):
            preview = raw.decode("utf-8", errors="replace")[:4000]
        out.append({"type": "file", "name": m.document.file_name or "document", "preview": preview})

    if m.voice:
        out.append({"type": "audio"})

    return out


async def send_outputs(m: Message, result: dict):
    for p in result.get("screenshots", []) or []:
        if p:
            try:
                await m.answer_photo(FSInputFile(p))
            except Exception:
                pass
    for p in result.get("audio_files", []) or []:
        if p:
            try:
                await m.answer_voice(FSInputFile(p))
            except Exception:
                pass
    for p in result.get("files", []) or []:
        if p:
            try:
                await m.answer_document(FSInputFile(p))
            except Exception:
                pass


@dp.message(CommandStart())
async def start_cmd(m: Message):
    DB.upsert_user(m.from_user.id, m.from_user.username or "", m.from_user.first_name or "")
    buttons = await supabase_store.get_buttons()
    txt = (
        "AgentForge online.\n"
        "Group: reply/mention me to talk.\n"
        "Private: owners have agent mode, others assistant mode by default.\n"
        "Use /alarm <seconds> | <task prompt>."
    )
    if buttons:
        txt += "\n\nStart buttons loaded from Supabase: " + ", ".join([b.get("label", "") for b in buttons[:6]])
    await m.answer(txt)


@dp.message(Command("agent_on"))
async def agent_on_cmd(m: Message):
    if not is_owner_user(m):
        return
    DB.q("CREATE TABLE IF NOT EXISTS kv (key TEXT PRIMARY KEY, value TEXT)")
    DB.q("INSERT INTO kv(key,value) VALUES('public_agent_mode','1') ON CONFLICT(key) DO UPDATE SET value='1'")
    await m.answer("Public agent mode enabled")


@dp.message(Command("agent_off"))
async def agent_off_cmd(m: Message):
    if not is_owner_user(m):
        return
    DB.q("CREATE TABLE IF NOT EXISTS kv (key TEXT PRIMARY KEY, value TEXT)")
    DB.q("INSERT INTO kv(key,value) VALUES('public_agent_mode','0') ON CONFLICT(key) DO UPDATE SET value='0'")
    await m.answer("Public agent mode disabled")


@dp.message(Command("alarm"))
async def alarm_cmd(m: Message):
    text = (m.text or "").replace("/alarm", "", 1).strip()
    if "|" not in text:
        await m.answer("Format: /alarm <seconds> | <task prompt>")
        return
    left, right = [x.strip() for x in text.split("|", 1)]
    try:
        secs = int(left)
    except ValueError:
        await m.answer("Seconds must be integer")
        return

    settings = {"mode": "agent", "is_owner": is_owner_user(m)}
    result = await engine.run(user_id=m.from_user.id, chat_id=m.chat.id, message=f"Schedule an alarm in {secs} seconds and do: {right}", user_settings=settings)
    await m.answer(result.get("text", "Done")[:3900])
    await supabase_store.save_memory(m.from_user.id, (m.from_user.username or ""), "assistant", result.get("text", "")[:3900])


@dp.message(Command("status"))
async def status_cmd(m: Message):
    rows = DB.q(
        "SELECT id,task_prompt,run_at,status,message FROM scheduled_tasks WHERE user_id=? ORDER BY id DESC LIMIT 10",
        (m.from_user.id,),
        fetch=True,
    )
    if not rows:
        await m.answer("No scheduled tasks")
        return
    out = [f"#{r['id']} [{r['status']}] {r['message']} @ {r['run_at']}\n{r['task_prompt'][:120]}" for r in rows]
    await m.answer("\n\n".join(out)[:3900])


@dp.message(F.video)
async def handle_video(m: Message):
    if not is_owner_user(m):
        await m.answer("Only owners can upload videos for YouTube publishing")
        return

    if not Config.ENABLE_YOUTUBE_UPLOAD:
        await m.answer("YouTube upload disabled by config")
        return

    await m.answer("Downloading video...")
    bot = m.bot
    file = await bot.get_file(m.video.file_id)
    path = f"{Config.DATA_DIR}/video_{m.video.file_id}.mp4"
    await bot.download_file(file.file_path, path)

    settings = {"mode": "agent", "is_owner": True}
    prompt = f"Upload this local file to youtube_upload tool. file_path={path}. title=Bot Upload {m.date.isoformat()} description={(m.caption or 'Uploaded by owner')}"
    result = await engine.run(user_id=m.from_user.id, chat_id=m.chat.id, message=prompt, user_settings=settings)
    await m.answer(result.get("text", "done")[:3900])


@dp.message(F.text | F.photo | F.document | F.voice)
async def on_any_message(m: Message, bot: Bot):
    DB.upsert_user(m.from_user.id, m.from_user.username or "", m.from_user.first_name or "")
    user = DB.get_user(m.from_user.id)
    if user and user["is_banned"]:
        return

    if m.chat.type == ChatType.PRIVATE and not is_owner_user(m):
        await m.answer("Private chat is owner-only. Please use group mention/reply mode.")
        return

    # Group behavior: respond only if replied to bot or mentioned.
    if m.chat.type in (ChatType.GROUP, ChatType.SUPERGROUP):
        me = await bot.get_me()
        mentioned = bool(m.text and f"@{(me.username or '').lower()}" in m.text.lower())
        replied = bool(m.reply_to_message and m.reply_to_message.from_user and m.reply_to_message.from_user.id == me.id)
        if not (mentioned or replied):
            return

    mode = user_mode(m)
    settings = {
        "preferred_model": user["preferred_model"] if user else Config.DEFAULT_MODEL,
        "system_prompt": user["system_prompt"] if user else "",
        "temperature": user["temperature"] if user else 0.7,
        "mode": mode,
        "is_owner": is_owner_user(m),
    }

    text = m.text or m.caption or "Analyze uploaded content and respond clearly."
    attachments = await collect_attachments(bot, m)

    # If direct chat and unknown user, assistant behavior only.
    if m.chat.type == ChatType.PRIVATE and not is_owner_user(m) and mode == "assistant":
        settings["system_prompt"] = (settings.get("system_prompt", "") + "\nYou are assistant only for this user.").strip()

    uname = (m.from_user.username or "")
    await supabase_store.save_memory(m.from_user.id, uname, "user", text)
    result = await engine.run(user_id=m.from_user.id, chat_id=m.chat.id, message=text, attachments=attachments, user_settings=settings)
    await m.answer(result.get("text", "Done")[:3900])
    await supabase_store.save_memory(m.from_user.id, (m.from_user.username or ""), "assistant", result.get("text", "")[:3900])
    await send_outputs(m, result)

    if result.get("_restart") and is_owner_user(m):
        await m.answer("Restart requested by tool. Exiting process for supervisor restart.")
        raise SystemExit(0)


async def notify_boss_messages(bot: Bot):
    while True:
        try:
            rows = DB.q("SELECT id,sender_username,sender_id,content FROM boss_messages WHERE notified=0 ORDER BY id ASC LIMIT 20", fetch=True)
            if rows:
                for admin_id in Config.ADMIN_IDS:
                    for r in rows:
                        txt = f"📨 Boss message #{r['id']} from @{r['sender_username'] or 'unknown'} ({r['sender_id']}):\n{r['content'][:3500]}"
                        try:
                            await bot.send_message(admin_id, txt)
                        except Exception:
                            pass
                DB.q("UPDATE boss_messages SET notified=1 WHERE notified=0")
        except Exception:
            pass
        await asyncio.sleep(20)


async def main():
    from aiogram.client.default import DefaultBotProperties
    
    if not Config.BOT_TOKEN:
        raise RuntimeError("BOT_TOKEN is missing")
    
    # Connect the bot to the local Flask bridge we just built
    custom_session = AiohttpSession(
        api=TelegramAPIServer.from_base(f"http://127.0.0.1:{BRIDGE_PORT}")
    )
    
    # ⚡️ THE FIX: aiogram 3.7.0+ syntax for parse_mode ⚡️
    bot = Bot(
        token=Config.BOT_TOKEN, 
        session=custom_session, 
        default=DefaultBotProperties(parse_mode="HTML")
    )
    
    scheduler.set_bot(bot)
    await scheduler.start()
    asyncio.create_task(notify_boss_messages(bot))
    
    print("🤖 BRUKGUARDIAN AGENTFORGE IS STARTING...")
    try:
        await bot.delete_webhook(drop_pending_updates=True)
    except Exception:
        pass

    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())