from __future__ import annotations import httpx from datetime import datetime from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware import starlette.exceptions from starlette.exceptions import HTTPException as StarletteHTTPException import asyncio # Logging import logging # Config from app.config.settings import settings # Core Modules from app.core import cache, cache_cleanup, pool, rate_limit_middleware, rate_limit_cleanup # Exception handlers from app.exception_handlers import not_found_handler, internal_error_handler, general_exception_handler # API Routers from app.api.endpoints import recommendations, hls, media, explore, thumbnails # We will define new standardized routers here or import them if we moved them. # For this refactor, we will define them inline or in a new api module. # To keep it clean, I will implement the Router structure within main.py for now, # ensuring they obey the /api/v1/ prefix. from fastapi import APIRouter # Scrapers & Models from app.scrapers import masa49, xhamster, xnxx, xvideos, pornhub, youporn, redtube, beeg, spankbang, fapnut, pornxp, hqporner, xxxparodyhd, pornwex, tube8, pornhat, brazzpw, gosexpod, watcherotic from app.models.schemas import ScrapeResponse, ListItem, CategoryItem, ScrapeRequest, ListRequest logging.basicConfig(level=logging.INFO) # NOTE: Lifespan REMOVED for cPanel/Passenger compatibility. # a2wsgi runs ASGI lifespans synchronously, which blocks the first request. # Background tasks can be added back later if using a true ASGI server like uvicorn. # Create FastAPI app app = FastAPI( title="AppHub API", description="Professional Standard API with Versioning, Plural Naming, and Queue Services", version="2.1.0", docs_url="/docs", redoc_url="/redoc", openapi_url="/openapi.json" ) # Register exception handlers app.add_exception_handler(404, not_found_handler) app.add_exception_handler(500, internal_error_handler) app.add_exception_handler(StarletteHTTPException, general_exception_handler) app.add_exception_handler(HTTPException, general_exception_handler) app.add_middleware( CORSMiddleware, allow_origins=settings.CORS_ORIGINS, allow_credentials=settings.CORS_ALLOW_CREDENTIALS, allow_methods=settings.CORS_ALLOW_METHODS, allow_headers=settings.CORS_ALLOW_HEADERS, ) # Mount static files app.mount("/static", StaticFiles(directory="app/static"), name="static") # app.middleware("http")(rate_limit_middleware) # ============================================================================== # API V1 Router # ============================================================================== api_v1_router = APIRouter(prefix="/api/v1") # --- Scraper / Resources Endpoints --- from pydantic import BaseModel, HttpUrl, field_validator, Field from typing import Any, Optional class ScrapeRequestV1(BaseModel): url: HttpUrl class CrawlRequestV1(BaseModel): base_url: HttpUrl start_page: int = Field(1, ge=1) max_pages: int = Field(5, ge=1, le=20) per_page_limit: int = Field(0, ge=0, le=200) max_items: int = Field(500, ge=1, le=1000) # Import loose dispatch functions (re-using existing ones for now) # Ideally these should be in services/scraper_service.py async def _scrape_dispatch(url: str, host: str) -> dict[str, object]: if xhamster.can_handle(host): return await xhamster.scrape(url) if masa49.can_handle(host): return await masa49.scrape(url) if xnxx.can_handle(host): return await xnxx.scrape(url) if xvideos.can_handle(host): return await xvideos.scrape(url) if pornhub.can_handle(host): return await pornhub.scrape(url) if youporn.can_handle(host): return await youporn.scrape(url) if redtube.can_handle(host): return await redtube.scrape(url) if beeg.can_handle(host): return await beeg.scrape(url) if spankbang.can_handle(host): return await spankbang.scrape(url) if fapnut.can_handle(host): return await fapnut.scrape(url) if pornxp.can_handle(host): return await pornxp.scrape(url) if hqporner.can_handle(host): return await hqporner.scrape(url) if xxxparodyhd.can_handle(host): return await xxxparodyhd.scrape(url) if pornwex.can_handle(host): return await pornwex.scrape(url) if tube8.can_handle(host): return await tube8.scrape(url) if pornhat.can_handle(host): return await pornhat.scrape(url) if brazzpw.can_handle(host): return await brazzpw.scrape(url) if gosexpod.can_handle(host): return await gosexpod.scrape(url) if watcherotic.can_handle(host): return await watcherotic.scrape(url) raise HTTPException(status_code=400, detail="Unsupported host") async def _list_dispatch(base_url: str, host: str, page: int, limit: int) -> list[dict[str, object]]: if xhamster.can_handle(host): return await xhamster.list_videos(base_url=base_url, page=page, limit=limit) if masa49.can_handle(host): return await masa49.list_videos(base_url=base_url, page=page, limit=limit) if xnxx.can_handle(host): return await xnxx.list_videos(base_url=base_url, page=page, limit=limit) if xvideos.can_handle(host): return await xvideos.list_videos(base_url=base_url, page=page, limit=limit) if pornhub.can_handle(host): return await pornhub.list_videos(base_url=base_url, page=page, limit=limit) if youporn.can_handle(host): return await youporn.list_videos(base_url=base_url, page=page, limit=limit) if redtube.can_handle(host): return await redtube.list_videos(base_url=base_url, page=page, limit=limit) if beeg.can_handle(host): return await beeg.list_videos(base_url=base_url, page=page, limit=limit) if spankbang.can_handle(host): return await spankbang.list_videos(base_url=base_url, page=page, limit=limit) if fapnut.can_handle(host): return await fapnut.list_videos(base_url=base_url, page=page, limit=limit) if pornxp.can_handle(host): return await pornxp.list_videos(base_url=base_url, page=page, limit=limit) if hqporner.can_handle(host): return await hqporner.list_videos(base_url=base_url, page=page, limit=limit) if xxxparodyhd.can_handle(host): return await xxxparodyhd.list_videos(base_url=base_url, page=page, limit=limit) if pornwex.can_handle(host): return await pornwex.list_videos(base_url=base_url, page=page, limit=limit) if tube8.can_handle(host): return await tube8.list_videos(base_url=base_url, page=page, limit=limit) if pornhat.can_handle(host): return await pornhat.list_videos(base_url=base_url, page=page, limit=limit) if brazzpw.can_handle(host): return await brazzpw.list_videos(base_url=base_url, page=page, limit=limit) if gosexpod.can_handle(host): return await gosexpod.list_videos(base_url=base_url, page=page, limit=limit) if watcherotic.can_handle(host): return await watcherotic.list_videos(base_url=base_url, page=page, limit=limit) raise HTTPException(status_code=400, detail="Unsupported host") async def _crawl_dispatch(base_url: str, host: str, start_page: int, max_pages: int, per_page_limit: int, max_items: int) -> list[dict[str, object]]: if xhamster.can_handle(host): return await xhamster.crawl_videos(base_url=base_url, start_page=start_page, max_pages=max_pages, per_page_limit=per_page_limit, max_items=max_items) raise HTTPException(status_code=400, detail="Unsupported host") @api_v1_router.post("/scrapes", response_model=ScrapeResponse, tags=["Scraping"]) async def create_scrape(request: Request, body: ScrapeRequestV1) -> ScrapeResponse: """ Scrape a single video URL. Renamed from /scrape to POST /scrapes (create a scrape). """ from app.config.settings import settings api_base = settings.BASE_URL or str(request.base_url) cache_key = f"scrape:{str(body.url)}" cached_result = await cache.get(cache_key) if cached_result: logging.info(f"⚡ Cache HIT for scrape {body.url}") return ScrapeResponse(**cached_result) try: data = await _scrape_dispatch(str(body.url), body.url.host or "") if "thumbnail_url" in data: data["thumbnail_url"] = thumbnails.wrap_thumbnail_url(data["thumbnail_url"], api_base) await cache.set(cache_key, data, ttl_seconds=7200) # Cache scrapes for 2 hours except httpx.HTTPStatusError as e: raise HTTPException(status_code=e.response.status_code, detail="Upstream returned error") from e except Exception as e: raise HTTPException(status_code=502, detail="Failed to fetch url") from e return ScrapeResponse(**data) @api_v1_router.get("/videos", response_model=list[ListItem], response_model_exclude_unset=True, tags=["Videos"]) async def list_videos(request: Request, base_url: str, page: int = 1, limit: int = 100) -> list[ListItem]: """ List videos from a category/channel URL. Renamed from /list to GET /videos. """ if page < 1: page = 1 if limit < 1: limit = 1 if limit > 200: limit = 200 # Check cache (v2 optimization) cache_key = f"list:{base_url}:p{page}:l{limit}" cached_items = await cache.get(cache_key) if cached_items: logging.info(f"⚡ Cache HIT for list {base_url} page {page}") return [ListItem(**it) for it in cached_items] host = "" try: parsed = HttpUrl(base_url) host = parsed.host or "" except: pass try: items = await _list_dispatch(base_url, host, page, limit) except httpx.HTTPStatusError as e: raise HTTPException(status_code=e.response.status_code, detail="Upstream returned error") from e except Exception as e: raise HTTPException(status_code=502, detail="Failed to fetch url") from e if items: # Wrap thumbnails in proxy for certain sources (like HQPorner) from app.config.settings import settings api_base = settings.BASE_URL or str(request.base_url) for it in items: if "thumbnail_url" in it: it["thumbnail_url"] = thumbnails.wrap_thumbnail_url(it["thumbnail_url"], api_base) await cache.set(cache_key, items, ttl_seconds=3600) # Cache for 1 hour (aggressive) return [ListItem(**it) for it in items] @api_v1_router.post("/crawls", response_model=list[ListItem], tags=["Crawling"]) async def create_crawl(request: Request, body: CrawlRequestV1) -> list[ListItem]: """ Crawl a site for videos. Renamed from /crawl to POST /crawls. """ try: items = await _crawl_dispatch( str(body.base_url), body.base_url.host or "", body.start_page, body.max_pages, body.per_page_limit, body.max_items, ) except httpx.HTTPStatusError as e: raise HTTPException(status_code=e.response.status_code, detail="Upstream returned error") from e except Exception as e: raise HTTPException(status_code=502, detail="Failed to fetch url") from e if items: # Wrap thumbnails in proxy for certain sources (like HQPorner) from app.config.settings import settings api_base = settings.BASE_URL or str(request.base_url) for it in items: if "thumbnail_url" in it: it["thumbnail_url"] = thumbnails.wrap_thumbnail_url(it["thumbnail_url"], api_base) return [ListItem(**it) for it in items] # --- Categories --- # Aggregating categories into a cleaned up endpoint # GET /api/v1/categories?source=xnxx @api_v1_router.get("/categories", response_model=list[CategoryItem], tags=["Categories"]) async def get_categories(source: str) -> list[CategoryItem]: """ Get categories for a specific source. """ s = source.lower() try: if s == "xnxx": return [CategoryItem(**c) for c in xnxx.get_categories()] if s == "masa": return [CategoryItem(**c) for c in masa49.get_categories()] if s == "xvideos": return [CategoryItem(**c) for c in xvideos.get_categories()] if s == "xhamster": return [CategoryItem(**c) for c in xhamster.get_categories()] if s == "youporn": return [CategoryItem(**c) for c in youporn.get_categories()] if s == "pornhub": return [CategoryItem(**c) for c in pornhub.get_categories()] if s == "redtube": return [CategoryItem(**c) for c in redtube.get_categories()] if s == "beeg": return [CategoryItem(**c) for c in beeg.get_categories()] if s == "spankbang": return [CategoryItem(**c) for c in spankbang.get_categories()] if s == "onlyfans" or s == "fapnut": return [CategoryItem(**c) for c in await fapnut.get_categories()] if s == "pornxp": return [CategoryItem(**c) for c in pornxp.get_categories()] if s == "hqporner": return [CategoryItem(**c) for c in hqporner.get_categories()] if s == "xxxparodyhd": return [CategoryItem(**c) for c in xxxparodyhd.get_categories()] if s == "pornwex": return [CategoryItem(**c) for c in pornwex.get_categories()] if s == "tube8": return [CategoryItem(**c) for c in tube8.get_categories()] if s == "pornhat": return [CategoryItem(**c) for c in pornhat.get_categories()] if s == "brazzpw": return [CategoryItem(**c) for c in brazzpw.get_categories()] if s == "gosexpod": return [CategoryItem(**c) for c in gosexpod.get_categories()] if s == "watcherotic": return [CategoryItem(**c) for c in watcherotic.get_categories()] raise HTTPException(status_code=400, detail="Unknown source") except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to load categories: {str(e)}") # --- Global Search & Trending (Pro Features) --- from app.services.global_search import global_search as _global_search, global_trending from fastapi import Query @api_v1_router.get("/search/global", tags=["Search"]) async def global_search_endpoint( request: Request, query: str = Query(..., description="Search keyword"), sites: Optional[list[str]] = Query(None, description="Sites to search"), limit_per_site: int = Query(10, ge=1, le=50), max_sites: int = Query(30, ge=1, le=50) ): from app.config.settings import settings api_base = settings.BASE_URL or str(request.base_url) res = await _global_search(query, sites, limit_per_site, max_sites) if "results" in res: for item in res["results"]: if "thumbnail_url" in item: item["thumbnail_url"] = thumbnails.wrap_thumbnail_url(item["thumbnail_url"], api_base) return res @api_v1_router.get("/trending/global", tags=["Trending"]) async def global_trending_endpoint( request: Request, sites: Optional[list[str]] = Query(None), limit_per_site: int = Query(10, ge=1, le=50) ): from app.config.settings import settings api_base = settings.BASE_URL or str(request.base_url) res = await global_trending(sites, limit_per_site) if "results" in res: for item in res["results"]: if "thumbnail_url" in item: item["thumbnail_url"] = thumbnails.wrap_thumbnail_url(item["thumbnail_url"], api_base) return res # --- Video Streaming Info --- from app.services.video_streaming import get_video_info, get_stream_url from fastapi import Request @api_v1_router.get("/videos/info", tags=["Streaming"]) async def video_info_endpoint(request: Request, url: str = Query(..., description="Video page URL")): from app.config.settings import settings api_base = settings.BASE_URL or str(request.base_url) try: return await get_video_info(url, api_base_url=api_base) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to fetch video info: {str(e)}") @api_v1_router.get("/videos/stream", tags=["Streaming"]) async def direct_stream_endpoint( request: Request, url: str = Query(..., description="Video page URL"), quality: str = Query("default") ): from app.config.settings import settings api_base = settings.BASE_URL or str(request.base_url) try: return await get_stream_url(url, quality, api_base_url=api_base) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to fetch stream URL: {str(e)}") @api_v1_router.get("/videos/download", tags=["Streaming"]) async def video_download_endpoint(request: Request, url: str = Query(..., description="Video page URL")): """ Returns only MP4 download links for a given video URL. Filters out HLS/adaptive streams. """ from app.config.settings import settings api_base = settings.BASE_URL or str(request.base_url) try: info = await get_video_info(url, api_base_url=api_base) video_data = info.get("video", {}) streams = video_data.get("streams", []) # Filter for MP4 only mp4_links = [] for s in streams: fmt = s.get("format", "").lower() stream_url = s.get("url", "") # Skip explicit HLS streams or m3u8 playlists, which may contain .mp4 in path if fmt == "hls" or ".m3u8" in stream_url.lower(): continue if fmt == "mp4" or ".mp4" in stream_url.lower(): mp4_links.append({ "quality": s.get("quality", "unknown"), "url": stream_url, "format": "mp4" }) return { "status": "success", "url": url, "title": info.get("title"), "downloads": mp4_links } except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to fetch download links: {str(e)}") # include routers api_v1_router.include_router(explore.router) api_v1_router.include_router(recommendations.router, prefix="/recommendations", tags=["AI Recommendations"]) api_v1_router.include_router(hls.router, prefix="/hls", tags=["HLS Proxy"]) api_v1_router.include_router(thumbnails.router, prefix="/thumbnails", tags=["Thumbnail Proxy"]) api_v1_router.include_router(media.router) # --- Notifications --- from app.models.schemas import NotificationResponse, NotificationItem @api_v1_router.get("/notifications", response_model=NotificationResponse, tags=["Notifications"]) async def get_notifications(): """ Get app notifications and announcements. In a real app, these would come from a database. """ sample_notifications = [ NotificationItem( id="1", title="Welcome to AppHub", message="Thank you for using our app! Stay tuned for more features.", type="info", created_at=datetime.now() ) ] return NotificationResponse( notifications=sample_notifications, total=len(sample_notifications) ) # --- AppHub Version --- import importlib from app import apphub_version @app.get("/api/apphub/version", tags=["System"]) async def get_apphub_version(): importlib.reload(apphub_version) return { "version": apphub_version.VERSION, "buildNumber": apphub_version.BUILD_NUMBER, "minSupportedBuild": getattr(apphub_version, "MIN_SUPPORTED_BUILD", 1), "releaseDate": getattr(apphub_version, "RELEASE_DATE", ""), "downloadUrl": apphub_version.DOWNLOAD_URL, "apkHash": getattr(apphub_version, "APK_HASH", ""), "changelog": apphub_version.CHANGELOG.strip(), "changelogTitle": apphub_version.CHANGELOG_TITLE, "isMandatory": apphub_version.IS_MANDATORY, "sizeBytes": apphub_version.SIZE_BYTES, } # Include Main V1 Router app.include_router(api_v1_router) @app.get("/health", tags=["System"]) async def health() -> dict[str, str]: logging.info("Health route CALLED successfully") return {"status": "ok", "mode": "safe"}