homelab-codex-ws/jobs/documents-ingest/src/documents_ingest/paperless_adapter.py

"""Paperless -> envelope adapter — module 5 phase 2 (kb/phases/kb-m5-faza2.md,
§4.2-4.3, §6 step 5).

Reads documents from the Paperless REST API (read-only — GET only, this job never
writes to Paperless) and inserts them as `source='paperless'` rows into the `envelope`
table on kb-postgres, reusing `kb_mail.envelope.Envelope` / `kb_mail.db.insert_envelope`
(packages/kb-mail, untouched by this change — see plan §1.6). Existing `source='gmail'`
rows and `document_chunk` are never touched.

Cross-source proof (plan §1.9, §4.2): when a document's `original_file_name` matches a
`consume_name` in the documents-ingest registry.json (produced by phase 1,
jobs/documents-ingest/extractor.py), a `source_mail` entity links the document envelope
back to the originating mail envelope. No match = no `source_mail` entity — a normal,
expected outcome for documents added outside the faktury-1 pipeline.

Runs on PIHA (needs a route to both the Paperless API and kb-postgres):

Install (from repo root):
    pip install -e packages/kb-mail/
    pip install -e jobs/documents-ingest/

Usage:
    # Dry run (default) — fetch, map, count; no DB writes:
    documents-ingest-paperless --dsn postgresql://kb:<pw>@localhost:5433/kb \\
        --paperless-token <token>

    # Real run:
    documents-ingest-paperless --dsn ... --paperless-token ... --apply

DSN can come from KB_DSN, token from PAPERLESS_API_TOKEN, URL from PAPERLESS_URL
(defaults to Paperless' fixed LAN address per services/paperless/env.example).
"""
from __future__ import annotations

import argparse
import asyncio
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import AsyncIterator, Optional

import aiohttp
import asyncpg
import structlog

from kb_mail.db import insert_envelope
from kb_mail.envelope import Envelope

from .extractor import DEFAULT_REGISTRY, load_registry

_log = structlog.get_logger(__name__)

DEFAULT_PAPERLESS_URL = "http://192.168.31.5:8210"
DEFAULT_PAGE_SIZE = 200


def parse_document_ts(value: str) -> datetime:
    """Parse Paperless `created` (ISO 8601, always carries an offset) to UTC-aware."""
    dt = datetime.fromisoformat(value)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)


def build_registry_index(registry: dict[str, dict]) -> dict[str, dict]:
    """Invert the sha256-keyed registry to consume_name -> {envelope_id, sha256, filename}.

    consume_name (== Paperless `original_file_name` for faktury-1 uploads) is unique in
    the registry (verified in plan §1.9), so this is a safe 1:1 index for the join.
    """
    return {
        entry["consume_name"]: {
            "envelope_id": entry["envelope_id"],
            "sha256": sha256,
            "filename": entry["filename"],
        }
        for sha256, entry in registry.items()
    }


def find_source_mail(original_file_name: Optional[str], registry_index: dict[str, dict]) -> Optional[dict]:
    """Deterministic cross-source join: original_file_name == registry consume_name (plan §1.9).

    Returns None when there is no match — a normal outcome for documents added outside
    the faktury-1 pipeline (plan §4.2), never guessed via filename similarity or content.
    """
    if not original_file_name:
        return None
    return registry_index.get(original_file_name)


def build_entities(
    doc: dict,
    correspondents: dict[int, Optional[str]],
    tags: dict[int, Optional[str]],
    source_mail: Optional[dict],
) -> list[dict]:
    """Build the `entities[]` list for one document, per plan §4.2."""
    entities: list[dict] = [
        {"type": "content", "text": doc.get("content") or ""},
        {"type": "correspondent", "name": correspondents.get(doc.get("correspondent"))},
    ]
    for tag_id in doc.get("tags") or []:
        entities.append({"type": "tag", "name": tags.get(tag_id)})
    entities.append({"type": "filename", "value": doc.get("original_file_name")})
    entities.append({"type": "content_type", "value": doc.get("mime_type")})
    if source_mail is not None:
        entities.append({
            "type": "source_mail",
            "envelope_id": source_mail["envelope_id"],
            "attachment_sha256": source_mail["sha256"],
            "attachment_filename": source_mail["filename"],
        })
    return entities


def map_document(
    doc: dict,
    correspondents: dict[int, Optional[str]],
    tags: dict[int, Optional[str]],
    source_mail: Optional[dict],
) -> Envelope:
    """Map one Paperless API document record to a KB envelope (plan §4.2-4.3)."""
    return Envelope(
        id=f"paperless:{doc['id']}",
        source="paperless",
        ts=parse_document_ts(doc["created"]),
        raw_ref=str(doc["id"]),
        geo=None,
        entities=build_entities(doc, correspondents, tags, source_mail),
    )


async def fetch_lookup(session: aiohttp.ClientSession, base_url: str, path: str) -> dict[int, Optional[str]]:
    """Fetch a paginated Paperless id->name table (correspondents, tags) into a dict."""
    result: dict[int, Optional[str]] = {}
    url: Optional[str] = f"{base_url}{path}"
    while url:
        async with session.get(url) as resp:
            resp.raise_for_status()
            data = await resp.json()
        for item in data.get("results", []):
            result[item["id"]] = item.get("name")
        url = data.get("next")
    return result


async def iter_documents(
    session: aiohttp.ClientSession,
    base_url: str,
    page_size: int = DEFAULT_PAGE_SIZE,
    limit: Optional[int] = None,
) -> AsyncIterator[dict]:
    """Yield every document from `GET /api/documents/`, following pagination.

    `limit` caps the total number yielded (across pages) — for smoke-testing a slice
    without pulling the whole collection. Full re-runs are cheap and idempotent (~186
    docs today), so there is no --offset: unlike the 225k-row header backfill, this job
    does not need resumable partitioning.
    """
    url: Optional[str] = f"{base_url}/api/documents/?page_size={page_size}&ordering=id"
    count = 0
    while url:
        async with session.get(url) as resp:
            resp.raise_for_status()
            data = await resp.json()
        for doc in data.get("results", []):
            if limit is not None and count >= limit:
                return
            yield doc
            count += 1
        url = data.get("next")


async def run(
    dsn: str,
    paperless_url: str,
    paperless_token: str,
    registry_path: Path = DEFAULT_REGISTRY,
    limit: Optional[int] = None,
    page_size: int = DEFAULT_PAGE_SIZE,
    apply: bool = False,
) -> dict[str, int]:
    """Fetch Paperless documents, map to envelopes, and (if apply) insert new ones.

    Returns stats that must always balance:

        fetched = already_in_db + inserted + errors

    `source_mail_linked` and `empty_content` are informational subsets of `fetched`,
    not separate outcome buckets. dry-run (apply=False) never writes — `inserted`
    reports what *would* be written, mirroring documents-ingest's extractor.py.
    Idempotent via a pre-fetched set of existing `source='paperless'` envelope ids
    (insert_envelope's own ON CONFLICT DO NOTHING is the second line of defense).
    """
    stats = {
        "fetched": 0,
        "already_in_db": 0,
        "inserted": 0,
        "source_mail_linked": 0,
        "empty_content": 0,
        "errors": 0,
    }

    registry_index = build_registry_index(load_registry(registry_path))

    conn = await asyncpg.connect(dsn)
    try:
        existing_ids = {
            r["id"] for r in await conn.fetch("SELECT id FROM envelope WHERE source = 'paperless'")
        }

        headers = {"Authorization": f"Token {paperless_token}"}
        async with aiohttp.ClientSession(headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as session:
            correspondents = await fetch_lookup(session, paperless_url, "/api/correspondents/")
            tags = await fetch_lookup(session, paperless_url, "/api/tags/")

            async for doc in iter_documents(session, paperless_url, page_size=page_size, limit=limit):
                stats["fetched"] += 1

                try:
                    source_mail = find_source_mail(doc.get("original_file_name"), registry_index)
                    env = map_document(doc, correspondents, tags, source_mail)
                except Exception:
                    _log.warning("skip.map_error", document_id=doc.get("id"), exc_info=True)
                    stats["errors"] += 1
                    continue

                if not (doc.get("content") or "").strip():
                    stats["empty_content"] += 1
                if source_mail is not None:
                    stats["source_mail_linked"] += 1

                if env.id in existing_ids:
                    stats["already_in_db"] += 1
                    continue

                if apply:
                    await insert_envelope(conn, env)
                    existing_ids.add(env.id)
                stats["inserted"] += 1
    finally:
        await conn.close()

    balance = stats["already_in_db"] + stats["inserted"] + stats["errors"]
    if balance != stats["fetched"]:
        _log.error("stats_mismatch", fetched=stats["fetched"], balance=balance, **stats)

    _log.info("run_complete", apply=apply, **stats)
    return stats


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Adapt Paperless documents into KB envelopes (module 5, phase 2 — plan §4.2-4.3)."
    )
    parser.add_argument("--dsn", default=os.environ.get("KB_DSN"),
                        help="asyncpg DSN for kb-postgres (or set KB_DSN env var)")
    parser.add_argument("--paperless-url", default=os.environ.get("PAPERLESS_URL", DEFAULT_PAPERLESS_URL),
                        help=f"Paperless base URL (default: {DEFAULT_PAPERLESS_URL}, or set PAPERLESS_URL)")
    parser.add_argument("--paperless-token", default=os.environ.get("PAPERLESS_API_TOKEN"),
                        help="Paperless API token (or set PAPERLESS_API_TOKEN env var)")
    parser.add_argument("--registry", type=Path, default=DEFAULT_REGISTRY,
                        help=f"documents-ingest phase-1 registry JSON (default: {DEFAULT_REGISTRY})")
    parser.add_argument("--limit", type=int, default=None,
                        help="Max documents to process (default: all)")
    parser.add_argument("--page-size", type=int, default=DEFAULT_PAGE_SIZE,
                        help=f"Paperless API page size (default: {DEFAULT_PAGE_SIZE})")
    parser.add_argument("--apply", action="store_true",
                        help="Actually insert envelope rows. Default is dry-run.")
    args = parser.parse_args()

    if not args.dsn:
        _log.error("missing_dsn", hint="pass --dsn or set KB_DSN")
        sys.exit(1)
    if not args.paperless_token:
        _log.error("missing_token", hint="pass --paperless-token or set PAPERLESS_API_TOKEN")
        sys.exit(1)

    stats = asyncio.run(
        run(
            dsn=args.dsn,
            paperless_url=args.paperless_url,
            paperless_token=args.paperless_token,
            registry_path=args.registry,
            limit=args.limit,
            page_size=args.page_size,
            apply=args.apply,
        )
    )

    mode = "APPLY" if args.apply else "DRY-RUN"
    _log.info("summary", mode=mode, **stats)

    balanced = stats["fetched"] == stats["already_in_db"] + stats["inserted"] + stats["errors"]
    failed = stats["errors"] > 0 or not balanced
    sys.exit(1 if failed else 0)


if __name__ == "__main__":
    main()