bot/app.py

"""Application entry point — FastAPI + Pyrogram in one process."""

from __future__ import annotations

import asyncio
import secrets
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Annotated, AsyncGenerator

import httpx
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse
from fastapi.security import APIKeyHeader
from fastapi.templating import Jinja2Templates
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from loguru import logger
from pyrogram import Client
from starlette.responses import Response as StarletteResponse

from bot.client import create_client
from config import settings
import secrets
import mimetypes
from pathlib import Path

from fastapi.responses import FileResponse

from database.connection import db
from services.queue import get_queue
from services.streamtape_service import _get_accounts, StreamTapeError

import bot.handlers.message_handler  # noqa: F401
import bot.handlers.callback_handler  # noqa: F401


_client: Client | None = None

_templates_dir = Path(__file__).resolve().parent.parent / "templates"
templates = Jinja2Templates(directory=str(_templates_dir))

_api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


async def verify_api_key(
    x_api_key: Annotated[str | None, Depends(_api_key_header)],
) -> None:
    """Validate the X-API-Key header using constant-time comparison."""
    if x_api_key is None:
        raise HTTPException(status_code=401, detail="Missing API key")
    if not secrets.compare_digest(x_api_key, settings.api_key):
        raise HTTPException(status_code=403, detail="Invalid API key")


def get_client() -> Client:
    global _client
    if _client is None:
        _client = create_client()
    return _client


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator:
    """Start Pyrogram client on boot, stop on shutdown."""
    logger.info("🚀 Starting Upload Bot...")
    client = get_client()

    await db.ensure_tables()

    await client.start()
    logger.info("✅ Pyrogram client started.")

    try:
        await client.set_bot_commands([
            {"command": "start", "description": "🔄 بدء / إعادة تعيين البوت"},
        ])
    except Exception as exc:
        logger.warning("Failed to set bot commands: {}", exc)

    # ── Background disk cleanup task ─────────────────────────────────
    async def _periodic_cleanup():
        """ينظف ملفات الرفع القديمة كل 15 دقيقة."""
        from utils.file_utils import smart_disk_cleanup
        from config import settings

        while True:
            try:
                await asyncio.sleep(900)  # 15 دقيقة
                result = await asyncio.to_thread(
                    smart_disk_cleanup,
                    settings.upload_dir,
                    min_free_mb=200,
                    max_age_minutes=120,
                )
                if result["deleted_files"] > 0 or result["deleted_dirs"] > 0:
                    logger.info("Periodic cleanup: {msg}", msg=result["message"])
            except asyncio.CancelledError:
                break
            except Exception as exc:
                logger.warning("Cleanup task error: {}", exc)

    cleanup_task = asyncio.create_task(_periodic_cleanup())

    yield

    cleanup_task.cancel()
    try:
        await cleanup_task
    except asyncio.CancelledError:
        pass

    logger.info("🛑 Shutting down...")
    if client.is_connected:
        await client.stop()


app = FastAPI(
    title="Upload Bot",
    description="Telegram upload bot with MTProto, FFmpeg, StreamTape, Turso",
    version="1.0.0",
    lifespan=lifespan,
)


@app.middleware("http")
async def add_security_headers(request: Request, call_next) -> StarletteResponse:
    """Attach security hardening headers to every response."""
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Content-Security-Policy"] = (
        "default-src 'self'; "
        "script-src 'self' https://unpkg.com https://cdn.tailwindcss.com 'unsafe-inline'; "
        "style-src 'self' 'unsafe-inline'; "
        "img-src 'self' data:; "
        "connect-src 'self'; "
        "frame-ancestors 'none';"
    )
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
    response.headers["Permissions-Policy"] = (
        "camera=(), microphone=(), geolocation=()"
    )
    return response


app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=[
        settings.host,
        "localhost",
        "127.0.0.1",
        "::1",
        "*.onrender.com",
        "*.hf.space",
        "khaled-alim-archiver.hf.space",
    ],
)


@app.get("/")
async def root():
    return {"status": "alive", "bot": "running"}


@app.get("/health")
async def health():
    client = get_client()
    return {
        "status": "ok",
        "pyrogram_connected": client.is_connected,
    }


@app.get("/status", dependencies=[Depends(verify_api_key)])
async def status():
    client = get_client()
    me = await client.get_me()
    return {
        "bot": me.username or me.first_name,
        "connected": client.is_connected,
    }


@app.get("/api/stats", dependencies=[Depends(verify_api_key)])
async def api_stats():
    return await db.get_stats()


@app.get("/api/queue-status", dependencies=[Depends(verify_api_key)])
async def api_queue_status():
    q = get_queue()
    return q.status()


@app.get("/api/streamtape-play/{file_id}")
async def api_streamtape_play(file_id: str):
    """توليد رابط مباشر للتشغيل من معرف ملف StreamTape.

    يستخدم dlticket + dl API لتوليد رابط tapecontent.net مؤقت.
    الرابط صالح لدقائق معدودة — كرر الاستدعاء عند الحاجة.
    """
    from services.streamtape_service import generate_direct_url, _get_accounts, extract_streamtape_file_id

    # استخراج file_id من الرابط إذا كان المستخدم أرسل رابطاً كاملاً
    parsed_id = extract_streamtape_file_id(file_id)
    if not parsed_id:
        return {"error": f"Invalid StreamTape file ID: {file_id}"}
    file_id = parsed_id

    accounts = _get_accounts()
    results: list[dict] = []
    errors: list[str] = []

    for acc in accounts:
        try:
            url = await generate_direct_url(file_id, acc.username, acc.password)
            results.append({
                "account": acc.index,
                "url": url,
                "is_direct": "tapecontent.net" in url,
            })
        except Exception as exc:
            errors.append(f"Account {acc.index}: {exc}")

    return {
        "file_id": file_id,
        "results": results,
        "errors": errors if errors else None,
    }


@app.get("/api/diagnose-streamtape", dependencies=[Depends(verify_api_key)])
async def api_diagnose_streamtape():
    """Test StreamTape API connectivity from this HF Space."""
    import httpx
    from services.streamtape_service import _get_accounts

    # اختبار حساب واحد فقط (الأول) — الاختبارات المتعددة تسبب بطئاً
    results: list[dict] = []
    try:
        acc = _get_accounts()[0]
        async with httpx.AsyncClient(timeout=5) as client:
            resp = await client.get(
                "https://api.streamtape.com/file/info",
                params={"login": acc.username, "key": acc.password, "file": "test"},
            )
            data = resp.json()
            results.append({
                "account": acc.index,
                "api_status": data.get("status"),
                "reachable": data.get("status") == 200,
            })
    except Exception as exc:
        results.append({"account": 1, "error": str(exc)[:100], "reachable": False})

    # اختبار CDN
    cdn_reachable = False
    try:
        async with httpx.AsyncClient(timeout=3) as c:
            r = await c.get("https://tapecontent.net")
            cdn_reachable = r.status_code < 500
    except Exception:
        cdn_reachable = False

    return {
        "streamtape_diagnostic": results,
        "cdn_reachable": cdn_reachable,
        "note": "API ✅ | CDN ❌ محجوب → البوت يحول لـ FTP تلقائياً"
    }


@app.get("/dashboard", response_class=HTMLResponse)
async def dashboard(request: Request):
    stats = await db.get_stats()
    q = get_queue()
    return templates.TemplateResponse(
        "dashboard.html",
        {"request": request, "stats": stats, "queue": q.status()},
    )