MediaFusion/api/frontend_api.py

import math
from typing import Dict, List, Optional, Any, Literal

from fastapi import APIRouter, Depends, Request, HTTPException, BackgroundTasks
from pydantic import BaseModel

from db import crud, schemas
from db.config import settings
from utils import const
from utils.network import get_user_data, get_user_public_ip

# Create a router with more appropriate naming for the frontend
router = APIRouter()


# ---- Data Models ----


class AppConfig(BaseModel):
    """Base configuration data for the application"""

    addon_name: str
    logo_url: str
    host_url: str
    poster_host_url: Optional[str]
    version: str
    description: str
    branding_description: str
    is_public_instance: bool
    disabled_providers: List[str]
    authentication_required: bool


class UserConfig(BaseModel):
    """User-specific configuration data"""

    user_data: Dict[str, Any]
    configured_fields: List[str]


class ValidationResult(BaseModel):
    """Data model for validation results"""

    status: str
    message: Optional[str] = None


class StreamData(BaseModel):
    """Data model for stream information"""

    title: str
    year: int
    poster: str
    background: str
    description: str
    streams: List[Dict[str, Any]]
    catalog_type: str
    season: Optional[int] = None
    episode: Optional[int] = None
    video_id: str
    secret_str: str
    series_data: Optional[str] = None


# ---- API Endpoints ----


@router.get("/app-config", response_model=AppConfig)
async def get_app_config():
    """
    Get basic application configuration.
    Replaces the former home endpoint with more specific naming.
    """
    return {
        "addon_name": settings.addon_name,
        "logo_url": settings.logo_url,
        "host_url": settings.host_url,
        "poster_host_url": settings.poster_host_url,
        "version": settings.version,
        "description": settings.description,
        "branding_description": settings.branding_description,
        "is_public_instance": settings.is_public_instance,
        "disabled_providers": settings.disabled_providers,
        "authentication_required": settings.api_password is not None
        and not settings.is_public_instance,
    }


@router.get("/user-config", response_model=UserConfig)
async def get_user_config(
    user_data: schemas.UserData = Depends(get_user_data),
    secret_str: Optional[str] = None,
):
    """
    Get user-specific configuration.
    Returns only user configuration without default values from constants.
    """
    configured_fields = []

    # Handle sensitive data masking
    if user_data.streaming_provider:
        user_data.streaming_provider.password = "••••••••"
        user_data.streaming_provider.token = "••••••••"
        configured_fields.extend(["provider_token", "password"])

        if user_data.streaming_provider.qbittorrent_config:
            user_data.streaming_provider.qbittorrent_config.qbittorrent_password = (
                "••••••••"
            )
            user_data.streaming_provider.qbittorrent_config.webdav_password = "••••••••"
            configured_fields.extend(["qbittorrent_password", "webdav_password"])

    if user_data.mediaflow_config:
        user_data.mediaflow_config.api_password = "••••••••"
        configured_fields.append("mediaflow_api_password")

    if user_data.rpdb_config:
        user_data.rpdb_config.api_key = "••••••••"
        configured_fields.append("rpdb_api_key")

    user_data.api_password = None

    return {
        "user_data": user_data.model_dump(),
        "configured_fields": configured_fields,
    }


@router.get("/system-constants")
async def get_system_constants():
    """
    Get all system constants needed by the frontend.
    Renamed from 'constants' to be more descriptive.
    """
    return {
        "CATALOG_DATA": const.CATALOG_DATA,
        "RESOLUTIONS": const.RESOLUTIONS,
        "TORRENT_SORTING_PRIORITY_OPTIONS": const.TORRENT_SORTING_PRIORITY_OPTIONS,
        "SUPPORTED_LANGUAGES": const.SUPPORTED_LANGUAGES,
        "QUALITY_GROUPS": const.QUALITY_GROUPS,
    }


@router.get(
    "/download/{secret_str}/{catalog_type}/{video_id}", response_model=StreamData
)
@router.get(
    "/download/{secret_str}/{catalog_type}/{video_id}/{season}/{episode}",
    response_model=StreamData,
)
async def get_download_info(
    request: Request,
    secret_str: str,
    catalog_type: Literal["movie", "series"],
    video_id: str,
    user_data: schemas.UserData = Depends(get_user_data),
    background_tasks: BackgroundTasks = BackgroundTasks(),
    season: int = None,
    episode: int = None,
):
    """
    Get download information for the React frontend.
    Replaces the Jinja template rendering for the download page.
    """
    if (
        not user_data.streaming_provider
        or not user_data.streaming_provider.download_via_browser
    ):
        raise HTTPException(
            status_code=403,
            detail="Download option is not enabled or no streaming provider configured",
        )

    metadata = (
        await crud.get_movie_data_by_id(video_id)
        if catalog_type == "movie"
        else await crud.get_series_data_by_id(video_id)
    )
    if not metadata:
        raise HTTPException(status_code=404, detail="Metadata not found")

    user_ip = await get_user_public_ip(request, user_data)

    if catalog_type == "movie":
        streams = await crud.get_movie_streams(
            user_data, secret_str, video_id, user_ip, background_tasks
        )
    else:
        streams = await crud.get_series_streams(
            user_data, secret_str, video_id, season, episode, user_ip, background_tasks
        )

    streaming_provider_path = f"{settings.host_url}/streaming_provider/"
    downloadable_streams = [
        stream.model_dump()
        for stream in streams
        if stream.url and stream.url.startswith(streaming_provider_path)
    ]

    if video_id.startswith("tt") and user_data.rpdb_config:
        _poster = f"https://api.ratingposterdb.com/{user_data.rpdb_config.api_key}/imdb/poster-default/{video_id}.jpg?fallback=true"
    else:
        _poster = f"{settings.poster_host_url}/poster/{catalog_type}/{video_id}.jpg"

    background = (
        metadata.background
        if metadata.background
        else f"{settings.host_url}/static/images/background.jpg"
    )

    return {
        "title": metadata.title,
        "year": metadata.year,
        "poster": _poster,
        "background": background,
        "description": metadata.description,
        "streams": downloadable_streams,
        "catalog_type": catalog_type,
        "season": season,
        "episode": episode,
        "video_id": video_id,
        "secret_str": secret_str,
        "series_data": (
            metadata.model_dump_json(
                include={"episodes": {"__all__": {"season_number", "episode_number"}}}
            )
            if catalog_type == "series"
            else None
        ),
    }