Merge pull request 'Tier-1 video dedup: import-time + retroactive cleanup (#871)' (#110) from dev into main

alembic/versions/0052_image_duration_seconds.py

@@ -0,0 +1,32 @@
+"""image_record: duration_seconds (Tier-1 video near-dup key)
+
+#871. Videos previously deduped on sha256 only (pHash is images-only), so a
+different encode/remux of the same video imported as a distinct record. Persist
+the container duration so the importer can treat same-artist videos with matching
+duration (+ aspect ratio) as the same content and dedup/supersede like images.
+NULL for images and for video rows imported before this column existed (a
+backfill re-probes those so they participate in dedup).
+
+Revision ID: 0052
+Revises: 0051
+Create Date: 2026-06-16
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0052"
+down_revision: Union[str, None] = "0051"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "image_record", sa.Column("duration_seconds", sa.Float(), nullable=True)
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("image_record", "duration_seconds")

backend/app/api/admin.py

@@ -361,3 +361,34 @@ async def trigger_prune_missing_files():
 
     async_result = prune_missing_file_records_task.delay()
     return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+
+
+@admin_bp.route("/maintenance/dedup-videos", methods=["POST"])
+async def trigger_dedup_videos():
+    """Tier-1 video dedup (#871). Body {"dry_run": bool}: dry_run=true previews
+    what would be removed (groups / redundant count / reclaimable bytes) WITHOUT
+    deleting; dry_run=false applies it (re-link posts to the keeper, then delete
+    the redundant copies). Either way it first re-probes NULL-duration videos so
+    the existing library participates. Returns the Celery task id — poll
+    /maintenance/task-result/<id> for the summary."""
+    from ..tasks.admin import dedup_videos_task
+
+    body = await request.get_json(silent=True) or {}
+    dry_run = bool(body.get("dry_run", True))  # default to the SAFE preview
+    async_result = dedup_videos_task.delay(dry_run=dry_run)
+    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+
+
+@admin_bp.route("/maintenance/task-result/<task_id>", methods=["GET"])
+async def maintenance_task_result(task_id: str):
+    """Poll a maintenance Celery task's result (the summary dict it returns).
+    Used by the video-dedup card to show the dry-run projection before apply."""
+    from ..celery_app import celery
+
+    res = celery.AsyncResult(task_id)
+    ready = res.ready()
+    return jsonify({
+        "ready": ready,
+        "successful": res.successful() if ready else None,
+        "result": res.result if (ready and res.successful()) else None,
+    })

backend/app/models/image_record.py

@@ -13,6 +13,7 @@ from sqlalchemy import (
     BigInteger,
     DateTime,
     Enum,
+    Float,
     ForeignKey,
     Integer,
     String,
@@ -39,6 +40,10 @@ class ImageRecord(Base):
     mime: Mapped[str] = mapped_column(String(64), nullable=False)
     width: Mapped[int | None] = mapped_column(Integer, nullable=True)
     height: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # Video container duration (seconds); NULL for images. The Tier-1 video
+    # near-dup key (#871): two videos of the same artist with matching duration
+    # (+ aspect) are the same content across re-encodes — dedup like image pHash.
+    duration_seconds: Mapped[float | None] = mapped_column(Float, nullable=True)
 
     # Integrity verification status. FC-2e populates this; FC-2a leaves rows at 'unknown'.
     # Values: 'unknown' (default), 'ok', 'corrupt', 'failed_verification'.

backend/app/services/cleanup_service.py

@@ -32,9 +32,17 @@ from ..models import (
 from ..models.series_chapter import SeriesChapter
 from ..models.series_page import SeriesPage
 from ..models.tag import image_tag
+from ..utils import safe_probe
+from .importer import _VIDEO_DUP_ASPECT_TOL, _VIDEO_DUP_DURATION_TOL_SECONDS
 
 log = logging.getLogger(__name__)
 
+# Sentinel written to duration_seconds when a video was probed but ffprobe
+# reported no usable duration (missing/corrupt file) — distinct from NULL (never
+# probed) so the backfill doesn't re-probe it forever, and < 0 so it can never
+# match a real duration in the dedup grouping (#871).
+_VIDEO_DURATION_UNKNOWN = -1.0
+
 
 def project_artist_cascade(session: Session, *, slug: str) -> dict:
     """Read-only projection of what delete_artist_cascade would touch.
@@ -940,3 +948,193 @@ def reextract_archive_attachments(
         except Exception as exc:
             log.warning("re-extract enqueue failed for image %s: %s", img_id, exc)
     return summary
+
+
+# ---- Tier-1 video dedup (#871) ------------------------------------------
+
+
+def _aspect_matches(w, h, cw, ch) -> bool:
+    """Same aspect ratio within tolerance; missing dims don't block (duration is
+    the primary signal). Mirrors Importer._video_aspect_matches."""
+    if not (w and h and cw and ch):
+        return True
+    return abs((w / h) - (cw / ch)) <= _VIDEO_DUP_ASPECT_TOL
+
+
+def backfill_video_durations(session: Session) -> int:
+    """Populate image_record.duration_seconds for video rows imported before #871
+    (NULL). Idempotent — only NULL rows are touched, so a re-run after a timeout
+    naturally resumes. A probe that yields no duration writes the
+    _VIDEO_DURATION_UNKNOWN sentinel so the file isn't re-probed forever (and can
+    never match a real duration). Returns the count of rows given a real duration.
+    """
+    populated = 0
+    while True:
+        rows = session.execute(
+            select(ImageRecord.id, ImageRecord.path)
+            .where(
+                ImageRecord.mime.like("video/%"),
+                ImageRecord.duration_seconds.is_(None),
+            )
+            .order_by(ImageRecord.id)
+            .limit(500)
+        ).all()
+        if not rows:
+            break
+        for rid, path in rows:
+            probe = safe_probe.probe_video(Path(path))
+            dur = probe.duration if probe.ok and probe.duration else None
+            session.execute(
+                update(ImageRecord)
+                .where(ImageRecord.id == rid)
+                .values(
+                    duration_seconds=dur if dur is not None else _VIDEO_DURATION_UNKNOWN
+                )
+            )
+            if dur is not None:
+                populated += 1
+        session.commit()
+    return populated
+
+
+def _video_dup_group(members: list) -> dict:
+    """Pick the keeper (highest pixel area, then largest bytes, then lowest id for
+    stability) and describe the group."""
+    keeper = max(
+        members,
+        key=lambda m: ((m.width or 0) * (m.height or 0), m.size_bytes or 0, -m.id),
+    )
+    losers = [m for m in members if m.id != keeper.id]
+    return {
+        "artist_id": keeper.artist_id,
+        "keeper_id": keeper.id,
+        "loser_ids": [m.id for m in losers],
+        "duration": keeper.duration_seconds,
+        "count": len(members),
+        "reclaim_bytes": sum((m.size_bytes or 0) for m in losers),
+    }
+
+
+def find_video_dup_groups(session: Session) -> list[dict]:
+    """Cluster videos that are the same content (#871): same artist, duration
+    within tolerance, matching aspect ratio. Returns groups of >1 member. Greedy
+    sweep over duration-sorted rows, anchored to each cluster's first member so the
+    cluster's duration span never exceeds the tolerance (no chain drift)."""
+    rows = session.execute(
+        select(
+            ImageRecord.id, ImageRecord.artist_id, ImageRecord.duration_seconds,
+            ImageRecord.width, ImageRecord.height, ImageRecord.size_bytes,
+        )
+        .where(
+            ImageRecord.mime.like("video/%"),
+            ImageRecord.duration_seconds.is_not(None),
+            ImageRecord.duration_seconds > 0,
+            ImageRecord.artist_id.is_not(None),
+        )
+        .order_by(
+            ImageRecord.artist_id, ImageRecord.duration_seconds, ImageRecord.id
+        )
+    ).all()
+    groups: list[dict] = []
+    cluster: list = []
+    anchor = None
+    for r in rows:
+        if (
+            anchor is not None
+            and r.artist_id == anchor.artist_id
+            and (r.duration_seconds - anchor.duration_seconds)
+            <= _VIDEO_DUP_DURATION_TOL_SECONDS
+            and _aspect_matches(r.width, r.height, anchor.width, anchor.height)
+        ):
+            cluster.append(r)
+        else:
+            if len(cluster) > 1:
+                groups.append(_video_dup_group(cluster))
+            cluster = [r]
+            anchor = r
+    if len(cluster) > 1:
+        groups.append(_video_dup_group(cluster))
+    return groups
+
+
+def _relink_provenance_to_keeper(
+    session: Session, *, loser_id: int, keeper_id: int
+) -> int:
+    """Ensure the keeper has an ImageProvenance row for every post the loser was
+    linked to, so deleting the loser never drops the video off a post. Returns the
+    number of new keeper↔post links added."""
+    rows = session.execute(
+        select(ImageProvenance.post_id, ImageProvenance.source_id)
+        .where(ImageProvenance.image_record_id == loser_id)
+    ).all()
+    added = 0
+    for post_id, source_id in rows:
+        exists = session.execute(
+            select(ImageProvenance.id).where(
+                ImageProvenance.image_record_id == keeper_id,
+                ImageProvenance.post_id == post_id,
+            )
+        ).scalar_one_or_none()
+        if exists is None:
+            session.add(ImageProvenance(
+                image_record_id=keeper_id, post_id=post_id, source_id=source_id,
+            ))
+            session.flush()
+            added += 1
+    return added
+
+
+def dedup_videos(
+    session: Session, *, images_root: Path, dry_run: bool = False
+) -> dict:
+    """Find and (unless dry_run) collapse Tier-1 video duplicates (#871).
+
+    Re-probes NULL-duration videos first so the existing library participates,
+    then clusters by artist + duration + aspect and keeps the highest-res copy per
+    cluster. On apply, each loser's post links are re-pointed to the keeper BEFORE
+    the loser record + file are deleted, so no post loses the video. dry_run shares
+    the same discovery predicate and returns the projection without deleting
+    (rule 93).
+
+    NOTE: tags/curation on a loser are NOT merged onto the keeper — videos rarely
+    carry hand-curation and merging would add FK-juggling risk. Flagged as a
+    follow-up if it ever matters.
+    """
+    backfill_video_durations(session)
+    groups = find_video_dup_groups(session)
+    redundant = sum(len(g["loser_ids"]) for g in groups)
+    reclaim = sum(g["reclaim_bytes"] for g in groups)
+    sample = [
+        {
+            "keeper_id": g["keeper_id"],
+            "redundant": len(g["loser_ids"]),
+            "duration": round(g["duration"], 1),
+        }
+        for g in groups[:50]
+    ]
+    if dry_run:
+        return {
+            "groups": len(groups), "redundant": redundant,
+            "reclaim_bytes": reclaim, "sample": sample,
+        }
+
+    relinked = 0
+    for g in groups:
+        for loser_id in g["loser_ids"]:
+            relinked += _relink_provenance_to_keeper(
+                session, loser_id=loser_id, keeper_id=g["keeper_id"]
+            )
+        session.commit()
+
+    loser_ids = [lid for g in groups for lid in g["loser_ids"]]
+    deleted = delete_images(session, image_ids=loser_ids, images_root=images_root)
+    log.info(
+        "video dedup: %d group(s), %d redundant removed, %d post link(s) re-pointed",
+        len(groups), deleted["images_deleted"], relinked,
+    )
+    return {
+        "groups": len(groups), "redundant": redundant,
+        "reclaim_bytes": reclaim, "deleted": deleted["images_deleted"],
+        "files_deleted": deleted["files_deleted"],
+        "relinked_posts": relinked, "sample": sample,
+    }

backend/app/services/importer.py

@@ -85,6 +85,15 @@ ALL_EXTS = IMAGE_EXTS | VIDEO_EXTS
 # deep-nested archive can't recurse forever; the bomb-probe re-runs per level.
 _ARCHIVE_MAX_DEPTH = 3
 
+# Tier-1 video near-dup (#871). pHash is images-only, so videos deduped on sha256
+# alone — a different encode/remux of the same clip imported as a distinct record.
+# Identity = container duration within a tight tolerance + matching aspect ratio,
+# scoped to the same artist (codec/bitrate are NOT part of identity — the point is
+# to match ACROSS re-encodes). Quality axis for supersede = pixel dimensions, same
+# as image pHash. Tight tolerances because a wrong video merge is destructive.
+_VIDEO_DUP_DURATION_TOL_SECONDS = 1.0
+_VIDEO_DUP_ASPECT_TOL = 0.02
+
 
 def is_supported(path: Path) -> bool:
     return path.suffix.lower() in ALL_EXTS
@@ -604,6 +613,48 @@ class Importer:
                 archive_path.name, depth, exc,
             )
 
+    @staticmethod
+    def _video_aspect_matches(w, h, cw, ch) -> bool:
+        """True when two (w,h) pairs share an aspect ratio within tolerance.
+        Missing dims → don't block (duration is the primary signal)."""
+        if not (w and h and cw and ch):
+            return True
+        return abs((w / h) - (cw / ch)) <= _VIDEO_DUP_ASPECT_TOL
+
+    def _find_similar_video(
+        self, duration: float | None, width: int | None, height: int | None,
+        artist_id: int | None,
+    ) -> tuple[str, int | None]:
+        """Tier-1 video near-dup (#871). Find a same-artist video whose duration
+        matches within tolerance AND whose aspect ratio matches — treat it as the
+        same content (a different encode/remux). Returns find_similar's contract:
+        ("none"|"larger_exists"|"smaller_exists", id), quality judged by pixel
+        dimensions so a higher-res copy supersedes a smaller one."""
+        if duration is None or artist_id is None:
+            return ("none", None)
+        lo = duration - _VIDEO_DUP_DURATION_TOL_SECONDS
+        hi = duration + _VIDEO_DUP_DURATION_TOL_SECONDS
+        rows = self.session.execute(
+            select(ImageRecord.id, ImageRecord.width, ImageRecord.height)
+            .where(
+                ImageRecord.artist_id == artist_id,
+                ImageRecord.mime.like("video/%"),
+                ImageRecord.duration_seconds.is_not(None),
+                ImageRecord.duration_seconds >= lo,
+                ImageRecord.duration_seconds <= hi,
+            )
+        ).all()
+        w, h = width or 0, height or 0
+        for cid, cw, ch in rows:
+            if not self._video_aspect_matches(width, height, cw, ch):
+                continue
+            cw0, ch0 = cw or 0, ch or 0
+            if cw0 >= w and ch0 >= h:
+                return ("larger_exists", cid)
+            if w > cw0 or h > ch0:
+                return ("smaller_exists", cid)
+        return ("none", None)
+
     def _import_media(
         self, source: Path, attribution_path: Path,
         *, explicit_source: Source | None = None,
@@ -622,13 +673,13 @@ class Importer:
 
         # Compute file dimensions (images only) and apply filters.
         width = height = None
+        duration = None  # video container duration (Tier-1 near-dup key, #871)
         has_alpha = False
         if is_video(source):
             # Layer-3 isolation: validate the container via ffprobe (a
             # separate process) before the rest of the pipeline touches
             # it. A corrupt video that would crash a decoder is rejected
-            # cleanly here, and we capture width/height for free (the
-            # importer didn't previously record video dimensions).
+            # cleanly here, and we capture width/height + duration for free.
             probe = safe_probe.probe_video(source)
             if not probe.ok:
                 if probe.crashed:
@@ -640,7 +691,7 @@ class Importer:
                     status="skipped", skip_reason=SkipReason.invalid_image,
                     error=probe.reason,
                 )
-            width, height = probe.width, probe.height
+            width, height, duration = probe.width, probe.height, probe.duration
         else:
             try:
                 with Image.open(source) as im:
@@ -752,6 +803,32 @@ class Importer:
                     return ImportResult(
                         status="superseded", image_id=match_id
                     )
+        elif duration is not None:
+            # Tier-1 video near-dup (#871): same-artist, matching duration+aspect
+            # → same content across re-encodes. Mirror the image flow.
+            rel, match_id = self._find_similar_video(
+                duration, width, height,
+                path_artist.id if path_artist else None,
+            )
+            if rel == "larger_exists":
+                larger = self.session.get(ImageRecord, match_id)
+                if larger is not None:
+                    self._apply_sidecar(
+                        larger, attribution_path, path_artist,
+                        explicit_source=explicit_source,
+                    )
+                    self.session.commit()
+                return ImportResult(
+                    status="skipped", skip_reason=SkipReason.duplicate_phash,
+                    image_id=match_id,
+                    error="video near-duplicate (duration+aspect) of existing equal/larger video",
+                )
+            if rel == "smaller_exists":
+                target = self.session.get(ImageRecord, match_id)
+                self._supersede(
+                    target, source, sha, None, width, height, duration=duration,
+                )
+                return ImportResult(status="superseded", image_id=match_id)
 
         dest = self._copy_to_library(source, sha, attribution_path)
 
@@ -763,6 +840,7 @@ class Importer:
             mime=_mime_for(source),
             width=width,
             height=height,
+            duration_seconds=duration,
             origin="imported_filesystem",
             integrity_status="unknown",
         )
@@ -956,6 +1034,7 @@ class Importer:
 
         # Format / dimension / transparency filters (mirror _import_media).
         width = height = None
+        duration = None  # video container duration (Tier-1 near-dup key, #871)
         has_alpha = False
         if not is_video(path):
             try:
@@ -987,6 +1066,14 @@ class Importer:
                         status="skipped", skip_reason=SkipReason.too_transparent,
                         error=f"{pct:.2%} transparent",
                     )
+        else:
+            # Best-effort probe for dims + duration so downloaded videos can dedup
+            # (#871). LENIENT: unlike _import_media this path does not reject on a
+            # probe failure — preserve the existing "import the downloaded video"
+            # behavior; we just skip dedup when we can't read the duration.
+            vp = safe_probe.probe_video(path)
+            if vp.ok:
+                width, height, duration = vp.width, vp.height, vp.duration
 
         # Hash dedup
         sha = _sha256_of(path)
@@ -1045,6 +1132,30 @@ class Importer:
                         new_path=path, artist=artist, source_row=source,
                     )
                     return ImportResult(status="superseded", image_id=match_id)
+        elif duration is not None:
+            # Tier-1 video near-dup (#871): same content re-downloaded from another
+            # source/encode. artist is the subscription artist (passed explicitly).
+            rel, match_id = self._find_similar_video(
+                duration, width, height, artist.id if artist else None,
+            )
+            if rel == "larger_exists":
+                larger = self.session.get(ImageRecord, match_id)
+                if larger is not None:
+                    self._apply_sidecar(larger, path, artist, explicit_source=source)
+                    self.session.commit()
+                return ImportResult(
+                    status="skipped", skip_reason=SkipReason.duplicate_phash,
+                    image_id=match_id,
+                    error="video near-duplicate (duration+aspect) of existing equal/larger video",
+                )
+            if rel == "smaller_exists":
+                target = self.session.get(ImageRecord, match_id)
+                self._supersede(
+                    target, path, sha, None, width, height,
+                    new_path=path, artist=artist, source_row=source,
+                    duration=duration,
+                )
+                return ImportResult(status="superseded", image_id=match_id)
 
         # Create record — the path IS where the file lives.
         record = ImageRecord(
@@ -1055,6 +1166,7 @@ class Importer:
             mime=_mime_for(path),
             width=width,
             height=height,
+            duration_seconds=duration,
             origin="downloaded",
             integrity_status="unknown",
         )
@@ -1270,6 +1382,7 @@ class Importer:
         *, new_path: Path | None = None,
         artist: Artist | None = None,
         source_row: Source | None = None,
+        duration: float | None = None,
     ) -> None:
         """Replace `existing`'s file with the larger `source`, keeping the
         row id (so tags/series/curation stay attached). ML is cleared so
@@ -1302,6 +1415,7 @@ class Importer:
         existing.mime = _mime_for(source)
         existing.width = width
         existing.height = height
+        existing.duration_seconds = duration  # #871: keep the kept copy's duration
         existing.thumbnail_path = None
         existing.integrity_status = "unknown"
         existing.tagger_model_version = None

backend/app/tasks/admin.py

@@ -122,6 +122,27 @@ def prune_missing_file_records_task(self) -> dict:
     return {"checked": checked, "missing": len(missing_ids), "deleted": deleted}
 
 
+@celery.task(
+    name="backend.app.tasks.admin.dedup_videos_task",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError),
+    retry_backoff=15, retry_backoff_max=180, max_retries=1,
+    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
+)
+def dedup_videos_task(self, dry_run: bool = False) -> dict:
+    """Tier-1 video dedup (#871): re-probe NULL-duration videos, cluster by
+    artist + duration + aspect, keep the highest-res copy per cluster. dry_run
+    returns the projection (groups/redundant/reclaimable bytes) WITHOUT deleting;
+    apply re-points each loser's post links to the keeper then deletes the
+    redundant records + files. Operator-triggered; the summary lands in
+    task_run.metadata (FC-3i) for the Maintenance card to surface."""
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        return cleanup_service.dedup_videos(
+            session, images_root=IMAGES_ROOT, dry_run=dry_run,
+        )
+
+
 @celery.task(
     name="backend.app.tasks.admin.bulk_delete_images_task",
     bind=True,

backend/app/utils/safe_probe.py

@@ -60,6 +60,9 @@ class ProbeResult:
     reason: str | None = None
     width: int | None = None
     height: int | None = None
+    # Container duration in seconds (videos only) — the Tier-1 video near-dup
+    # key (#871). None when not a video / ffprobe didn't report it.
+    duration: float | None = None
 
 
 def probe_video(path: Path, *, timeout: float = VIDEO_PROBE_TIMEOUT_SECONDS) -> ProbeResult:
@@ -69,7 +72,10 @@ def probe_video(path: Path, *, timeout: float = VIDEO_PROBE_TIMEOUT_SECONDS) ->
             [
                 "ffprobe", "-v", "error",
                 "-select_streams", "v:0",
-                "-show_entries", "stream=width,height",
+                # format=duration is the reliable container duration (stream
+                # duration is often N/A for remuxed/streamed mp4) — the #871
+                # video near-dup key.
+                "-show_entries", "stream=width,height:format=duration",
                 "-of", "json", str(path),
             ],
             capture_output=True, text=True, timeout=timeout,
@@ -87,13 +93,21 @@ def probe_video(path: Path, *, timeout: float = VIDEO_PROBE_TIMEOUT_SECONDS) ->
             reason=f"ffprobe rejected the file: {out.stderr.strip()[:200]}",
         )
     try:
-        streams = (json.loads(out.stdout) or {}).get("streams") or []
+        parsed = json.loads(out.stdout) or {}
     except json.JSONDecodeError as exc:
         return ProbeResult(ok=False, crashed=False, reason=f"ffprobe output parse failed: {exc}")
+    streams = parsed.get("streams") or []
     if not streams:
         return ProbeResult(ok=False, crashed=False, reason="no decodable video stream")
+    duration = None
+    raw_dur = (parsed.get("format") or {}).get("duration")
+    try:
+        duration = float(raw_dur) if raw_dur is not None else None
+    except (TypeError, ValueError):
+        duration = None
     return ProbeResult(
         ok=True, width=streams[0].get("width"), height=streams[0].get("height"),
+        duration=duration,
     )
 
 

frontend/src/components/settings/MaintenancePanel.vue

@@ -17,6 +17,7 @@
     <DbMaintenanceCard class="mt-6" />
     <ArchiveReextractCard class="mt-6" />
     <MissingFileRepairCard class="mt-6" />
+    <VideoDedupCard class="mt-6" />
     <BackupCard class="mt-6" />
     <!-- TagMaintenanceCard moved to Cleanup tab (v26.05.25.7) — it
          operates on the existing library which fits the Cleanup-tab
@@ -37,6 +38,7 @@ import AliasTable from './AliasTable.vue'
 import DbMaintenanceCard from './DbMaintenanceCard.vue'
 import ArchiveReextractCard from './ArchiveReextractCard.vue'
 import MissingFileRepairCard from './MissingFileRepairCard.vue'
+import VideoDedupCard from './VideoDedupCard.vue'
 import BackupCard from './BackupCard.vue'
 import { useSystemActivityStore } from '../../stores/systemActivity.js'
 

frontend/src/components/settings/VideoDedupCard.vue

@@ -0,0 +1,149 @@
+<template>
+  <!-- #871: Tier-1 video dedup — collapse same-artist videos that match on
+       duration + aspect (the same clip re-encoded / pulled from multiple
+       sources). Preview first, then apply (destructive: removes the redundant
+       copies, keeping the highest-resolution one). -->
+  <v-card>
+    <v-card-title>Deduplicate videos</v-card-title>
+    <v-card-text>
+      <p class="text-body-2 mb-3">
+        Finds videos of the same artist that are the same content across
+        re-encodes (matching duration + aspect ratio) and keeps the
+        highest-resolution copy. <strong>Preview</strong> first to see what would
+        be removed; <strong>Apply</strong> re-points each post to the kept copy,
+        then deletes the redundant videos and their files. The first run also
+        computes durations for older videos, so it may take a while.
+      </p>
+
+      <div class="d-flex align-center flex-wrap" style="gap: 12px;">
+        <v-btn
+          color="primary" variant="tonal" rounded="pill"
+          :loading="previewing" :disabled="applying" @click="preview"
+        >
+          <v-icon start>mdi-magnify</v-icon> Preview
+        </v-btn>
+        <v-btn
+          color="error" rounded="pill"
+          :loading="applying"
+          :disabled="previewing || !canApply"
+          @click="confirmOpen = true"
+        >
+          <v-icon start>mdi-content-duplicate</v-icon> Apply
+        </v-btn>
+      </div>
+
+      <v-alert
+        v-if="summary" :type="summaryType" variant="tonal" class="mt-4"
+        density="comfortable"
+      >
+        <span v-if="applied">
+          Removed {{ summary.deleted }} redundant video(s) across
+          {{ summary.groups }} group(s); re-pointed {{ summary.relinked_posts }}
+          post link(s); reclaimed {{ humanBytes(summary.reclaim_bytes) }}.
+        </span>
+        <span v-else-if="summary.redundant > 0">
+          {{ summary.redundant }} redundant video(s) across {{ summary.groups }}
+          group(s) — {{ humanBytes(summary.reclaim_bytes) }} reclaimable. Click
+          <strong>Apply</strong> to remove them (keeping the best copy).
+        </span>
+        <span v-else>No duplicate videos found.</span>
+      </v-alert>
+
+      <QueueStatusBar queue="maintenance" queue-label="Maintenance" />
+    </v-card-text>
+
+    <v-dialog v-model="confirmOpen" max-width="440">
+      <v-card>
+        <v-card-title>Remove duplicate videos?</v-card-title>
+        <v-card-text class="text-body-2">
+          This permanently deletes
+          <strong>{{ summary?.redundant ?? 0 }}</strong> redundant video file(s),
+          keeping the highest-resolution copy in each group. Posts are re-pointed
+          to the kept copy first, so nothing disappears from a post.
+        </v-card-text>
+        <v-card-actions>
+          <v-spacer />
+          <v-btn variant="text" @click="confirmOpen = false">Cancel</v-btn>
+          <v-btn color="error" @click="apply">Remove duplicates</v-btn>
+        </v-card-actions>
+      </v-card>
+    </v-dialog>
+  </v-card>
+</template>
+
+<script setup>
+import { computed, ref } from 'vue'
+
+import { useApi } from '../../composables/useApi.js'
+import { toast } from '../../utils/toast.js'
+import QueueStatusBar from './QueueStatusBar.vue'
+
+const api = useApi()
+const previewing = ref(false)
+const applying = ref(false)
+const confirmOpen = ref(false)
+const summary = ref(null)
+const applied = ref(false)
+
+const canApply = computed(() => !!summary.value && !applied.value && summary.value.redundant > 0)
+const summaryType = computed(() => {
+  if (applied.value) return 'success'
+  return summary.value && summary.value.redundant > 0 ? 'info' : 'success'
+})
+
+function humanBytes (n) {
+  const b = Number(n || 0)
+  if (b >= 1 << 30) return (b / (1 << 30)).toFixed(1) + ' GB'
+  if (b >= 1 << 20) return (b / (1 << 20)).toFixed(1) + ' MB'
+  if (b >= 1 << 10) return (b / (1 << 10)).toFixed(1) + ' KB'
+  return b + ' B'
+}
+
+// Poll the maintenance task result until ready (or give up). The dedup task can
+// re-probe the whole video library on its first run, so allow a generous window.
+async function pollResult (taskId) {
+  for (let i = 0; i < 150; i++) {
+    await new Promise(r => setTimeout(r, 2000))
+    const r = await api.get(`/api/admin/maintenance/task-result/${taskId}`)
+    if (r.ready) {
+      if (!r.successful) throw new Error('the dedup task failed — check the worker logs')
+      return r.result
+    }
+  }
+  throw new Error('timed out waiting for the dedup task — check the task dashboard')
+}
+
+async function run (dryRun) {
+  const { task_id: taskId } = await api.post(
+    '/api/admin/maintenance/dedup-videos', { body: { dry_run: dryRun } },
+  )
+  return pollResult(taskId)
+}
+
+async function preview () {
+  previewing.value = true
+  applied.value = false
+  summary.value = null
+  try {
+    summary.value = await run(true)
+  } catch (e) {
+    toast({ text: e?.body?.detail || e?.message || 'Preview failed', type: 'error' })
+  } finally {
+    previewing.value = false
+  }
+}
+
+async function apply () {
+  confirmOpen.value = false
+  applying.value = true
+  try {
+    summary.value = await run(false)
+    applied.value = true
+    toast({ text: 'Duplicate videos removed', type: 'success' })
+  } catch (e) {
+    toast({ text: e?.body?.detail || e?.message || 'Apply failed', type: 'error' })
+  } finally {
+    applying.value = false
+  }
+}
+</script>

tests/test_cleanup_service.py

@@ -511,3 +511,99 @@ def test_reset_content_tagging_deletes_content_keeps_fandom_series(db_sync, tmp_
     # Only the series image_tag association survived (content ones cascaded).
     remaining = db_sync.execute(select(image_tag.c.tag_id)).scalars().all()
     assert remaining == [s.id]
+
+
+# ---- Tier-1 video dedup (#871) ------------------------------------------
+
+
+def _make_video(db_sync, *, artist, path, sha256, duration, w, h, size=2000):
+    img = ImageRecord(
+        artist_id=artist.id, path=path, sha256=sha256, size_bytes=size,
+        mime="video/mp4", origin="downloaded",
+        width=w, height=h, duration_seconds=duration,
+    )
+    db_sync.add(img)
+    db_sync.flush()
+    return img
+
+
+def test_dedup_videos_collapses_and_relinks(db_sync, tmp_path):
+    """Apply: a same-artist, same-duration, same-aspect video pair collapses to the
+    higher-res keeper, and the loser's post is re-pointed to the keeper first so it
+    doesn't lose the video."""
+    a = _make_artist(db_sync, slug="vd", name="VD")
+    keeper = _make_video(
+        db_sync, artist=a, path=str(tmp_path / "k.mp4"), sha256="k" * 64,
+        duration=120.0, w=1920, h=1080,
+    )
+    loser = _make_video(
+        db_sync, artist=a, path=str(tmp_path / "l.mp4"), sha256="l" * 64,
+        duration=120.3, w=1280, h=720,  # within 1s, same 16:9, smaller
+    )
+    (tmp_path / "k.mp4").write_bytes(b"k")
+    (tmp_path / "l.mp4").write_bytes(b"l")
+    pa = Post(artist_id=a.id, external_post_id="pa")
+    pb = Post(artist_id=a.id, external_post_id="pb")
+    db_sync.add_all([pa, pb])
+    db_sync.flush()
+    db_sync.add(ImageProvenance(image_record_id=keeper.id, post_id=pa.id))
+    db_sync.add(ImageProvenance(image_record_id=loser.id, post_id=pb.id))
+    db_sync.commit()
+    keeper_id, loser_id, pb_id = keeper.id, loser.id, pb.id
+
+    out = cleanup_service.dedup_videos(db_sync, images_root=tmp_path, dry_run=False)
+    assert out["groups"] == 1
+    assert out["deleted"] == 1
+    assert out["relinked_posts"] == 1
+
+    db_sync.expire_all()
+    assert db_sync.get(ImageRecord, loser_id) is None
+    assert db_sync.get(ImageRecord, keeper_id) is not None
+    linked = db_sync.execute(
+        select(ImageProvenance.post_id)
+        .where(ImageProvenance.image_record_id == keeper_id)
+    ).scalars().all()
+    assert pb_id in set(linked)  # keeper inherited the loser's post
+    assert not (tmp_path / "l.mp4").exists()  # redundant file removed
+
+
+def test_dedup_videos_dry_run_keeps_everything(db_sync, tmp_path):
+    """Preview reports the group without deleting (rule 93 shared predicate)."""
+    a = _make_artist(db_sync, slug="vd2", name="VD2")
+    _make_video(
+        db_sync, artist=a, path=str(tmp_path / "k.mp4"), sha256="a" * 64,
+        duration=90.0, w=1920, h=1080,
+    )
+    _make_video(
+        db_sync, artist=a, path=str(tmp_path / "l.mp4"), sha256="b" * 64,
+        duration=90.2, w=1280, h=720,
+    )
+    db_sync.commit()
+
+    out = cleanup_service.dedup_videos(db_sync, images_root=tmp_path, dry_run=True)
+    assert out["groups"] == 1
+    assert out["redundant"] == 1
+    assert "deleted" not in out
+    assert db_sync.execute(
+        select(func.count(ImageRecord.id))
+    ).scalar_one() == 2
+
+
+def test_dedup_videos_distinct_durations_not_grouped(db_sync, tmp_path):
+    """False-merge guard: clearly different durations stay separate."""
+    a = _make_artist(db_sync, slug="vd3", name="VD3")
+    _make_video(
+        db_sync, artist=a, path=str(tmp_path / "x.mp4"), sha256="c" * 64,
+        duration=60.0, w=1920, h=1080,
+    )
+    _make_video(
+        db_sync, artist=a, path=str(tmp_path / "y.mp4"), sha256="d" * 64,
+        duration=200.0, w=1920, h=1080,
+    )
+    db_sync.commit()
+
+    out = cleanup_service.dedup_videos(db_sync, images_root=tmp_path, dry_run=True)
+    assert out["groups"] == 0
+    assert db_sync.execute(
+        select(func.count(ImageRecord.id))
+    ).scalar_one() == 2

tests/test_importer_attach_in_place.py

@@ -3,11 +3,13 @@
 from pathlib import Path
 
 import pytest
-from sqlalchemy import select
+from sqlalchemy import func, select
 
 from backend.app.models import Artist, ImageRecord, ImportSettings
 from backend.app.services.importer import Importer
 from backend.app.services.thumbnailer import Thumbnailer
+from backend.app.utils import safe_probe
+from backend.app.utils.safe_probe import ProbeResult
 
 pytestmark = pytest.mark.integration
 
@@ -303,6 +305,100 @@ def test_attach_in_place_post_first_false_applies_post_fields(importer, db_sync)
     assert post.post_title == "Applies"
 
 
+def _fake_video_probe(monkeypatch, dims):
+    """Patch safe_probe.probe_video to return controlled (w,h,duration) keyed by
+    filename — so video dedup (#871) can be tested without real mp4s/ffprobe."""
+    def _probe(path, **_kw):
+        w, h, d = dims[Path(path).name]
+        return ProbeResult(ok=True, width=w, height=h, duration=d)
+    monkeypatch.setattr(safe_probe, "probe_video", _probe)
+
+
+def _mkvideo(images_root, rel, content):
+    p = images_root / rel
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(content)
+    return p
+
+
+def test_attach_in_place_video_dedup_skips_smaller(importer, db_sync, monkeypatch):
+    """#871 Tier-1: a same-artist video with matching duration+aspect but smaller
+    resolution is a near-dup of the existing larger video → skipped + linked, not
+    re-imported (the #859 'same video from multiple sources' case)."""
+    images_root = importer.images_root
+    artist = Artist(name="Vid", slug="vid")
+    db_sync.add(artist)
+    db_sync.flush()
+    _fake_video_probe(monkeypatch, {
+        "a.mp4": (1920, 1080, 300.0),
+        "b.mp4": (1280, 720, 300.4),  # within 1s, same 16:9 aspect, smaller
+    })
+
+    big = _mkvideo(images_root, "vid/patreon/p/a.mp4", b"AAAA")
+    r1 = importer.attach_in_place(big, artist=artist)
+    assert r1.status == "imported"
+
+    small = _mkvideo(images_root, "vid/patreon/p2/b.mp4", b"BBBB")
+    r2 = importer.attach_in_place(small, artist=artist)
+    assert r2.status == "skipped"
+    assert r2.skip_reason.value == "duplicate_phash"
+    assert r2.image_id == r1.image_id
+    assert db_sync.execute(
+        select(func.count()).select_from(ImageRecord)
+    ).scalar_one() == 1
+
+
+def test_attach_in_place_video_dedup_supersedes_larger(importer, db_sync, monkeypatch):
+    """A higher-res re-download supersedes the smaller existing video, keeping the
+    row id + adopting the new duration."""
+    images_root = importer.images_root
+    artist = Artist(name="Vid2", slug="vid2")
+    db_sync.add(artist)
+    db_sync.flush()
+    _fake_video_probe(monkeypatch, {
+        "s.mp4": (640, 360, 120.0),
+        "l.mp4": (1920, 1080, 120.2),  # within 1s, same aspect, larger
+    })
+
+    small = _mkvideo(images_root, "vid2/p/s.mp4", b"SSSS")
+    r1 = importer.attach_in_place(small, artist=artist)
+    assert r1.status == "imported"
+
+    big = _mkvideo(images_root, "vid2/p/l.mp4", b"LLLL")
+    r2 = importer.attach_in_place(big, artist=artist)
+    assert r2.status == "superseded"
+    assert r2.image_id == r1.image_id
+
+    db_sync.expire_all()
+    rec = db_sync.get(ImageRecord, r1.image_id)
+    assert rec.width == 1920
+    assert rec.duration_seconds == 120.2
+    assert db_sync.execute(
+        select(func.count()).select_from(ImageRecord)
+    ).scalar_one() == 1
+
+
+def test_attach_in_place_video_distinct_durations_not_merged(importer, db_sync, monkeypatch):
+    """Guard against false-merge: two videos with clearly different durations stay
+    separate even at the same artist + resolution."""
+    images_root = importer.images_root
+    artist = Artist(name="Vid3", slug="vid3")
+    db_sync.add(artist)
+    db_sync.flush()
+    _fake_video_probe(monkeypatch, {
+        "a.mp4": (1920, 1080, 60.0),
+        "b.mp4": (1920, 1080, 200.0),  # 140s apart — different content
+    })
+
+    a = _mkvideo(images_root, "vid3/p/a.mp4", b"AAAA")
+    b = _mkvideo(images_root, "vid3/p/b.mp4", b"BBBB")
+    assert importer.attach_in_place(a, artist=artist).status == "imported"
+    assert importer.attach_in_place(b, artist=artist).status == "imported"
+    assert db_sync.execute(
+        select(func.count()).select_from(ImageRecord)
+    ).scalar_one() == 2
+
+
 def test_attach_in_place_invalid_image_returns_skipped(importer):
     images_root = importer.images_root
     bad = images_root / "fred" / "bad.jpg"

tests/test_tasks_admin.py

@@ -42,6 +42,10 @@ def test_bulk_delete_images_task_registered():
     )
 
 
+def test_dedup_videos_task_registered():
+    assert "backend.app.tasks.admin.dedup_videos_task" in celery.tasks
+
+
 # --- delete_artist_cascade_task -------------------------------------