diff --git a/backend/services/notifications.py b/backend/services/notifications.py index d6fd4373..9de5f6d1 100644 --- a/backend/services/notifications.py +++ b/backend/services/notifications.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections import defaultdict from collections.abc import Iterable from datetime import UTC, datetime, timedelta from typing import Any @@ -19,8 +20,15 @@ ) from backend.core.logger import LOG +from backend.core.rule_engine import ( + FIELD_LABELS, + OPERATOR_LABELS, + VALUELESS_OPERATORS, + format_rule_value, +) from backend.database import async_db from backend.database.models import ( + Episode, GeneralSettings, Movie, MovieVersion, @@ -281,73 +289,235 @@ def _compose_title( return f"{title}: {subject}" -def _format_cleanup_candidate_line( - candidate: dict[str, Any], *, include_reason: bool = False -) -> str: - """Format a single cleanup candidate as one scannable bullet.""" - media_type = str(candidate.get("media_type") or "").lower() - season_number = candidate.get("season_number") - episode_number = candidate.get("episode_number") - version_file_name = str(candidate.get("version_file_name") or "").strip() +# Conditions that only scope a rule to part of the library. They say where a +# rule looks, not why a title matched, and their values are opaque server ids. +_SCOPE_ONLY_REASON_FIELDS = frozenset({"library.id"}) - parts = [ - _media_label( - candidate.get("media_title") or "Unknown", candidate.get("media_year") + +def _notification_reasons(reason_data: Any) -> list[dict[str, Any]]: + """Turn stored candidate reasons into what a notification can show. + + Each reason keeps its rule name plus two renderings of its conditions: + "full" with the measured value, as the candidates page shows it, and + "short" without it, so episodes matched by the same rule collapse into one + line. Library scoping conditions are dropped; see _SCOPE_ONLY_REASON_FIELDS. + """ + reasons: list[dict[str, Any]] = [] + for part in reason_data if isinstance(reason_data, list) else []: + if not isinstance(part, dict): + continue + full: list[str] = [] + short: list[str] = [] + for condition in part.get("conditions") or []: + if not isinstance(condition, dict): + continue + field = str(condition.get("field") or "") + if field in _SCOPE_ONLY_REASON_FIELDS: + continue + operator = str(condition.get("operator") or "") + label = str(condition.get("field_label") or FIELD_LABELS.get(field, field)) + op = str( + condition.get("operator_label") + or OPERATOR_LABELS.get(operator, operator) + ) + display = str(condition.get("display") or "").strip() + if operator in VALUELESS_OPERATORS: + brief = f"{label} {op}".strip() + else: + expected = condition.get("expected") + values = expected if isinstance(expected, list) else [expected] + rendered = ", ".join( + format_rule_value(field, value) + for value in values + if value is not None + ) + brief = f"{label} {op} {rendered}".strip() + if brief: + short.append(brief) + full.append(display or brief) + rule = str(part.get("rule_name") or "").strip() + if full or rule: + reasons.append({"rule": rule, "full": full, "short": short}) + return reasons + + +def _episode_codes(pairs: Iterable[tuple[int, int]]) -> str: + """Compress season/episode pairs into codes such as "S01E05-E07, S02E01".""" + by_season: dict[int, list[int]] = defaultdict(list) + for season_number, episode_number in sorted(set(pairs)): + by_season[season_number].append(episode_number) + codes: list[str] = [] + for season_number, episodes in by_season.items(): + start = prev = episodes[0] + runs: list[tuple[int, int]] = [] + for episode_number in episodes[1:]: + if episode_number == prev + 1: + prev = episode_number + continue + runs.append((start, prev)) + start = prev = episode_number + runs.append((start, prev)) + for first, last in runs: + code = f"S{season_number:02d}E{first:02d}" + codes.append(code if first == last else f"{code}-E{last:02d}") + return ", ".join(codes) + + +def _candidate_group_scope(items: list[dict[str, Any]]) -> str: + """Describe which part of a title the grouped candidates cover.""" + media_type = str(items[0].get("media_type") or "").lower() + if media_type == "series": + if any( + item.get("season_number") is None and item.get("episode_number") is None + for item in items + ): + return "Entire series" + episodes = [ + (item["season_number"], item["episode_number"]) + for item in items + if isinstance(item.get("season_number"), int) + and isinstance(item.get("episode_number"), int) + ] + seasons = sorted( + { + item["season_number"] + for item in items + if isinstance(item.get("season_number"), int) + and not isinstance(item.get("episode_number"), int) + } ) - ] - if media_type == "series" and isinstance(season_number, int): - if isinstance(episode_number, int): - parts.append(f"S{season_number:02d}E{episode_number:02d}") + parts: list[str] = [] + if seasons: + parts.append( + f"Season {seasons[0]}" + if len(seasons) == 1 and not episodes + else ", ".join(f"S{number:02d}" for number in seasons) + ) + if episodes: + parts.append(_episode_codes(episodes)) + return ", ".join(parts) + if len(items) > 1: + return f"{len(items)} versions" + return "" + + +def _candidate_reason_lines(items: list[dict[str, Any]]) -> list[str]: + """Return the distinct reasons behind a group of candidates. + + A lone candidate keeps its measured values ("Days since watched > 7 (8)"). + A group drops them, since every episode carries a slightly different number + and listing each would repeat the same rule once per episode. + """ + key = "full" if len(items) == 1 else "short" + seen: list[str] = [] + for item in items: + reasons = item.get("reasons") + if isinstance(reasons, list) and reasons: + texts = [] + for reason in reasons: + if not isinstance(reason, dict): + continue + conditions = [c for c in reason.get(key) or [] if c] + rule = str(reason.get("rule") or "").strip() + if conditions: + texts.append(", ".join(conditions)) + elif rule: + texts.append(rule) else: - parts.append(f"Season {season_number}") - elif version_file_name: - parts.append(_code(version_file_name)) - parts.append(_format_bytes(int(candidate.get("estimated_space_bytes") or 0))) - - line = f"- {_SEPARATOR.join(parts)}" - if include_reason: - reasons = candidate.get("reason_tokens") - if isinstance(reasons, list): - tokens = [ - str(reason).strip() - for reason in reasons[:_MAX_REASON_TOKENS] - if str(reason).strip() - ] - if tokens: - line += f" - {', '.join(tokens)}" - return line + # contexts built before structured reasons existed + tokens = item.get("reason_tokens") + texts = [str(t).strip() for t in tokens or [] if str(t).strip()] + for text in texts[:_MAX_REASON_TOKENS]: + if text not in seen: + seen.append(text) + return seen[:_MAX_REASON_TOKENS] + + +def _group_cleanup_candidates( + candidates: list[dict[str, Any]], +) -> list[tuple[str, list[dict[str, Any]]]]: + """Group candidates by title so one show reads as one line, biggest first.""" + groups: dict[tuple[str, Any], list[dict[str, Any]]] = {} + for candidate in candidates: + if not isinstance(candidate, dict): + continue + key = ( + str(candidate.get("media_type") or "").lower(), + candidate["media_id"] + if candidate.get("media_id") is not None + # Older contexts did not carry the database identity. + else ( + str(candidate.get("media_title") or "Unknown"), + candidate.get("media_year"), + ), + ) + groups.setdefault(key, []).append(candidate) + ordered = sorted( + groups.items(), + key=lambda entry: ( + -sum(int(item.get("estimated_space_bytes") or 0) for item in entry[1]) + ), + ) + return [(key[0], items) for key, items in ordered] + + +def _format_cleanup_group( + items: list[dict[str, Any]], *, include_reasons: bool +) -> list[str]: + """Render one title as a bullet, with its reasons as sub-bullets.""" + first = items[0] + parts = [ + f"**{_media_label(first.get('media_title') or 'Unknown', first.get('media_year'))}**" + ] + scope = _candidate_group_scope(items) + if scope: + parts.append(scope) + parts.append( + _format_bytes( + sum(int(item.get("estimated_space_bytes") or 0) for item in items) + ) + ) + lines = [f"- {_SEPARATOR.join(parts)}"] + if include_reasons: + if len(items) == 1: + file_name = str(first.get("version_file_name") or "").strip() + if file_name: + lines.append(f" - {_code(file_name)}") + lines.extend(f" - _{reason}_" for reason in _candidate_reason_lines(items)) + return lines def _cleanup_candidate_lines( - candidates: list[dict[str, Any]], *, include_reasons: bool + groups: list[tuple[str, list[dict[str, Any]]]], *, include_reasons: bool ) -> list[str]: - """Group the shown candidates by media type so the list reads in blocks.""" - grouped: dict[str, list[dict[str, Any]]] = {"Movies": [], "Series": [], "Other": []} - for candidate in candidates: - media_type = str(candidate.get("media_type") or "").lower() - key = ( + """Render grouped titles, with Movies/Series headings when both appear.""" + sections: dict[str, list[list[dict[str, Any]]]] = { + "Movies": [], + "Series": [], + "Other": [], + } + for media_type, items in groups: + label = ( "Movies" if media_type == "movie" else "Series" if media_type == "series" else "Other" ) - grouped[key].append(candidate) + sections[label].append(items) - populated = [(label, items) for label, items in grouped.items() if items] - # a lone group needs no heading; the bullets already read as one list + populated = [(label, entries) for label, entries in sections.items() if entries] + # a lone section needs no heading; the bullets already read as one list show_headings = len(populated) > 1 lines: list[str] = [] - for label, items in populated: + for label, entries in populated: if show_headings: if lines: lines.append("") lines.append(f"**{label}**") - lines.extend( - _format_cleanup_candidate_line(item, include_reason=include_reasons) - for item in items - ) + for items in entries: + lines.extend(_format_cleanup_group(items, include_reasons=include_reasons)) return lines @@ -406,8 +576,10 @@ def _compose_body( max_items = int(pref.get("max_items") or 5) max_items = min(max(max_items, 1), 20) - shown = [item for item in candidates[:max_items] if isinstance(item, dict)] - extra = max(0, len(candidates) - len(shown)) + # max_items counts titles: ten new episodes of one show are one entry + groups = _group_cleanup_candidates(candidates) + shown = groups[:max_items] + extra = max(0, len(groups) - len(shown)) lines = [ f"**{count}** new {noun}{_SEPARATOR}" f"**{_format_bytes(total_bytes)}** reclaimable" @@ -421,7 +593,7 @@ def _compose_body( ) if extra > 0: lines.append("") - lines.append(f"_and {extra} more_") + lines.append(f"_and {extra} more {_plural(extra, 'title')}_") return title, "\n".join(lines), _DEFAULT_BODY_FORMAT if detail == "compact": @@ -578,6 +750,7 @@ async def build_cleanup_notification_context( r.movie_version_id for r in rows if r.movie_version_id is not None } season_ids = {r.season_id for r in rows if r.season_id is not None} + episode_ids = {r.episode_id for r in rows if r.episode_id is not None} movies = { m.id: m @@ -626,10 +799,26 @@ async def build_cleanup_notification_context( ) } + episodes = { + e.id: e + for e in ( + ( + await session.execute( + select(Episode).where(Episode.id.in_(episode_ids)) + ) + ) + .scalars() + .all() + if episode_ids + else [] + ) + } + candidates: list[dict[str, Any]] = [] total_reclaimable_bytes = 0 for row in rows: season = None + episode = None if row.media_type.value == "movie": m = movies.get(row.movie_id) v = versions.get(row.movie_version_id) @@ -639,6 +828,7 @@ async def build_cleanup_notification_context( else: s = series.get(row.series_id) season = seasons.get(row.season_id) + episode = episodes.get(row.episode_id) media_title = s.title if s else "Unknown" media_year = s.year if s else None version_file_name = None @@ -656,15 +846,21 @@ async def build_cleanup_notification_context( candidates.append( { "media_type": row.media_type.value, + "media_id": row.movie_id + if row.media_type.value == "movie" + else row.series_id, "media_title": media_title, "media_year": media_year, "season_number": season.season_number if row.media_type.value == "series" and season else None, - "episode_number": None, + "episode_number": episode.episode_number + if row.media_type.value == "series" and episode + else None, "version_file_name": version_file_name, "estimated_space_bytes": size, "reason_tokens": reason_tokens, + "reasons": _notification_reasons(reason_data), } ) @@ -675,6 +871,37 @@ async def build_cleanup_notification_context( } +# Apprise sends Telegram legacy Markdown, which has no list syntax, so a "- " +# bullet arrives as a literal hyphen. Swapping in bullet glyphs keeps the +# structure readable there; every other transport gets proper markdown lists. +_TELEGRAM_SCHEMES = ("tgram://", "telegram://") + + +def _is_telegram_url(url: str) -> bool: + return url.strip().lower().startswith(_TELEGRAM_SCHEMES) + + +def _telegram_bullets(body: str) -> str: + """Replace markdown list markers with bullet glyphs outside code fences.""" + out: list[str] = [] + in_fence = False + for line in body.split("\n"): + if line.startswith("```"): + in_fence = not in_fence + elif not in_fence: + if line.startswith("- "): + line = f"• {line[2:]}" + elif line.startswith(" - "): + line = f" ◦ {line[4:]}" + out.append(line) + return "\n".join(out) + + +def _destination_key(url: str) -> str: + """Identity of a destination, used to send each event once per target.""" + return url.strip().rstrip("/") + + @retry( stop=stop_after_attempt(4), wait=wait_exponential(multiplier=1, min=1, max=10), @@ -700,6 +927,8 @@ async def send_notification( ap = apprise.Apprise(asset=_APPRISE_ASSET) ap.add(url) label = log_label or url + if body_format == apprise.NotifyFormat.MARKDOWN and _is_telegram_url(url): + message = _telegram_bullets(message) try: result = await ap.async_notify( @@ -759,8 +988,14 @@ async def notify_user( message: str, body_format: apprise.NotifyFormat = _DEFAULT_BODY_FORMAT, context: dict[str, Any] | None = None, + delivered: set[str] | None = None, ) -> dict[str, int]: - """Send a notification to a specific user based on their notification preferences.""" + """Send a notification to a specific user based on their notification preferences. + + `delivered` is shared by callers fanning one event out to several users. + Two accounts pointing at the same destination (an admin and their own user + account, both on one Telegram chat) then receive the event once, not twice. + """ results = {"sent": 0, "failed": 0} async with async_db() as session: @@ -793,6 +1028,15 @@ async def notify_user( if destination is None: continue url, log_label = destination + if delivered is not None: + key = _destination_key(url) + if key in delivered: + LOG.debug( + f"Skipping {notification_type} to {log_label}: already " + "delivered to this destination for this event" + ) + continue + delivered.add(key) composed_title, composed_message, composed_format = _compose_notification( notification_type=notification_type, @@ -870,6 +1114,7 @@ async def notify_admins( LOG.warning("No active admin users found to send notification to") return results + delivered: set[str] = set() for user in admin_users: user_results = await notify_user( user_id=user.id, @@ -878,6 +1123,7 @@ async def notify_admins( message=message, body_format=body_format, context=context, + delivered=delivered, ) results["sent"] += user_results["sent"] results["failed"] += user_results["failed"] @@ -896,6 +1142,7 @@ async def notify_users( """Send a notification to multiple specific users.""" results = {"sent": 0, "failed": 0} + delivered: set[str] = set() for user_id in user_ids: user_results = await notify_user( user_id=user_id, @@ -904,6 +1151,7 @@ async def notify_users( message=message, body_format=body_format, context=context, + delivered=delivered, ) results["sent"] += user_results["sent"] results["failed"] += user_results["failed"] @@ -937,6 +1185,7 @@ async def notify_all_users( return results type_field = _notification_type_to_field(notification_type) + delivered: set[str] = set() for user in users: eligible_settings = [ @@ -956,6 +1205,14 @@ async def notify_all_users( if destination is None: continue url, log_label = destination + key = _destination_key(url) + if key in delivered: + LOG.debug( + f"Skipping {notification_type} to {log_label}: already " + "delivered to this destination for this event" + ) + continue + delivered.add(key) composed_title, composed_message, composed_format = _compose_notification( notification_type=notification_type, diff --git a/tests/test_notification_delivery.py b/tests/test_notification_delivery.py new file mode 100644 index 00000000..551d409c --- /dev/null +++ b/tests/test_notification_delivery.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from backend.database import Base +from backend.database.models import ( + Movie, + NotificationSetting, + ReclaimCandidate, + Series, + User, +) +from backend.enums import MediaType, NotificationType, UserRole +from backend.services import notifications + +_SHARED_CHAT = "tgram://123456:ABCdef/987654" + + +@pytest.fixture +async def sent(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]: + """Two admins whose destinations both point at one Telegram chat.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:", future=True) + session_maker = async_sessionmaker( + engine, expire_on_commit=False, class_=AsyncSession + ) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + async with session_maker() as session: + for index, url in enumerate((_SHARED_CHAT, f"{_SHARED_CHAT}/", _SHARED_CHAT)): + user = User( + username=f"admin{index}", + password_hash="x", + role=UserRole.ADMIN, + is_active=True, + ) + session.add(user) + await session.flush() + session.add( + NotificationSetting( + user_id=user.id, + name=f"telegram {index}", + url=url, + enabled=True, + admin_message=True, + new_cleanup_candidates=True, + ) + ) + # someone on a different chat must still hear about it + other = User( + username="other", + password_hash="x", + role=UserRole.ADMIN, + is_active=True, + ) + session.add(other) + await session.flush() + session.add( + NotificationSetting( + user_id=other.id, + name="discord", + url="discord://webhook_id/webhook_token", + enabled=True, + admin_message=True, + new_cleanup_candidates=True, + ) + ) + await session.commit() + + calls: list[dict[str, Any]] = [] + + async def fake_send(**kwargs: Any) -> bool: + calls.append(kwargs) + return True + + monkeypatch.setattr(notifications, "async_db", session_maker) + monkeypatch.setattr(notifications, "send_notification", fake_send) + return calls + + +@pytest.mark.anyio +async def test_admins_sharing_a_chat_get_one_message( + sent: list[dict[str, Any]], +) -> None: + results = await notifications.notify_admins( + NotificationType.ADMIN_MESSAGE, "Heads up", "Something happened" + ) + + assert sorted(call["url"].rstrip("/") for call in sent) == sorted( + [_SHARED_CHAT, "discord://webhook_id/webhook_token"] + ) + assert results == {"sent": 2, "failed": 0} + + +@pytest.mark.anyio +async def test_all_users_fan_out_is_deduplicated_too( + sent: list[dict[str, Any]], +) -> None: + await notifications.notify_all_users( + NotificationType.NEW_CLEANUP_CANDIDATES, + "New Cleanup Candidates Found", + "There are 2 new cleanup candidates", + context={"created_count": 2, "total_reclaimable_bytes": 0, "candidates": []}, + ) + + assert len(sent) == 2 + + +@pytest.mark.anyio +async def test_cleanup_context_carries_episode_numbers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from backend.database.models import Episode, Season + + engine = create_async_engine("sqlite+aiosqlite:///:memory:", future=True) + session_maker = async_sessionmaker( + engine, expire_on_commit=False, class_=AsyncSession + ) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + monkeypatch.setattr(notifications, "async_db", session_maker) + + started = datetime.now(UTC) + async with session_maker() as session: + series = Series(title="Carrie", tmdb_id=1, year=2026) + session.add(series) + await session.flush() + season = Season(series_id=series.id, season_number=1) + session.add(season) + await session.flush() + for number in (5, 6): + episode = Episode(season_id=season.id, episode_number=number) + session.add(episode) + await session.flush() + session.add( + ReclaimCandidate( + media_type=MediaType.SERIES, + series_id=series.id, + season_id=season.id, + episode_id=episode.id, + estimated_space_bytes=500_000_000, + matched_rule_ids=[], + matched_criteria={}, + reason="Old episodes", + reason_data=[ + { + "rule_name": "Old episodes", + "conditions": [ + { + "field": "library.id", + "operator": "contains_any", + "expected": ["b539bda9"], + "display": "Library matches any b539bda9", + } + ], + } + ], + ) + ) + await session.commit() + + context = await notifications.build_cleanup_notification_context( + created_count=2, created_since=started + ) + + numbers = sorted(c["episode_number"] for c in context["candidates"]) + assert numbers == [5, 6] + assert {c["media_id"] for c in context["candidates"]} == {series.id} + assert len(notifications._group_cleanup_candidates(context["candidates"])) == 1 + # a rule whose only condition is library scoping falls back to its name + assert context["candidates"][0]["reasons"] == [ + {"rule": "Old episodes", "full": [], "short": []} + ] + + +@pytest.mark.anyio +@pytest.mark.parametrize("media_type", [MediaType.MOVIE, MediaType.SERIES]) +async def test_cleanup_keeps_same_named_media_separate( + monkeypatch: pytest.MonkeyPatch, media_type: MediaType +) -> None: + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + session_maker = async_sessionmaker(engine, expire_on_commit=False) + try: + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + monkeypatch.setattr(notifications, "async_db", session_maker) + started = datetime.now(UTC) + async with session_maker() as session: + for tmdb_id in (100, 200): + model = Movie if media_type == MediaType.MOVIE else Series + media = model(title="Same Name", year=2020, tmdb_id=tmdb_id) + session.add(media) + await session.flush() + session.add( + ReclaimCandidate( + media_type=media_type, + movie_id=media.id if media_type == MediaType.MOVIE else None, + series_id=media.id if media_type == MediaType.SERIES else None, + estimated_space_bytes=1024**3, + matched_rule_ids=[], + matched_criteria={}, + reason="Test", + reason_data=[], + ) + ) + await session.commit() + + context = await notifications.build_cleanup_notification_context( + created_count=2, created_since=started + ) + assert len({c["media_id"] for c in context["candidates"]}) == 2 + _, body, _ = notifications._compose_body( + notification_type=NotificationType.NEW_CLEANUP_CANDIDATES, + setting=NotificationSetting(user_id=1, enabled=True), + fallback_title="New candidates", + fallback_message="", + context=context, + ) + assert body.count("**Same Name (2020)**") == 2 + assert "versions" not in body + finally: + await engine.dispose() diff --git a/tests/test_notification_formatting.py b/tests/test_notification_formatting.py index 7e5d8dbb..e61183a5 100644 --- a/tests/test_notification_formatting.py +++ b/tests/test_notification_formatting.py @@ -3,13 +3,18 @@ from typing import Any import apprise +import pytest from backend.database.models import NotificationSetting from backend.enums import NotificationType from backend.services.notifications import ( _compose_body, + _episode_codes, _format_bytes, + _is_telegram_url, + _notification_reasons, _notify_type_for, + _telegram_bullets, _truncate_body, ) @@ -183,8 +188,10 @@ def test_cleanup_body_groups_by_media_type() -> None: assert body.startswith("**3** new cleanup candidates · **8.4 GB** reclaimable") assert "**Movies**" in body assert "**Series**" in body - assert "- The Matrix (1999) · `The.Matrix.1999_x265-GRP.mkv` · 5.1 GB" in body - assert "- Breaking Bad (2008) · Season 2 · 3.3 GB" in body + assert "- **The Matrix (1999)** · 5.1 GB" in body + assert "- **Breaking Bad (2008)** · Season 2 · 3.3 GB" in body + # file names only appear alongside reasons, where they explain the match + assert "The.Matrix.1999_x265-GRP.mkv" not in body # reasons stay out of the default summary view assert "Watched 90+ days ago" not in body @@ -214,12 +221,22 @@ def test_cleanup_body_appends_reasons_when_requested() -> None: } }, ) - assert "- Watched 90+ days ago, Not requested" in body + assert " - `The.Matrix.1999_x265-GRP.mkv`" in body + assert " - _Watched 90+ days ago_" in body + assert " - _Not requested_" in body def test_cleanup_body_reports_the_remainder() -> None: context = _cleanup_context(created_count=9) - context["candidates"] = context["candidates"] * 5 + context["candidates"] = [ + { + "media_type": "movie", + "media_title": f"Movie {index}", + "media_year": 2000 + index, + "estimated_space_bytes": 1_000_000_000 - index, + } + for index in range(9) + ] _, body = _compose( NotificationType.NEW_CLEANUP_CANDIDATES, title="New Cleanup Candidates Found", @@ -231,7 +248,7 @@ def test_cleanup_body_reports_the_remainder() -> None: } }, ) - assert body.endswith("_and 6 more_") + assert body.endswith("_and 5 more titles_") def test_cleanup_count_only_detail() -> None: @@ -260,3 +277,166 @@ def test_format_bytes() -> None: assert _format_bytes(512) == "512 B" assert _format_bytes(1024) == "1.0 KB" assert _format_bytes(5_476_083_302) == "5.1 GB" + + +def _episode( + title: str, + season: int, + episode: int, + size: int, + days: int, + year: int = 2020, +) -> dict[str, Any]: + """An episode candidate shaped like build_cleanup_notification_context output.""" + return { + "media_type": "series", + "media_title": title, + "media_year": year, + "season_number": season, + "episode_number": episode, + "estimated_space_bytes": size, + "reason_tokens": [ + f"Tag tv series episodes (S{season:02d}E{episode:02d}): Library matches " + "any b539bda9 (b539bda9), Days since watched > 7 " + f"({days})" + ], + "reasons": _notification_reasons( + [ + { + "rule_name": "Tag tv series episodes", + "conditions": [ + { + "field": "library.id", + "field_label": "Library", + "operator": "contains_any", + "operator_label": "matches any", + "expected": ["b539bda9"], + "actual": "b539bda9", + "display": "Library matches any b539bda9 (b539bda9)", + }, + { + "field": "watch.days_since_last_watched", + "field_label": "Days since watched", + "operator": "greater_than", + "operator_label": ">", + "expected": 7, + "actual": days, + "display": f"Days since watched > 7 ({days})", + }, + ], + } + ] + ), + } + + +def test_episode_candidates_name_the_episode_not_just_the_season() -> None: + context = { + "created_count": 3, + "total_reclaimable_bytes": 1_430_000_000, + "candidates": [ + _episode("Carrie", 1, 5, 546_900_000, 8, year=2026), + _episode("Carrie", 1, 6, 437_100_000, 8, year=2026), + _episode("Carrie", 1, 7, 428_900_000, 9, year=2026), + ], + } + _, body = _compose( + NotificationType.NEW_CLEANUP_CANDIDATES, + title="New Cleanup Candidates Found", + context=context, + ) + # one line per show, episodes compressed into a range + assert body.count("Carrie") == 1 + assert "- **Carrie (2026)** · S01E05-E07 · 1.3 GB" in body + + +def test_episode_codes_compress_runs_per_season() -> None: + assert ( + _episode_codes([(1, 10), (2, 1), (1, 8), (1, 9), (2, 3)]) + == "S01E08-E10, S02E01, S02E03" + ) + + +def test_episode_reasons_hide_library_ids_and_collapse_per_show() -> None: + context = { + "created_count": 2, + "total_reclaimable_bytes": 2_500_000_000, + "candidates": [ + _episode("Ted Lasso", 2, 1, 1_300_000_000, 8), + _episode("Ted Lasso", 1, 10, 1_200_000_000, 9), + ], + } + _, body = _compose( + NotificationType.NEW_CLEANUP_CANDIDATES, + title="New Cleanup Candidates Found", + context=context, + preferences={ + NotificationType.NEW_CLEANUP_CANDIDATES.value: { + "detail": "top_n_with_reasons", + } + }, + ) + assert "b539bda9" not in body + assert "- **Ted Lasso (2020)** · S01E10, S02E01 · 2.3 GB" in body + # the shared rule shows once, without each episode's own day count + assert body.count("Days since watched > 7") == 1 + assert " - _Days since watched > 7_" in body + + +def test_single_episode_keeps_the_measured_value() -> None: + context = { + "created_count": 1, + "total_reclaimable_bytes": 1_100_000_000, + "candidates": [_episode("Ted Lasso", 1, 9, 1_100_000_000, 8)], + } + _, body = _compose( + NotificationType.NEW_CLEANUP_CANDIDATES, + title="New Cleanup Candidates Found", + context=context, + preferences={ + NotificationType.NEW_CLEANUP_CANDIDATES.value: { + "detail": "top_n_with_reasons", + } + }, + ) + assert "- **Ted Lasso (2020)** · S01E09 · 1.0 GB" in body + assert " - _Days since watched > 7 (8)_" in body + assert "b539bda9" not in body + + +def test_telegram_gets_bullet_glyphs_outside_code_fences() -> None: + body = "lead\n\n- one\n - two\n```\n- keep\n```" + assert _telegram_bullets(body) == "lead\n\n• one\n ◦ two\n```\n- keep\n```" + assert _is_telegram_url("tgram://123:abc/456") + assert not _is_telegram_url("discord://a/b") + + +@pytest.mark.parametrize("episode_number", [None, 3]) +def test_whole_series_scope_remains_visible_in_mixed_group( + episode_number: int | None, +) -> None: + whole_series = { + "media_type": "series", + "media_id": 42, + "media_title": "Example Show", + "media_year": 2020, + "season_number": None, + "episode_number": None, + "estimated_space_bytes": 10 * 1024**3, + } + child = { + **whole_series, + "season_number": 1, + "episode_number": episode_number, + "estimated_space_bytes": 2 * 1024**3, + } + _, body = _compose( + NotificationType.NEW_CLEANUP_CANDIDATES, + context={ + "created_count": 2, + "total_reclaimable_bytes": 12 * 1024**3, + "candidates": [child, whole_series], + }, + ) + assert body.count("**Example Show (2020)**") == 1 + assert "**Example Show (2020)** · Entire series · 12.0 GB" in body