feat: ingest pipeline (Candidate/Fetcher/ingest_candidate)

Per sub-project-2 plan Task 8 / spec §6. ingest_candidate dedupes, downloads, mechanically tags, drafts coordinates, builds a proposed Record and appends it; ingest_search loops over fetcher hits. All boundaries (download/prober/color_fn) injectable, so the pipeline is fully hermetic. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-04 06:29:09 -07:00
parent 3d5c34491c
commit 2db2b7ac16
3 changed files with 341 additions and 0 deletions
@@ -0,0 +1,193 @@
+import pytest
+
+from hef.catalog import CatalogError, load_catalog, validate_catalog
+from tools.drafting import HeuristicProposer
+from tools.ingest.base import Candidate, ingest_candidate, ingest_search
+from tools.probe import Probe
+
+
+def make_candidate(**o):
+    base = dict(
+        source_archive="nasa",
+        source_url="https://example.org/landing/x",
+        media_url="https://example.org/media/x.mp4",
+        title="Earthrise",
+        license="public_domain",
+        attribution="",
+        suggested_id="nasa-earthrise",
+        media_ext="mp4",
+        description="a view of earth",
+    )
+    base.update(o)
+    return Candidate(**base)
+
+
+def video_probe():
+    return Probe(
+        streams=[
+            {
+                "codec_type": "video",
+                "width": 1920,
+                "height": 1080,
+                "disposition": {"attached_pic": 0},
+            },
+            {"codec_type": "audio"},
+        ],
+        format={"duration": "10.0"},
+    )
+
+
+def make_prober(probe):
+    def prober(path):
+        return probe
+
+    return prober
+
+
+def make_downloader(content=b"FAKEMEDIA"):
+    def downloader(url, dest):
+        dest.write_bytes(content)
+
+    return downloader
+
+
+def test_ingest_appends_one_proposed_record(tmp_path):
+    catalog = tmp_path / "library.jsonl"
+    media_root = tmp_path / "media"
+    rec = ingest_candidate(
+        make_candidate(),
+        catalog_path=catalog,
+        media_root=media_root,
+        proposer=HeuristicProposer(),
+        prober=make_prober(video_probe()),
+        downloader=make_downloader(),
+    )
+    assert rec is not None
+    records = load_catalog(catalog)
+    assert len(records) == 1
+    r = records[0]
+    assert r.id == "nasa-earthrise"
+    assert r.mode == "av"
+    assert r.resolution == "1920x1080"
+    assert r.duration_s == 10
+    assert r.review_status == "proposed"
+    assert r.reviewed_at is None
+    assert r.rationale != ""
+    assert r.file_path == "nasa/nasa-earthrise.mp4"
+    assert r.notes == "a view of earth"
+    # media written under media_root/<archive>/<id>.<ext>
+    assert (media_root / "nasa" / "nasa-earthrise.mp4").exists()
+
+
+def test_ingest_is_idempotent(tmp_path):
+    catalog = tmp_path / "library.jsonl"
+    media_root = tmp_path / "media"
+    kw = dict(
+        catalog_path=catalog,
+        media_root=media_root,
+        proposer=HeuristicProposer(),
+        prober=make_prober(video_probe()),
+        downloader=make_downloader(),
+    )
+    first = ingest_candidate(make_candidate(), **kw)
+    second = ingest_candidate(make_candidate(), **kw)
+    assert first is not None
+    assert second is None  # skipped as duplicate
+    assert len(load_catalog(catalog)) == 1
+
+
+def test_dominant_color_off_by_default(tmp_path):
+    catalog = tmp_path / "library.jsonl"
+    rec = ingest_candidate(
+        make_candidate(),
+        catalog_path=catalog,
+        media_root=tmp_path / "media",
+        proposer=HeuristicProposer(),
+        prober=make_prober(video_probe()),
+        downloader=make_downloader(),
+    )
+    assert rec.dominant_color == ""
+
+
+def test_dominant_color_computed_when_enabled(tmp_path):
+    catalog = tmp_path / "library.jsonl"
+    color_calls = []
+
+    def fake_color_fn(path, *, midpoint_s=0.0):
+        color_calls.append(midpoint_s)
+        return "#abcdef"
+
+    rec = ingest_candidate(
+        make_candidate(),
+        catalog_path=catalog,
+        media_root=tmp_path / "media",
+        proposer=HeuristicProposer(),
+        prober=make_prober(video_probe()),
+        downloader=make_downloader(),
+        compute_color=True,
+        color_fn=fake_color_fn,
+    )
+    assert rec.dominant_color == "#abcdef"
+    assert color_calls  # was invoked
+
+
+def test_audio_keeps_empty_color_even_when_enabled(tmp_path):
+    audio = Probe(streams=[{"codec_type": "audio"}], format={"duration": "30.0"})
+
+    def boom(path, *, midpoint_s=0.0):
+        raise AssertionError("must not compute color for audio")
+
+    rec = ingest_candidate(
+        make_candidate(media_ext="mp3", source_archive="librivox", license="public_domain"),
+        catalog_path=tmp_path / "library.jsonl",
+        media_root=tmp_path / "media",
+        proposer=HeuristicProposer(),
+        prober=make_prober(audio),
+        downloader=make_downloader(),
+        compute_color=True,
+        color_fn=boom,
+    )
+    assert rec.mode == "audio" and rec.dominant_color == ""
+
+
+def test_unmappable_license_raises_and_writes_nothing(tmp_path):
+    catalog = tmp_path / "library.jsonl"
+    with pytest.raises(CatalogError):
+        ingest_candidate(
+            make_candidate(license="all_rights_reserved"),
+            catalog_path=catalog,
+            media_root=tmp_path / "media",
+            proposer=HeuristicProposer(),
+            prober=make_prober(video_probe()),
+            downloader=make_downloader(),
+        )
+    assert not catalog.exists() or load_catalog(catalog) == []
+
+
+def test_ingest_search_runs_over_hits(tmp_path):
+    catalog = tmp_path / "library.jsonl"
+
+    class FakeFetcher:
+        archive = "nasa"
+
+        def search(self, query, *, limit):
+            return [
+                make_candidate(suggested_id="nasa-a", media_url="u/a.mp4"),
+                make_candidate(suggested_id="nasa-b", media_url="u/b.mp4"),
+            ][:limit]
+
+        def resolve(self, identifier):  # pragma: no cover - not used here
+            raise NotImplementedError
+
+    recs = ingest_search(
+        FakeFetcher(),
+        "earth",
+        limit=2,
+        catalog_path=catalog,
+        media_root=tmp_path / "media",
+        proposer=HeuristicProposer(),
+        prober=make_prober(video_probe()),
+        downloader=make_downloader(),
+    )
+    assert [r.id for r in recs] == ["nasa-a", "nasa-b"]
+    validate_catalog(load_catalog(catalog))
@@ -0,0 +1,148 @@
+"""Ingest seam: a Candidate shape, the Fetcher protocol, and the pipeline.
+
+Every external boundary — archive HTTP (download), ffprobe (prober), and the
+optional ffmpeg dominant-color (color_fn) — is injected, so the whole pipeline
+is exercised hermetically (no network, no binaries, no disk beyond tmp).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Protocol
+
+from hef.catalog import (
+    Record,
+    append_record,
+    load_catalog,
+    validate,
+    validate_catalog,
+)
+from tools.drafting import Signals
+from tools.http import HttpClient
+from tools.mediatools import compute_dominant_color
+from tools.probe import probe_file
+from tools.tagging import derive_tags
+
+log = logging.getLogger(__name__)
+
+_VIDEO_MODES = {"video", "av"}
+
+
+@dataclass
+class Candidate:
+    source_archive: str  # origin label, e.g. internet_archive | librivox | nasa
+    source_url: str  # human/landing URL recorded in the record
+    media_url: str  # direct download URL of the chosen file
+    title: str
+    license: str  # normalized (tools.licensing) -> hef LICENSES vocab
+    attribution: str  # "" unless the license requires it
+    suggested_id: str  # stable id derived from archive + archive-identifier
+    media_ext: str  # file extension for the download target
+    description: str = ""  # free text used by drafting signals; lands in notes
+
+
+class Fetcher(Protocol):
+    archive: str
+
+    def search(self, query: str, *, limit: int) -> list[Candidate]: ...
+
+    def resolve(self, identifier: str) -> Candidate: ...
+
+
+def _default_downloader(url, dest) -> None:
+    client = HttpClient()
+    dest.write_bytes(client.get_bytes(url))
+
+
+def ingest_candidate(
+    candidate: Candidate,
+    *,
+    catalog_path,
+    media_root,
+    proposer,
+    prober=probe_file,
+    downloader=_default_downloader,
+    compute_color: bool = False,
+    color_fn=compute_dominant_color,
+):
+    """Ingest one candidate into the catalog as a `proposed` record.
+
+    Returns the appended Record, or None if it was skipped as a duplicate.
+    Raises CatalogError if the resulting record is invalid (e.g. an unmappable
+    license) — nothing is appended in that case.
+    """
+    catalog_path = Path(catalog_path)
+    media_root = Path(media_root)
+
+    # 1. Dedupe.
+    existing = load_catalog(catalog_path) if catalog_path.exists() else []
+    validate_catalog(existing)
+    if any(r.id == candidate.suggested_id for r in existing):
+        log.info("already in catalog, skipping: %s", candidate.suggested_id)
+        return None
+
+    # 2. Download (skip if already present).
+    rel_path = f"{candidate.source_archive}/{candidate.suggested_id}.{candidate.media_ext}"
+    dest = media_root / rel_path
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    if not dest.exists():
+        downloader(candidate.media_url, dest)
+
+    # 3. Mechanical tagging.
+    tags = derive_tags(prober(dest))
+    dominant_color = ""
+    if compute_color and tags.mode in _VIDEO_MODES:
+        dominant_color = color_fn(dest, midpoint_s=tags.duration_s / 2)
+
+    # 4. Draft coordinates.
+    draft = proposer.propose(
+        Signals(
+            title=candidate.title,
+            description=candidate.description,
+            source_archive=candidate.source_archive,
+            mode=tags.mode,
+            duration_s=tags.duration_s,
+        )
+    )
+    coord = draft.coordinate
+
+    # 5. Build the record — proposed/reviewed_at=None falls out of the defaults.
+    record = Record(
+        id=candidate.suggested_id,
+        title=candidate.title,
+        source_url=candidate.source_url,
+        source_archive=candidate.source_archive,
+        license=candidate.license,
+        mode=tags.mode,
+        left=coord.left,
+        right=coord.right,
+        dark=coord.dark,
+        light=coord.light,
+        duration_s=tags.duration_s,
+        file_path=rel_path,
+        attribution=candidate.attribution,
+        resolution=tags.resolution,
+        dominant_color=dominant_color,
+        rationale=draft.rationale,
+        notes=candidate.description,
+    )
+
+    # 6. Validate + re-check uniqueness + append.
+    validate(record)
+    if any(r.id == record.id for r in existing):  # racey re-check
+        log.info("already in catalog (race), skipping: %s", record.id)
+        return None
+    append_record(record, catalog_path)
+    return record
+
+
+def ingest_search(fetcher: Fetcher, query: str, *, limit: int, **kw):
+    """Run ingest_candidate over each search hit. Returns the appended records."""
+    appended = []
+    for candidate in fetcher.search(query, limit=limit):
+        rec = ingest_candidate(candidate, **kw)
+        if rec is not None:
+            appended.append(rec)
+    return appended