siftlode/backend/app/plex/overlay.py

"""The HU overlay: a Hungarian Plex section written onto its English twin's rows, split out of
`plex/sync.py`, which runs it once per paired section during the full sync.

The pairing is admin-configured (`plex_overlay_map`, ``huKey:enKey,...``). An overlay section creates
no browsable rows of its own; it only fills the `*_hu` columns of the EN rows it matches by guid.
`plex/sync.py` and this module import each other as MODULES (attribute access at call time), which is
what lets either be imported first.
"""

from collections.abc import Iterable

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import logs, sysconfig
from app.models import PlexItem, PlexLibrary, PlexShow
from app.plex import sync as plex_sync
from app.plex.client import PlexClient
from app.plex.genres import hu_genres
from app.plex.plaintext import plain_text

log = logs.get_logger(__name__)


def overlay_map(db: Session) -> dict[str, str]:
    """The configured HU-overlay pairing: {huSectionKey: enSectionKey} from `plex_overlay_map`
    ('huKey:enKey,...'). Each HU section is synced as a non-browsable overlay onto its EN twin.
    Malformed pairs are skipped."""
    raw = sysconfig.get_str(db, "plex_overlay_map")
    out: dict[str, str] = {}
    for pair in raw.split(","):
        hu, sep, en = pair.partition(":")
        hu, en = hu.strip(), en.strip()
        if sep and hu and en:
            out[hu] = en
    return out


def sync_overlay(
    db: Session, plex: PlexClient, lib: PlexLibrary, en_lib_id: int, stats: dict
) -> None:
    """Write an overlay section's Hungarian metadata onto its EN twin's rows, matched by guid. Creates
    NO browsable rows — an overlay library is never listed on its own. Only title/summary/tagline/genres
    differ by language; the language-neutral people_text feeds both search vectors. A show section's
    episodes are overlaid too: the HU section lists every episode with its Hungarian title and
    summary, and the episode rows live in the EN show library.

    This is a full RECONCILE: after a COMPLETE HU listing every guid-bearing EN row's HU fields are
    cleared, then the matches are repopulated — so an EN row whose HU twin vanished (or was re-guid'd)
    falls back to EN instead of keeping stale Hungarian metadata. A listing cut short only adds. HU
    items with no EN match are skipped."""
    if lib.kind == "movie":
        passes = [
            (
                plex_sync._TYPE_MOVIE,
                select(PlexItem).where(
                    PlexItem.library_id == en_lib_id,
                    PlexItem.kind == "movie",
                    PlexItem.guid.is_not(None),
                ),
            )
        ]
    else:
        passes = [
            (
                plex_sync._TYPE_SHOW,
                select(PlexShow).where(
                    PlexShow.library_id == en_lib_id, PlexShow.guid.is_not(None)
                ),
            ),
            (
                plex_sync._TYPE_EPISODE,
                select(PlexItem).where(
                    PlexItem.library_id == en_lib_id,
                    PlexItem.kind == "episode",
                    PlexItem.guid.is_not(None),
                ),
            ),
        ]
    for item_type, rows_q in passes:
        _overlay_pass(db, plex, lib, item_type, db.execute(rows_q).scalars(), stats)
    db.flush()


def _overlay_pass(
    db: Session,
    plex: PlexClient,
    lib: PlexLibrary,
    item_type: int,
    rows: Iterable[PlexItem] | Iterable[PlexShow],
    stats: dict,
) -> None:
    """One kind of the overlay (movies, shows or episodes) — see `sync_overlay`."""
    # guid -> the EN rows carrying it (a list: two editions of one film can share a guid; both overlay).
    by_guid: dict[str, list[PlexItem | PlexShow]] = {}
    for r in rows:
        if r.guid:  # the passes select guid-bearing rows only
            by_guid.setdefault(r.guid, []).append(r)
    # The HU listing is read first and applied after, so the clear can wait for its verdict.
    hu_by_guid, complete = _read_hu_listing(plex, lib, item_type, by_guid)
    episodes = item_type == plex_sync._TYPE_EPISODE
    if complete:
        _clear_hu(by_guid, episodes)
    else:
        stats["overlay_incomplete"] = stats.get("overlay_incomplete", 0) + 1
        log.warning(
            "the HU listing ended before the server's total; existing HU text is kept",
            extra=logs.event("plex.sync.overlay_incomplete", library=lib.plex_key),
        )
    for g, meta in hu_by_guid.items():
        targets = by_guid[g]
        title_hu = (meta.get("title") or "").strip() or None
        summary_hu = plain_text(meta.get("summary"))
        tagline_hu = plain_text(meta.get("tagline"))
        genres_hu = hu_genres(plex_sync._tags(meta, "Genre"))
        for row in targets:
            row.title_hu = title_hu
            row.summary_hu = summary_hu
            if not episodes:
                row.tagline_hu = tagline_hu
                row.genres_hu = genres_hu or row.genres_hu
        stats["overlaid"] = stats.get("overlaid", 0) + 1


def _read_hu_listing(
    plex: PlexClient, lib: PlexLibrary, item_type: int, by_guid: dict
) -> tuple[dict[str, dict], bool]:
    """The HU section's entries for the guids the EN side carries, and whether the walk reached the
    server's total. The first HU entry for a guid wins; a dup HU item can't overwrite or double-count."""
    progress: dict = {}
    hu_by_guid: dict[str, dict] = {}
    for meta in plex_sync._paginate(plex, lib.plex_key, item_type, progress):
        g = plex_sync._canonical_guid(meta)
        if g and g in by_guid and g not in hu_by_guid:
            hu_by_guid[g] = meta
    return hu_by_guid, bool(progress.get("complete"))


def _clear_hu(by_guid: dict, episodes: bool) -> None:
    """Reconcile step: clear HU fields on all guid-bearing rows; the matches are re-set after. Only
    rows that actually held HU data become dirty (SQLAlchemy skips a no-net-change UPDATE), so this is
    cheap. The genres are the exception: every row gets Hungarian ones, from its HU twin or, lacking
    one, from its own English list — otherwise the HU genre list mixed "Drama" (no twin) with "Dráma".
    Only after a COMPLETE listing: one a Plex hiccup cut short would clear the HU text of every title
    it failed to reach, until the next full sync (the watch reconcile's twin)."""
    for group in by_guid.values():
        for row in group:
            row.title_hu = row.summary_hu = None
            if not episodes:
                row.tagline_hu = None
                row.genres_hu = hu_genres(row.genres)