MEGAINFRA-WEBAPP/scripts/migrate-delivery-notes-ocr.py

#!/usr/bin/env python3
"""
One-off migration to add OCR pipeline fields to the delivery_notes collection
and to extend the data_import_batches batch_type select with delivery_notes.

Run once after deploying the OCR webhook code. Idempotent — re-running on a
collection that already has the field is a no-op.

The python bootstrap script (pb-bootstrap-schema.py) only creates collections
that don't exist; it can't add fields to an existing one. This script does.

Required: POCKETBASE_ADMIN_TOKEN (or email + password) in env / .env.local.
"""
import json
import os
import sys
import urllib.error
import urllib.request

PB = os.environ.get("NEXT_PUBLIC_POCKETBASE_URL", "https://pb.mega.infra.apphub.co.za")
EMAIL = os.environ.get("POCKETBASE_ADMIN_EMAIL", "admin@mega.co.za")
PASSWORD = os.environ.get("POCKETBASE_ADMIN_PASSWORD", "12345678")


def http(method, path, body=None, token=None):
    url = f"{PB}{path}"
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method)
    if body is not None:
        req.add_header("Content-Type", "application/json")
    if token:
        req.add_header("Authorization", token)
    try:
        with urllib.request.urlopen(req) as r:
            txt = r.read().decode()
            return r.status, json.loads(txt) if txt else None
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read().decode() or "{}")


def login() -> str:
    token = os.environ.get("POCKETBASE_ADMIN_TOKEN")
    if token:
        # Validate it; if it's expired, fall through to password auth.
        code, _ = http("GET", "/api/collections/_superusers/auth-refresh", token=token)
        if code == 200:
            print("auth ok (existing token)")
            return token
        print("existing token rejected, falling back to email/password…")
    code, auth = http(
        "POST",
        "/api/collections/_superusers/auth-with-password",
        {"identity": EMAIL, "password": PASSWORD},
    )
    if code != 200:
        sys.exit(f"AUTH FAILED: {code} {auth}")
    print(f"auth ok, admin id={auth['record']['id']}")
    return auth["token"]


def get_collection(token: str, name: str):
    code, resp = http("GET", f"/api/collections/{name}", token=token)
    if code != 200:
        sys.exit(f"GET collection {name} failed: {code} {resp}")
    return resp


def patch_collection(token: str, collection_id: str, patch: dict):
    code, resp = http("PATCH", f"/api/collections/{collection_id}", patch, token)
    if code != 200:
        sys.exit(f"PATCH collection {collection_id} failed: {code} {resp}")
    return resp


def field(name, type_, **kw):
    f = {"name": name, "type": type_}
    f.update(kw)
    return f


def ensure_field(existing_fields: list, want: dict) -> bool:
    """Return True if the field was added."""
    name = want["name"]
    if any(f.get("name") == name for f in existing_fields):
        print(f"  • {name:25s} already exists — skipping")
        return False
    return True


def migrate_delivery_notes(token: str):
    print("\n→ delivery_notes")
    coll = get_collection(token, "delivery_notes")
    fields = list(coll.get("fields", []))
    additions = []

    if ensure_field(fields, field("processing_status", "select", required=False,
                                  values=["pending", "ocr-complete", "auto-verified",
                                          "needs-review", "manual-verified", "failed"],
                                  maxSelect=1)):
        additions.append(field("processing_status", "select", required=False,
                                values=["pending", "ocr-complete", "auto-verified",
                                        "needs-review", "manual-verified", "failed"],
                                maxSelect=1))
    if ensure_field(fields, field("ocr_confidence", "number", required=False)):
        additions.append(field("ocr_confidence", "number", required=False))
    if ensure_field(fields, field("ocr_completed_at", "date", required=False)):
        additions.append(field("ocr_completed_at", "date", required=False))
    if ensure_field(fields, field("received_by_name", "text", required=False)):
        additions.append(field("received_by_name", "text", required=False))

    if additions:
        patch_collection(token, coll["id"], {"fields": fields + additions})
        for a in additions:
            print(f"  ✓ added {a['name']}")
    else:
        print("  • no changes needed")


def migrate_data_import_batches(token: str):
    print("\n→ data_import_batches")
    coll = get_collection(token, "data_import_batches")
    fields = list(coll.get("fields", []))
    # Find the batch_type field; extend its `values` to include delivery_notes.
    bt = next((f for f in fields if f.get("name") == "batch_type"), None)
    if not bt:
        sys.exit("batch_type field not found on data_import_batches")
    if "delivery_notes" in (bt.get("values") or []):
        print("  • batch_type already includes delivery_notes — skipping")
        return
    bt["values"] = list(bt.get("values") or []) + ["delivery_notes"]
    patch_collection(token, coll["id"], {"fields": fields})
    print("  ✓ added delivery_notes to batch_type select")


if __name__ == "__main__":
    token = login()
    migrate_delivery_notes(token)
    migrate_data_import_batches(token)
    print("\nDone.")