filebot-media-browser/api/main.py

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",
    "cover_mime_type",
]


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/{record_id}/cover")
async def get_record_cover(record_id: int):
    """Cover-Bild aus DB liefern (BYTEA). 404 wenn kein Bild gespeichert."""
    if pool is None:
        raise HTTPException(503, detail="Datenbank nicht verbunden")

    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            f"SELECT cover_image, cover_mime_type FROM {qualified_table} WHERE id = $1",
            record_id,
        )

    if row is None:
        raise HTTPException(404, detail="Eintrag nicht gefunden")
    if not row["cover_image"]:
        raise HTTPException(404, detail="Kein Bild gespeichert")

    mime = row["cover_mime_type"] or "image/jpeg"
    return Response(
        content=bytes(row["cover_image"]),
        media_type=mime,
        headers={"Cache-Control": "public, max-age=604800"},
    )


@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,
    }


class SetCoverUrlBody(BaseModel):
    cover_url: str


@app.patch("/api/records/{record_id}/cover_url")
async def set_record_cover_url(record_id: int, body: SetCoverUrlBody):
    if pool is None:
        raise HTTPException(503, detail="Datenbank nicht verbunden")
    url = body.cover_url.strip()
    if not url:
        raise HTTPException(422, detail="cover_url darf nicht leer sein")
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            f"UPDATE {qualified_table} SET cover_url = $1, cover_image = NULL, cover_mime_type = NULL WHERE id = $2 RETURNING id, cover_url",
            url,
            record_id,
        )
    if row is None:
        raise HTTPException(404, detail=f"Eintrag mit ID {record_id} nicht gefunden")
    return {"id": row["id"], "cover_url": row["cover_url"]}


@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}