import json import logging import os import re from contextlib import asynccontextmanager from datetime import date, datetime, timezone from decimal import Decimal from pathlib import Path from typing import Any from urllib.parse import urlparse from uuid import UUID import asyncpg import httpx from asyncpg import PostgresError from fastapi import FastAPI, HTTPException, Query, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, Response from pydantic import BaseModel from config import load_config logging.basicConfig(level=logging.INFO) logger = logging.getLogger("filebot-api") COLUMNS = [ "id", "file_name_original", "file_name_new", "object_type", "cover_url", "movie_rating", "movie_votes", "video_compression_format", "video_codec_library", "audio_codec", "container_format", "standard_video_format", "exact_video_format", "fourk_resolution", "audio_codec_profile", "audio_channel_format", "audio_channel_count", "audio_channel_layout", "video_resolution", "video_width", "video_height", "video_bitdepth", "high_dynamic_range", "dolby_vision", "overall_bitrate", "video_bitrate", "audio_bitrate", "created_at", "seen_at", ] class SetSeenBody(BaseModel): seen: bool pool: asyncpg.Pool | None = None qualified_table: str = '"public"."n8n_filebot"' config_path: Path = Path("/app/config/config.yaml") config_loaded_from_file: bool = False db_settings: dict[str, Any] = {} def serialize_value(val: Any) -> Any: if val is None: return None if isinstance(val, Decimal): return float(val) if isinstance(val, (datetime, date)): return val.isoformat() if isinstance(val, UUID): return str(val) return val def normalize_cover_url(val: Any) -> str | None: """Cover-URL aus DB/n8n bereinigen (JSON-Strings, Anführungszeichen, eingebettete URLs).""" if val is None: return None if isinstance(val, dict): for key in ("url", "cover_url", "medium", "original", "poster", "href", "src"): if key in val and val[key]: nested = normalize_cover_url(val[key]) if nested: return nested return None if isinstance(val, list) and val: return normalize_cover_url(val[0]) if not isinstance(val, str): val = str(val) s = val.strip().strip('"').strip("'") if not s: return None if s.startswith("{") or s.startswith("["): try: return normalize_cover_url(json.loads(s)) except json.JSONDecodeError: pass match = re.search(r"https?://[^\s\"'<>]+", s) if match: return match.group(0).rstrip(".,)") if s.startswith("http://") or s.startswith("https://"): return s return None def row_to_dict(record: asyncpg.Record) -> dict[str, Any]: data: dict[str, Any] = {} for col in COLUMNS: val = serialize_value(record[col]) if col == "cover_url": val = normalize_cover_url(val) data[col] = val return data def build_record_conditions( *, search: str | None = None, format_filter: str | None = None, codec: str | None = None, container: str | None = None, created_from: str | None = None, created_to: str | None = None, unseen_only: bool = False, object_type: str | None = None, ) -> tuple[list[str], list[Any]]: conditions: list[str] = [] params: list[Any] = [] idx = 1 if search: conditions.append( f"(file_name_original ILIKE ${idx} OR file_name_new ILIKE ${idx})" ) params.append(f"%{search}%") idx += 1 if format_filter: conditions.append(f"standard_video_format = ${idx}") params.append(format_filter) idx += 1 if codec: conditions.append(f"video_codec_library = ${idx}") params.append(codec) idx += 1 if container: conditions.append(f"container_format = ${idx}") params.append(container) idx += 1 if created_from: conditions.append(f"created_at >= ${idx}") params.append(parse_filter_datetime(created_from, "von")) idx += 1 if created_to: conditions.append(f"created_at <= ${idx}") params.append(parse_filter_datetime(created_to, "bis")) idx += 1 if unseen_only: conditions.append("seen_at IS NULL") if object_type: conditions.append(f"LOWER(TRIM(object_type::text)) = ${idx}") params.append(object_type.strip().lower()) idx += 1 return conditions, params def parse_filter_datetime(value: str, param_name: str) -> datetime: try: normalized = value.strip().replace("Z", "+00:00") dt = datetime.fromisoformat(normalized) if dt.tzinfo is None: return dt.replace(tzinfo=timezone.utc) return dt except ValueError as exc: raise HTTPException(400, detail=f"Ungültiges Datum ({param_name})") from exc @asynccontextmanager async def lifespan(app: FastAPI): global pool, qualified_table, config_path, config_loaded_from_file, db_settings cfg_path = os.getenv("CONFIG_PATH", "/app/config/config.yaml") cfg, config_path = load_config(cfg_path) config_loaded_from_file = config_path.is_file() db = cfg.database qualified_table = db.qualified_table db_settings = { "host": db.host, "port": db.port, "database": db.name, "user": db.user, "table": db.table, "db_schema": db.db_schema, "qualified_table": db.qualified_table, "config_file": str(config_path), "config_exists": config_loaded_from_file, } if not config_loaded_from_file: logger.warning("Config-Datei fehlt: %s – Standardwerte aktiv", config_path) try: pool = await asyncpg.create_pool( host=db.host, port=db.port, database=db.name, user=db.user, password=db.password, ssl=db.ssl, min_size=1, max_size=5, command_timeout=30, ) async with pool.acquire() as conn: await conn.fetchval("SELECT 1") logger.info( "PostgreSQL verbunden: %s:%s/%s → %s", db.host, db.port, db.name, qualified_table, ) except Exception as exc: logger.exception("PostgreSQL-Verbindung fehlgeschlagen: %s", exc) pool = None yield if pool: await pool.close() pool = None app = FastAPI(title="Filebot Media Browser", lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) @app.exception_handler(PostgresError) async def postgres_error_handler(_request: Request, exc: PostgresError): logger.exception("PostgreSQL-Fehler: %s", exc) return JSONResponse( status_code=503, content={"detail": f"Datenbankfehler: {exc}"}, ) @app.exception_handler(Exception) async def generic_error_handler(_request: Request, exc: Exception): if isinstance(exc, HTTPException): raise exc logger.exception("Unerwarteter Fehler: %s", exc) return JSONResponse( status_code=500, content={"detail": f"Serverfehler: {type(exc).__name__}: {exc}"}, ) @app.get("/api/health") async def health(): if pool is None: raise HTTPException(503, detail="Datenbank nicht verbunden") async with pool.acquire() as conn: await conn.fetchval("SELECT 1") return {"status": "ok"} @app.get("/api/status") async def status(): """Diagnose: Config + DB + Tabellenzugriff.""" info: dict[str, Any] = { "database": db_settings, "pool": pool is not None, "table": qualified_table, } if pool is None: info["error"] = "Kein Connection-Pool – Zugangsdaten oder Netzwerk prüfen" return info try: async with pool.acquire() as conn: info["ping"] = await conn.fetchval("SELECT 1") count = await conn.fetchval(f"SELECT COUNT(*) FROM {qualified_table}") info["row_count"] = count except PostgresError as exc: info["error"] = str(exc) if pool is not None: try: async with pool.acquire() as conn: schema = db_settings.get("db_schema", "public") in_schema = await conn.fetch( """ SELECT table_name FROM information_schema.tables WHERE table_schema = $1 AND table_type = 'BASE TABLE' ORDER BY table_name """, schema, ) similar = await conn.fetch( """ SELECT table_schema, table_name FROM information_schema.tables WHERE table_type = 'BASE TABLE' AND ( table_name ILIKE '%filebot%' OR table_name ILIKE '%file%' ) ORDER BY table_schema, table_name LIMIT 30 """ ) info["tables_in_configured_schema"] = [r["table_name"] for r in in_schema] info["similar_tables"] = [ f'{r["table_schema"]}.{r["table_name"]}' for r in similar ] except PostgresError: pass return info @app.get("/api/columns") async def get_columns(): return {"columns": COLUMNS} @app.get("/api/records/meta") async def get_records_meta( search: str | None = Query(None), format_filter: str | None = Query(None, alias="format"), codec: str | None = None, container: str | None = None, created_from: str | None = Query(None), created_to: str | None = Query(None), unseen_only: bool = Query(False), object_type: str | None = None, ): """Leichtgewichtiger Check für Auto-Refresh (ohne volle Datensätze).""" if pool is None: raise HTTPException(503, detail="Datenbank nicht verbunden") conditions, params = build_record_conditions( search=search, format_filter=format_filter, codec=codec, container=container, created_from=created_from, created_to=created_to, unseen_only=unseen_only, object_type=object_type, ) where = f"WHERE {' AND '.join(conditions)}" if conditions else "" if conditions: unseen_conditions = [*conditions, "seen_at IS NULL"] unseen_where = f"WHERE {' AND '.join(unseen_conditions)}" else: unseen_where = "WHERE seen_at IS NULL" async with pool.acquire() as conn: row = await conn.fetchrow( f""" SELECT COUNT(*)::int AS total, COALESCE(MAX(id), 0)::int AS max_id, MAX(created_at) AS latest_created_at, ( SELECT md5(COALESCE(string_agg(sig, '|'), '')) FROM ( SELECT ( id::text || ':' || COALESCE(cover_url, '') || ':' || COALESCE(file_name_new, '') || ':' || COALESCE(object_type::text, '') || ':' || COALESCE(seen_at::text, '') ) AS sig FROM {qualified_table} {where} ORDER BY created_at DESC NULLS LAST, id DESC LIMIT 300 ) recent ) AS revision FROM {qualified_table} {where} """, *params, ) unseen_total = await conn.fetchval( f"SELECT COUNT(*)::int FROM {qualified_table} {unseen_where}", *params, ) latest = row["latest_created_at"] return { "total": row["total"], "unseen_total": unseen_total, "max_id": row["max_id"], "latest_created_at": serialize_value(latest) if latest else None, "revision": row["revision"] or "", } @app.get("/api/cover") async def proxy_cover(url: str = Query(..., min_length=8)): """Cover-Bilder über die API laden (Hotlink/Referrer-Probleme umgehen).""" normalized = normalize_cover_url(url) if not normalized: raise HTTPException(400, detail="Ungültige Cover-URL") parsed = urlparse(normalized) if parsed.scheme not in ("http", "https") or not parsed.netloc: raise HTTPException(400, detail="Ungültige Cover-URL") try: async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client: resp = await client.get( normalized, headers={"User-Agent": "FilebotMediaBrowser/1.0"}, ) except httpx.HTTPError as exc: raise HTTPException(502, detail=f"Cover nicht ladbar: {exc}") from exc if resp.status_code != 200: raise HTTPException(502, detail=f"Cover nicht ladbar ({resp.status_code})") content_type = resp.headers.get("content-type", "image/jpeg") if not content_type.startswith("image/"): content_type = "image/jpeg" return Response( content=resp.content, media_type=content_type, headers={"Cache-Control": "public, max-age=86400"}, ) @app.get("/api/records") async def get_records( search: str | None = Query(None, description="Suche in Dateinamen"), format_filter: str | None = Query(None, alias="format"), codec: str | None = None, container: str | None = None, created_from: str | None = Query(None, description="created_at >= (ISO)"), created_to: str | None = Query(None, description="created_at <= (ISO)"), unseen_only: bool = Query(False, description="Nur ungesehene Einträge"), object_type: str | None = None, limit: int = Query(5000, le=10000), offset: int = Query(0, ge=0), ): if pool is None: raise HTTPException(503, detail="Datenbank nicht verbunden") conditions, params = build_record_conditions( search=search, format_filter=format_filter, codec=codec, container=container, created_from=created_from, created_to=created_to, unseen_only=unseen_only, object_type=object_type, ) idx = len(params) + 1 where = f"WHERE {' AND '.join(conditions)}" if conditions else "" cols = ", ".join(COLUMNS) query = f""" SELECT {cols} FROM {qualified_table} {where} ORDER BY created_at DESC NULLS LAST, id DESC LIMIT ${idx} OFFSET ${idx + 1} """ params.extend([limit, offset]) count_query = f"SELECT COUNT(*) FROM {qualified_table} {where}" count_params = params[:-2] if conditions: unseen_conditions = [*conditions, "seen_at IS NULL"] unseen_where = f"WHERE {' AND '.join(unseen_conditions)}" else: unseen_where = "WHERE seen_at IS NULL" async with pool.acquire() as conn: rows = await conn.fetch(query, *params) total = await conn.fetchval(count_query, *count_params) unseen_total = await conn.fetchval( f"SELECT COUNT(*) FROM {qualified_table} {unseen_where}", *count_params, ) return { "records": [row_to_dict(r) for r in rows], "total": total, "unseen_total": unseen_total, "limit": limit, "offset": offset, } @app.patch("/api/records/{record_id}/seen") async def set_record_seen(record_id: int, body: SetSeenBody): """Einzelnen Eintrag als gesehen (seen=true) oder ungesehen (seen=false) markieren.""" if pool is None: raise HTTPException(503, detail="Datenbank nicht verbunden") if body.seen: sql = f""" UPDATE {qualified_table} SET seen_at = NOW() WHERE id = $1 RETURNING id, seen_at """ else: sql = f""" UPDATE {qualified_table} SET seen_at = NULL WHERE id = $1 RETURNING id, seen_at """ async with pool.acquire() as conn: row = await conn.fetchrow(sql, record_id) if row is None: raise HTTPException(404, detail=f"Eintrag mit ID {record_id} nicht gefunden") return { "id": row["id"], "seen_at": serialize_value(row["seen_at"]), } @app.post("/api/records/mark-all-seen") async def mark_all_seen( search: str | None = Query(None), format_filter: str | None = Query(None, alias="format"), codec: str | None = None, container: str | None = None, created_from: str | None = None, created_to: str | None = None, object_type: str | None = None, ): """Alle sichtbaren (gefilterten) Einträge als gesehen markieren.""" if pool is None: raise HTTPException(503, detail="Datenbank nicht verbunden") conditions, params = build_record_conditions( search=search, format_filter=format_filter, codec=codec, container=container, created_from=created_from, created_to=created_to, unseen_only=False, object_type=object_type, ) if conditions: where = f"WHERE {' AND '.join(conditions)} AND seen_at IS NULL" else: where = "WHERE seen_at IS NULL" async with pool.acquire() as conn: result = await conn.execute( f"UPDATE {qualified_table} SET seen_at = NOW() {where}", *params, ) count = int(result.split()[-1]) if result else 0 return {"marked": count} @app.get("/api/filters") async def get_filter_options(): if pool is None: raise HTTPException(503, detail="Datenbank nicht verbunden") async with pool.acquire() as conn: formats = await conn.fetch( f""" SELECT DISTINCT standard_video_format AS value FROM {qualified_table} WHERE standard_video_format IS NOT NULL AND standard_video_format::text <> '' ORDER BY 1 """ ) codecs = await conn.fetch( f""" SELECT DISTINCT video_codec_library AS value FROM {qualified_table} WHERE video_codec_library IS NOT NULL AND video_codec_library::text <> '' ORDER BY 1 """ ) containers = await conn.fetch( f""" SELECT DISTINCT container_format AS value FROM {qualified_table} WHERE container_format IS NOT NULL AND container_format::text <> '' ORDER BY 1 """ ) audio_codecs = await conn.fetch( f""" SELECT DISTINCT audio_codec AS value FROM {qualified_table} WHERE audio_codec IS NOT NULL AND audio_codec::text <> '' ORDER BY 1 """ ) object_types = await conn.fetch( f""" SELECT DISTINCT LOWER(TRIM(object_type::text)) AS value FROM {qualified_table} WHERE object_type IS NOT NULL AND TRIM(object_type::text) <> '' ORDER BY 1 """ ) return { "formats": [r["value"] for r in formats], "codecs": [r["value"] for r in codecs], "containers": [r["value"] for r in containers], "audio_codecs": [r["value"] for r in audio_codecs], "object_types": [r["value"] for r in object_types], } @app.delete("/api/records/{record_id}") async def delete_record(record_id: int): if pool is None: raise HTTPException(503, detail="Datenbank nicht verbunden") async with pool.acquire() as conn: result = await conn.execute( f"DELETE FROM {qualified_table} WHERE id = $1", record_id, ) if result == "DELETE 0": raise HTTPException(404, detail=f"Eintrag mit ID {record_id} nicht gefunden") return {"deleted": True, "id": record_id}