Ben Stull
1d7f821ab8
Per sub-project-2 plan Task 10. argparse entry wiring named fetcher + HeuristicProposer into ingest_search/ingest_candidate; --query/--resolve, --limit, --catalog, --media-root (env HEF_MEDIA_ROOT), --dominant-color. Unknown archive -> exit 2, deferred archive -> exit 3. Secrets env-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
"""CLI entry point for ingest: `python -m tools.ingest_cli <archive> ...`.
|
|
|
|
Wires a named fetcher + the HeuristicProposer into the ingest pipeline. Secrets
|
|
(e.g. FREESOUND_API_TOKEN) are read from the environment only, never as flags.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
|
|
from tools.drafting import HeuristicProposer
|
|
from tools.http import HttpClient
|
|
from tools.ingest.base import ingest_candidate, ingest_search
|
|
from tools.ingest.fma import FmaFetcher
|
|
from tools.ingest.freesound import FreesoundFetcher
|
|
from tools.ingest.internet_archive import InternetArchiveFetcher
|
|
from tools.ingest.librivox import LibriVoxFetcher
|
|
from tools.ingest.musopen import MusopenFetcher
|
|
from tools.ingest.nasa import NasaFetcher
|
|
|
|
FETCHERS = {
|
|
"librivox": LibriVoxFetcher,
|
|
"nasa": NasaFetcher,
|
|
"internet_archive": InternetArchiveFetcher,
|
|
"musopen": MusopenFetcher,
|
|
"fma": FmaFetcher,
|
|
"freesound": FreesoundFetcher,
|
|
}
|
|
|
|
DEFAULT_CATALOG = "catalog/library.jsonl"
|
|
DEFAULT_MEDIA_ROOT = "./media"
|
|
|
|
|
|
def _build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(
|
|
prog="tools.ingest_cli",
|
|
description="Ingest candidates from an archive into the catalog as proposed records.",
|
|
)
|
|
p.add_argument("archive", help=f"source archive ({', '.join(sorted(FETCHERS))})")
|
|
p.add_argument("--query", help="search query")
|
|
p.add_argument("--resolve", metavar="IDENTIFIER", help="ingest a single item by id/URL")
|
|
p.add_argument("--limit", type=int, default=5, help="max search hits (default 5)")
|
|
p.add_argument("--catalog", default=DEFAULT_CATALOG, help=f"catalog JSONL (default {DEFAULT_CATALOG})")
|
|
p.add_argument(
|
|
"--media-root",
|
|
default=os.environ.get("HEF_MEDIA_ROOT", DEFAULT_MEDIA_ROOT),
|
|
help="download root (env HEF_MEDIA_ROOT; default ./media)",
|
|
)
|
|
p.add_argument(
|
|
"--dominant-color",
|
|
action="store_true",
|
|
help="compute dominant_color for video/av (opt-in, ffmpeg-only)",
|
|
)
|
|
return p
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
args = _build_parser().parse_args(argv)
|
|
|
|
cls = FETCHERS.get(args.archive)
|
|
if cls is None:
|
|
print(
|
|
f"error: unknown archive {args.archive!r}; "
|
|
f"choose one of {', '.join(sorted(FETCHERS))}",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
if not args.query and not args.resolve:
|
|
print("error: provide --query or --resolve", file=sys.stderr)
|
|
return 2
|
|
|
|
fetcher = cls(HttpClient())
|
|
kw = dict(
|
|
catalog_path=args.catalog,
|
|
media_root=args.media_root,
|
|
proposer=HeuristicProposer(),
|
|
compute_color=args.dominant_color,
|
|
)
|
|
|
|
try:
|
|
if args.resolve:
|
|
rec = ingest_candidate(fetcher.resolve(args.resolve), **kw)
|
|
appended = [rec] if rec is not None else []
|
|
else:
|
|
appended = ingest_search(fetcher, args.query, limit=args.limit, **kw)
|
|
except NotImplementedError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 3
|
|
|
|
print(f"ingested {len(appended)} proposed record(s) into {args.catalog}")
|
|
for rec in appended:
|
|
print(f" {rec.id} [{rec.mode}] {rec.title}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
raise SystemExit(main())
|