feat(import): Tier-1 video near-dup by duration+aspect (#871)

Videos deduped on sha256 only (pHash is images-only), so a different encode/remux
of the same clip imported as a distinct record — the "same video from multiple
sources" clutter surfaced by #859.

Tier-1 metadata fingerprint: identity = container duration (±1.0s) + matching
aspect ratio, scoped to the same artist; quality axis = pixel dimensions (mirrors
image pHash: larger_exists→skip+link, smaller_exists→supersede). Codec/bitrate
are deliberately NOT part of identity (the point is matching across re-encodes).
Tight tolerances because a wrong video merge is destructive.

- image_record.duration_seconds (Float, nullable; migration 0052). NULL for images.
- safe_probe.probe_video also reads format=duration (one extra ffprobe field on the
  call that already runs); ProbeResult.duration.
- _find_similar_video(duration,w,h,artist) shared by both import pipelines.
- _import_media (filesystem/archive path): captures duration, video near-dup
  branch, persists duration.
- attach_in_place (download path — handles #859's videos, previously didn't probe
  video at all): best-effort probe for dims+duration (LENIENT — never newly rejects
  a downloaded video on probe failure), video near-dup branch, persists duration.
- _supersede carries duration onto the kept row.

Reuses SkipReason.duplicate_phash so the existing download/external dup-cleanup
(path-safe unlink, #859) applies unchanged. Tests: skip-smaller, supersede-larger
(+ duration adopted), and distinct-durations-not-merged (false-merge guard).

Follow-up (Phase 2, #871): a backfill to re-probe NULL-duration existing videos so
the current library participates in dedup; retroactive merge of existing dups is a
separate destructive maintenance action.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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"