src/pageparse/cli.py

from __future__ import annotations

import hashlib
from pathlib import Path

import typer

from pageparse.config import settings
from pageparse.extract import Extractor
from pageparse.ingest import IMAGE_EXTENSIONS, load_image
from pageparse.ocr.handwriting import HandwritingOCR
from pageparse.ocr.printed import PrintedOCR
from pageparse.preprocess import preprocess
from pageparse.store import Store

app = typer.Typer()
store = Store()


@app.command()
def process(
    path: str = typer.Argument(..., help="Path to image or directory"),
    batch: bool = typer.Option(False, "--batch", help="Process all images in directory"),
    airgap: bool = typer.Option(False, "--airgap", help="Fail if any network call attempted"),
    parallel: bool = typer.Option(False, "--parallel", help="Process files in parallel"),
    schema: str = typer.Option("auto", "--schema", help="Schema type (auto, todo, meeting, recipe)"),
    diff_sync: bool = typer.Option(True, "--diff-sync/--no-diff-sync", help="Skip duplicate files"),
) -> None:
    if airgap:
        settings.airgap = True

    settings.auto_schema = schema == "auto"
    settings.diff_sync_enabled = diff_sync

    paths = [Path(path)]
    if batch and Path(path).is_dir():
        paths = []
        for ext in IMAGE_EXTENSIONS:
            paths.extend(Path(path).glob(f"*{ext}"))
            paths.extend(Path(path).glob(f"*{ext.upper()}"))

    store.init_db()

    if parallel and len(paths) > 1:
        from pageparse import pipelines
        typer.echo(f"Processing {len(paths)} files in parallel ({settings.max_workers} workers)...")
        results = pipelines.process_batch(paths)
        for r in results:
            if "error" in r:
                typer.echo(f"  FAILED: {Path(r['path']).name} — {r['error']}")
            else:
                typer.echo(f"  Processed: {Path(r['path']).name} — {len(r.get('raw_text', ''))} chars")
        return

    ocr = HandwritingOCR()
    printed_ocr = PrintedOCR()
    extractor = Extractor()
    store.init_db()

    for p in paths:
        typer.echo(f"Processing: {p.name}")

        content_bytes = p.read_bytes()
        content_hash = hashlib.sha256(content_bytes).hexdigest()

        if diff_sync:
            existing = store.get_source_by_hash(content_hash)
            if existing:
                typer.echo(f"  Skipped (duplicate): {p.name}")
                continue

        img = load_image(p)
        cleaned = preprocess(img)
        raw_text = ocr.recognize(cleaned)
        if not raw_text.strip():
            raw_text = printed_ocr.recognize(cleaned)

        if schema == "auto":
            schema = extractor.detect_schema_type(raw_text)

        result = extractor.extract(raw_text, p.name, schema)
        source_id = store.save(result, raw_text, content_hash=content_hash)
        typer.echo(f"  Saved source #{source_id} with {len(result.records)} records")


@app.command()
def list(
    priority: str | None = typer.Option(None, "--priority", help="Filter by priority"),
    type: str | None = typer.Option(None, "--type", help="Filter by record type"),
) -> None:
    records = store.get_records(priority=priority, type=type)
    for r in records:
        prio = r["priority"] or "None"
        typer.echo(f"  [{r['type']}] [{prio}] {r['content']} — {r['filename']}")


@app.command()
def search(
    query: str = typer.Argument(..., help="Search query"),
) -> None:
    from pageparse.search import SemanticSearch

    searcher = SemanticSearch()
    results = searcher.search(query)
    for r in results:
        typer.echo(f"  [{r['type']}] {r['content']} (confidence: {r.get('confidence', 'N/A')})")


@app.command()
def export(
    format: str = typer.Option("json", "--format", help="Export format (json or csv)"),
) -> None:
    import csv
    import io
    import json

    sources = store.list_sources()
    for s in sources:
        s["records"] = store.get_records(source_id=s["id"])

    if format == "json":
        typer.echo(json.dumps(sources, indent=2))
    elif format == "csv":
        records = store.get_records()
        if not records:
            typer.echo("No records to export.")
            return
        output = io.StringIO()
        writer = csv.DictWriter(output, fieldnames=records[0].keys())
        writer.writeheader()
        writer.writerows(records)
        typer.echo(output.getvalue().rstrip())


@app.command()
def serve(
    host: str = "127.0.0.1",
    port: int = 8000,
) -> None:
    import os
    import uvicorn
    env_port = os.environ.get("PORT")
    if env_port:
        try:
            port = int(env_port)
        except ValueError:
            pass
    uvicorn.run("pageparse.web:app", host=host, port=port, reload=False)


@app.command()
def benchmark(
    path: str = typer.Argument(..., help="Path to image or directory"),
    iterations: int = typer.Option(3, "--iterations", "-n", help="Number of iterations"),
) -> None:
    import time

    from pageparse.ocr.handwriting import HandwritingOCR
    from pageparse.ocr.printed import PrintedOCR
    from pageparse.preprocess import preprocess

    typer.echo(f"Benchmarking: {path} ({iterations} iterations)")

    img = load_image(path)
    ocr = HandwritingOCR()
    printed_ocr = PrintedOCR()

    times = {"preprocess": [], "handwriting_ocr": [], "printed_ocr": []}

    for i in range(iterations):
        t0 = time.time()
        cleaned = preprocess(img)
        times["preprocess"].append(time.time() - t0)

        t0 = time.time()
        ocr.recognize(cleaned)
        times["handwriting_ocr"].append(time.time() - t0)

        t0 = time.time()
        printed_ocr.recognize(cleaned)
        times["printed_ocr"].append(time.time() - t0)

    for stage, durations in times.items():
        avg = sum(durations) / len(durations)
        typer.echo(f"  {stage}: avg={avg:.3f}s, min={min(durations):.3f}s, max={max(durations):.3f}s")


@app.command()
def stats() -> None:
    store = Store()
    stats = store.get_stats()
    typer.echo(f"Sources: {stats['sources']}")
    typer.echo(f"Records: {stats['records']}")


if __name__ == "__main__":
    app()