routers/stream.py

from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import Response
from pydantic import BaseModel
import aiohttp
import time
import asyncio
from urllib.parse import quote
from database import fetch_one, fetch_all, execute
from auth import verify_init_data
from config import ACCOUNTS, resolve_proxy_url
from routers.utils import progress_achievement

router = APIRouter()

_stream_cache = {}
_stream_cache_lock = asyncio.Lock()


class ProgressRequest(BaseModel):
    content_type: str
    content_id: int
    season: int = None
    episode: int = None
    position: float
    duration: float
    completed: bool = False


async def _route_to_account_proxy(file_name: str, account_key: str, dataset_name: str) -> dict:
    if not account_key or not dataset_name:
        return {"ok": False, "error": "Missing account or dataset info"}
    proxy_url = resolve_proxy_url(account_key)
    cache_key = f"{dataset_name}|{file_name}"
    encoded = quote(cache_key, safe='')
    stream_url = f"{proxy_url}/stream/{encoded}"
    return {"ok": True, "url": stream_url, "cached": False, "proxy_name": account_key}


@router.get("/movie/{content_id}")
async def get_movie_stream(content_id: int):
    row = await fetch_one(
        "SELECT * FROM archives WHERE media_type='movie' AND media_id = ? AND status='archived' ORDER BY quality DESC LIMIT 1",
        [str(content_id)],
    )
    if not row:
        raise HTTPException(status_code=404, detail="Stream not available")

    file_name = row.get("file_name", "")
    dataset_name = row.get("dataset_name", "")
    account_key = row.get("account_key", "")
    if not file_name:
        raise HTTPException(status_code=404, detail="Archive info incomplete")

    proxy_resp = await _route_to_account_proxy(file_name, account_key, dataset_name)
    if not proxy_resp.get("ok"):
        raise HTTPException(status_code=502, detail=f"Proxy error: {proxy_resp.get('error', 'unknown')}")

    stream_url = proxy_resp.get("url", "")
    return {
        "ok": True,
        "data": {
            "url": stream_url,
            "proxy_cached": proxy_resp.get("cached", False),
            "proxy_name": proxy_resp.get("proxy_name", ""),
            "ttl_remaining": proxy_resp.get("ttl_remaining_seconds", 0),
            "quality": row.get("quality", ""),
            "file_size": row.get("file_size", 0),
            "archive_id": row.get("archive_id", ""),
            "fallback_used": False,
            "expires_at": None,
        },
    }


@router.get("/episode/{content_id}")
async def get_episode_stream(content_id: int, season: int = None, episode: int = None):
    where = "media_type='tv' AND media_id = ? AND status='archived'"
    params = [str(content_id)]
    if season is not None:
        where += " AND season_number = ?"
        params.append(str(season))
    if episode is not None:
        where += " AND episode_number = ?"
        params.append(str(episode))

    row = await fetch_one(
        f"SELECT * FROM archives WHERE {where} ORDER BY quality DESC LIMIT 1", params
    )
    if not row:
        raise HTTPException(status_code=404, detail="Stream not available")

    file_name = row.get("file_name", "")
    dataset_name = row.get("dataset_name", "")
    account_key = row.get("account_key", "")
    if not file_name:
        raise HTTPException(status_code=404, detail="Archive info incomplete")

    proxy_resp = await _route_to_account_proxy(file_name, account_key, dataset_name)
    if not proxy_resp.get("ok"):
        raise HTTPException(status_code=502, detail=f"Proxy error: {proxy_resp.get('error', 'unknown')}")

    stream_url = proxy_resp.get("url", "")
    return {
        "ok": True,
        "data": {
            "url": stream_url,
            "proxy_cached": proxy_resp.get("cached", False),
            "proxy_name": proxy_resp.get("proxy_name", ""),
            "ttl_remaining": proxy_resp.get("ttl_remaining_seconds", 0),
            "quality": row.get("quality", ""),
            "file_size": row.get("file_size", 0),
            "archive_id": row.get("archive_id", ""),
            "fallback_used": False,
            "expires_at": None,
        },
    }


@router.get("/proxy/status")
async def proxy_status():
    results = {}
    for key in [k for k in ACCOUNTS if k != "main"]:
        proxy_url = resolve_proxy_url(key)
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(f"{proxy_url}/health", timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    if resp.status == 200:
                        data = await resp.json()
                        results[key] = {"ok": True, **data}
                    else:
                        results[key] = {"ok": False, "status": resp.status}
        except Exception as e:
            results[key] = {"ok": False, "error": str(e)}
    return {"ok": True, "data": results}


@router.get("/proxy/cache/{account_key:str}")
async def proxy_cache_status(account_key: str):
    if account_key not in ACCOUNTS or account_key == "main":
        raise HTTPException(status_code=404, detail="Account not found")
    proxy_url = resolve_proxy_url(account_key)
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{proxy_url}/list", timeout=aiohttp.ClientTimeout(total=10)) as resp:
                if resp.status == 200:
                    return await resp.json()
                return {"ok": False, "error": f"Proxy returned {resp.status}"}
    except Exception as e:
        raise HTTPException(status_code=502, detail=str(e))


@router.get("/vast")
async def get_vast():
    vast_url = "https://physicaldad.com/d/m.F/zWd/GfN/vYZ/GFUN/EeQmP9su/ZuURlakRPuTGcTxQNtz/Ue1HMdjQkct-NmzvE/3ZNWT/UVzAMXwh"
    try:
        async with aiohttp.ClientSession() as sess:
            async with sess.get(vast_url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                if resp.status == 200:
                    body = await resp.read()
                    if body and len(body) > 100:
                        return Response(content=body, media_type="application/xml", headers={
                            "Access-Control-Allow-Origin": "*",
                            "Cache-Control": "public, max-age=60",
                        })
    except:
        pass
    return Response(
        content=b'<?xml version="1.0"?><VAST version="3.0"></VAST>',
        media_type="application/xml",
        headers={"Access-Control-Allow-Origin": "*"},
    )


@router.post("/progress")
async def report_progress(body: ProgressRequest, request: Request):
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("tma "):
        raise HTTPException(status_code=401, detail="Unauthorized")

    user_data = verify_init_data(auth_header[4:])
    if not user_data:
        raise HTTPException(status_code=401, detail="Invalid init data")

    tg_id = user_data.get("id")
    user = await fetch_one("SELECT id FROM users WHERE user_id = ?", [str(tg_id)])
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    user_db_id = user["id"]

    await execute(
        """INSERT INTO watch_history
           (user_id, tmdb_id, media_type, season_number, episode_number,
            progress_seconds, duration_seconds, watch_count, last_watched_at)
           VALUES (?, ?, ?, ?, ?, ?, ?, 1, datetime('now'))
           ON CONFLICT(user_id, tmdb_id, media_type, COALESCE(season_number, 0), COALESCE(episode_number, 0))
           DO UPDATE SET progress_seconds = ?, duration_seconds = ?,
           watch_count = watch_count + 1, last_watched_at = datetime('now')""",
        [
            str(user_db_id), str(body.content_id), body.content_type,
            str(body.season) if body.season else None, str(body.episode) if body.episode else None,
            str(int(body.position)), str(int(body.duration)),
            str(int(body.position)), str(int(body.duration)),
        ],
    )

    if body.completed:
        try:
            await execute("UPDATE users SET stars_balance = stars_balance + 1 WHERE id = ?", [str(user_db_id)])
            watch_count = await fetch_one("SELECT COUNT(*) as cnt FROM watch_history WHERE user_id=? AND media_type='movie' AND completed=1", [str(user_db_id)])
            if watch_count:
                await progress_achievement(user_db_id, "watching", watch_count["cnt"])
        except Exception:
            pass

    return {"ok": True, "data": {"saved": True}}