"""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