"""Mirror a locally-reachable Plex library into Siftlode's own catalog (plex_* tables).

Playback is from the local physical file, so this only mirrors METADATA + the physical-media facts
(codecs/container/path) needed to browse, search, and later play. It reuses Plex's own codec info
(no ffprobe needed) to classify each item's browser playability. Movies + episodes become playable
LEAF rows (plex_items); shows/seasons are mirrored for the drill-down + episode ordering.

The sync is a full reconcile of the enabled sections (upsert by rating_key) AND prunes rows for
titles that disappeared from Plex: each section diffs the rating_keys it just listed against the ones
stored, and deletes the stale ones (their watch-state / playlist links cascade). The prune is guarded
on a NON-empty listing — a section that returns zero items is left untouched, so a transient Plex
glitch that yields an empty page can't wipe a library (the metadata is cheap to re-sync; an accidental
mass-delete is the only real risk). Library-level reconcile (dropping a whole removed section, the
dead `enabled` column) is a separate follow-up. Cast + intro/credit markers are fetched lazily on the
item-detail/player path (P2), not here.
"""

from datetime import datetime, timezone

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app import logs, sysconfig
from app.models import PlexCollection, PlexItem, PlexLibrary, PlexSeason, PlexShow
from app.plex import detail, people
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.plex.lookups import by_rating_key

log = logs.get_logger(__name__)
# Codecs/containers a browser can play in a native <video> without any transcoding.
_BROWSER_VIDEO = {"h264"}
_BROWSER_AUDIO = {"aac", "mp3", "opus", "vorbis"}
_BROWSER_CONTAINER = {"mp4", "mov", "m4v"}

# Plex section item types.
_TYPE_MOVIE = 1
_TYPE_SHOW = 2
_TYPE_SEASON = 3
_TYPE_EPISODE = 4

_PAGE = 200


def classify_playable(vcodec: str | None, acodec: str | None, container: str | None) -> str:
    """How the browser can play this file: 'direct' (native <video>), 'remux' (copy video, fix
    container / transcode audio — cheap), or 'transcode' (re-encode video — expensive)."""
    v = (vcodec or "").lower()
    a = (acodec or "").lower()
    c = (container or "").lower()
    if not v or v not in _BROWSER_VIDEO:
        return "transcode"
    if a and a not in _BROWSER_AUDIO:
        return "remux"
    if c and c not in _BROWSER_CONTAINER:
        return "remux"
    return "direct"


# Video codecs the prod iGPU (UHD 630 / Quick Sync Gen 9.5) can hardware-DECODE — the precondition for
# the full-GPU VAAPI transcode pipeline. THE single source of this set: plex.stream imports it here so
# the runtime encoder choice (`_want_hw`) and this catalog facet can never disagree by codec. A source
# outside it (AV1, MPEG-4 ASP/XviD…) is software-only even on the iGPU (it can't decode it to feed its
# own encoder).
HW_DECODABLE_CODECS = frozenset({"h264", "hevc", "h265", "vp9", "vc1", "mpeg2video"})
# Color-transfer values marking an HDR source (PQ / HLG). A 10-bit/HDR source breaks the VAAPI
# scale=nv12 path on Gen 9.5 (see plex.stream) → software transcode, so it counts as NOT HW-capable.
_HDR_TRANSFERS = frozenset({"smpte2084", "arib-std-b67"})


def transcode_hw_capable(
    playable: str | None,
    vcodec: str | None,
    bit_depth: int | None = None,
    hdr: bool = False,
) -> bool | None:
    """The stored `PlexItem.transcode_hw` facet: for a `playable="transcode"` item, whether the prod
    iGPU can HARDWARE-transcode it (else software libx264 only); None for a non-transcode item. HW needs
    a HW-decodable codec AND an 8-bit SDR source — a 10-bit/HDR source (bit_depth>8 or an HDR transfer)
    breaks the VAAPI nv12 pipeline and falls back to software. `bit_depth`/`hdr` are unknown at sync
    time (the cheap listing carries no Stream), so they default to 8-bit SDR (codec-only) and the
    bit-depth backfill refines a 10-bit/HDR HEVC to False later."""
    if playable != "transcode":
        return None
    if (vcodec or "").lower() not in HW_DECODABLE_CODECS:
        return False
    if bit_depth is not None and bit_depth > 8:
        return False
    return not hdr


def _video_stream_props(meta: dict) -> tuple[int | None, bool]:
    """(bit_depth, is_hdr) from a FULL Plex metadata dict's video Stream (Media→Part→Stream, streamType
    1). Only the rich metadata call carries Stream; the cheap listing does not — so this is read by the
    bit-depth backfill, not the listing sync. (None, False) when absent/unparseable."""
    for m in meta.get("Media") or []:
        for part in m.get("Part") or []:
            for s in part.get("Stream") or []:
                if s.get("streamType") != 1:  # 1 = video
                    continue
                bd = s.get("bitDepth")
                try:
                    depth = int(bd) if bd else None
                except (TypeError, ValueError):
                    depth = None
                hdr = str(s.get("colorTrc") or "").lower() in _HDR_TRANSFERS
                return depth, hdr
    return None, False


def _epoch(v) -> datetime | None:
    try:
        return datetime.fromtimestamp(int(v), tz=timezone.utc) if v else None
    except (TypeError, ValueError, OSError):
        return None


def _date(v):
    """Parse a Plex 'YYYY-MM-DD' string (originallyAvailableAt) → date, or None."""
    try:
        return datetime.strptime(v, "%Y-%m-%d").date() if v else None
    except (TypeError, ValueError):
        return None


def _tags(meta: dict, key: str, limit: int | None = None) -> list | None:
    """The .tag values of a Plex child-tag array (Genre/Director/Role) present in the listing."""
    vals = [t.get("tag") for t in (meta.get(key) or []) if t.get("tag")]
    if limit:
        vals = vals[:limit]
    return vals or None


def _rating(meta: dict) -> float | None:
    try:
        return float(meta.get("audienceRating") or meta.get("rating"))
    except (TypeError, ValueError):
        return None


# Provider preference for the canonical guid (a stable id shared across a title's language twins).
_GUID_PREFERENCE = ("tmdb", "imdb", "tvdb")


def _canonical_guid(meta: dict) -> str | None:
    """The canonical external id used to pair an EN title with its HU-overlay twin (E5 S5). Reads the
    `Guid` array (present only with includeGuids=1 in the listing), preferring tmdb > imdb > tvdb,
    normalised to '<provider>:<id>' (e.g. 'tmdb:12345'). None when no known provider id is present."""
    found: dict[str, str] = {}
    for g in meta.get("Guid") or []:
        raw = str(g.get("id") or "")
        for prov in _GUID_PREFERENCE:
            if raw.startswith(f"{prov}://"):
                found.setdefault(prov, f"{prov}:{raw.split('://', 1)[1]}")
    for prov in _GUID_PREFERENCE:
        if prov in found:
            return found[prov]
    return None


def _media_facts(meta: dict) -> dict:
    media = meta.get("Media") or []
    if not media:
        return {}
    m = media[0]
    part = (m.get("Part") or [{}])[0]
    return {
        "container": m.get("container") or part.get("container"),
        "codec_video": m.get("videoCodec"),
        "codec_audio": m.get("audioCodec"),
        "size_bytes": part.get("size"),
        "file_path": part.get("file"),
        "part_key": part.get("key"),
    }


def _dur_s(meta: dict) -> int | None:
    d = meta.get("duration")
    return int(d / 1000) if d else None


def _enabled_section_keys(db: Session) -> set[str] | None:
    """The configured allowlist of section keys, or None to mean 'all movie/show sections'."""
    raw = sysconfig.get_str(db, "plex_libraries")
    keys = {k.strip() for k in raw.split(",") if k.strip()}
    return keys or None


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(db: Session) -> dict:
    """Full reconcile of the enabled movie/show sections. Returns a small stats dict."""
    if not sysconfig.get_bool(db, "plex_enabled"):
        return {"skipped": "disabled"}
    stats = {
        "libraries": 0,
        "movies": 0,
        "shows": 0,
        "seasons": 0,
        "episodes": 0,
        "collections": 0,
        "overlaid": 0,
        "pruned": 0,
    }
    wanted = _enabled_section_keys(db)
    overlay_map = _overlay_map(db)  # {huKey: enKey}
    overlay_keys = set(overlay_map)
    try:
        with PlexClient(db) as plex:
            sections = {
                str(s.get("key")): s for s in plex.sections() if s.get("type") in ("movie", "show")
            }
            libs_by_key: dict[str, PlexLibrary] = {}
            # Pass 1 — browsable (non-overlay) sections. HU-overlay sections are handled in pass 2 and
            # are NEVER mirrored as browsable libraries of their own.
            for key, s in sections.items():
                if key in overlay_keys:
                    continue
                if wanted is not None and key not in wanted:
                    continue
                lib = _upsert_library(db, key, s.get("title") or key, s.get("type"))
                libs_by_key[key] = lib
                stats["libraries"] += 1
                if s.get("type") == "movie":
                    _sync_movies(db, plex, lib, stats)
                else:
                    _sync_shows(db, plex, lib, stats)
                _sync_collections(db, plex, lib, stats)
                # Commit each section as it completes: a PlexError on a LATER section then keeps the
                # libraries already mirrored (idempotent — the next run reconciles the rest), instead
                # of the old single end-of-loop commit that a late failure rolled back entirely.
                db.commit()
            # Pass 2 — HU overlays. The EN twins are now synced (with guids), so write each overlay
            # section's Hungarian metadata onto the matched EN rows. Requires the EN section to have been
            # synced in pass 1 (in the allowlist / all) and to share the overlay's kind.
            for hu_key, en_key in overlay_map.items():
                s = sections.get(hu_key)
                en_lib = libs_by_key.get(en_key)
                if s is None or en_lib is None or s.get("type") != en_lib.kind:
                    log.warning(
                        "Plex overlay skipped: hu=%s en=%s (missing section/EN lib or kind mismatch)",
                        hu_key,
                        en_key,
                    )
                    continue
                lib = _upsert_library(db, hu_key, s.get("title") or hu_key, s.get("type"))
                lib.metadata_lang = "hu"
                lib.overlay_of = en_lib.id
                db.flush()
                _sync_overlay(db, plex, lib, en_lib.id, stats)
                db.commit()
            # Rich-fetch a bounded batch of not-yet-enriched titles' FULL cast, so the whole library
            # becomes filterable over successive runs (the cheap listing above only stored the top ~3).
            try:
                stats["people_enriched"] = _enrich_people_batch(db, plex)
            except Exception as e:  # best-effort — never fail the sync over the backfill
                log.warning("Plex people enrich batch failed: %s", e)
            # Probe a bounded batch of HW-decodable transcode items for bit-depth/HDR, so the
            # playback-class filter's HW/SW split is precise (a 10-bit/HDR HEVC → software) over
            # successive runs. Only HW-decodable-codec transcodes are probed — the rest are SW by codec.
            try:
                stats["media_probed"] = _probe_media_batch(db, plex)
            except Exception as e:  # best-effort — never fail the sync over the backfill
                log.warning("Plex media probe batch failed: %s", e)
    except PlexNotConfigured as e:
        return {"skipped": str(e)}
    except PlexError as e:
        db.rollback()
        log.warning("Plex sync failed: %s", e)
        return {"error": str(e)}
    # Prune the headshot cache: a reconcile may have removed titles, orphaning people who no longer
    # appear in ANY item's cast/directors — drop those rows, mirroring how Plex drops orphaned metadata.
    try:
        stats["people_pruned"] = people.prune_orphan_people(db)
    except Exception as e:  # best-effort housekeeping — never fail the sync over it
        log.warning("Plex people prune failed: %s", e)
    log.info("Plex sync done: %s", stats)
    return stats


# How many titles to rich-fetch per sync run for the cast backfill — bounds the extra Plex load; the
# whole library is covered over successive runs (each fetched item is stamped people_enriched_at once).
_ENRICH_BATCH = 60
# Same bounded-per-run pattern for the bit-depth probe. Only HW-decodable-codec transcode items are
# probed (typically a small minority — the bulk of transcodes are old codecs that are SW by codec), so
# a modest batch covers the work-list in a run or two; a stuck item retries next run (watermark unset).
_PROBE_BATCH = 60


def _probe_media_batch(db: Session, plex: PlexClient, limit: int = _PROBE_BATCH) -> int:
    """Rich-fetch up to `limit` not-yet-probed HW-decodable transcode items, read the video Stream's
    bit-depth + HDR, and store `video_bit_depth` + the refined `transcode_hw` (a 10-bit/HDR HEVC →
    False/software). Stamps `media_probed_at` on every attempt (success OR a Plex miss with no Stream),
    so an item is probed once and not re-fetched every sync; a transient fetch FAILURE leaves the
    watermark unset to retry. Oldest-id first. Returns how many items were probed this run."""
    rows = (
        db.execute(
            select(PlexItem)
            .where(
                PlexItem.playable == "transcode",
                func.lower(PlexItem.codec_video).in_(HW_DECODABLE_CODECS),
                PlexItem.media_probed_at.is_(None),
            )
            .order_by(PlexItem.id)
            .limit(limit)
        )
        .scalars()
        .all()
    )
    now = datetime.now(timezone.utc)
    n = 0
    for row in rows:
        try:
            meta = plex.metadata(row.rating_key) or {}
        except Exception as e:  # transient — leave media_probed_at NULL so it retries next run
            log.debug("Plex media probe fetch failed for %s: %s", row.rating_key, e)
            continue
        bit_depth, hdr = _video_stream_props(meta)
        row.video_bit_depth = bit_depth
        row.transcode_hw = transcode_hw_capable("transcode", row.codec_video, bit_depth, hdr)
        row.media_probed_at = (
            now  # probed (even if Plex carried no Stream — don't re-fetch forever)
        )
        n += 1
    db.commit()
    return n


def _enrich_people_batch(db: Session, plex: PlexClient, limit: int = _ENRICH_BATCH) -> int:
    """Rich-fetch the FULL cast of up to `limit` not-yet-enriched movies/shows and persist it (via the
    shared enrich_row_people). Oldest-id first, movies before shows. Stamps people_enriched_at only on
    a SUCCESSFUL fetch, so a transient per-item failure retries next run rather than being lost.
    Returns how many items were enriched this run."""
    movies = (
        db.execute(
            select(PlexItem)
            .where(PlexItem.kind == "movie", PlexItem.people_enriched_at.is_(None))
            .order_by(PlexItem.id)
            .limit(limit)
        )
        .scalars()
        .all()
    )
    remaining = limit - len(movies)
    shows = (
        db.execute(
            select(PlexShow)
            .where(PlexShow.people_enriched_at.is_(None))
            .order_by(PlexShow.id)
            .limit(remaining)
        )
        .scalars()
        .all()
        if remaining > 0
        else []
    )
    now = datetime.now(timezone.utc)
    n = 0
    for row in [*movies, *shows]:
        try:
            meta = plex.metadata(row.rating_key) or {}
        except Exception as e:
            log.debug("Plex people enrich fetch failed for %s: %s", row.rating_key, e)
            continue  # leave people_enriched_at NULL so it's retried on a later run
        people.enrich_row_people(db, row, people.cast_from_meta(meta))
        detail.apply_rich_detail(
            row, meta
        )  # imdb/tagline/full_cast → served from the row, not live
        row.people_enriched_at = now
        n += 1
    db.commit()
    return n


def _upsert_library(db: Session, key: str, title: str, kind: str) -> PlexLibrary:
    lib = db.execute(select(PlexLibrary).where(PlexLibrary.plex_key == key)).scalars().first()
    if lib is None:
        lib = PlexLibrary(plex_key=key)
        db.add(lib)
    lib.title = title
    lib.kind = kind
    lib.synced_at = datetime.now(timezone.utc)
    db.flush()
    return lib


def _paginate(plex: PlexClient, key: str, item_type: int, progress: dict | None = None):
    """Yield a section's items page by page. If `progress` is given, set `progress["complete"]` True
    ONLY when the walk reached the server's reported `total` — a truncated fetch (an empty/short page
    mid-sequence, e.g. a transient Plex hiccup, before `total` is reached) leaves it False so the
    caller can skip pruning. Pruning on a partial listing would delete rows the listing merely failed
    to mention (the metadata is re-syncable, but that still cascades watch-state)."""
    start = 0
    total = None
    while True:
        items, total = plex.section_items(key, item_type, start, _PAGE)
        if not items:
            break
        for it in items:
            yield it
        start += len(items)
        if total and start >= total:
            break
        if len(items) < _PAGE:
            break
    if progress is not None:
        # A truthy `total` we reached = a trustworthy complete listing. `total` unknown/zero-with-rows
        # (a Plex that doesn't report totalSize) is treated as NOT complete, so prune stays conservative.
        progress["complete"] = bool(total) and start >= total


# The section listing exposes at most the top ~20 cast (Plex's own cap); a stored cast longer than this
# can only have come from the rich enrich, never the listing.
_LISTING_CAST_CAP = 20


def _apply_facet_fields(row, meta: dict) -> None:
    """Set the filterable/orderable metadata columns shared by plex_items + plex_shows (identical
    column names) from the cheap section listing: rating, content_rating, studio, release date,
    genres/directors/cast, and the searchable people blob folded into search_vector at sync time."""
    row.rating = _rating(meta)
    row.content_rating = meta.get("contentRating") or None
    row.studio = (meta.get("studio") or None) and str(meta.get("studio"))[:255]
    row.originally_available_at = _date(meta.get("originallyAvailableAt"))
    row.genres = _tags(meta, "Genre")
    row.directors = _tags(meta, "Director")
    # The listing carries only the top ~20 cast, so re-applying it would DOWNGRADE an already-enriched
    # full cast (up to 24, grown by the rich backfill/info-page enrich), silently breaking the
    # person-page cast filter for names past the top ~20. The watermark gate below (`people_enriched_at
    # is None`) normally protects this, but it is NOT enough on its own: a migration that resets the
    # watermark re-opens the regression (0065 nulled it, which is exactly how the S5b fix got undone in
    # prod). So guard STRUCTURALLY too, by LENGTH against the listing cap: a stored cast LONGER than the
    # cap can only be an enriched fuller list, so keep it (robust to any watermark reset). A stored cast
    # at or below the cap is itself listing-sourced, so take the fresh listing — which tracks Plex cast
    # EDITS in EITHER direction (comparing against this listing's length instead would silently keep a
    # stale longer cast when Plex TRIMS one).
    if row.people_enriched_at is None:
        listing_cast = _tags(meta, "Role", limit=_LISTING_CAST_CAP)
        if len(row.cast_names or []) <= _LISTING_CAST_CAP:
            row.cast_names = listing_cast
    people = list(dict.fromkeys((row.directors or []) + (row.cast_names or [])))
    row.people_text = " ".join(people) or None


def _apply_item(row: PlexItem, lib_id: int, kind: str, meta: dict) -> None:
    row.library_id = lib_id
    row.kind = kind
    # EN<->HU pairing key (movies; harmless/unused on episodes). Keep the prior guid if this listing
    # carries none (a server ignoring includeGuids mustn't wipe the overlay's only match key).
    row.guid = _canonical_guid(meta) or row.guid
    row.title = (meta.get("title") or "").strip() or "Untitled"
    row.summary = meta.get("summary")
    row.year = meta.get("year")
    row.duration_s = _dur_s(meta)
    row.thumb_key = meta.get("thumb")
    row.art_key = meta.get("art") or meta.get("grandparentArt")
    row.added_at = _epoch(meta.get("addedAt"))
    _apply_facet_fields(row, meta)  # filterable/orderable extras (shared with shows)
    mf = _media_facts(meta)
    new_vcodec = mf.get("codec_video")
    new_part_key = mf.get("part_key")
    # The physical file was replaced when its video codec changes OR — for a SINGLE-version item — when its
    # Plex Part key changes (a re-encode / quality upgrade gets a new part id). Either way any probed
    # bit-depth is stale, so reset the probe watermark and re-derive transcode_hw from codec alone until the
    # backfill re-probes. Codec alone is NOT enough: a same-codec bit-depth re-encode (8-bit→10-bit HEVC)
    # keeps the codec `hevc` but flips the HW class, and only the part-key move catches it.
    #
    # The part-key check is gated on single-version because `_media_facts` reads media[0]/part[0], and for a
    # MULTI-version item (an original + a Plex "Optimized" copy) that order isn't guaranteed stable across
    # listings — an unguarded part-key compare would then reset+re-probe every sync and oscillate the
    # transcode_hw bucket between the two versions. A multi-version same-codec re-encode not self-healing is
    # an acceptable rare edge; a per-sync re-probe loop is not.
    single_version = len(meta.get("Media") or []) <= 1
    codec_changed = (row.codec_video or "").lower() != (new_vcodec or "").lower()
    if codec_changed or (single_version and row.part_key != new_part_key):
        row.video_bit_depth = None
        row.media_probed_at = None
    row.file_path = mf.get("file_path")
    row.part_key = new_part_key
    row.container = mf.get("container")
    row.codec_video = new_vcodec
    row.codec_audio = mf.get("codec_audio")
    row.size_bytes = mf.get("size_bytes")
    row.playable = classify_playable(
        mf.get("codec_video"), mf.get("codec_audio"), mf.get("container")
    )
    # transcode_hw: keep the bit-depth backfill's authoritative value (it folds in bit-depth + HDR) once
    # probed; otherwise (re)derive the provisional codec-only class. A non-transcode item is always None,
    # even if it had been probed as a transcode before (an audio/container flip to remux without a codec
    # change wouldn't reset the watermark, so pin it here).
    if row.playable != "transcode":
        row.transcode_hw = None
    elif row.media_probed_at is None:
        row.transcode_hw = transcode_hw_capable("transcode", row.codec_video, row.video_bit_depth)


def _prune_stale(
    db: Session, existing: dict, seen: set[str], stats: dict, *, complete: bool
) -> None:
    """Delete stored rows whose rating_key is no longer in the freshly-listed `seen` set — the titles
    that disappeared from Plex (their watch-state / playlist links cascade via the FKs). GUARDED on a
    COMPLETE, non-empty listing: a truncated fetch (`complete` False — a mid-sequence empty/short page
    before the server's total) or a zero-item listing prunes NOTHING, so a transient Plex hiccup can
    never wipe a library over rows it merely failed to list. Callers pass the rating_key -> row map
    they already built for the upsert."""
    if not complete or not seen:
        return
    for rk, row in existing.items():
        if rk not in seen:
            db.delete(row)
            stats["pruned"] = stats.get("pruned", 0) + 1
    db.flush()


def _sync_movies(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
    existing = {
        r.rating_key: r
        for r in db.execute(
            select(PlexItem).where(PlexItem.library_id == lib.id, PlexItem.kind == "movie")
        ).scalars()
    }
    seen: set[str] = set()
    progress: dict = {}
    for meta in _paginate(plex, lib.plex_key, _TYPE_MOVIE, progress):
        rk = str(meta.get("ratingKey") or "")
        if not rk:
            continue
        row = existing.get(rk) or PlexItem(rating_key=rk)
        if row.id is None:
            db.add(row)
        _apply_item(row, lib.id, "movie", meta)
        seen.add(rk)
        stats["movies"] += 1
    db.flush()
    _prune_stale(db, existing, seen, stats, complete=progress.get("complete", False))


def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
    # 1) shows
    shows = {
        r.rating_key: r
        for r in db.execute(select(PlexShow).where(PlexShow.library_id == lib.id)).scalars()
    }
    shows_seen: set[str] = set()
    shows_progress: dict = {}
    for meta in _paginate(plex, lib.plex_key, _TYPE_SHOW, shows_progress):
        rk = str(meta.get("ratingKey") or "")
        if not rk:
            continue
        shows_seen.add(rk)
        sh = shows.get(rk) or PlexShow(rating_key=rk)
        if sh.id is None:
            db.add(sh)
            shows[rk] = sh
        sh.library_id = lib.id
        sh.guid = (
            _canonical_guid(meta) or sh.guid
        )  # EN<->HU pairing key; keep prior if listing has none
        sh.title = (meta.get("title") or "").strip() or "Untitled"
        sh.summary = meta.get("summary")
        sh.year = meta.get("year")
        sh.thumb_key = meta.get("thumb")
        sh.art_key = meta.get("art")
        sh.child_count = meta.get("childCount")
        sh.added_at = _epoch(meta.get("addedAt"))
        _apply_facet_fields(sh, meta)  # same filterable/orderable tags as movies
        stats["shows"] += 1
    db.flush()
    show_id = {rk: sh.id for rk, sh in shows.items()}

    # 2) seasons
    seasons = {
        r.rating_key: r
        for r in db.execute(
            select(PlexSeason).join(PlexShow).where(PlexShow.library_id == lib.id)
        ).scalars()
    }
    seasons_seen: set[str] = set()
    seasons_progress: dict = {}
    for meta in _paginate(plex, lib.plex_key, _TYPE_SEASON, seasons_progress):
        rk = str(meta.get("ratingKey") or "")
        sid = show_id.get(str(meta.get("parentRatingKey") or ""))
        if not rk or sid is None:
            continue
        seasons_seen.add(rk)
        se = seasons.get(rk) or PlexSeason(rating_key=rk)
        if se.id is None:
            db.add(se)
            seasons[rk] = se
        se.show_id = sid
        se.title = meta.get("title")
        se.season_number = meta.get("index")
        se.thumb_key = meta.get("thumb")
        stats["seasons"] += 1
    db.flush()
    season_id = {rk: se.id for rk, se in seasons.items()}

    # 3) episodes (playable leaves)
    episodes = {
        r.rating_key: r
        for r in db.execute(
            select(PlexItem).where(PlexItem.library_id == lib.id, PlexItem.kind == "episode")
        ).scalars()
    }
    episodes_seen: set[str] = set()
    episodes_progress: dict = {}
    for meta in _paginate(plex, lib.plex_key, _TYPE_EPISODE, episodes_progress):
        rk = str(meta.get("ratingKey") or "")
        if not rk:
            continue
        episodes_seen.add(rk)
        row = episodes.get(rk) or PlexItem(rating_key=rk)
        if row.id is None:
            db.add(row)
        _apply_item(row, lib.id, "episode", meta)
        row.show_id = show_id.get(str(meta.get("grandparentRatingKey") or ""))
        row.season_id = season_id.get(str(meta.get("parentRatingKey") or ""))
        row.season_number = meta.get("parentIndex")
        row.episode_number = meta.get("index")
        stats["episodes"] += 1
    db.flush()

    # Prune leaf-first (episodes -> seasons -> shows) so a stale row is deleted directly, before its
    # parent's cascade would reach it — avoids deleting the same ORM object twice. Each level is
    # independently guarded on a non-empty listing by _prune_stale.
    _prune_stale(
        db, episodes, episodes_seen, stats, complete=episodes_progress.get("complete", False)
    )
    _prune_stale(db, seasons, seasons_seen, stats, complete=seasons_progress.get("complete", False))
    _prune_stale(db, shows, shows_seen, stats, complete=shows_progress.get("complete", False))


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/genres differ
    by language; the language-neutral people_text feeds both search vectors.

    This is a full RECONCILE: every guid-bearing EN row's HU fields are cleared first, then the HU
    section repopulates the matches — so an EN row whose HU twin vanished (or was re-guid'd) falls back
    to EN instead of keeping stale Hungarian metadata. HU items with no EN match are skipped."""
    if lib.kind == "movie":
        rows = db.execute(
            select(PlexItem).where(
                PlexItem.library_id == en_lib_id,
                PlexItem.kind == "movie",
                PlexItem.guid.is_not(None),
            )
        ).scalars()
        item_type = _TYPE_MOVIE
    else:
        rows = db.execute(
            select(PlexShow).where(PlexShow.library_id == en_lib_id, PlexShow.guid.is_not(None))
        ).scalars()
        item_type = _TYPE_SHOW
    # guid -> the EN rows carrying it (a list: two editions of one film can share a guid; both overlay).
    by_guid: dict[str, list] = {}
    for r in rows:
        by_guid.setdefault(r.guid, []).append(r)
    # Reconcile step: clear HU fields on all guid-bearing rows. Only rows that actually held HU data
    # become dirty (SQLAlchemy skips a no-net-change UPDATE), so this is cheap; matches are re-set below.
    for group in by_guid.values():
        for row in group:
            row.title_hu = row.summary_hu = row.genres_hu = None
    seen: set[str] = (
        set()
    )  # first HU entry for a guid wins; a dup HU item can't overwrite or double-count
    for meta in _paginate(plex, lib.plex_key, item_type):
        g = _canonical_guid(meta)
        targets = by_guid.get(g) if g else None
        if not targets or g in seen:
            continue
        seen.add(g)
        title_hu = (meta.get("title") or "").strip() or None
        summary_hu = meta.get("summary")
        genres_hu = _tags(meta, "Genre")
        for row in targets:
            row.title_hu = title_hu
            row.summary_hu = summary_hu
            row.genres_hu = genres_hu
        stats["overlaid"] = stats.get("overlaid", 0) + 1
    db.flush()


def _upsert_collection(
    db: Session, existing: PlexCollection | None, rk: str, lib_id: int, meta: dict
) -> PlexCollection:
    """Get-or-create a PlexCollection row and apply its metadata from a Plex `meta` dict. `editable`
    is user-managed and preserved (never set here). Shared by the full sync + the targeted re-sync."""
    col = existing or PlexCollection(rating_key=rk)
    if col.id is None:
        db.add(col)
    col.library_id = lib_id
    col.title = (meta.get("title") or "").strip() or "Untitled"
    col.summary = meta.get("summary")
    col.thumb_key = meta.get("thumb")
    col.child_count = meta.get("childCount")
    col.smart = str(meta.get("smart")) == "1"
    col.synced_at = datetime.now(timezone.utc)
    return col


def resync_collection(db: Session, plex: PlexClient, lib: PlexLibrary, rk: str) -> PlexCollection:
    """Targeted re-sync of ONE collection after a write-back (create/add/remove/rename) — avoids the
    full ~4-min library sync. Upserts the row (preserving `editable`) and reconciles ONLY this
    collection's membership on the affected member rows."""
    meta = plex.metadata(rk) or {}
    col = _upsert_collection(db, by_rating_key(db, PlexCollection, rk), rk, lib.id, meta)
    Model = PlexItem if lib.kind == "movie" else PlexShow
    new_members = {
        str(c.get("ratingKey")) for c in plex.collection_children(rk) if c.get("ratingKey")
    }
    current = db.execute(select(Model).where(Model.collection_keys.contains([rk]))).scalars().all()
    for row in current:  # drop rk from rows that are no longer members
        if row.rating_key not in new_members:
            row.collection_keys = [k for k in (row.collection_keys or []) if k != rk] or None
    to_add = new_members - {r.rating_key for r in current}
    if to_add:
        for row in db.execute(select(Model).where(Model.rating_key.in_(to_add))).scalars():
            row.collection_keys = list(dict.fromkeys([*(row.collection_keys or []), rk]))
    db.flush()
    return col


def delete_collection_local(db: Session, lib: PlexLibrary, rk: str) -> None:
    """Local cleanup after a collection is deleted from Plex: strip rk from members + drop the row."""
    Model = PlexItem if lib.kind == "movie" else PlexShow
    for row in db.execute(select(Model).where(Model.collection_keys.contains([rk]))).scalars():
        row.collection_keys = [k for k in (row.collection_keys or []) if k != rk] or None
    col = by_rating_key(db, PlexCollection, rk)
    if col is not None:
        db.delete(col)
    db.flush()


def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
    """Mirror a library's collections + membership. Reads never touch Plex afterwards. Membership is
    written as `collection_keys` on member movies (plex_items) / shows (plex_shows). A full reconcile:
    collections gone from Plex are pruned, and each member row's keys are rebuilt from scratch."""
    cols = plex.collections(lib.plex_key)
    existing = {
        c.rating_key: c
        for c in db.execute(
            select(PlexCollection).where(PlexCollection.library_id == lib.id)
        ).scalars()
    }
    seen: set[str] = set()
    membership: dict[str, list[str]] = {}  # member rating_key -> [collection rating_key, ...]
    for meta in cols:
        rk = str(meta.get("ratingKey") or "")
        if not rk:
            continue
        _upsert_collection(db, existing.get(rk), rk, lib.id, meta)
        seen.add(rk)
        for child in plex.collection_children(rk):
            crk = str(child.get("ratingKey") or "")
            if crk:
                membership.setdefault(crk, []).append(rk)
        stats["collections"] += 1
    db.flush()
    for rk, col in existing.items():
        if rk not in seen:
            db.delete(col)
    # Rebuild collection_keys on every member row of this library (clear + set).
    for row in db.execute(select(PlexItem).where(PlexItem.library_id == lib.id)).scalars():
        row.collection_keys = membership.get(row.rating_key) or None
    for row in db.execute(select(PlexShow).where(PlexShow.library_id == lib.id)).scalars():
        row.collection_keys = membership.get(row.rating_key) or None
    db.flush()