human-experience-filter-art/tools/ingest/base.py

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