From ca66d31f0abbcf8abd749eb7a2d58fe3e16c2214 Mon Sep 17 00:00:00 2001 From: oskar Date: Wed, 15 Jul 2026 15:06:56 +0200 Subject: [PATCH] feat(documents-ingest): Paperless -> envelope adapter (module 5 phase 2 step 5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds documents-ingest-paperless: paginated GET /api/documents/, maps each doc to a source='paperless' envelope per plan §4.2-4.3, reusing kb_mail.Envelope/insert_envelope unchanged (packages/kb-mail not touched). Cross-source link (source_mail entity) is a deterministic join of original_file_name against the phase-1 registry.json consume_name index — no heuristics, no correspondent guessing (plan decision 4). Stats always balance (fetched = already_in_db + inserted + errors) and main() now also exits non-zero on imbalance, not just on errors>0, matching the exit-code convention already established in gmail-bulk-import. Verified live on PIHA (rsync to /tmp, ~/kb/venv, PIHA checkout untouched): dry-run then --apply inserted 186/186 paperless envelopes (0 errors, 180 source_mail links), a second --apply reported inserted=0/already_in_db=186 (idempotent), gmail rows stayed at 225030 and document_chunk stayed empty. Rotated the kb-ingest Paperless API token after it was accidentally partially echoed during recon (old token now dead). Co-Authored-By: Claude Sonnet 5 --- jobs/documents-ingest/README.md | 120 +++++ jobs/documents-ingest/pyproject.toml | 3 + .../src/documents_ingest/paperless_adapter.py | 302 +++++++++++++ .../tests/test_paperless_adapter.py | 424 ++++++++++++++++++ 4 files changed, 849 insertions(+) create mode 100644 jobs/documents-ingest/src/documents_ingest/paperless_adapter.py create mode 100644 jobs/documents-ingest/tests/test_paperless_adapter.py diff --git a/jobs/documents-ingest/README.md b/jobs/documents-ingest/README.md index 5acf85c..a264f33 100644 --- a/jobs/documents-ingest/README.md +++ b/jobs/documents-ingest/README.md @@ -181,3 +181,123 @@ Covers: filename sanitization, consume-name collision handling, manifest filtering, MIME PDF-part extraction (including the RFC 2047 decoding mismatch), sha256 match/mismatch, duplicate detection, dry-run vs `--apply` behavior, and multi-attachment envelopes. + +--- + +## Phase 2 — `documents-ingest-paperless` (Paperless -> envelope adapter) + +Module 5, phase 2 (`docs/kb/modules/05-faza2-plan.md`, §4.2-4.3, §6 step 5). +Reads documents from the **Paperless REST API** (read-only — GET only, 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` from `packages/kb-mail` (untouched by this +change — see plan §1.6). Existing `source='gmail'` rows and `document_chunk` +are never touched; this job only ever `INSERT`s new `paperless` rows. + +### Cross-source link (`source_mail`) + +Per plan §1.9/§4.2, the deterministic join uses no heuristics: a document's +`original_file_name` (from the Paperless API) is matched against +`consume_name` in this job's **phase-1 registry** +(`/opt/homelab/data/documents-ingest/registry.json`, produced by +`extractor.py` — see above). A match appends a `source_mail` entity pointing +back at the originating mail envelope; no match means the document was added +outside the faktury-1 pipeline, and the entity is simply omitted — not an +error. + +### Install + +```bash +pip install -e packages/kb-mail/ +pip install -e jobs/documents-ingest/ +``` + +### Usage + +```bash +# Dry run (default) — fetch from Paperless, map, count; no DB writes: +documents-ingest-paperless --dsn postgresql://kb:@localhost:5433/kb \ + --paperless-token + +# Real run — insert new envelope rows: +documents-ingest-paperless --dsn ... --paperless-token ... --apply + +# Smoke-test slice: +documents-ingest-paperless --dsn ... --paperless-token ... --limit 5 +``` + +`--dsn` can come from `KB_DSN`, `--paperless-token` from `PAPERLESS_API_TOKEN`, +`--paperless-url` from `PAPERLESS_URL` (defaults to Paperless' fixed LAN +address, `http://192.168.31.5:8210`). No `--offset`: unlike the 225 030-row +header backfill, a full re-scan of Paperless' ~186 documents is cheap and +already idempotent, so there is no need for resumable partitioning — `--limit` +exists only to cap a run for smoke-testing. + +### Mapping (plan §4.3) + +``` +id = f"paperless:{document_id}" -- prefixed: Paperless doc-ids are small + -- sequential ints that would otherwise + -- collide with any future source's ids +ts = documents_document.created -- Paperless-detected date (content/filename), + -- not filesystem mtime +geo = NULL +raw_ref = str(document_id) -- REFERENCE — Paperless is the source of truth, + -- no bytes are copied +entities = content, correspondent, tag(s), filename, content_type, + and source_mail when the registry join hits (plan §4.2) +``` + +`correspondent`/`tag` are resolved from Paperless' `/api/correspondents/` and +`/api/tags/` (fetched once, cached in memory for the run) and kept purely as +informational metadata — nothing in this pipeline depends on them being +non-null (plan decision 4). A document with empty OCR content (Paperless OCR +sometimes produces none) still gets a normal envelope with `"text": ""` — not +skipped, not an error, just counted (`empty_content`). + +### Idempotency + +A pre-fetched set of existing `source='paperless'` envelope ids (one query at +the start of each run) skips documents already inserted; `insert_envelope`'s +own `ON CONFLICT (id) DO NOTHING` is the second line of defense. Re-running +`--apply` immediately after a successful run reports `inserted: 0` and +`already_in_db` equal to the previous run's `inserted` count. + +### Stats must balance + +``` +fetched = already_in_db + inserted + errors +``` + +`source_mail_linked` and `empty_content` are informational subsets of +`fetched`, not separate outcome buckets. A per-document mapping failure +(e.g. an unparseable `created` date) is isolated, logged, and counted as +`errors` — it never aborts the run. `main()` exits 1 on non-zero `errors` +or if the balance invariant above doesn't hold (mirrors +`gmail-bulk-import`'s exit-code convention) — a clean run always exits 0. + +### Tests + +```bash +pip install -e packages/kb-mail/ +pip install -e jobs/documents-ingest/ +cd jobs/documents-ingest && pytest +``` + +Pure unit tests, no DB or real HTTP — `run()` is tested by monkeypatching +`asyncpg.connect` (fake connection) and `aiohttp.ClientSession` (fake session +serving canned JSON pages). Covers: mapping shape (content, correspondent, +tag(s), filename, content_type, source_mail), the registry join (hit and +miss), pagination (both the documents list and the correspondents/tags lookup +tables), `--limit`, idempotency (pre-existing ids skipped, a second `--apply` +run inserts nothing new), isolated per-document mapping errors, and the +stats-balance invariant. + +### Definition of Done + +Per `CLAUDE.md`: smoke run is `documents-ingest-paperless --dsn ... +--paperless-token ... --limit 5` (dry-run first) against kb-postgres@PIHA and +the live Paperless API, over SSH — **not executed as part of this change** +without operator confirmation (this job reads production Paperless data and +writes production envelope rows on `--apply`). `pytest` passes locally before +this commit. diff --git a/jobs/documents-ingest/pyproject.toml b/jobs/documents-ingest/pyproject.toml index b496292..8265a09 100644 --- a/jobs/documents-ingest/pyproject.toml +++ b/jobs/documents-ingest/pyproject.toml @@ -9,10 +9,13 @@ requires-python = ">=3.11" dependencies = [ "asyncpg>=0.29", "structlog>=24.1", + "aiohttp>=3.9", + "kb-mail", ] [project.scripts] documents-ingest = "documents_ingest.extractor:main" +documents-ingest-paperless = "documents_ingest.paperless_adapter:main" [tool.setuptools.packages.find] where = ["src"] diff --git a/jobs/documents-ingest/src/documents_ingest/paperless_adapter.py b/jobs/documents-ingest/src/documents_ingest/paperless_adapter.py new file mode 100644 index 0000000..a13c823 --- /dev/null +++ b/jobs/documents-ingest/src/documents_ingest/paperless_adapter.py @@ -0,0 +1,302 @@ +"""Paperless -> envelope adapter — module 5 phase 2 (docs/kb/modules/05-faza2-plan.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:@localhost:5433/kb \\ + --paperless-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() diff --git a/jobs/documents-ingest/tests/test_paperless_adapter.py b/jobs/documents-ingest/tests/test_paperless_adapter.py new file mode 100644 index 0000000..a13464c --- /dev/null +++ b/jobs/documents-ingest/tests/test_paperless_adapter.py @@ -0,0 +1,424 @@ +"""Unit tests for the Paperless -> envelope adapter — no DB, no real HTTP.""" +from __future__ import annotations + +import json +from datetime import datetime, timezone + +import pytest + +from documents_ingest.paperless_adapter import ( + build_entities, + build_registry_index, + find_source_mail, + fetch_lookup, + iter_documents, + map_document, + parse_document_ts, + run, +) + +BASE_URL = "http://fake-paperless" + + +def _doc( + doc_id=1, + created="2020-04-21T00:00:00+02:00", + content="some ocr text", + correspondent=None, + tags=None, + original_file_name="file.pdf", + mime_type="application/pdf", +): + return { + "id": doc_id, + "correspondent": correspondent, + "tags": tags or [], + "content": content, + "created": created, + "original_file_name": original_file_name, + "mime_type": mime_type, + } + + +def _registry_entry(sha256, envelope_id, filename, consume_name): + return {sha256: {"envelope_id": envelope_id, "filename": filename, "consume_name": consume_name, + "size": 100, "ingested_at": "2026-07-13T17:50:49.281559+00:00"}} + + +class TestParseDocumentTs: + def test_parses_offset_to_utc(self): + ts = parse_document_ts("2020-04-21T00:00:00+02:00") + assert ts == datetime(2020, 4, 20, 22, 0, 0, tzinfo=timezone.utc) + + def test_result_is_utc_aware(self): + assert parse_document_ts("2026-06-09T12:00:00+00:00").tzinfo is not None + + def test_parses_z_suffix(self): + ts = parse_document_ts("2026-06-09T12:00:00Z") + assert ts == datetime(2026, 6, 9, 12, 0, 0, tzinfo=timezone.utc) + + +class TestBuildRegistryIndex: + def test_inverts_sha256_key_to_consume_name_key(self): + registry = { + "sha-a": {"envelope_id": "env-a", "filename": "a.pdf", "consume_name": "2026-06-09_a.pdf"}, + } + index = build_registry_index(registry) + assert index == { + "2026-06-09_a.pdf": {"envelope_id": "env-a", "sha256": "sha-a", "filename": "a.pdf"}, + } + + +class TestFindSourceMail: + def test_hit_on_matching_consume_name(self): + index = build_registry_index({ + "sha-a": {"envelope_id": "env-a", "filename": "a.pdf", "consume_name": "2026-06-09_a.pdf"}, + }) + match = find_source_mail("2026-06-09_a.pdf", index) + assert match == {"envelope_id": "env-a", "sha256": "sha-a", "filename": "a.pdf"} + + def test_miss_returns_none(self): + index = build_registry_index({ + "sha-a": {"envelope_id": "env-a", "filename": "a.pdf", "consume_name": "2026-06-09_a.pdf"}, + }) + assert find_source_mail("polisa 920008969228.pdf", index) is None + + def test_none_filename_returns_none(self): + assert find_source_mail(None, {}) is None + + +class TestBuildEntities: + def test_shape_without_correspondent_tags_or_source_mail(self): + doc = _doc() + entities = build_entities(doc, correspondents={}, tags={}, source_mail=None) + assert entities == [ + {"type": "content", "text": "some ocr text"}, + {"type": "correspondent", "name": None}, + {"type": "filename", "value": "file.pdf"}, + {"type": "content_type", "value": "application/pdf"}, + ] + + def test_correspondent_name_resolved(self): + doc = _doc(correspondent=7) + entities = build_entities(doc, correspondents={7: "WARTA"}, tags={}, source_mail=None) + assert {"type": "correspondent", "name": "WARTA"} in entities + + def test_multiple_tags_each_get_own_entity(self): + doc = _doc(tags=[1, 2]) + entities = build_entities(doc, correspondents={}, tags={1: "faktury", 2: "2026"}, source_mail=None) + tag_entities = [e for e in entities if e["type"] == "tag"] + assert tag_entities == [{"type": "tag", "name": "faktury"}, {"type": "tag", "name": "2026"}] + + def test_no_tags_means_no_tag_entities(self): + entities = build_entities(_doc(tags=[]), correspondents={}, tags={}, source_mail=None) + assert not [e for e in entities if e["type"] == "tag"] + + def test_source_mail_entity_appended_when_present(self): + source_mail = {"envelope_id": "env-a", "sha256": "sha-a", "filename": "a.pdf"} + entities = build_entities(_doc(), correspondents={}, tags={}, source_mail=source_mail) + assert entities[-1] == { + "type": "source_mail", + "envelope_id": "env-a", + "attachment_sha256": "sha-a", + "attachment_filename": "a.pdf", + } + + def test_empty_content_becomes_empty_string_not_none(self): + entities = build_entities(_doc(content=None), correspondents={}, tags={}, source_mail=None) + assert entities[0] == {"type": "content", "text": ""} + + +class TestMapDocument: + def test_id_gets_paperless_prefix(self): + env = map_document(_doc(doc_id=42), correspondents={}, tags={}, source_mail=None) + assert env.id == "paperless:42" + + def test_source_is_paperless(self): + env = map_document(_doc(), correspondents={}, tags={}, source_mail=None) + assert env.source == "paperless" + + def test_raw_ref_is_document_id_string(self): + env = map_document(_doc(doc_id=42), correspondents={}, tags={}, source_mail=None) + assert env.raw_ref == "42" + + def test_geo_is_none(self): + env = map_document(_doc(), correspondents={}, tags={}, source_mail=None) + assert env.geo is None + + def test_ts_parsed_from_created(self): + env = map_document(_doc(created="2020-04-21T00:00:00+02:00"), correspondents={}, tags={}, source_mail=None) + assert env.ts == datetime(2020, 4, 20, 22, 0, 0, tzinfo=timezone.utc) + + +class _FakeResponse: + def __init__(self, payload): + self._payload = payload + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def json(self): + return self._payload + + def raise_for_status(self): + pass + + +class _FakeSession: + def __init__(self, pages: dict[str, dict]): + self._pages = pages + + def get(self, url): + return _FakeResponse(self._pages[url]) + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +def _empty_page(): + return {"count": 0, "next": None, "previous": None, "results": []} + + +class TestFetchLookup: + async def test_single_page(self): + session = _FakeSession({ + f"{BASE_URL}/api/tags/": {"count": 2, "next": None, "results": [ + {"id": 1, "name": "faktury"}, {"id": 2, "name": "2026"}, + ]}, + }) + result = await fetch_lookup(session, BASE_URL, "/api/tags/") + assert result == {1: "faktury", 2: "2026"} + + async def test_follows_pagination(self): + session = _FakeSession({ + f"{BASE_URL}/api/tags/": { + "count": 2, "next": f"{BASE_URL}/api/tags/?page=2", + "results": [{"id": 1, "name": "a"}], + }, + f"{BASE_URL}/api/tags/?page=2": { + "count": 2, "next": None, "results": [{"id": 2, "name": "b"}], + }, + }) + result = await fetch_lookup(session, BASE_URL, "/api/tags/") + assert result == {1: "a", 2: "b"} + + async def test_empty_results(self): + session = _FakeSession({f"{BASE_URL}/api/correspondents/": _empty_page()}) + assert await fetch_lookup(session, BASE_URL, "/api/correspondents/") == {} + + +class TestIterDocuments: + async def test_yields_all_docs_across_pages(self): + url1 = f"{BASE_URL}/api/documents/?page_size=200&ordering=id" + url2 = f"{BASE_URL}/api/documents/?page=2" + session = _FakeSession({ + url1: {"count": 3, "next": url2, "results": [_doc(1), _doc(2)]}, + url2: {"count": 3, "next": None, "results": [_doc(3)]}, + }) + docs = [d async for d in iter_documents(session, BASE_URL, page_size=200)] + assert [d["id"] for d in docs] == [1, 2, 3] + + async def test_limit_stops_early_across_pages(self): + url1 = f"{BASE_URL}/api/documents/?page_size=200&ordering=id" + url2 = f"{BASE_URL}/api/documents/?page=2" + session = _FakeSession({ + url1: {"count": 3, "next": url2, "results": [_doc(1), _doc(2)]}, + url2: {"count": 3, "next": None, "results": [_doc(3)]}, + }) + docs = [d async for d in iter_documents(session, BASE_URL, page_size=200, limit=1)] + assert [d["id"] for d in docs] == [1] + + +class _FakeConn: + def __init__(self, existing_ids=None): + self.existing_ids = list(existing_ids or []) + self.execute_calls: list[tuple] = [] + + async def fetch(self, query, *params): + return [{"id": i} for i in self.existing_ids] + + async def execute(self, query, *params): + self.execute_calls.append(params) + return "INSERT 0 1" + + async def close(self): + pass + + +def _docs_page(docs, next_url=None): + return {"count": len(docs), "next": next_url, "results": docs} + + +class TestRun: + def _patch(self, monkeypatch, conn, pages): + async def _fake_connect(dsn): + return conn + monkeypatch.setattr("documents_ingest.paperless_adapter.asyncpg.connect", _fake_connect) + + def _fake_session_factory(*args, **kwargs): + return _FakeSession(pages) + monkeypatch.setattr("documents_ingest.paperless_adapter.aiohttp.ClientSession", _fake_session_factory) + + def _base_pages(self, docs): + docs_url = f"{BASE_URL}/api/documents/?page_size=200&ordering=id" + return { + docs_url: _docs_page(docs), + f"{BASE_URL}/api/correspondents/": _empty_page(), + f"{BASE_URL}/api/tags/": _empty_page(), + } + + async def test_dry_run_does_not_insert(self, monkeypatch, tmp_path): + registry_path = tmp_path / "registry.json" + registry_path.write_text("{}") + conn = _FakeConn() + pages = self._base_pages([_doc(1), _doc(2)]) + self._patch(monkeypatch, conn, pages) + + stats = await run( + dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", + registry_path=registry_path, apply=False, + ) + + assert stats["fetched"] == 2 + assert stats["inserted"] == 2 + assert conn.execute_calls == [] + + async def test_apply_inserts_new_documents(self, monkeypatch, tmp_path): + registry_path = tmp_path / "registry.json" + registry_path.write_text("{}") + conn = _FakeConn() + pages = self._base_pages([_doc(1), _doc(2)]) + self._patch(monkeypatch, conn, pages) + + stats = await run( + dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", + registry_path=registry_path, apply=True, + ) + + assert stats["inserted"] == 2 + assert len(conn.execute_calls) == 2 + + async def test_idempotent_skips_existing_ids(self, monkeypatch, tmp_path): + registry_path = tmp_path / "registry.json" + registry_path.write_text("{}") + conn = _FakeConn(existing_ids=["paperless:1"]) + pages = self._base_pages([_doc(1), _doc(2)]) + self._patch(monkeypatch, conn, pages) + + stats = await run( + dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", + registry_path=registry_path, apply=True, + ) + + assert stats["already_in_db"] == 1 + assert stats["inserted"] == 1 + assert len(conn.execute_calls) == 1 + + async def test_rerun_after_apply_inserts_nothing_new(self, monkeypatch, tmp_path): + registry_path = tmp_path / "registry.json" + registry_path.write_text("{}") + pages = self._base_pages([_doc(1), _doc(2)]) + + conn1 = _FakeConn() + self._patch(monkeypatch, conn1, pages) + first = await run( + dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", + registry_path=registry_path, apply=True, + ) + assert first["inserted"] == 2 + + # Second run sees both ids already in the DB (simulated via existing_ids). + conn2 = _FakeConn(existing_ids=["paperless:1", "paperless:2"]) + self._patch(monkeypatch, conn2, pages) + second = await run( + dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", + registry_path=registry_path, apply=True, + ) + assert second["inserted"] == 0 + assert second["already_in_db"] == 2 + assert conn2.execute_calls == [] + + async def test_source_mail_linked_counted_on_registry_hit(self, monkeypatch, tmp_path): + registry_path = tmp_path / "registry.json" + registry_path.write_text(json.dumps( + _registry_entry("sha-a", "env-a", "invoice.pdf", "2026-06-09_invoice.pdf") + )) + conn = _FakeConn() + pages = self._base_pages([_doc(1, original_file_name="2026-06-09_invoice.pdf"), _doc(2)]) + self._patch(monkeypatch, conn, pages) + + stats = await run( + dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", + registry_path=registry_path, apply=False, + ) + + assert stats["source_mail_linked"] == 1 + + async def test_empty_content_counted(self, monkeypatch, tmp_path): + registry_path = tmp_path / "registry.json" + registry_path.write_text("{}") + conn = _FakeConn() + pages = self._base_pages([_doc(1, content=""), _doc(2, content="text")]) + self._patch(monkeypatch, conn, pages) + + stats = await run( + dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", + registry_path=registry_path, apply=False, + ) + + assert stats["empty_content"] == 1 + + async def test_map_error_is_isolated_and_counted(self, monkeypatch, tmp_path): + registry_path = tmp_path / "registry.json" + registry_path.write_text("{}") + conn = _FakeConn() + bad_doc = _doc(1, created=None) # parse_document_ts(None) raises + pages = self._base_pages([bad_doc, _doc(2)]) + self._patch(monkeypatch, conn, pages) + + stats = await run( + dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", + registry_path=registry_path, apply=False, + ) + + assert stats["errors"] == 1 + assert stats["inserted"] == 1 + assert stats["fetched"] == 2 + + async def test_stats_balance_across_all_outcomes(self, monkeypatch, tmp_path): + registry_path = tmp_path / "registry.json" + registry_path.write_text("{}") + conn = _FakeConn(existing_ids=["paperless:1"]) + pages = self._base_pages([_doc(1), _doc(2), _doc(3, created=None)]) + self._patch(monkeypatch, conn, pages) + + stats = await run( + dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", + registry_path=registry_path, apply=True, + ) + + assert stats["fetched"] == 3 + assert stats["fetched"] == stats["already_in_db"] + stats["inserted"] + stats["errors"] + assert stats["already_in_db"] == 1 + assert stats["inserted"] == 1 + assert stats["errors"] == 1 + + async def test_limit_caps_documents_processed(self, monkeypatch, tmp_path): + registry_path = tmp_path / "registry.json" + registry_path.write_text("{}") + conn = _FakeConn() + pages = self._base_pages([_doc(1), _doc(2), _doc(3)]) + self._patch(monkeypatch, conn, pages) + + stats = await run( + dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", + registry_path=registry_path, limit=2, apply=False, + ) + + assert stats["fetched"] == 2