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'', 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}}