ingest/cli.py

"""Aurelius ingestion CLI.

    python -m ingest.cli finance
    python -m ingest.cli news --topics "AI,climate,markets"
    python -m ingest.cli biomed
    python -m ingest.cli finance --tickers "AAPL,MSFT,NVDA,F,GM"
    python -m ingest.cli status

Each source command runs the domain ingester (writes nodes+edges) then the
shared embed_and_fuse step (text + node2vec structural embeddings). After
that the matching adapter answers from the store and the source appears in
the app's source picker.
"""

from __future__ import annotations

import argparse
import asyncio

from core.store import get_store
from ingest.pipeline import embed_and_fuse


async def _run(args):
    if args.source == "status":
        store = get_store()
        for name in ("wikipedia", "openalex",
                     "biomed", "news", "finance"):
            n, e = store.node_count(name), store.edge_count(name)
            state = "ingested" if n else "live/empty"
            print(f"  {name:10s} {n:>9,} nodes  {e:>10,} edges   [{state}]")
        return

    if args.source == "finance":
        from ingest.finance import ingest
        tickers = args.tickers.split(",") if args.tickers else None
        await ingest([t.strip() for t in tickers] if tickers else None)
    elif args.source == "news":
        # news_intel's pipeline already runs its own index stage
        # (embed_and_fuse included) — don't run it twice.
        from ingest.news import ingest
        topics = [t.strip() for t in args.topics.split(",")] if args.topics else None
        await ingest(topics)
        return
    elif args.source == "biomed":
        if args.full or args.path:
            from ingest.biomed import ingest
            await ingest(args.path)
        else:
            from ingest.biomed_curated import ingest
            ingest()
    else:
        raise SystemExit(f"Unknown source: {args.source}")

    await embed_and_fuse(args.source)


def main():
    p = argparse.ArgumentParser(prog="aurelius-ingest")
    p.add_argument("source",
                   choices=["finance", "news", "biomed", "status"])
    p.add_argument("--tickers", help="finance: comma-separated symbols")
    p.add_argument("--topics", help="news: comma-separated topics")
    p.add_argument("--path", help="biomed: local hetionet json[.bz2] path")
    p.add_argument("--full", action="store_true",
                   help="biomed: download the full Hetionet graph instead of "
                        "the curated demo set")
    args = p.parse_args()
    asyncio.run(_run(args))


if __name__ == "__main__":
    main()