feat(video-library): implement video library processing and API integration

domain/plugins/routes/__init__.py

@@ -0,0 +1,2 @@
+"""插件专属服务端路由集合。"""
+

domain/plugins/routes/video_player.py

@@ -0,0 +1,142 @@
+import json
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from api.response import success
+from domain.auth.service import get_current_active_user
+
+
+router = APIRouter(
+    prefix="/video-player",
+    tags=["plugins"],
+    dependencies=[Depends(get_current_active_user)],
+)
+
+DATA_ROOT = Path("data/.video")
+
+
+def _read_json(path: Path) -> Dict[str, Any]:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _file_mtime_iso(path: Path) -> str:
+    try:
+        ts = path.stat().st_mtime
+    except FileNotFoundError:
+        return ""
+    return datetime.fromtimestamp(ts, tz=UTC).isoformat()
+
+
+def _extract_title(payload: Dict[str, Any]) -> str:
+    detail = (payload.get("tmdb") or {}).get("detail") or {}
+    if payload.get("type") == "tv":
+        return str(detail.get("name") or detail.get("original_name") or "")
+    return str(detail.get("title") or detail.get("original_title") or "")
+
+
+def _extract_year(payload: Dict[str, Any]) -> Optional[str]:
+    detail = (payload.get("tmdb") or {}).get("detail") or {}
+    value = detail.get("first_air_date") if payload.get("type") == "tv" else detail.get("release_date")
+    if not value or not isinstance(value, str):
+        return None
+    return value[:4] if len(value) >= 4 else value
+
+
+def _extract_genres(payload: Dict[str, Any]) -> List[str]:
+    detail = (payload.get("tmdb") or {}).get("detail") or {}
+    genres = detail.get("genres") or []
+    out: List[str] = []
+    if isinstance(genres, list):
+        for g in genres:
+            if isinstance(g, dict) and g.get("name"):
+                out.append(str(g["name"]))
+    return out
+
+
+def _summarize(item_id: str, payload: Dict[str, Any], mtime_iso: str) -> Dict[str, Any]:
+    detail = (payload.get("tmdb") or {}).get("detail") or {}
+    media_type = payload.get("type") or "unknown"
+    episodes = payload.get("episodes") or []
+    seasons = {e.get("season") for e in episodes if isinstance(e, dict) and e.get("season") is not None}
+
+    return {
+        "id": item_id,
+        "type": media_type,
+        "title": _extract_title(payload),
+        "year": _extract_year(payload),
+        "overview": detail.get("overview"),
+        "poster_path": detail.get("poster_path"),
+        "backdrop_path": detail.get("backdrop_path"),
+        "genres": _extract_genres(payload),
+        "tmdb_id": (payload.get("tmdb") or {}).get("id"),
+        "source_path": payload.get("source_path"),
+        "scraped_at": payload.get("scraped_at"),
+        "updated_at": mtime_iso,
+        "episodes_count": len(episodes) if isinstance(episodes, list) else 0,
+        "seasons_count": len(seasons),
+        "vote_average": detail.get("vote_average"),
+        "vote_count": detail.get("vote_count"),
+    }
+
+
+def _iter_library_files() -> List[tuple[str, Path]]:
+    files: List[tuple[str, Path]] = []
+    for sub in ("tv", "movie"):
+        folder = DATA_ROOT / sub
+        if not folder.exists():
+            continue
+        for p in folder.glob("*.json"):
+            if not p.is_file():
+                continue
+            files.append((sub, p))
+    return files
+
+
+@router.get("/library")
+async def list_library(
+    q: str | None = Query(None, description="搜索关键字（标题/简介）"),
+    media_type: str | None = Query(None, alias="type", description="tv 或 movie"),
+):
+    items: List[Dict[str, Any]] = []
+    keyword = (q or "").strip().lower()
+    type_filter = (media_type or "").strip().lower()
+    if type_filter and type_filter not in {"tv", "movie"}:
+        raise HTTPException(status_code=400, detail="type must be tv or movie")
+
+    for _sub, path in _iter_library_files():
+        item_id = path.stem
+        try:
+            payload = _read_json(path)
+        except Exception:
+            continue
+        if type_filter and str(payload.get("type") or "").lower() != type_filter:
+            continue
+        summary = _summarize(item_id, payload, _file_mtime_iso(path))
+        if keyword:
+            haystack = f"{summary.get('title') or ''} {summary.get('overview') or ''}".lower()
+            if keyword not in haystack:
+                continue
+        items.append(summary)
+
+    items.sort(key=lambda x: x.get("updated_at") or "", reverse=True)
+    return success(items)
+
+
+@router.get("/library/{item_id}")
+async def get_library_item(item_id: str):
+    candidates = [
+        DATA_ROOT / "tv" / f"{item_id}.json",
+        DATA_ROOT / "movie" / f"{item_id}.json",
+    ]
+    path = next((p for p in candidates if p.exists()), None)
+    if not path:
+        raise HTTPException(status_code=404, detail="Item not found")
+
+    payload = _read_json(path)
+    payload["id"] = item_id
+    payload["updated_at"] = _file_mtime_iso(path)
+    return success(payload)
+