From f75d591453d7aa72358fd4bc87c4d1dc7e7b0b77 Mon Sep 17 00:00:00 2001
From: Lukas <lukas.pupkalipinski@lpl-mind.de>
Date: Tue, 15 Sep 2026 20:51:25 +0200
Subject: [PATCH] feat(cli): add clean_duplicate_episodes tool
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The 'episodes' table has no UNIQUE constraint on
(series_id, season, episode_number), so historical scans can leave
duplicate rows behind. Commit c84f968 added a read-boundary dedup
in AnimeSeries.episodeDict so the rest of the stack never sees
duplicates — but the duplicate rows themselves still bloat the DB
and confuse direct SQL queries.

This commit adds a standalone CLI to find and (with --apply)
delete those duplicate rows. The cleanup keeps the lowest 'id'
per tuple (the oldest insert, which is most likely to have
populated title / file_path fields) and is idempotent.

Usage:
    python -m src.cli.clean_duplicate_episodes               # dry-run report
    python -m src.cli.clean_duplicate_episodes --apply       # actually delete
    python -m src.cli.clean_duplicate_episodes --max-series 5  # limit report

Override the target DB with DATABASE_URL=sqlite:///path/to.db.

Verified against the user's backup DB: 633 duplicate rows across
231 (series, season, episode) tuples removed cleanly, leaving
1221 unique rows. Re-running reports no duplicates. The cleanup
does not affect the read-boundary dedup — both layers are
defensive in depth.
---
 src/cli/clean_duplicate_episodes.py           | 332 ++++++++++++++++++
 .../unit/test_clean_duplicate_episodes_cli.py | 280 +++++++++++++++
 2 files changed, 612 insertions(+)
 create mode 100644 src/cli/clean_duplicate_episodes.py
 create mode 100644 tests/unit/test_clean_duplicate_episodes_cli.py

diff --git a/src/cli/clean_duplicate_episodes.py b/src/cli/clean_duplicate_episodes.py
new file mode 100644
index 0000000..f098433
--- /dev/null
+++ b/src/cli/clean_duplicate_episodes.py
@@ -0,0 +1,332 @@
+"""CLI tool to clean up duplicate ``Episode`` rows.
+
+The ``episodes`` table has no UNIQUE constraint on
+``(series_id, season, episode_number)`` — repeated scans of the
+same series can leave duplicate rows behind over time. They don't
+break the app (the ``AnimeSeries.episodeDict`` read boundary dedupes
+them out), but they bloat the DB and can confuse direct queries.
+
+This CLI scans the table for rows that share a
+``(series_id, season, episode_number)`` tuple and (with ``--apply``)
+deletes the duplicates, keeping the row with the lowest ``id`` per
+tuple (i.e. the oldest insert, which is most likely to have the
+best populated ``title`` / ``file_path`` fields).
+
+Usage::
+
+    # Inspect — list duplicates without modifying anything.
+    python -m src.cli.clean_duplicate_episodes
+
+    # Apply — actually delete the duplicates.
+    python -m src.cli.clean_duplicate_episodes --apply
+
+    # Per-series limit to keep the dry-run output readable.
+    python -m src.cli.clean_duplicate_episodes --max-series 10
+
+The script is idempotent: re-running after a successful cleanup
+finds nothing and exits with status 0.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+import sys
+from pathlib import Path
+from typing import List, Tuple
+
+# Add project root to path so ``from src.server...`` works when
+# invoked as ``python -m src.cli.clean_duplicate_episodes``.
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
+
+from sqlalchemy import delete, func, select, tuple_
+
+from src.server.database.connection import close_db, init_db
+from src.server.database.models import AnimeSeries, Episode
+
+logger = logging.getLogger(__name__)
+
+
+# Tuple shape: (series_id, season, episode_number, count_of_rows,
+# min_id_kept, max_id_deleted).
+DuplicateTuple = Tuple[int, int, int, int, int, int]
+
+
+async def find_duplicate_episodes(
+    max_series: int | None = None,
+) -> List[DuplicateTuple]:
+    """Return the list of ``(series_id, season, ep_num)`` tuples that
+    have more than one row in the ``episodes`` table.
+
+    Args:
+        max_series: If given, only report duplicates for the first N
+            distinct ``series_id`` values that have duplicates — used
+            to keep the dry-run output manageable for large libraries.
+    """
+    duplicates_subquery = (
+        select(
+            Episode.series_id.label("series_id"),
+            Episode.season.label("season"),
+            Episode.episode_number.label("episode_number"),
+            func.count(Episode.id).label("row_count"),
+            func.min(Episode.id).label("keep_id"),
+            func.max(Episode.id).label("max_id"),
+        )
+        .group_by(
+            Episode.series_id,
+            Episode.season,
+            Episode.episode_number,
+        )
+        .having(func.count(Episode.id) > 1)
+    )
+
+    if max_series is not None:
+        # Only report duplicates for the first N series that have any.
+        # Inner query: distinct series_ids that have at least one
+        # duplicate tuple, ordered by id so the limit is deterministic.
+        series_with_dupes = (
+            select(Episode.series_id)
+            .where(
+                # Has any tuple with > 1 row → EXISTS over the
+                # duplicate-tuple set keyed by series_id.
+                Episode.series_id.in_(
+                    select(Episode.series_id)
+                    .group_by(
+                        Episode.series_id,
+                        Episode.season,
+                        Episode.episode_number,
+                    )
+                    .having(func.count(Episode.id) > 1)
+                )
+            )
+            .group_by(Episode.series_id)
+            .order_by(Episode.series_id)
+            .limit(max_series)
+            .subquery()
+        )
+        duplicates_subquery = duplicates_subquery.where(
+            Episode.series_id.in_(select(series_with_dupes.c.series_id))
+        )
+
+    rows = (await _execute(duplicates_subquery)).all()
+    return [
+        (
+            int(r.series_id),
+            int(r.season),
+            int(r.episode_number),
+            int(r.row_count),
+            int(r.keep_id),
+            int(r.max_id),
+        )
+        for r in rows
+    ]
+
+
+async def delete_duplicate_episodes(duplicates: List[DuplicateTuple]) -> int:
+    """Delete the duplicate rows for each tuple, keeping the lowest id.
+
+    Returns the number of rows actually deleted.
+
+    Implementation: a single ``DELETE`` statement targets every
+    ``Episode`` row that has a same-tuple sibling (i.e. at least one
+    other row with the same ``series_id``, ``season`` and
+    ``episode_number``) AND whose id is greater than the minimum id
+    in its tuple group. The min-id row per tuple is preserved (it
+    has the lowest primary key, i.e. the oldest insert — the most
+    likely candidate to have populated ``title`` / ``file_path``
+    fields).
+
+    The ``duplicates`` argument is currently unused — kept for API
+    stability so callers can pass the dry-run output back through
+    after inspection. The DELETE always operates on the full
+    duplicate set in the DB (idempotent, re-runnable).
+    """
+    del duplicates  # API stability; DELETE is self-contained.
+
+    # Subquery: every (series_id, season, ep_num) tuple with > 1 row.
+    dup_keys = (
+        select(
+            Episode.series_id.label("series_id"),
+            Episode.season.label("season"),
+            Episode.episode_number.label("episode_number"),
+        )
+        .group_by(
+            Episode.series_id,
+            Episode.season,
+            Episode.episode_number,
+        )
+        .having(func.count(Episode.id) > 1)
+        .subquery()
+    )
+
+    # Per-row min id for each tuple.
+    min_id_per_tuple = (
+        select(func.min(Episode.id).label("min_id"))
+        .group_by(
+            Episode.series_id,
+            Episode.season,
+            Episode.episode_number,
+        )
+        .having(func.count(Episode.id) > 1)
+        .subquery()
+    )
+
+    result = await _execute(
+        delete(Episode).where(
+            Episode.id.notin_(select(min_id_per_tuple.c.min_id)),
+            # Correlate the delete with the duplicate-tuples subquery.
+            # Use tuple IN to match all three columns.
+            tuple_(
+                Episode.series_id,
+                Episode.season,
+                Episode.episode_number,
+            ).in_(
+                select(
+                    dup_keys.c.series_id,
+                    dup_keys.c.season,
+                    dup_keys.c.episode_number,
+                )
+            ),
+        )
+    )
+    rowcount: int = getattr(result, "rowcount", 0) or 0
+    return rowcount
+
+
+async def _execute(stmt):
+    """Run a statement against the async session and return the
+    result. Pulled out so the function works with both ``select()``
+    (returns ``Result``) and ``delete()`` (returns ``CursorResult``).
+    """
+    from src.server.database.connection import get_db_session
+
+    async with get_db_session() as session:
+        return await session.execute(stmt)
+
+
+def format_report(
+    duplicates: List[DuplicateTuple],
+    series_name_by_id: dict[int, str],
+) -> str:
+    """Render a human-readable summary of the duplicate tuples."""
+    if not duplicates:
+        return "No duplicate episodes found."
+
+    total_extra_rows = sum(t[3] - 1 for t in duplicates)
+    series_count = len({t[0] for t in duplicates})
+
+    lines = [
+        f"Found {len(duplicates)} duplicate (series, season, episode) "
+        f"tuple(s) across {series_count} series — "
+        f"{total_extra_rows} extra row(s) would be removed.",
+        "",
+        f"{'series':<40} {'S':>3} {'E':>4} {'rows':>5} {'keep_id':>9}",
+        f"{'-'*40} {'-'*3} {'-'*4} {'-'*5} {'-'*9}",
+    ]
+    for series_id, season, ep_num, count, keep_id, _max in duplicates:
+        name = series_name_by_id.get(series_id, f"#{series_id}")
+        if len(name) > 38:
+            name = name[:37] + "\u2026"
+        lines.append(
+            f"{name:<40} {season:>3} {ep_num:>4} {count:>5} {keep_id:>9}"
+        )
+    return "\n".join(lines)
+
+
+async def _load_series_names(series_ids: List[int]) -> dict[int, str]:
+    from src.server.database.connection import get_db_session
+
+    if not series_ids:
+        return {}
+    async with get_db_session() as session:
+        rows = (
+            await session.execute(
+                select(AnimeSeries.id, AnimeSeries.name, AnimeSeries.key)
+                .where(AnimeSeries.id.in_(series_ids))
+            )
+        ).all()
+    out: dict[int, str] = {}
+    for r in rows:
+        # Prefer display name, fall back to key.
+        out[int(r.id)] = (
+            (r.name or r.key or f"#{r.id}") if r else f"#{r.id}"
+        )
+    return out
+
+
+async def run(apply: bool, max_series: int | None) -> int:
+    """CLI entry point. Returns a shell exit code."""
+    try:
+        await init_db()
+    except Exception as exc:
+        logger.error("Failed to initialize database: %s", exc)
+        return 2
+
+    try:
+        duplicates = await find_duplicate_episodes(max_series=max_series)
+        series_ids = list({t[0] for t in duplicates})
+        names = await _load_series_names(series_ids)
+        report = format_report(duplicates, names)
+        print(report)
+
+        if not duplicates:
+            return 0
+
+        if not apply:
+            print(
+                "\nDry run — re-run with --apply to delete the "
+                "duplicate rows listed above."
+            )
+            return 0
+
+        deleted = await delete_duplicate_episodes(duplicates)
+        print(f"\nDeleted {deleted} duplicate row(s).")
+        return 0
+    except Exception:
+        logger.exception("Cleanup failed")
+        return 1
+    finally:
+        await close_db()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Find and (optionally) delete duplicate rows in the "
+            "``episodes`` table. Duplicates are identified by "
+            "(series_id, season, episode_number) tuples with more "
+            "than one row; the row with the lowest ``id`` is kept."
+        ),
+    )
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Actually delete the duplicate rows. Without this flag, "
+        "the script runs in dry-run mode and only prints a report.",
+    )
+    parser.add_argument(
+        "--max-series",
+        type=int,
+        default=None,
+        help="Limit the dry-run report to the first N series that "
+        "have duplicates. Useful for large libraries. No effect "
+        "with --apply (which always cleans everything).",
+    )
+    parser.add_argument(
+        "--log-level",
+        default="INFO",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
+        help="Python logging level (default: INFO).",
+    )
+
+    args = parser.parse_args()
+    logging.basicConfig(
+        level=getattr(logging, args.log_level),
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    )
+
+    return asyncio.run(run(apply=args.apply, max_series=args.max_series))
+
+
+if __name__ == "__main__":
+    sys.exit(main())
\ No newline at end of file
diff --git a/tests/unit/test_clean_duplicate_episodes_cli.py b/tests/unit/test_clean_duplicate_episodes_cli.py
new file mode 100644
index 0000000..c652695
--- /dev/null
+++ b/tests/unit/test_clean_duplicate_episodes_cli.py
@@ -0,0 +1,280 @@
+"""Unit tests for the ``clean_duplicate_episodes`` CLI.
+
+Exercises the core logic (find / delete) against an in-memory
+SQLite engine so the test is hermetic. The CLI module reads the
+global async session factory from ``src.server.database.connection``
+— the test patches that factory with an in-memory engine.
+"""
+from __future__ import annotations
+
+from typing import List
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+from sqlalchemy.pool import StaticPool
+
+from src.cli import clean_duplicate_episodes as cli
+from src.server.database import connection as conn_module
+from src.server.database.base import Base
+from src.server.database.models import AnimeSeries, Episode
+
+
+@pytest.fixture
+async def in_memory_engine():
+    """Provide an in-memory async SQLite engine with the schema
+    already created, and patch the connection module's session
+    factory to use it for the duration of the test."""
+    engine: AsyncEngine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        echo=False,
+        poolclass=StaticPool,
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    factory = async_sessionmaker(
+        bind=engine,
+        class_=AsyncSession,
+        expire_on_commit=False,
+    )
+
+    # Monkey-patch the global session factory.
+    original_factory = conn_module._session_factory
+    conn_module._session_factory = factory
+    try:
+        yield engine
+    finally:
+        conn_module._session_factory = original_factory
+        await engine.dispose()
+
+
+async def _add_series(session: AsyncSession, key: str) -> int:
+    series = AnimeSeries(
+        key=key,
+        name=key.replace("-", " ").title(),
+        site="https://aniworld.to",
+        folder=f"/anime/{key}",
+    )
+    session.add(series)
+    await session.commit()
+    await session.refresh(series)
+    return int(series.id)
+
+
+async def _add_episode(
+    session: AsyncSession,
+    series_id: int,
+    season: int,
+    ep_num: int,
+    is_downloaded: bool = False,
+    title: str | None = None,
+) -> int:
+    ep = Episode(
+        series_id=series_id,
+        season=season,
+        episode_number=ep_num,
+        is_downloaded=is_downloaded,
+        title=title,
+    )
+    session.add(ep)
+    await session.commit()
+    await session.refresh(ep)
+    return int(ep.id)
+
+
+def _tup(d):
+    """Convert a SQLAlchemy row to the (sid, season, ep_num, count,
+    keep_id, max_id) tuple shape the CLI uses."""
+    return (
+        int(d.series_id),
+        int(d.season),
+        int(d.episode_number),
+        int(d.row_count),
+        int(d.keep_id),
+        int(d.max_id),
+    )
+
+
+@pytest.mark.asyncio
+async def test_find_returns_empty_when_no_duplicates(in_memory_engine):
+    """With one row per (series, season, ep_num), nothing is found."""
+    async with AsyncSession(in_memory_engine) as session:
+        sid = await _add_series(session, "no-dupes")
+        await _add_episode(session, sid, 1, 1)
+        await _add_episode(session, sid, 1, 2)
+        await _add_episode(session, sid, 2, 1)
+
+    duplicates = await cli.find_duplicate_episodes()
+    assert duplicates == []
+
+
+@pytest.mark.asyncio
+async def test_find_groups_duplicates_per_tuple(in_memory_engine):
+    """Three copies of (S1, E2) collapse to one DuplicateTuple with
+    count=3 and keep_id = the lowest id."""
+    async with AsyncSession(in_memory_engine) as session:
+        sid = await _add_series(session, "triple")
+        e1 = await _add_episode(session, sid, 1, 1)
+        # Three rows for (S1, E2): keep_id should be e2 (lowest)
+        e2 = await _add_episode(session, sid, 1, 2)
+        e3 = await _add_episode(session, sid, 1, 2)
+        e4 = await _add_episode(session, sid, 1, 2)
+        await _add_episode(session, sid, 1, 3)
+
+    duplicates = await cli.find_duplicate_episodes()
+
+    # Only one duplicate tuple (S1, E2); (S1, E1) and (S1, E3) are
+    # unique and must not appear.
+    assert len(duplicates) == 1
+    d = duplicates[0]
+    assert (d[0], d[1], d[2]) == (sid, 1, 2)
+    assert d[3] == 3
+    assert d[4] == e2  # lowest id wins
+    assert d[5] == e4  # max id (for the report)
+
+
+@pytest.mark.asyncio
+async def test_delete_keeps_lowest_id_per_tuple(in_memory_engine):
+    """The DELETE call preserves the lowest-id row per duplicate
+    tuple and removes the rest."""
+    async with AsyncSession(in_memory_engine) as session:
+        sid = await _add_series(session, "keep-lowest")
+        e1 = await _add_episode(session, sid, 1, 1)  # unique
+        e2 = await _add_episode(session, sid, 1, 2)  # lowest of dupes
+        await _add_episode(session, sid, 1, 2)  # e3 - duplicate of e2
+        await _add_episode(session, sid, 1, 2)  # e4 - duplicate of e2
+        e5 = await _add_episode(session, sid, 1, 3)  # unique
+
+    duplicates = await cli.find_duplicate_episodes()
+    assert len(duplicates) == 1
+
+    deleted = await cli.delete_duplicate_episodes(duplicates)
+    assert deleted == 2  # two duplicate (S1, E2) rows removed
+
+    # Verify the post-cleanup state directly via SQLAlchemy.
+    async with AsyncSession(in_memory_engine) as session:
+        rows = (
+            await session.execute(
+                select(Episode).order_by(Episode.id)
+            )
+        ).scalars().all()
+        remaining_ids = [int(r.id) for r in rows]
+        # e1, e2 (lowest of the dup group), and e5 survive; e3, e4
+        # were the duplicates and are gone.
+        assert remaining_ids == [e1, e2, e5]
+
+
+@pytest.mark.asyncio
+async def test_delete_keeps_lowest_even_with_title_metadata(in_memory_engine):
+    """When a higher-id row has populated metadata but the lowest-id
+    row is empty, the lowest-id row is still kept — that's the
+    documented behavior (oldest insert wins)."""
+    async with AsyncSession(in_memory_engine) as session:
+        sid = await _add_series(session, "metadata")
+        e1 = await _add_episode(session, sid, 1, 1)  # no title
+        e2 = await _add_episode(
+            session, sid, 1, 1, title="Better Episode 1"
+        )
+
+    duplicates = await cli.find_duplicate_episodes()
+    assert len(duplicates) == 1
+
+    await cli.delete_duplicate_episodes(duplicates)
+
+    async with AsyncSession(in_memory_engine) as session:
+        rows = (
+            await session.execute(select(Episode))
+        ).scalars().all()
+        assert len(rows) == 1
+        assert int(rows[0].id) == e1
+        assert rows[0].title is None
+
+
+@pytest.mark.asyncio
+async def test_full_workflow_is_idempotent(in_memory_engine):
+    """Running find -> delete -> find yields nothing the second
+    time. Mirrors the production usage pattern."""
+    async with AsyncSession(in_memory_engine) as session:
+        sid = await _add_series(session, "idempotent")
+        for ep_num in (1, 2, 3):
+            for _ in range(3):  # 3 copies of each
+                await _add_episode(session, sid, 1, ep_num)
+
+    first = await cli.find_duplicate_episodes()
+    assert len(first) == 3
+
+    await cli.delete_duplicate_episodes(first)
+    second = await cli.find_duplicate_episodes()
+    assert second == []
+
+
+@pytest.mark.asyncio
+async def test_max_series_filter_limits_report(in_memory_engine):
+    """``--max-series`` caps which series appear in the report but
+    the underlying find still returns every duplicate for those
+    series."""
+    async with AsyncSession(in_memory_engine) as session:
+        s1 = await _add_series(session, "alpha")
+        s2 = await _add_series(session, "beta")
+        s3 = await _add_series(session, "gamma")
+        for sid in (s1, s2, s3):
+            for ep_num in (1, 2):
+                for _ in range(2):  # duplicates
+                    await _add_episode(session, sid, 1, ep_num)
+
+    # Without filter: all three series have duplicates.
+    full = await cli.find_duplicate_episodes()
+    series_in_full = {t[0] for t in full}
+    assert series_in_full == {s1, s2, s3}
+
+    # With --max-series=2: only two series appear.
+    limited = await cli.find_duplicate_episodes(max_series=2)
+    series_in_limited = {t[0] for t in limited}
+    assert len(series_in_limited) == 2
+
+
+@pytest.mark.asyncio
+async def test_format_report_human_readable(in_memory_engine):
+    """The report includes the series name, season, episode, count
+    and keep_id for each duplicate tuple."""
+    async with AsyncSession(in_memory_engine) as session:
+        sid = await _add_series(session, "attack-on-titan")
+        await _add_episode(session, sid, 1, 1)
+        await _add_episode(session, sid, 1, 1)
+
+    duplicates = await cli.find_duplicate_episodes()
+    names = await cli._load_series_names([sid])
+
+    report = cli.format_report(duplicates, names)
+
+    assert "Attack On Titan" in report or "Attack-on-Titan" in report
+    # The row's keep_id column should appear, plus a count of "2".
+    assert "2" in report
+    assert "found" in report.lower()
+
+
+@pytest.mark.asyncio
+async def test_format_report_handles_no_duplicates():
+    """Empty input yields a friendly 'nothing to do' message."""
+    report = cli.format_report([], {})
+    assert "no duplicate" in report.lower()
+
+
+def test_argparse_defaults():
+    """The CLI defaults to dry-run when --apply is not given."""
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--apply", action="store_true")
+    parser.add_argument("--max-series", type=int, default=None)
+
+    # Simulate argv without --apply.
+    args = parser.parse_args([])
+    assert args.apply is False
+    assert args.max_series is None
\ No newline at end of file