import os import shutil from pathlib import Path from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from typing import Optional from config import settings from routes.auth import require_auth router = APIRouter(prefix="/api/settings", tags=["settings"]) class BrandUpdate(BaseModel): brand_name: Optional[str] = None brand_tagline: Optional[str] = None brand_logo_url: Optional[str] = None class TranscodeQuality(BaseModel): label: str = Field(min_length=1, max_length=32) width: int = Field(ge=160, le=7680) height: int = Field(ge=90, le=4320) bitrate: str = Field(pattern=r"^[1-9][0-9]*(?:k|M)$") class TranscodeProfilesUpdate(BaseModel): profiles: dict[str, list[TranscodeQuality]] @router.get("/brand") async def get_brand(): return { "brand_name": settings.brand_name, "brand_tagline": settings.brand_tagline, "brand_logo_url": settings.brand_logo_url, } @router.put("/brand") async def update_brand(body: BrandUpdate, _: bool = Depends(require_auth)): if body.brand_name is not None: settings.brand_name = body.brand_name if body.brand_tagline is not None: settings.brand_tagline = body.brand_tagline if body.brand_logo_url is not None: settings.brand_logo_url = body.brand_logo_url return {"ok": True} @router.get("/transcode-profiles") async def get_transcode_profiles(_: bool = Depends(require_auth)): return settings.transcode_profiles @router.put("/transcode-profiles") async def update_transcode_profiles( body: TranscodeProfilesUpdate, _: bool = Depends(require_auth), ): if not body.profiles: raise HTTPException(status_code=422, detail="至少需要一个转码配置") normalized: dict[str, list[dict]] = {} for raw_name, qualities in body.profiles.items(): name = raw_name.strip() if not name or len(name) > 64: raise HTTPException(status_code=422, detail="转码配置名称无效") if not qualities: raise HTTPException(status_code=422, detail=f"转码配置 {name} 至少需要一个清晰度") labels = [quality.label.strip() for quality in qualities] if any(not label for label in labels) or len(labels) != len(set(labels)): raise HTTPException(status_code=422, detail=f"转码配置 {name} 的清晰度名称无效或重复") normalized[name] = [ { "label": quality.label.strip(), "width": quality.width, "height": quality.height, "bitrate": quality.bitrate, } for quality in sorted(qualities, key=lambda item: item.height, reverse=True) ] settings.transcode_profiles = normalized settings.qualities = next(iter(normalized.values())) return {"ok": True, "profiles": normalized} @router.get("/system-metrics") async def get_system_metrics(_: bool = Depends(require_auth)): load_1m, load_5m, load_15m = os.getloadavg() memory_total = 0 memory_available = 0 meminfo = Path("/proc/meminfo") if meminfo.exists(): values = {} for line in meminfo.read_text(encoding="ascii").splitlines(): key, _, value = line.partition(":") if value: values[key] = int(value.strip().split()[0]) * 1024 memory_total = values.get("MemTotal", 0) memory_available = values.get("MemAvailable", 0) metrics_path = Path(settings.replay_media_dir).resolve() disk_path = metrics_path if metrics_path.exists() else Path.cwd() disk = shutil.disk_usage(disk_path) cpu_count = os.cpu_count() or 1 return { "cpu": { "count": cpu_count, "load_1m": load_1m, "load_5m": load_5m, "load_15m": load_15m, "load_percent": round(min(100, load_1m / cpu_count * 100), 1), }, "memory": { "total": memory_total, "available": memory_available, "used_percent": round((1 - memory_available / memory_total) * 100, 1) if memory_total else 0, }, "disk": { "total": disk.total, "free": disk.free, "used_percent": round(disk.used / disk.total * 100, 1) if disk.total else 0, }, }