youclonedl/backend/routers/settings.py

from pathlib import Path
from typing import Optional
from fastapi import APIRouter, Depends, UploadFile, File, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from ..auth_utils import get_current_user
from ..config import settings
from ..database import get_db
from ..models import User, UserSettings
from ..services import ytdlp

router = APIRouter()


VALID_BROWSERS = {"", "chrome", "chromium", "firefox", "brave", "edge", "opera", "safari"}
VALID_REGIONS = {"US", "SE", "GB", "DE", "JP", "FR", "CA", "AU", "BR", "IN", "KR", "MX"}


class SettingsOut(BaseModel):
    preferred_quality: str
    max_concurrent_downloads: int
    hide_watched_from_feed: bool
    mark_watched_at_percent: int
    auto_download_on_sync: bool
    cookies_browser: str = ""
    cookies_file: str = ""
    theater_mode: bool = False
    discovery_regions: str = "US,SE"
    calm_mode: bool = False
    hide_subscriber_counts: bool = False
    autoplay_enabled: bool = False
    feed_weight_recency: float = 5.0
    feed_weight_affinity: float = 5.0
    feed_weight_channel: float = 5.0

    model_config = {"from_attributes": True}


class SettingsPatch(BaseModel):
    preferred_quality: Optional[str] = None
    max_concurrent_downloads: Optional[int] = Field(None, ge=1, le=5)
    hide_watched_from_feed: Optional[bool] = None
    mark_watched_at_percent: Optional[int] = Field(None, ge=50, le=100)
    auto_download_on_sync: Optional[bool] = None
    cookies_browser: Optional[str] = None
    cookies_file: Optional[str] = None
    theater_mode: Optional[bool] = None
    discovery_regions: Optional[str] = None
    calm_mode: Optional[bool] = None
    hide_subscriber_counts: Optional[bool] = None
    autoplay_enabled: Optional[bool] = None
    feed_weight_recency: Optional[float] = Field(None, ge=0.0, le=10.0)
    feed_weight_affinity: Optional[float] = Field(None, ge=0.0, le=10.0)
    feed_weight_channel: Optional[float] = Field(None, ge=0.0, le=10.0)


def _get_or_create(db: Session, user_id: int) -> UserSettings:
    s = db.query(UserSettings).filter_by(user_id=user_id).first()
    if not s:
        s = UserSettings(user_id=user_id)
        db.add(s)
        db.commit()
        db.refresh(s)
    return s


@router.get("", response_model=SettingsOut)
def get_settings(
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    return _get_or_create(db, current_user.id)


@router.patch("", response_model=SettingsOut)
def update_settings(
    body: SettingsPatch,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    s = _get_or_create(db, current_user.id)

    if body.preferred_quality is not None and body.preferred_quality in ytdlp.QUALITY_FORMATS:
        s.preferred_quality = body.preferred_quality
    if body.max_concurrent_downloads is not None:
        s.max_concurrent_downloads = body.max_concurrent_downloads
        ytdlp.set_max_concurrent(body.max_concurrent_downloads)
    if body.hide_watched_from_feed is not None:
        s.hide_watched_from_feed = body.hide_watched_from_feed
    if body.mark_watched_at_percent is not None:
        s.mark_watched_at_percent = body.mark_watched_at_percent
    if body.auto_download_on_sync is not None:
        s.auto_download_on_sync = body.auto_download_on_sync
    if body.cookies_browser is not None and body.cookies_browser in VALID_BROWSERS:
        s.cookies_browser = body.cookies_browser
        ytdlp.set_cookies_browser(body.cookies_browser)
    if body.cookies_file is not None:
        s.cookies_file = body.cookies_file.strip()
        ytdlp.set_cookies_file(body.cookies_file)
    if body.theater_mode is not None:
        s.theater_mode = body.theater_mode
    if body.discovery_regions is not None:
        # Validate: comma-separated list of known region codes
        codes = [r.strip().upper() for r in body.discovery_regions.split(",") if r.strip()]
        valid = [c for c in codes if c in VALID_REGIONS]
        if valid:
            s.discovery_regions = ",".join(valid)
    if body.calm_mode is not None:
        s.calm_mode = body.calm_mode
    if body.hide_subscriber_counts is not None:
        s.hide_subscriber_counts = body.hide_subscriber_counts
    if body.autoplay_enabled is not None:
        s.autoplay_enabled = body.autoplay_enabled
    if body.feed_weight_recency is not None:
        s.feed_weight_recency = body.feed_weight_recency
    if body.feed_weight_affinity is not None:
        s.feed_weight_affinity = body.feed_weight_affinity
    if body.feed_weight_channel is not None:
        s.feed_weight_channel = body.feed_weight_channel

    db.commit()
    db.refresh(s)
    return s


def _cookies_path() -> Path:
    db_file = settings.database_url.replace("sqlite:///", "")
    return Path(db_file).parent / "cookies.txt"


@router.post("/cookies-file", response_model=SettingsOut)
async def upload_cookies_file(
    file: UploadFile = File(...),
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    if not file.filename or not file.filename.endswith(".txt"):
        raise HTTPException(status_code=400, detail="Upload a .txt cookies file")
    content = await file.read()
    if len(content) > 5 * 1024 * 1024:
        raise HTTPException(status_code=400, detail="File too large (max 5 MB)")
    path = _cookies_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(content)
    s = _get_or_create(db, current_user.id)
    s.cookies_file = str(path)
    ytdlp.set_cookies_file(str(path))
    db.commit()
    db.refresh(s)
    return s


@router.delete("/cookies-file", response_model=SettingsOut)
def delete_cookies_file(
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    path = _cookies_path()
    if path.exists():
        path.unlink()
    s = _get_or_create(db, current_user.id)
    s.cookies_file = ""
    ytdlp.set_cookies_file("")
    db.commit()
    db.refresh(s)
    return s