From 2db2b7ac16d8c8420b07708bf6d30273d9beabcf Mon Sep 17 00:00:00 2001 From: Ben Stull Date: Thu, 4 Jun 2026 06:29:09 -0700 Subject: [PATCH] feat: ingest pipeline (Candidate/Fetcher/ingest_candidate) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- tests/test_ingest_pipeline.py | 193 ++++++++++++++++++++++++++++++++++ tools/ingest/__init__.py | 0 tools/ingest/base.py | 148 ++++++++++++++++++++++++++ 3 files changed, 341 insertions(+) create mode 100644 tests/test_ingest_pipeline.py create mode 100644 tools/ingest/__init__.py create mode 100644 tools/ingest/base.py diff --git a/tests/test_ingest_pipeline.py b/tests/test_ingest_pipeline.py new file mode 100644 index 0000000..e4446ab --- /dev/null +++ b/tests/test_ingest_pipeline.py @@ -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//. + 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)) diff --git a/tools/ingest/__init__.py b/tools/ingest/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tools/ingest/base.py b/tools/ingest/base.py new file mode 100644 index 0000000..24cc2d6 --- /dev/null +++ b/tools/ingest/base.py @@ -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