apps/api/scripts/seed.py

"""Seed the ledger with realistic vendor history (spec §8).

Loads 30 historical invoices across 6 vendors so the vendor-spend chart and the
per-vendor z-scores are meaningful the first time the dashboard is opened, and so
the planted near-duplicate pair guarantees the anomaly demo fires.

Documents are rendered as **real PDFs and real degraded scans** and pushed through
the **real pipeline** in the same process — the same hashing, routing, extraction,
validation, screening and persistence a browser upload takes. Nothing is inserted
straight into the tables.

Order matters: a duplicate can only be found against something already in the
ledger, so the corpus is processed oldest-first.

    python scripts/seed.py            # add to whatever is already there
    python scripts/seed.py --reset    # wipe the ledger first
    python scripts/seed.py --scans    # render a subset as photographed scans
"""

from __future__ import annotations

import argparse
import asyncio
import sys
import time
from pathlib import Path

# Allow `python scripts/seed.py` from the apps/api directory.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from sqlalchemy import text

from app.core.bootstrap import init_schema
from app.core.db import dispose_engine, init_engine, transaction
from app.core.logging import configure_logging
from app.core.settings import get_settings
from app.core.tracing import get_tracer
from app.deps import get_claude_client
from app.devtools.corpus import build_seed_corpus
from app.devtools.documents import degrade_to_scan, render_invoice_pdf
from app.models.enums import DocumentStatus
from app.pipeline.orchestrator import PipelineOrchestrator

_TABLES = ("anomalies", "extractions", "llm_traces", "audit_log", "failed_jobs", "documents")


async def reset_ledger() -> None:
    """Empty every table. The audit-log trigger blocks DELETE, so use TRUNCATE."""
    async with transaction() as session:
        await session.execute(text(f"TRUNCATE {', '.join(_TABLES)} RESTART IDENTITY CASCADE"))


async def seed(*, reset: bool, with_scans: bool) -> int:
    settings = get_settings()
    configure_logging("WARNING")  # the progress table below is the useful output

    engine = init_engine(settings)
    await init_schema(engine)
    if reset:
        await reset_ledger()
        print("Ledger reset.\n")

    client = get_claude_client()
    orchestrator = PipelineOrchestrator(client=client, settings=settings, tracer=get_tracer())

    corpus = build_seed_corpus()
    print(f"Seeding {len(corpus)} invoices across 6 vendors (mode: {client.mode})\n")
    print(f"{'#':>3}  {'invoice':16} {'vendor':30} {'total':>12}  {'status':13} {'ms':>5}  flags")
    print("-" * 96)

    failures = 0
    started = time.perf_counter()

    for index, item in enumerate(corpus, start=1):
        pdf = render_invoice_pdf(item.spec)
        # A slice of the corpus is photographed so the vision lane has real input.
        as_scan = with_scans and index % 7 == 0
        payload = degrade_to_scan(pdf) if as_scan else pdf
        filename = f"{item.stem}{'.jpg' if as_scan else '.pdf'}"
        content_type = "image/jpeg" if as_scan else "application/pdf"

        outcome = await orchestrator.ingest(
            data=payload, filename=filename, declared_content_type=content_type
        )
        if outcome.duplicate:
            print(f"{index:>3}  {item.spec.invoice_number:16} {'(already ingested)':30}")
            continue

        await orchestrator.process(
            document_id=outcome.document_id,
            data=payload,
            filename=filename,
            media_type=outcome.media_type,
        )

        async with transaction() as session:
            row = (
                await session.execute(
                    text(
                        "SELECT d.status, d.latency_ms, "
                        "  (SELECT count(*) FROM anomalies a WHERE a.document_id = d.id) AS flags "
                        "FROM documents d WHERE d.id = :id"
                    ),
                    {"id": outcome.document_id},
                )
            ).one()

        status = DocumentStatus(row.status)
        if status is DocumentStatus.FAILED:
            failures += 1
        marker = "  <-- ANOMALY" if row.flags else ""
        print(
            f"{index:>3}  {item.spec.invoice_number:16} {item.spec.vendor:30} "
            f"{item.spec.total:>12,.2f}  {status.value:13} {row.latency_ms or 0:>5}  "
            f"{row.flags}{marker}"
        )

    elapsed = time.perf_counter() - started

    async with transaction() as session:
        summary = (
            await session.execute(
                text(
                    "SELECT "
                    "  (SELECT count(*) FROM documents) AS docs, "
                    "  (SELECT count(*) FROM documents WHERE status='DONE') AS done, "
                    "  (SELECT count(*) FROM documents WHERE status='NEEDS_REVIEW') AS review, "
                    "  (SELECT count(*) FROM documents WHERE status='FAILED') AS failed, "
                    "  (SELECT count(*) FROM anomalies) AS anomalies, "
                    "  (SELECT count(*) FROM audit_log) AS audit, "
                    "  (SELECT coalesce(sum(cost_usd),0) FROM documents) AS cost, "
                    "  (SELECT coalesce(avg(latency_ms),0) FROM documents "
                    "     WHERE latency_ms IS NOT NULL) AS avg_ms"
                )
            )
        ).one()

    print("-" * 96)
    print(
        f"\n{summary.docs} documents · {summary.done} DONE · {summary.review} NEEDS_REVIEW · "
        f"{summary.failed} FAILED"
    )
    print(
        f"{summary.anomalies} anomaly flag(s) · {summary.audit} audit events · "
        f"avg {float(summary.avg_ms):.0f} ms/doc · ${float(summary.cost):.4f} total"
    )
    print(f"Seeded in {elapsed:.1f}s.\n")

    await dispose_engine()
    return 1 if failures else 0


def main() -> int:
    parser = argparse.ArgumentParser(description="Seed LedgerLens with vendor history.")
    parser.add_argument("--reset", action="store_true", help="Empty the ledger first.")
    parser.add_argument(
        "--scans",
        action="store_true",
        help="Render part of the corpus as photographed scans (needs a Claude key to read).",
    )
    args = parser.parse_args()
    return asyncio.run(seed(reset=args.reset, with_scans=args.scans))


if __name__ == "__main__":
    raise SystemExit(main())