"""The people are made local ahead of time. The people_enrich job fetches linked people's TMDB
bios (the most-credited first) and caches their portraits resized through Plex; the person page says
when a bio is still on its way (`enriching`), so it can ask again instead of staying empty.
DB lane (Postgres); the portrait cache runs in a temp directory.
"""

from datetime import datetime, timezone

import httpx
import pytest

from app import scheduler
from app.models import PlexItem, PlexLibrary, PlexPerson, User
from app.plex import imgcache
from app.plex import people as plex_people
from app.plex import tmdb as plex_tmdb

CDN = plex_people.PERSON_IMG_HOST + "people/"
DONE = datetime(2024, 1, 1, tzinfo=timezone.utc)


def _movie(db, lib, key, cast):
    db.add(PlexItem(rating_key=key, library_id=lib.id, kind="movie", title=key, cast_names=cast))


def _library(db):
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    return lib


class _Tmdb:
    """A TmdbClient stand-in: records the ids asked for, fails on `fail_on`."""

    def __init__(self, fail_on=()):
        self.asked: list[int] = []
        self.fail_on = set(fail_on)

    def fetch_person(self, pid):
        self.asked.append(pid)
        if pid in self.fail_on:
            raise plex_tmdb.TmdbError("429")
        return {"biography": f"Bio {pid}."}


def test_the_batch_takes_the_most_credited_linked_people_first(db):
    lib = _library(db)
    _movie(db, lib, "a", ["Often", "Sometimes"])
    _movie(db, lib, "b", ["Often", "Sometimes"])
    _movie(db, lib, "c", ["Often", "Unlinked"])
    db.add_all(
        [
            PlexPerson(name="Sometimes", tmdb_person_id=2),
            PlexPerson(name="Often", tmdb_person_id=1),
            PlexPerson(name="Unlinked"),  # no TMDB id: nothing to fetch
            PlexPerson(name="Done", tmdb_person_id=3, tmdb_enriched_at=DONE),
            PlexPerson(name="Rare", tmdb_person_id=4),  # credited nowhere: last
        ]
    )
    db.commit()
    tmdb = _Tmdb()
    assert plex_tmdb.enrich_people_batch(db, tmdb, 2) == {"enriched": 2}  # type: ignore[arg-type]
    assert tmdb.asked == [1, 2]
    db.expire_all()
    often = db.query(PlexPerson).filter_by(name="Often").one()
    assert often.biography == "Bio 1." and often.tmdb_enriched_at is not None
    assert db.query(PlexPerson).filter_by(name="Rare").one().tmdb_enriched_at is None  # next run


def test_a_failed_lookup_is_skipped_and_the_rest_still_land(db):
    db.add_all([PlexPerson(name="A", tmdb_person_id=1), PlexPerson(name="B", tmdb_person_id=2)])
    db.add(PlexPerson(name="C", tmdb_person_id=3))
    db.commit()
    tmdb = _Tmdb(fail_on={2})
    assert plex_tmdb.enrich_people_batch(db, tmdb, 10) == {"enriched": 2, "failed": 1}  # type: ignore[arg-type]
    assert tmdb.asked == [1, 2, 3]  # B refused: skipped, not the end of the batch
    db.expire_all()
    stamped = {p.name: p.tmdb_enriched_at is not None for p in db.query(PlexPerson)}
    assert stamped == {"A": True, "B": False, "C": True}  # B is tried again next run


def test_failures_in_a_row_end_the_batch(db):
    db.add_all([PlexPerson(name=f"P{i}", tmdb_person_id=i) for i in range(1, 7)])
    db.commit()
    tmdb = _Tmdb(fail_on={1, 2, 3, 4, 5, 6})  # TMDB is down
    assert plex_tmdb.enrich_people_batch(db, tmdb, 10) == {"enriched": 0, "failed": 3}  # type: ignore[arg-type]
    assert tmdb.asked == [1, 2, 3]


def test_failures_between_successes_do_not_end_the_batch(db):
    db.add_all([PlexPerson(name=f"P{i}", tmdb_person_id=i) for i in range(1, 9)])
    db.commit()
    tmdb = _Tmdb(fail_on={1, 3, 5, 7})  # fail, ok, fail, ok, …: never three in a row
    assert plex_tmdb.enrich_people_batch(db, tmdb, 10) == {"enriched": 4, "failed": 4}  # type: ignore[arg-type]
    assert tmdb.asked == list(range(1, 9))


def test_the_batch_stores_the_hungarian_biography(db):
    db.add(PlexPerson(name="A", tmdb_person_id=1))
    db.commit()

    class _WithHu(_Tmdb):
        def fetch_person(self, pid):
            return {
                "biography": "English.",
                "translations": {
                    "translations": [{"iso_639_1": "hu", "data": {"biography": "Magyar."}}]
                },
            }

    assert plex_tmdb.enrich_people_batch(db, _WithHu(), 10) == {"enriched": 1}  # type: ignore[arg-type]
    db.expire_all()
    person = db.query(PlexPerson).one()
    assert (person.biography, person.biography_hu) == ("English.", "Magyar.")


class _Plex:
    """A PlexClient stand-in. `down`: urls Plex does not answer for (True: none); `dead`: urls Plex
    answers `gone` for (a dead CDN file: "not found"); `status`: an error status Plex answers every
    other url with."""

    def __init__(self, down=False, dead=(), status=None, gone=404):
        self.asked: list[tuple[str, int | None]] = []
        self.down = down
        self.dead = set(dead)
        self.status = status
        self.gone = gone

    def __enter__(self):
        return self

    def __exit__(self, *a):
        return False

    def image_bytes(self, url, width=None, height=None):
        self.asked.append((url, width))
        if self.down is True or (self.down and url in self.down):
            raise plex_people.PlexError("down") from httpx.ConnectError("refused")
        code = self.gone if url in self.dead else self.status
        if code:
            req = httpx.Request("GET", url)
            status = httpx.HTTPStatusError(
                str(code), request=req, response=httpx.Response(code, request=req)
            )
            raise plex_people.PlexError(str(code)) from status
        return b"small", "image/jpeg"


@pytest.fixture
def cache_dir(tmp_path, monkeypatch):
    monkeypatch.setattr(imgcache, "CACHE_DIR", tmp_path)
    return tmp_path


@pytest.fixture
def plex_on(monkeypatch):
    """The Plex module switched on (the job's gate), with `plex` as its client."""
    state = {"enabled": True, "plex": _Plex()}
    monkeypatch.setattr(plex_people.sysconfig, "get_bool", lambda d, k: state["enabled"])
    monkeypatch.setattr(plex_people, "PlexClient", lambda d: state["plex"])
    return state


def _faces(db, n):
    db.add_all(
        [
            PlexPerson(name=f"P{i}", photo_thumb=plex_people.person_thumb_url(CDN + f"{i}.jpg"))
            for i in range(n)
        ]
    )
    db.commit()


def test_portraits_are_cached_small_and_replace_the_full_size_original(db, cache_dir, plex_on):
    lib = _library(db)
    _movie(db, lib, "a", ["Often"])
    _movie(db, lib, "b", ["Often", "Cached"])
    db.add_all(
        [
            PlexPerson(name="Often", photo_thumb=plex_people.person_thumb_url(CDN + "often.jpg")),
            PlexPerson(name="Cached", photo_thumb=plex_people.person_thumb_url(CDN + "cached.jpg")),
            PlexPerson(name="Faceless"),
        ]
    )
    db.commit()
    small, original = plex_people._portrait_keys(CDN + "often.jpg")
    imgcache.write(original, b"a big original", "image/jpeg")  # cached without Plex
    imgcache.write(plex_people._portrait_keys(CDN + "cached.jpg")[0], b"x", "image/jpeg")
    plex = plex_on["plex"]

    assert plex_people.warm_portraits(db, 10) == {"cached": 1}
    assert plex.asked == [
        (CDN + "often.jpg", plex_people.PORTRAIT_BOX)
    ]  # resized, the cached one skipped
    assert imgcache.read(small) == (b"small", "image/jpeg")
    assert not imgcache.has(original)


def test_portraits_wait_for_plex(db, cache_dir, plex_on, monkeypatch):
    _faces(db, 1)

    def not_configured(d):
        raise plex_people.PlexNotConfigured()

    monkeypatch.setattr(plex_people, "PlexClient", not_configured)
    assert plex_people.warm_portraits(db, 10) == {"cached": 0, "skipped": "disabled"}
    assert list(cache_dir.iterdir()) == []  # no full-size originals behind Plex's back


def test_portraits_wait_for_the_plex_module_to_be_on(db, cache_dir, plex_on):
    _faces(db, 1)
    plex_on["enabled"] = False  # switched off, its server still set up
    assert plex_people.warm_portraits(db, 10) == {"cached": 0, "skipped": "disabled"}
    assert plex_on["plex"].asked == []


def test_plex_not_answering_ends_the_portrait_run(db, cache_dir, plex_on):
    _faces(db, 8)
    plex_on["plex"] = _Plex(down=True)
    assert plex_people.warm_portraits(db, 100) == {"cached": 0, "failed": 3}
    assert len(plex_on["plex"].asked) == 3
    assert not list(cache_dir.glob("*.miss"))  # Plex was down, the files are not to blame


@pytest.mark.parametrize("code", [401, 403, 429, 500, 503])
def test_plex_refusing_is_plex_failing_not_a_dead_file(db, cache_dir, plex_on, code):
    _faces(db, 8)
    plex_on["plex"] = _Plex(status=code)  # a rotated token, a busy or broken transcoder
    assert plex_people.warm_portraits(db, 100) == {"cached": 0, "failed": 3}
    assert len(plex_on["plex"].asked) == 3
    assert not list(cache_dir.glob("*.miss"))  # the files are not to blame: asked again next run


def test_plex_failing_between_successes_does_not_end_the_run(db, cache_dir, plex_on):
    _faces(db, 8)
    plex_on["plex"] = _Plex(down={CDN + f"{i}.jpg" for i in (0, 2, 4, 6)})  # down, ok, down, ok, …
    assert plex_people.warm_portraits(db, 100) == {"cached": 4, "failed": 4}


def test_plex_failing_between_dead_files_does_not_end_the_run(db, cache_dir, plex_on):
    _faces(db, 8)
    plex_on["plex"] = _Plex(  # down, dead, down, dead, …: Plex answers in between
        down={CDN + f"{i}.jpg" for i in (0, 2, 4, 6)}, dead={CDN + f"{i}.jpg" for i in (1, 3, 5, 7)}
    )
    assert plex_people.warm_portraits(db, 100) == {"cached": 0, "failed": 8}
    assert len(plex_on["plex"].asked) == 8


@pytest.mark.parametrize("gone", [404, 410])
def test_dead_headshots_are_left_alone_and_the_rest_still_land(db, cache_dir, plex_on, gone):
    _faces(db, 8)
    dead = {CDN + f"{i}.jpg" for i in range(6)}  # more dead files than any failure limit
    plex_on["plex"] = _Plex(dead=dead, gone=gone)
    assert plex_people.warm_portraits(db, 100) == {"cached": 2, "failed": 6}

    plex_on["plex"] = again = _Plex(dead=dead, gone=gone)
    assert plex_people.warm_portraits(db, 100) == {"cached": 0}
    assert again.asked == []  # dead ones marked, live ones cached: nothing to ask for


def test_the_person_image_route(db, cache_dir, monkeypatch):
    from fastapi.testclient import TestClient

    import app.state as state
    from app.main import app
    from app.routes.plex import playback

    monkeypatch.setattr(state, "_configured_cache", True)
    monkeypatch.setattr(playback, "_require_session", lambda request: 1)
    client = TestClient(app)
    url = CDN + "face.jpg"
    small, original = plex_people._portrait_keys(url)

    def get(u):
        return client.get("/api/plex/person-image", params={"u": u})

    # Only a headshot-shaped url on the CDN host: no query variants of one file, no spellings an HTTP
    # client folds onto it, no other host.
    for bad in (
        url + "?n=1",
        url + "#x",
        CDN + "x/../face.jpg",
        CDN + "./face.jpg",
        plex_people.PERSON_IMG_HOST + "people//face.jpg",
        CDN + ".face.jpg",
        "https://example.com/people/a.jpg",
    ):
        assert get(bad).status_code == 400, bad
        assert plex_people.person_thumb_url(bad) is None, bad  # never stored either

    # No Plex: the original from the CDN, kept under its own key (the job resizes it later).
    def not_configured(d):
        raise plex_people.PlexNotConfigured()

    class _Cdn:
        def __init__(self, *a, **k):
            pass

        def __enter__(self):
            return self

        def __exit__(self, *a):
            return False

        def get(self, u):
            return httpx.Response(
                200,
                content=b"original",
                headers={"content-type": "image/jpeg"},
                request=httpx.Request("GET", u),
            )

    monkeypatch.setattr(playback, "PlexClient", not_configured)
    monkeypatch.setattr(plex_people.httpx, "Client", _Cdn)
    resp = get(url)
    assert (resp.status_code, resp.content) == (200, b"original")
    assert imgcache.has(original) and not imgcache.has(small)
    monkeypatch.setattr(plex_people.httpx, "Client", None)  # the CDN is not asked again
    assert get(url).content == b"original"  # the kept original answers

    # With Plex: resized, and it replaces the original.
    monkeypatch.setattr(playback, "PlexClient", lambda d: _Plex())
    imgcache.drop(original)
    resp = get(url)
    assert (resp.status_code, resp.content) == (200, b"small")
    assert imgcache.has(small)
    assert get(url).content == b"small"  # the cache answers now

    monkeypatch.setattr(playback, "PlexClient", lambda d: _Plex(down=True))
    assert get(CDN + "other.jpg").status_code == 502


def test_the_people_job_runs_its_halves_without_a_tmdb_key(db, cache_dir, plex_on, monkeypatch):
    _faces(db, 1)
    monkeypatch.setattr(plex_tmdb.sysconfig, "get_str", lambda d, k: "")
    assert plex_tmdb.run_people_enrich(db) == {
        "bios": {"skipped": "tmdb_api_key not set"},
        "portraits": {"cached": 1},
    }


def test_a_phase_count_is_a_change_worth_an_audit_row():
    assert scheduler._run_changed({"bios": {"enriched": 3}, "portraits": {"cached": 0}})
    assert scheduler._run_changed({"recent": {"videos_added": 2}, "deep": {}})
    assert not scheduler._run_changed({"bios": {"skipped": "x"}, "portraits": {"cached": 0}})


def test_the_person_page_says_a_bio_is_on_its_way(db, monkeypatch, override_dep):
    from fastapi.testclient import TestClient

    import app.state as state
    from app.auth import current_user
    from app.db import get_db
    from app.main import app
    from app.routes.plex import people

    db.add_all(
        [
            PlexPerson(name="Pending", tmdb_person_id=1),
            PlexPerson(name="Known", tmdb_person_id=2, tmdb_enriched_at=DONE, biography="Bio."),
            PlexPerson(name="Nobody"),
        ]
    )
    user = User(email="enriching@example.com")
    db.add(user)
    db.commit()
    monkeypatch.setattr(state, "_configured_cache", True)
    override_dep(get_db, db.info["request"])
    override_dep(current_user, lambda: user)
    monkeypatch.setattr(people, "_enrich_person_bg", lambda pid: None)
    key = {"value": "key"}
    monkeypatch.setattr(people.sysconfig, "get_str", lambda d, k: key["value"])

    def enriching(name):
        resp = TestClient(app).get(f"/api/plex/person/{name}")
        assert resp.status_code == 200, resp.text
        return resp.json()["enriching"]

    assert enriching("Pending") is True
    assert enriching("Known") is False
    assert enriching("Nobody") is False
    key["value"] = ""  # no TMDB key: nothing will ever arrive, so the page must not wait
    assert enriching("Pending") is False


def test_known_for_is_the_most_voted_roles_not_guest_visits():
    payload = {
        "combined_credits": {
            "cast": [
                # Popular this week, but a visit as himself: not what he is known for.
                {
                    "id": 1,
                    "media_type": "tv",
                    "name": "Late Show",
                    "character": "Himself - Guest",
                    "popularity": 900,
                    "vote_count": 50,
                },
                {
                    "id": 2,
                    "media_type": "tv",
                    "name": "Talk Hour",
                    "character": "",
                    "genre_ids": [10767],
                    "popularity": 800,
                    "vote_count": 400,
                },
                {
                    "id": 3,
                    "media_type": "movie",
                    "title": "Small Film",
                    "character": "Bob",
                    "popularity": 90,
                    "vote_count": 300,
                },
                {
                    "id": 4,
                    "media_type": "movie",
                    "title": "Big Film",
                    "character": "Hero",
                    "popularity": 10,
                    "vote_count": 9000,
                },
            ],
            "crew": [
                {
                    "id": 5,
                    "media_type": "movie",
                    "title": "His Own Film",
                    "job": "Director",
                    "vote_count": 1000,
                },
                {
                    "id": 6,
                    "media_type": "movie",
                    "title": "Thanked In",
                    "job": "Thanks",
                    "vote_count": 99999,
                },
            ],
        }
    }
    kf = plex_tmdb.parse_person(payload)["known_for"]
    assert [k["title"] for k in kf] == ["Big Film", "His Own Film", "Small Film"]


def test_the_page_shows_at_most_fifteen_known_for_titles(db):
    from app.routes.plex import people

    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    for i in range(1, 21):  # TMDB ids start at 1
        db.add(
            PlexItem(
                rating_key=f"k{i}", library_id=lib.id, kind="movie", title=f"T{i}", guid=f"tmdb:{i}"
            )
        )
    db.commit()
    pool = [{"tmdb_id": i, "media_type": "movie", "title": f"T{i}"} for i in range(1, 21)]
    cards = people._resolve_known_for(db, pool, 0, [lib.id], [], "en")
    assert [c["title"] for c in cards] == [f"T{i}" for i in range(1, 16)]