diff --git a/scripts/pb-add-dn-staged-ingestion.py b/scripts/pb-add-dn-staged-ingestion.py new file mode 100644 index 0000000..94da319 --- /dev/null +++ b/scripts/pb-add-dn-staged-ingestion.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +"""Add the staged delivery-note ingestion schema. + +The migration is idempotent and is a dry run unless ``--apply`` is supplied. +It reads existing PocketBase fields and indexes before appending anything, so +running it against a live instance preserves unrelated schema changes. +""" +import sys + +from pb_env import PB, http, superuser_token + + +DRY_RUN = "--dry-run" in sys.argv or "--apply" not in sys.argv + + +def get_collection(token, name): + code, response = http("GET", f"/api/collections/{name}", token=token) + return response if code == 200 else None + + +def require_collection(token, name): + collection = get_collection(token, name) + if collection is None: + sys.exit(f"collection {name} not found") + return collection + + +def relation(name, collection_id, required=False): + return { + "name": name, + "type": "relation", + "required": required, + "collectionId": collection_id, + "maxSelect": 1, + "cascadeDelete": False, + } + + +def field(name, field_type="text", **options): + return {"name": name, "type": field_type, "required": False, **options} + + +def add_fields(token, name, wanted): + collection = require_collection(token, name) + existing = list(collection.get("fields", [])) + names = {item.get("name") for item in existing} + additions = [item for item in wanted if item["name"] not in names] + if not additions: + print(f" {name}: fields already present") + return + print(f" {name}: add fields {[item['name'] for item in additions]}") + if not DRY_RUN: + code, response = http( + "PATCH", f"/api/collections/{collection['id']}", + {"fields": existing + additions}, token, + ) + if code != 200: + sys.exit(f"PATCH {name} failed: {code} {response}") + + +def add_indexes(token, name, wanted): + collection = require_collection(token, name) + existing = list(collection.get("indexes", []) or []) + additions = [item for item in wanted if item not in existing] + if not additions: + print(f" {name}: indexes already present") + return + print(f" {name}: add indexes {additions}") + if not DRY_RUN: + code, response = http( + "PATCH", f"/api/collections/{collection['id']}", + {"indexes": existing + additions}, token, + ) + if code != 200: + sys.exit(f"PATCH {name} indexes failed: {code} {response}") + + +def create_collection(token, name, fields, indexes): + existing = get_collection(token, name) + if existing is not None: + print(f" {name}: already exists") + add_fields(token, name, fields) + add_indexes(token, name, indexes) + return existing["id"] + print(f" {name}: create collection") + if DRY_RUN: + return name + code, response = http( + "POST", "/api/collections", + {"name": name, "type": "base", "fields": fields, "indexes": indexes}, + token, + ) + if code != 200: + sys.exit(f"CREATE {name} failed: {code} {response}") + return response["id"] + + +def system_fields(): + return [ + field("created", "date", onCreate=True), + field("updated", "date", onCreate=True, onUpdate=True), + ] + + +def main(): + try: + token = superuser_token() + except SystemExit as error: + # A dry run is useful in CI and during review even when no PB + # credentials are mounted. It still reports the complete intended + # mutation and, by definition, performs no writes. + if not DRY_RUN: + raise + print(f"offline dry run: {error}") + print("would create: dn_batch_sources, dn_batch_pages, " + "supplier_document_aliases, dn_review_events, dn_pipeline_locks") + print("would extend: dn_batches, dn_batch_notes, delivery_notes") + print("would add unique indexes: source(batch,file_hash), " + "page(source,source_page,region), alias(normalised_key), " + "lock(batch), delivery_notes(batch_note)") + print("would add partial unique capture-session index on dn_batches") + print("\nDRY RUN. No schema changes written.") + return 0 + print(f"target: {PB} mode: {'DRY RUN' if DRY_RUN else 'APPLY'}") + + batches = require_collection(token, "dn_batches") + notes = require_collection(token, "dn_batch_notes") + suppliers = require_collection(token, "suppliers") + delivery_notes = require_collection(token, "delivery_notes") + batch_id = batches["id"] + notes_id = notes["id"] + supplier_id = suppliers["id"] + delivery_notes_id = delivery_notes["id"] + + source_id = create_collection( + token, "dn_batch_sources", + system_fields() + [ + relation("batch", batch_id, required=True), + field("file", "file", maxSelect=1, maxSize=20971520), + field("file_hash"), field("ordinal", "number"), field("mime"), + field("original_name"), field("capture_session_id"), + field("delivery_session_id"), field("captured_at", "date"), + field("latitude", "number"), field("longitude", "number"), + field("location_accuracy_m", "number"), + ], + ["CREATE UNIQUE INDEX `idx_dn_batch_sources_batch_file_hash` " + "ON `dn_batch_sources` (`batch`, `file_hash`)"] + ) + create_collection( + token, "dn_batch_pages", + system_fields() + [ + relation("batch", batch_id, required=True), + relation("source", source_id, required=True), + field("source_page", "number"), field("region", "number"), + field("image", "file", maxSelect=1, maxSize=20971520), + field("image_hash"), field("mime"), field("width", "number"), + field("height", "number"), field("rotation", "number"), + field("read_status"), field("extraction", "json", maxSize=2000000), + field("schema_version"), field("model"), field("prompt_version"), + field("read_error"), field("manual_group"), + ], + ["CREATE UNIQUE INDEX `idx_dn_batch_pages_source_page_region` " + "ON `dn_batch_pages` (`source`, `source_page`, `region`)"] + ) + create_collection( + token, "supplier_document_aliases", + system_fields() + [ + relation("supplier", supplier_id), field("raw_value"), + field("normalised_key"), field("source"), field("approved_by"), + ], + ["CREATE UNIQUE INDEX `idx_supplier_document_aliases_normalised_key` " + "ON `supplier_document_aliases` (`normalised_key`)"] + ) + create_collection( + token, "dn_review_events", + system_fields() + [ + relation("batch", batch_id, required=True), relation("note", notes_id), + field("action"), field("before", "json", maxSize=2000000), + field("after", "json", maxSize=2000000), field("reason"), field("actor"), + ], [], + ) + create_collection( + token, "dn_pipeline_locks", + system_fields() + [relation("batch", batch_id, required=True), + field("expires_at", "date"), field("worker")], + ["CREATE UNIQUE INDEX `idx_dn_pipeline_locks_batch` " + "ON `dn_pipeline_locks` (`batch`)"] + ) + + add_fields(token, "dn_batches", [ + field("stage", "select", maxSelect=1, + values=["queued", "preparing", "reading", "split_review", "complete", "failed"]), + field("source_count", "number"), field("processing_completed_at", "date"), + field("capture_session_id"), + ]) + add_fields(token, "dn_batch_notes", [ + field("stage", "select", maxSelect=1, + values=["matching", "needs_review", "ready_to_post", "posted", "dismissed"]), + field("assembly_revision", "number"), field("pages", "json", maxSize=2000000), + relation("supplier", supplier_id), field("supplier_name_raw"), + field("match_method"), field("match_evidence", "json", maxSize=2000000), + field("posted_at", "date"), + ]) + add_fields(token, "delivery_notes", [relation("batch_note", notes_id)]) + + add_indexes(token, "delivery_notes", [ + "CREATE UNIQUE INDEX `idx_delivery_notes_batch_note` " + "ON `delivery_notes` (`batch_note`)", + ]) + add_indexes(token, "dn_batches", [ + "CREATE UNIQUE INDEX `idx_dn_batches_capture_session_id` " + "ON `dn_batches` (`capture_session_id`) WHERE `capture_session_id` != ''", + ]) + + print("\nDRY RUN. No schema changes written." if DRY_RUN else "\nSchema applied.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/lib/dn-pipeline/stages.test.ts b/src/lib/dn-pipeline/stages.test.ts new file mode 100644 index 0000000..d9203ec --- /dev/null +++ b/src/lib/dn-pipeline/stages.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from "vitest"; +import { + displayBatchStage, + displayCandidateStage, + isCandidateTerminal, +} from "./stages"; + +describe("stage compatibility", () => { + it("maps legacy states without rewriting live rows", () => { + expect(displayBatchStage("analysing", "")).toBe("reading"); + expect(displayCandidateStage("blocked", "")).toBe("needs_review"); + expect(displayCandidateStage("promoted", "")).toBe("posted"); + }); + + it("prefers an explicit stage", () => { + expect(displayBatchStage("needs_attention", "split_review")).toBe("split_review"); + expect(displayCandidateStage("blocked", "matching")).toBe("matching"); + }); + + it("recognises terminal candidates", () => { + expect(isCandidateTerminal("posted")).toBe(true); + expect(isCandidateTerminal("dismissed")).toBe(true); + expect(isCandidateTerminal("needs_review")).toBe(false); + }); +}); diff --git a/src/lib/dn-pipeline/stages.ts b/src/lib/dn-pipeline/stages.ts new file mode 100644 index 0000000..bc689d0 --- /dev/null +++ b/src/lib/dn-pipeline/stages.ts @@ -0,0 +1,54 @@ +export type BatchStage = + | "queued" + | "preparing" + | "reading" + | "split_review" + | "complete" + | "failed"; + +export type CandidateStage = + | "matching" + | "needs_review" + | "ready_to_post" + | "posted" + | "dismissed"; + +const BATCH_STAGES: readonly BatchStage[] = [ + "queued", + "preparing", + "reading", + "split_review", + "complete", + "failed", +]; + +const CANDIDATE_STAGES: readonly CandidateStage[] = [ + "matching", + "needs_review", + "ready_to_post", + "posted", + "dismissed", +]; + +export function displayBatchStage(legacy: string, explicit: string): BatchStage { + if ((BATCH_STAGES as readonly string[]).includes(explicit)) { + return explicit as BatchStage; + } + if (legacy === "analysing") return "reading"; + if (legacy === "processed") return "complete"; + if (legacy === "needs_attention") return "split_review"; + return "queued"; +} + +export function displayCandidateStage(legacy: string, explicit: string): CandidateStage { + if ((CANDIDATE_STAGES as readonly string[]).includes(explicit)) { + return explicit as CandidateStage; + } + if (legacy === "promoted") return "posted"; + if (legacy === "dismissed") return "dismissed"; + if (legacy === "blocked") return "needs_review"; + return "ready_to_post"; +} + +export const isCandidateTerminal = (stage: CandidateStage) => + stage === "posted" || stage === "dismissed"; diff --git a/src/lib/pb-types.ts b/src/lib/pb-types.ts index 6e54887..e538eec 100644 --- a/src/lib/pb-types.ts +++ b/src/lib/pb-types.ts @@ -8,6 +8,7 @@ * RATE_PER_LITRE fallback when it is absent. */ import type { RecordModel } from "./pb"; +import type { BatchStage, CandidateStage } from "./dn-pipeline/stages"; export interface SiteRecord extends RecordModel { collectionName: "sites"; @@ -302,7 +303,12 @@ export interface DeliveryNoteRecord extends RecordModel { ocr_completed_at?: string; received_by_name?: string; document_type?: "delivery_note" | "invoice" | "collection_note" | "unknown"; - expand?: { site?: SiteRecord; purchase_order?: PurchaseOrderRecord }; + batch_note?: string; + expand?: { + site?: SiteRecord; + purchase_order?: PurchaseOrderRecord; + batch_note?: DnBatchNoteRecord; + }; } export type DeliveryNoteProcessingStatus = @@ -385,6 +391,10 @@ export interface DnBatchRecord extends RecordModel { /** Set explicitly at upload — PocketBase's `created` is unusable here. */ uploaded_at: string; open_pages?: unknown; + stage?: BatchStage; + source_count?: number; + processing_completed_at?: string; + capture_session_id?: string; expand?: { site?: SiteRecord }; } @@ -436,6 +446,84 @@ export interface DnBatchNoteRecord extends RecordModel { status: DnBatchNoteStatus; /** Back-reference to the DELIVERY_NOTES row this note was promoted into. Empty while a candidate or blocked. */ delivery_note: string; + stage?: CandidateStage; + assembly_revision?: number; + pages?: unknown; + supplier?: string; + supplier_name_raw?: string; + match_method?: string; + match_evidence?: unknown; + posted_at?: string; + expand?: { batch?: DnBatchRecord; supplier?: SupplierRecord; delivery_note?: DeliveryNoteRecord }; +} + +export interface DnBatchSourceRecord extends RecordModel { + collectionName: "dn_batch_sources"; + batch: string; + file: string; + file_hash: string; + ordinal: number; + mime: string; + original_name: string; + capture_session_id?: string; + delivery_session_id?: string; + captured_at?: string; + latitude?: number; + longitude?: number; + location_accuracy_m?: number; + expand?: { batch?: DnBatchRecord }; +} + +export interface DnBatchPageRecord extends RecordModel { + collectionName: "dn_batch_pages"; + batch: string; + source: string; + source_page: number; + region: number; + image: string; + image_hash: string; + mime: string; + width: number; + height: number; + rotation: number; + read_status: string; + extraction: unknown; + schema_version: string; + model: string; + prompt_version: string; + read_error: string; + manual_group: string; + expand?: { batch?: DnBatchRecord; source?: DnBatchSourceRecord }; +} + +export interface SupplierDocumentAliasRecord extends RecordModel { + collectionName: "supplier_document_aliases"; + supplier: string; + raw_value: string; + normalised_key: string; + source: string; + approved_by: string; + expand?: { supplier?: SupplierRecord }; +} + +export interface DnReviewEventRecord extends RecordModel { + collectionName: "dn_review_events"; + batch: string; + note: string; + action: string; + before: unknown; + after: unknown; + reason: string; + actor: string; + expand?: { batch?: DnBatchRecord; note?: DnBatchNoteRecord }; +} + +export interface DnPipelineLockRecord extends RecordModel { + collectionName: "dn_pipeline_locks"; + batch: string; + expires_at: string; + worker: string; + expand?: { batch?: DnBatchRecord }; } export type DocumentStatus =