Fix: give Plex guide titles a poster when Plex returns another entry

CHANGELOG.md

@@ -9,6 +9,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 ### Fixed
 
 - Upload Live TV posters again at the next sweep when Plex deletes them while cleaning bundles.
+- Give a Plex guide title a poster when Plex returns another entry for it, such as "News" for "News at Ten".
 
 ## [0.9.0] - 2026-10-12
 

docs/how-it-works.md

@@ -63,7 +63,7 @@ Built-in marks: Netflix, Prime Video, Apple TV, Disney+, HBO Max, Hulu, Paramoun
 - Plex's own guide, used with antenna tuners, comes with artwork. There only titles found on TMDB get a Posteryard poster; the rest keep Plex's image.
 - A block named after its channel gets a [service collection](#collections) tile: art of the channel's show with the most airtime that has TMDB art, over a dark grey band with the channel's logo. A channel without such a show gets the band with its logo alone.
 
-Plex rebuilds the guide at every guide refresh, about once a day, and shows the guide's own images again. Posteryard checks at the next sweep and selects its posters again without uploading them twice. Plex deletes these uploads when it cleans bundles, once a week by default and when a tool such as ImageMaid asks for it. Posteryard checks at every sweep and uploads them again. The channel logos left of the guide come from the guide; Plex takes no upload there. `posteryard preview --live-tv` saves all guide posters to a folder without changing the server.
+Plex rebuilds the guide at every guide refresh, about once a day, and shows the guide's own images again. Posteryard checks at the next sweep and selects its posters again without uploading them twice. Plex deletes these uploads when it cleans bundles, once a week by default and when a tool such as ImageMaid asks for it. Posteryard checks at every sweep and uploads them again. Plex can return the wrong entry for a title that starts another title, such as "News" and "News at Ten". Posteryard then puts the poster on the entry by its id. The channel logos left of the guide come from the guide; Plex takes no upload there. `posteryard preview --live-tv` saves all guide posters to a folder without changing the server.
 
 Jellyfin keeps each airing as its own item, so every airing in the next 4 days gets the show's poster, about 60 KB each. Jellyfin keeps it until the guide changes that airing. Posteryard reads guide pictures from the XMLTV guide Jellyfin uses. A guide set up as a file path, or one Posteryard cannot reach, gives no guide pictures; the posters are made without them. The channel logos in Jellyfin stay as they are.
 

src/posteryard/jellyfin.py

@@ -381,8 +381,8 @@ def _guide_pictures(self, start: int, end: int) -> dict[tuple[str, int], str]:
                 log.warning("guide pictures skipped", extra={"guide": http.redact(path), "reason": reason})
         return pictures
 
-    def guide_is(self, provider: str, key: str, guid: str) -> bool:
-        return True
+    def guide_target(self, provider: str, key: str, guid: str) -> str | None:
+        return key
 
     def guide_images(self, provider: str, key: str) -> list[Item]:
         tag = self.selected(key, "poster")

src/posteryard/livetv.py

@@ -261,25 +261,30 @@ def apply(self, show: GuideShow, fetch: Callable[[str], Image.Image]) -> Counter
         drawn: list[bytes] = []
         for target in show.targets:
             try:
-                outcome = self._apply_target(show, target, plan, drawn)
+                outcome, direct = self._apply_target(show, target, plan, drawn)
             except (http.RequestError, pipeline.NotFoundError, OSError, ValueError) as exc:
                 log.warning("live TV poster failed", extra={"title": show.title, "reason": str(exc)})
                 tally["failed"] += 1
                 continue
             tally[outcome] += 1
             record = self.store.guide(show.record_key(target))
-            if outcome in CONFIRMED and record and record.image_key.startswith("upload://"):
+            if outcome in CONFIRMED and direct and record and record.image_key.startswith("upload://"):
                 self._confirmed.append(record.key)
         if tally["uploaded"]:
             log.info("live TV poster uploaded", extra={"title": show.title, "using": "; ".join(plan.notes)})
         return tally
 
-    def _apply_target(self, show: GuideShow, target: str, plan: pipeline.Plan, drawn: list[bytes]) -> str:
-        if not self.server.guide_is(show.provider, target, show.guid):
+    def _apply_target(self, show: GuideShow, target: str, plan: pipeline.Plan, drawn: list[bytes]) -> tuple[str, bool]:
+        """The outcome, and whether the target itself reaches the entry. The record stays under the target."""
+        where = self.server.guide_target(show.provider, target, show.guid)
+        if where is None:
             log.info("the server returns another guide entry for this title, skipped", extra={"title": show.title})
-            return "skipped"
+            return "skipped", False
+        return self._apply_at(show, target, where, plan, drawn), where == target
+
+    def _apply_at(self, show: GuideShow, target: str, where: str, plan: pipeline.Plan, drawn: list[bytes]) -> str:
         record = self.store.guide(show.record_key(target))
-        images = self.server.guide_images(show.provider, target)
+        images = self.server.guide_images(show.provider, where)
         selected = next((str(i.get("ratingKey")) for i in images if i.get("selected")), "")
         ours = record.image_key if record else ""
         if selected.startswith("upload://") and selected != ours:
@@ -288,11 +293,13 @@ def _apply_target(self, show: GuideShow, target: str, plan: pipeline.Plan, drawn
         if record and ours in stored and record.fingerprint == plan.fingerprint:
             if selected == ours:
                 return "unchanged"
-            self.server.guide_select(show.provider, target, ours)
+            self.server.guide_select(show.provider, where, ours)
             return "selected again"
         if not drawn:
             drawn.append(jpeg(plan.draw()))
-        image_key = self.server.guide_upload(show.provider, target, drawn[0])
+        if self.server.guide_target(show.provider, where, show.guid) != where:
+            return "skipped"
+        image_key = self.server.guide_upload(show.provider, where, drawn[0])
         self.store.guide_uploaded(show.record_key(target), show.title, plan.fingerprint, image_key, show.guide_picture)
         return "uploaded"
 

src/posteryard/plex.py

@@ -19,6 +19,7 @@
 GUIDE_PROVIDER = re.compile(r"^tv\.plex\.providers\.epg\.[a-z0-9]+:\d{1,9}$")
 GUIDE_KEY = re.compile(r"^[A-Za-z0-9%._~-]{1,2000}$")
 GUIDE_PAGE = 2000
+GUIDE_IDS = 300
 
 
 class Plex:
@@ -253,13 +254,54 @@ def guide_airings(self, provider: str, start: int, end: int) -> list[Item]:
                 return airings
 
     def guide_item(self, provider: str, key: str) -> Item | None:
-        meta = self._get(f"/{_provider(provider)}/metadata/{_guide_key(key)}").get("Metadata") or []
+        try:
+            meta = self._get(f"/{_provider(provider)}/metadata/{_guide_key(key)}").get("Metadata") or []
+        except http.HttpError as exc:
+            if exc.status == 404:
+                return None
+            raise
         return meta[0] if meta else None
 
-    def guide_is(self, provider: str, key: str, guid: str) -> bool:
-        """Plex finds a guide entry by a case-insensitive prefix of its name: "atv" can return "ATV Ana Haber"."""
-        item = self.guide_item(provider, key)
-        return item is not None and str(item.get("guid") or "") == guid
+    def guide_target(self, provider: str, key: str, guid: str) -> str | None:
+        """The key that reaches the entry with this guid, or None.
+
+        Plex finds an entry by its name key loosely: "News" can return "News at Ten". The entry's numeric id is exact.
+        """
+
+        def reaches(target: str | None) -> bool:
+            item = self.guide_item(provider, target) if target else None
+            return item is not None and str(item.get("guid") or "") == guid
+
+        if not guid:
+            return None
+        if reaches(key):
+            return key
+        found = self._guide_id(provider, guid)
+        return found if reaches(found) else None
+
+    def _guide_id(self, provider: str, guid: str) -> str | None:
+        """Plex lists no numeric ids. It returns a list of ids in order and leaves out ids it does not have."""
+
+        def holds(start: int, end: int) -> bool | None:
+            ids = ",".join(str(n) for n in range(start, end))
+            try:
+                items = self._get(f"/{_provider(provider)}/metadata/{ids}").get("Metadata") or []
+            except http.HttpError as exc:
+                if exc.status != 404:
+                    raise
+                items = []
+            return any(str(i.get("guid") or "") == guid for i in items) if items else None
+
+        start = 1
+        while (found := holds(start, start + GUIDE_IDS)) is not True:
+            if found is None:
+                return None
+            start += GUIDE_IDS
+        end = start + GUIDE_IDS
+        while end - start > 1:
+            middle = (start + end) // 2
+            start, end = (start, middle) if holds(start, middle) else (middle, end)
+        return str(start)
 
     def guide_images(self, provider: str, key: str) -> list[Item]:
         images: list[Item] = self._get(f"/{_provider(provider)}/metadata/{_guide_key(key)}/posters").get("Metadata", [])

src/posteryard/server.py

@@ -79,7 +79,7 @@ class GuideServer(Protocol):
 
     def guides(self) -> list[Guide]: ...
     def guide_airings(self, provider: str, start: int, end: int) -> list[Item]: ...
-    def guide_is(self, provider: str, key: str, guid: str) -> bool: ...
+    def guide_target(self, provider: str, key: str, guid: str) -> str | None: ...
     def guide_images(self, provider: str, key: str) -> list[Item]: ...
     def guide_upload(self, provider: str, key: str, jpeg: bytes) -> str: ...
     def guide_select(self, provider: str, key: str, image_key: str) -> None: ...

tests/test_jellyfin.py

@@ -302,7 +302,7 @@ def test_guide_uploads_go_to_each_airing_and_cannot_be_selected_again(serve: Any
             f"POST /Items/{PROGRAMME}/Images/Primary": b"",
         }
     )
-    assert jellyfin.guide_is("guide", PROGRAMME, "anything")
+    assert jellyfin.guide_target("guide", PROGRAMME, "anything") == PROGRAMME
     assert jellyfin.guide_images("guide", PROGRAMME) == [{"ratingKey": "new", "selected": True}]
     assert jellyfin.guide_upload("guide", PROGRAMME, b"jpeg") == "new"
     assert base64.b64decode(next(c[3] for c in server.calls if c[0] == "POST") or b"") == b"jpeg"

tests/test_livetv.py

@@ -66,6 +66,9 @@ def __init__(self, airings: list[Item]) -> None:
     def guide_item(self, provider: str, key: str) -> Item | None:
         return {"guid": f"g/{self.mixed.get(key, key)}", "title": key}
 
+    def _guide_id(self, provider: str, guid: str) -> str | None:
+        return next((k for k, v in self.mixed.items() if f"g/{v}" == guid and k.isdigit()), None)
+
     def guides(self) -> list[Guide]:
         return [Guide(GUIDE, self.refreshed)]
 
@@ -392,7 +395,7 @@ def test_a_deleted_poster_the_run_cannot_restore_stops_making_the_guide_due(
     elif outcome == "failed":
         monkeypatch.setattr(live, "plan", lambda show, fetch: (_ for _ in ()).throw(ValueError("unreadable")))
     elif outcome == "skipped":
-        monkeypatch.setattr(plex, "guide_is", lambda provider, key, guid: False)
+        monkeypatch.setattr(plex, "guide_target", lambda provider, key, guid: None)
     else:
         plex.posters["a"] = [{"ratingKey": "upload://posters/by-hand", "selected": True}]
     assert live.due(now=2000)
@@ -526,13 +529,42 @@ def test_trim_bars_cuts_flat_padding_only() -> None:
     assert cards.trim_bars(Image.new("RGB", (40, 40), (5, 5, 5))).size == (40, 40)
 
 
-def test_a_guide_entry_plex_mixes_up_with_another_is_left_alone(tmp_path: Path) -> None:
+def test_a_guide_entry_plex_mixes_up_with_another_and_no_id_reaches_is_left_alone(tmp_path: Path) -> None:
     live, plex, _ = build(tmp_path, [airing("atv", "a", channel="atv"), airing("ATV Ana Haber", "b", channel="atv")])
     plex.mixed["a"] = "b"
     assert dict(live.run()) == {"skipped": 1, "uploaded": 1}
     assert plex.uploads == ["b"]
 
 
+def test_a_guide_entry_plex_mixes_up_with_another_gets_its_poster_by_numeric_id(tmp_path: Path) -> None:
+    live, plex, _ = build(tmp_path, [airing("News", "a"), airing("News at Ten", "b")])
+    plex.mixed |= {"a": "b", "7": "a"}
+    assert dict(live.run(now=1000)) == {"uploaded": 2}
+    assert sorted(plex.uploads) == ["7", "b"]
+    assert live.store.guide(f"{GUIDE}/a").image_key == plex.posters["7"][1]["ratingKey"]  # type: ignore[union-attr]
+    assert live.store.meta("livetv_confirmed") == f'["{GUIDE}/b"]'
+    plex.refreshed = 2
+    for images in plex.posters.values():
+        images[0]["selected"], images[1]["selected"] = True, False
+    assert dict(live.run(now=2000)) == {"selected again": 2}
+    assert sorted(key for key, _ in plex.selects) == ["7", "b"]
+
+
+def test_a_guide_entry_renumbered_before_the_upload_is_skipped(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    live, plex, _ = build(tmp_path, [airing("News", "a"), airing("News at Ten", "b")])
+    plex.mixed |= {"a": "b", "7": "a"}
+    images = plex.guide_images
+
+    def refresh(provider: str, key: str) -> list[Item]:
+        if key == "7":
+            plex.mixed |= {"7": "c", "8": "a"}
+        return images(provider, key)
+
+    monkeypatch.setattr(plex, "guide_images", refresh)
+    assert dict(live.run(now=1000)) == {"skipped": 1, "uploaded": 1}
+    assert plex.uploads == ["b"]
+
+
 def test_a_remembered_picture_is_used_only_while_plex_shows_the_upload(tmp_path: Path) -> None:
     live, plex, _ = build(tmp_path, [airing("Talk", "a", thumb="https://img.example/talk.jpg")])
     live.run(now=1000)

tests/test_plex.py

@@ -247,3 +247,49 @@ def test_guide_calls_refuse_other_paths(serve: Any, provider: str, key: str) ->
     with pytest.raises(ValueError, match="not a Plex guide"):
         plex.guide_select(provider, key, "upload://posters/x")
     assert server.calls == []
+
+
+def guide_by_id(present: dict[int, str]) -> tuple[Plex, list[str]]:
+    plex, asked = Plex(BASE, "example-token"), []
+
+    def get(path: str, **_: Any) -> dict[str, Any]:
+        asked.append(path)
+        last = path.rsplit("/", 1)[1]
+        ids = [int(n) for n in last.split(",")] if last[0].isdigit() else []
+        guid = present.get(ids[0]) if len(ids) == 1 else None
+        if path.endswith(f"/{SHOW}"):
+            guid = "g/news-at-ten"
+        items = [{"guid": present[n]} for n in ids if n in present] if guid is None else [{"guid": guid}]
+        if not items:
+            raise http.HttpError(404, path)
+        return {"Metadata": items}
+
+    plex._get = get  # type: ignore[method-assign]
+    return plex, asked
+
+
+def test_guide_target_is_the_name_key_when_it_reaches_the_entry() -> None:
+    plex, asked = guide_by_id({})
+    assert plex.guide_target(GUIDE, SHOW, "g/news-at-ten") == SHOW
+    assert len(asked) == 1
+
+
+def test_guide_target_finds_the_numeric_id_when_the_name_key_reaches_another_entry() -> None:
+    present = {n: f"g/{n}" for n in range(1, 700) if n % 7}
+    present[452] = "g/news"
+    plex, asked = guide_by_id(present)
+    assert plex.guide_target(GUIDE, SHOW, "g/news") == "452"
+    assert len(asked) < 15
+
+
+def test_guide_target_is_none_when_no_entry_has_the_guid() -> None:
+    plex, asked = guide_by_id({n: f"g/{n}" for n in range(1, 400)})
+    assert plex.guide_target(GUIDE, SHOW, "g/news") is None
+    assert len(asked) == 4
+    assert asked[-1].endswith(",900")
+
+
+def test_guide_target_is_none_for_an_entry_without_a_guid() -> None:
+    plex, asked = guide_by_id({1: ""})
+    assert plex.guide_target(GUIDE, SHOW, "") is None
+    assert asked == []