horror/backend/app/sync.py

import asyncio
import logging
from datetime import datetime, timezone
from urllib.parse import quote_plus

from . import canon, db, imagehash, netflix, search, teaser, tmdb

log = logging.getLogger("sync")

TMDB_CONCURRENCY = 8
TEASER_CONCURRENCY = 5


def netflix_url(title):
    return f"https://www.netflix.com/search?q={quote_plus(title)}"


def build_movie(data):
    release = data.get("release_date") or ""
    collection = data.get("belongs_to_collection") or {}
    return {
        "collection_id": collection.get("id"),
        "collection_name": collection.get("name"),
        "id": data["id"],
        "title": tmdb.pick_title(data),
        "original_title": data.get("original_title"),
        "release_date": release or None,
        "year": int(release[:4]) if release[:4].isdigit() else None,
        "overview": tmdb.pick_overview(data),
        "poster_path": data.get("poster_path"),
        "backdrop_path": data.get("backdrop_path"),
        "vote_average": data.get("vote_average"),
        "vote_count": data.get("vote_count"),
        "runtime": data.get("runtime"),
        "genres": [g["name"] for g in data.get("genres", [])],
        "images": tmdb.pick_backdrops(data),
        "trailer_key": tmdb.pick_trailer(data),
        "imdb_id": data.get("imdb_id"),
        "rentable": tmdb.pick_rentable(data),
        # Die echte Titel-ID trägt fill_netflix_links nach
        "netflix_id": None,
        "netflix_url": tmdb.pick_watch_link(data)
        or netflix_url(data.get("title") or data.get("original_title") or ""),
        # setzt run() je nach Quelle
        "on_netflix": 0,
        "canon_group": None,
        "canon_pos": None,
    }


async def fetch_details(http, ids):
    semaphore = asyncio.Semaphore(TMDB_CONCURRENCY)

    async def one(movie_id):
        async with semaphore:
            try:
                return build_movie(await tmdb.detail(http, movie_id))
            except Exception:
                log.warning("Details fehlgeschlagen für %s", movie_id)
                return None

    results = await asyncio.gather(*(one(i) for i in ids))
    return [m for m in results if m]


async def dedupe_images(movies):
    """Crops und Werbemotive desselben Frames fliegen raus."""
    all_paths = {p for m in movies for p in m["images"]}
    with db.connect() as conn:
        prints = db.known_prints(conn, all_paths)

    missing = sorted(all_paths - set(prints))
    if missing:
        log.info("Berechne Fingerabdrücke für %s Bilder", len(missing))
        fresh = await imagehash.fingerprint_many(missing)
        with db.connect() as conn:
            db.store_prints(conn, fresh)
        prints.update(fresh)

    before = sum(len(m["images"]) for m in movies)
    for movie in movies:
        movie["images"] = imagehash.dedupe(movie["images"], prints)
    after = sum(len(m["images"]) for m in movies)
    log.info("Bilder: %s → %s nach Dedup", before, after)


# TMDB drosselt die Watch-Seiten. Mehr als das pro Lauf bringt nur noch
# leere Antworten — der Rest kommt beim nächsten Sync dran.
LINKS_PRO_LAUF = 250


async def fill_netflix_links(movies, known):
    """Direktlink auf die Netflix-Titelseite, sofern die ID auffindbar ist.

    Sonst bleibt der Link auf TMDBs Watch-Seite stehen — von dort ist es
    ein Klick weiter.
    """
    for movie in movies:
        movie["netflix_id"] = known.get(movie["id"])

    offen = [m for m in movies if not m["netflix_id"]]
    # Die meistgesehenen zuerst, die klickt man am ehesten an
    offen.sort(key=lambda m: -(m["vote_count"] or 0))
    todo = [m["id"] for m in offen[:LINKS_PRO_LAUF]]
    if todo:
        log.info(
            "Suche Netflix-IDs für %s von %s offenen Filmen", len(todo), len(offen)
        )
        found = await netflix.fetch_ids(todo)
        for movie in movies:
            movie["netflix_id"] = movie["netflix_id"] or found.get(movie["id"])

    for movie in movies:
        if movie["netflix_id"]:
            movie["netflix_url"] = netflix.title_url(movie["netflix_id"])
    return sum(1 for m in movies if m["netflix_id"])


async def fill_teasers(movies, known):
    todo = [m for m in movies if not known.get(m["id"])]
    for movie in movies:
        movie["teaser"] = known.get(movie["id"], "")
    if not todo:
        return 0

    log.info("Erzeuge %s Teaser über MiniMax", len(todo))
    semaphore = asyncio.Semaphore(TEASER_CONCURRENCY)
    async with teaser.client() as http:
        texts = await asyncio.gather(
            *(teaser.generate(http, m, semaphore) for m in todo)
        )
    for movie, text in zip(todo, texts):
        movie["teaser"] = text
    return len(todo)


async def resolve_canon():
    """Kanon-Liste auf TMDB-IDs abbilden: {tmdb_id: (gruppe, position)}."""
    groups = await canon.resolve_all()
    places = {}
    for group_index, entries in enumerate(groups):
        for position, entry in enumerate(entries):
            if entry["hit"]:
                places.setdefault(entry["hit"]["id"], (group_index, position))
    return places


async def katalog_nachtragen():
    """Filme in den Katalog holen, die jemand gemerkt hat, die aber weder
    auf Netflix laufen noch im Kanon stehen.

    Ohne sie bleiben die Karten in Sammlung, Rad und Gesehen leer: kein
    Teaser, keine Note, keine Bilder. Die Detailansicht holt sich das
    jedes Mal neu von TMDB — hier landet es einmal in der Datenbank.
    """
    with db.connect() as conn:
        fehlen = sorted(
            db.benutzte_ids(conn)
            - {r["id"] for r in conn.execute("SELECT id FROM movies")}
        )
    if not fehlen:
        return 0

    bremse = asyncio.Semaphore(TMDB_CONCURRENCY)

    async def hole(movie_id):
        async with bremse:
            try:
                return await search.detail(movie_id)
            except Exception:
                # Ein gelöschter TMDB-Eintrag soll den Rest nicht aufhalten
                return None

    filme = [f for f in await asyncio.gather(*map(hole, fehlen)) if f]
    with db.connect() as conn:
        for film in filme:
            db.upsert(conn, film)
    log.info("%s Filme aus Sammlungen in den Katalog geholt", len(filme))
    return len(filme)


async def leihbarkeit_nachtragen():
    """Filme, bei denen noch nicht steht, ob man sie leihen kann — der
    Altbestand von vor dieser Spalte. Der tägliche Sync hält es danach
    aktuell, hier geht es nur um das erste Mal."""
    with db.connect() as conn:
        offen = [r["id"] for r in conn.execute("SELECT id FROM movies WHERE rentable IS NULL")]
    if not offen:
        return 0

    bremse = asyncio.Semaphore(TMDB_CONCURRENCY)

    async def hole(http, movie_id):
        async with bremse:
            try:
                resp = await http.get(f"/movie/{movie_id}/watch/providers")
                resp.raise_for_status()
                return movie_id, tmdb.pick_rentable({"watch/providers": resp.json()})
            except Exception:
                return movie_id, None

    async with tmdb.client() as http:
        stand = await asyncio.gather(*(hole(http, i) for i in offen))
    with db.connect() as conn:
        conn.executemany(
            "UPDATE movies SET rentable = ? WHERE id = ?",
            [(wert, i) for i, wert in stand if wert is not None],
        )
    log.info("Leihbarkeit für %s Filme nachgetragen", sum(1 for _, w in stand if w is not None))
    return len(offen)


async def run():
    log.info("Sync startet")
    canon_places = await resolve_canon()
    log.info("%s Kanon-Titel aufgelöst", len(canon_places))

    async with tmdb.client() as http:
        netflix_ids = await tmdb.discover_all(http)
        log.info("%s Filme bei Netflix DE gefunden", len(netflix_ids))
        wanted = sorted(netflix_ids | set(canon_places))
        movies = await fetch_details(http, wanted)

    for movie in movies:
        movie["on_netflix"] = 1 if movie["id"] in netflix_ids else 0
        place = canon_places.get(movie["id"])
        if place:
            movie["canon_group"], movie["canon_pos"] = place

    await dedupe_images(movies)

    with db.connect() as conn:
        known_ids = db.existing_netflix_ids(conn)
        known = db.existing_teasers(conn)
    on_netflix = [m for m in movies if m["on_netflix"]]
    linked = await fill_netflix_links(on_netflix, known_ids)
    new_teasers = await fill_teasers(movies, known)

    stamp = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with db.connect() as conn:
        for movie in movies:
            movie["synced_at"] = stamp
            db.upsert(conn, movie)
        removed = db.drop_missing(conn, [m["id"] for m in movies])
        db.set_meta(conn, "last_sync", stamp)
        db.set_meta(conn, "movie_count", len(on_netflix))
        db.set_meta(conn, "canon_count", sum(1 for m in movies if m["canon_group"] is not None))

    nachgetragen = await katalog_nachtragen()

    log.info(
        "Sync fertig: %s Filme (%s Netflix, %s Kanon), "
        "%s Netflix-Links, %s neue Teaser, %s entfernt, %s nachgetragen",
        len(movies),
        len(on_netflix),
        sum(1 for m in movies if m["canon_group"] is not None),
        linked,
        new_teasers,
        removed,
        nachgetragen,
    )
    return {
        "movies": len(movies),
        "netflix": len(on_netflix),
        "netflix_links": linked,
        "new_teasers": new_teasers,
        "removed": removed,
        "nachgetragen": nachgetragen,
    }


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
    db.init()
    asyncio.run(run())