feat(cli): wire all commands via Typer (ingest, search, discover, provenance, ask-stub)
- Implement all CLI commands as specified in F-07: ingest, search, and discover subcommands (leads, citing, cited-by, cocited), and provenance subcommands (scan, add-link, links, bib). - Add ask command stub that exits with code 1 (not yet implemented). - Implement comprehensive tests with mocked dependencies (no real DB/API calls). Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
155
codex/cli.py
155
codex/cli.py
@@ -1,8 +1,157 @@
|
||||
"""codex CLI — commands wired to domain modules."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Optional
|
||||
|
||||
import typer
|
||||
|
||||
app = typer.Typer(help="codex — personal knowledge base for scientific papers.")
|
||||
discover_app = typer.Typer(help="Discovery queries over the citation graph.")
|
||||
prov_app = typer.Typer(help="Provenance: @cite scan, code_links, bib export.")
|
||||
app.add_typer(discover_app, name="discover")
|
||||
app.add_typer(prov_app, name="provenance")
|
||||
|
||||
|
||||
@app.callback()
|
||||
def _main() -> None:
|
||||
"""codex CLI — commands land here in F-07."""
|
||||
@app.command()
|
||||
def ingest(
|
||||
paper_id: str = typer.Argument(..., help="arXiv ID, DOI, or OpenAlex W-ID"),
|
||||
source: Optional[str] = typer.Option(None, "--source", "-s", help="Path to .tex or .pdf"), # noqa: UP045
|
||||
) -> None:
|
||||
"""Ingest a paper into the knowledge base."""
|
||||
from codex.ingest import ingest_paper
|
||||
|
||||
result = ingest_paper(paper_id, source_path=source)
|
||||
typer.echo(
|
||||
f"Ingested {result.paper_id}: {result.chunks_upserted} chunks, "
|
||||
f"{result.citations_upserted} citations"
|
||||
)
|
||||
|
||||
|
||||
@app.command()
|
||||
def search(
|
||||
query: str = typer.Argument(..., help="Natural-language search query"),
|
||||
limit: int = typer.Option(10, "--limit", "-n", help="Number of results"),
|
||||
) -> None:
|
||||
"""Semantic similarity search over paper abstracts."""
|
||||
from codex.db import get_conn
|
||||
from codex.embed import get_embedder
|
||||
|
||||
emb = get_embedder().encode_dense([query])[0].tolist()
|
||||
with get_conn() as conn:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT id, title, year,
|
||||
abstract_emb <-> %(emb)s AS distance
|
||||
FROM papers
|
||||
WHERE abstract_emb IS NOT NULL
|
||||
ORDER BY abstract_emb <-> %(emb)s
|
||||
LIMIT %(limit)s
|
||||
""",
|
||||
{"emb": emb, "limit": limit},
|
||||
).fetchall()
|
||||
for row in rows:
|
||||
typer.echo(f"[{row['distance']:.3f}] {row['id']} ({row['year']}) — {row['title']}")
|
||||
|
||||
|
||||
@discover_app.command("leads")
|
||||
def discover_leads(
|
||||
limit: int = typer.Option(20, "--limit", "-n"),
|
||||
) -> None:
|
||||
"""Show discovery leads: cited papers not yet ingested, ranked by pull."""
|
||||
from codex.discover import discovery_leads
|
||||
|
||||
for item in discovery_leads(limit=limit):
|
||||
typer.echo(f"{item['pull']:>4}× {item['cited_id']}")
|
||||
|
||||
|
||||
@discover_app.command("citing")
|
||||
def discover_citing(paper_id: str = typer.Argument(...)) -> None:
|
||||
"""List papers that cite PAPER_ID."""
|
||||
from codex.discover import citing_papers
|
||||
|
||||
for pid in citing_papers(paper_id):
|
||||
typer.echo(pid)
|
||||
|
||||
|
||||
@discover_app.command("cited-by")
|
||||
def discover_cited_by(paper_id: str = typer.Argument(...)) -> None:
|
||||
"""List papers that PAPER_ID cites."""
|
||||
from codex.discover import cited_by
|
||||
|
||||
for pid in cited_by(paper_id):
|
||||
typer.echo(pid)
|
||||
|
||||
|
||||
@discover_app.command("cocited")
|
||||
def discover_cocited(
|
||||
paper_id: str = typer.Argument(...),
|
||||
limit: int = typer.Option(10, "--limit", "-n"),
|
||||
) -> None:
|
||||
"""List papers frequently co-cited with PAPER_ID."""
|
||||
from codex.discover import cocited_papers
|
||||
|
||||
for item in cocited_papers(paper_id, limit=limit):
|
||||
typer.echo(f"{item['co_citations']:>4}× {item['paper_id']}")
|
||||
|
||||
|
||||
@prov_app.command("scan")
|
||||
def prov_scan(root_dir: str = typer.Argument(..., help="Root directory to scan")) -> None:
|
||||
"""Scan C++ sources for @cite tags."""
|
||||
from codex.provenance import scan_cite_tags
|
||||
|
||||
hits = scan_cite_tags(root_dir)
|
||||
typer.echo(json.dumps(hits, indent=2))
|
||||
|
||||
|
||||
@prov_app.command("add-link")
|
||||
def prov_add_link(
|
||||
symbol: str = typer.Argument(...),
|
||||
paper_id: Optional[str] = typer.Option(None, "--paper-id"), # noqa: UP045
|
||||
role: Optional[str] = typer.Option(None, "--role"), # noqa: UP045
|
||||
note: Optional[str] = typer.Option(None, "--note"), # noqa: UP045
|
||||
) -> None:
|
||||
"""Add a code → paper link."""
|
||||
from codex.provenance import add_code_link
|
||||
|
||||
link = add_code_link(symbol, paper_id=paper_id, role=role, note=note)
|
||||
typer.echo(f"Created code_link id={link.id}: {symbol} → {paper_id}")
|
||||
|
||||
|
||||
@prov_app.command("links")
|
||||
def prov_links(
|
||||
paper_id: Optional[str] = typer.Option(None, "--paper-id"), # noqa: UP045
|
||||
) -> None:
|
||||
"""List code → paper links."""
|
||||
from codex.provenance import list_code_links
|
||||
|
||||
for link in list_code_links(paper_id=paper_id):
|
||||
typer.echo(f"[{link.id}] {link.symbol} → {link.paper_id} ({link.role})")
|
||||
|
||||
|
||||
@prov_app.command("bib")
|
||||
def prov_bib(
|
||||
paper_ids: Optional[list[str]] = typer.Argument(None), # noqa: UP045, B008
|
||||
output: Optional[str] = typer.Option( # noqa: UP045
|
||||
None, "--output", "-o", help="Write to file instead of stdout"
|
||||
),
|
||||
) -> None:
|
||||
"""Export papers as BibTeX."""
|
||||
from codex.provenance import export_bib
|
||||
|
||||
bib = export_bib(paper_ids if paper_ids else None)
|
||||
if output:
|
||||
from pathlib import Path
|
||||
|
||||
Path(output).write_text(bib, encoding="utf-8")
|
||||
typer.echo(f"Wrote {output}")
|
||||
else:
|
||||
typer.echo(bib)
|
||||
|
||||
|
||||
@app.command()
|
||||
def ask(question: str = typer.Argument(..., help="Question to answer")) -> None:
|
||||
"""Ask a question (RAG — not yet implemented)."""
|
||||
typer.echo("ask: not yet implemented", err=True)
|
||||
raise typer.Exit(1)
|
||||
|
||||
Reference in New Issue
Block a user