fix(plex): the rails' Restart comes from the one rollup; plain text decodes whole entity names only

Both Continue-watching rails built a show's card state inline; they now take it from rollup.show_state, so a show part watched with nothing to resume offers Play alone there too. A named reference counts only when its whole name is an HTML5 entity: html.unescape still decoded the legacy "&not" inside "&notice;". Tests cover the grid's in-progress-but-not-resumable case through its SQL and both rails.

backend/app/discover/queries.py

@@ -40,7 +40,7 @@ from app.models import (
     VideoState,
 )
 from app.plex import rollup
-from app.plex.cards import ShowState, display_title, leaf_card, movie_card, show_card
+from app.plex.cards import display_title, leaf_card, movie_card, show_card
 from app.plex.query import apply_show_state_filter, content_libs
 
 # How many rows a single rail may carry — Home's rails and the Plex module's Discover rails alike
@@ -256,7 +256,7 @@ def plex_in_progress(
     if show_libs:
         prog = rollup.agg_subquery(db, uid, include_resumed=True).subquery()
         sq = apply_show_state_filter(
-            select(PlexShow, prog.c.resumed_at, prog.c.inprog).join(
+            select(PlexShow, prog.c.resumed_at, prog.c.total, prog.c.watched, prog.c.inprog).join(
                 prog, prog.c.show_id == PlexShow.id
             ),
             prog.c.total,
@@ -264,11 +264,12 @@ def plex_in_progress(
             prog.c.inprog,
             "in_progress",
         ).where(PlexShow.library_id.in_(show_libs))
-        for sh, resumed_at, inprog in db.execute(
+        for sh, resumed_at, total, watched, inprog in db.execute(
             sq.order_by(prog.c.resumed_at.desc().nullslast()).limit(limit)
         ).all():
-            state = ShowState("in_progress", (inprog or 0) > 0)
-            out.append((resumed_at, show_card(sh, state, lang=lang)))
+            out.append(
+                (resumed_at, show_card(sh, rollup.show_state(total, watched, inprog), lang=lang))
+            )
 
     return out
 

backend/app/plex/plaintext.py

@@ -9,6 +9,7 @@ that cleaned the existing rows calls it too, so the two cannot disagree.
 
 import html
 import re
+from html.entities import html5
 
 _BREAK = re.compile(r"<br\s*/?>", re.IGNORECASE)
 # A paragraph or block boundary: a blank line between the two. Only a tag name ends the match, so a
@@ -18,10 +19,17 @@ _TAG = re.compile(r"</?[a-zA-Z][^>]*>")
 _BLANKS = re.compile(r"[ \t]*\n[ \t]*\n\s*")
 
 # A well-formed character reference only. `html.unescape` alone also decodes the legacy forms that
-# have no semicolon, so prose like "AT&notice" or "R&copy" came out as "AT¬ice" and "R©".
+# have no semicolon, so prose like "AT&notice" or "R&copy" came out as "AT¬ice" and "R©". A named
+# one counts only when its whole name is an entity: for "&notice;" `html.unescape` still falls back
+# to the legacy "&not" inside it.
 _REF = re.compile(r"&(?:#[0-9]+|#[xX][0-9a-fA-F]+|[A-Za-z][A-Za-z0-9]*);")
 
 
+def _decode(m: re.Match[str]) -> str:
+    ref = m.group()
+    return html.unescape(ref) if ref[1] == "#" or ref[1:] in html5 else ref
+
+
 def plain_text(value: str | None) -> str | None:
     """`value` with its markup gone: line breaks kept as newlines, paragraphs as a blank line,
     every other tag dropped and entities decoded. None or an all-blank result stays None."""
@@ -31,6 +39,6 @@ def plain_text(value: str | None) -> str | None:
         value = _BREAK.sub("\n", value)
         value = _BLOCK.sub("\n\n", value)
         value = _TAG.sub("", value)
-        value = _REF.sub(lambda m: html.unescape(m.group()), value)
+        value = _REF.sub(_decode, value)
         value = _BLANKS.sub("\n\n", value)
     return value.strip() or None

backend/app/plex/rollup.py

@@ -107,7 +107,7 @@ def state_map(db: Session, user_id: int, show_ids: list[int]) -> dict[int, ShowS
 
 def item_state_map(db: Session, user_id: int, item_ids: list[int]) -> dict[int, PlexState]:
     """A user's PlexState rows for a set of leaf items (movies or episodes), keyed by item_id — the
-    leaf analogue of `state_map` (which rolls a show's episodes up to a status string). One batched
+    leaf analogue of `state_map` (which rolls a show's episodes up to a `ShowState`). One batched
     fetch, so the caller serializes cards without a per-card query. Empty in → empty dict (no `IN ()`)."""
     if not item_ids:
         return {}

backend/app/routes/plex/discover.py

@@ -25,7 +25,6 @@ from app.models import PlexItem, PlexShow, PlexState, User
 from app.plex import rollup
 from app.plex.cards import (
     SHOW_NEW,
-    ShowState,
     display_title,
     leaf_card,
     movie_card,
@@ -239,7 +238,7 @@ def _continue_watching(
         # `apply_show_state_filter` the grid uses, so the rail can't drift from the grid's definition.
         prog = rollup.agg_subquery(db, uid, include_resumed=True).subquery()
         sq = apply_show_state_filter(
-            select(PlexShow, prog.c.resumed_at, prog.c.inprog).join(
+            select(PlexShow, prog.c.resumed_at, prog.c.total, prog.c.watched, prog.c.inprog).join(
                 prog, prog.c.show_id == PlexShow.id
             ),
             prog.c.total,
@@ -252,8 +251,8 @@ def _continue_watching(
             .order_by(prog.c.resumed_at.desc().nullslast())
             .limit(RAIL_LIMIT)
         ).all()
-        for sh, resumed_at, inprog in srows:
-            state = ShowState("in_progress", (inprog or 0) > 0)
+        for sh, resumed_at, total, watched, inprog in srows:
+            state = rollup.show_state(total, watched, inprog)
             rows.append((resumed_at, show_card(sh, state, lang=lang)))
 
     rows.sort(key=lambda r: r[0] or rollup.NO_TS, reverse=True)

backend/tests/test_plex_discover.py

@@ -165,6 +165,30 @@ def test_continue_watching_movies_and_shows(db, monkeypatch, override_dep):
     assert "m1" not in ids
 
 
+def test_a_continue_watching_show_says_whether_it_has_an_episode_to_resume(
+    db, monkeypatch, override_dep
+):
+    # Both in-progress rails (Discover's and Home's) take a show's card state from the one rollup:
+    # a show part watched with nothing to resume offers Play alone, not Continue + Restart.
+    from app.discover.queries import plex_in_progress
+
+    user = _seed(db)
+
+    def rails():
+        d = _get(db, user, monkeypatch, override_dep, "/api/plex/discover?scope=both").json()
+        home = {c["id"]: c for _, c in plex_in_progress(db, user.id, lang="en")}
+        return {c["id"]: c for c in d["continue_watching"]}["sh1"], home["sh1"]
+
+    assert [c["resumable"] for c in rails()] == [True, True]
+    ep1 = db.execute(select(PlexItem).where(PlexItem.rating_key == "sh1e1")).scalar_one()
+    st = db.execute(
+        select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == ep1.id)
+    ).scalar_one()
+    st.status, st.position_seconds = "watched", 0
+    db.commit()
+    assert [(c["status"], c["resumable"]) for c in rails()] == [("in_progress", False)] * 2
+
+
 def test_continue_watching_null_status_in_hidden_out(db, monkeypatch, override_dep):
     # A started movie whose PlexState.status is NULL is still "in progress": `NULL NOT IN (...)` is NULL
     # (not TRUE), so a bare notin_ would wrongly drop it — the or_(is_(None), notin_) branch keeps it.

backend/tests/test_plex_plaintext.py

@@ -33,6 +33,8 @@ HU_SUMMARY = "<p>A műsorban minden p&aacute;rosnak stoppal kell eljutnia.</p>"
         ("a < b and 1 <3 you", "a < b and 1 <3 you"),
         # Only a reference with its semicolon is one: an ampersand in prose stays as written.
         ("AT&notice, R&copy and H&M", "AT&notice, R&copy and H&M"),
+        # ...and only a whole entity name: a legacy one at its start ("&not", "&amp") is not.
+        ("AT&notice; and &ampx;", "AT&notice; and &ampx;"),
         ("&#233;t&#xE9; &copy;", "été ©"),
         ("Plain summary.", "Plain summary."),
         ("  padded  ", "padded"),

backend/tests/test_plex_query_conversion.py

@@ -220,15 +220,18 @@ class TestRollup:
             for i in _get(db, user, monkeypatch, override_dep, "/api/plex/library").json()["items"]
         }
         assert cards["s1"]["resumable"] is True
+        # e1 finished, e2 untouched: in progress, but nothing to resume
+        e1 = db.execute(select(PlexItem).where(PlexItem.rating_key == "e1")).scalar_one()
         db.execute(
-            delete(PlexState).where(PlexState.user_id == user.id, PlexState.position_seconds > 0)
+            delete(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == e1.id)
         )
+        db.add(PlexState(user_id=user.id, item_id=e1.id, status="watched"))
         db.commit()
         cards = {
             i["id"]: i
             for i in _get(db, user, monkeypatch, override_dep, "/api/plex/library").json()["items"]
         }
-        assert cards["s1"]["resumable"] is False
+        assert (cards["s1"]["status"], cards["s1"]["resumable"]) == ("in_progress", False)
 
 
 def _fake_paginate(by_type):