File size: 2,585 Bytes
75ce203 658d200 75ce203 19c84c2 75ce203 19c84c2 75ce203 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """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()
|