diff --git a/scripts/migrations/2026-08-31-data-import-batches-allow-pdf-images.py b/scripts/migrations/2026-08-31-data-import-batches-allow-pdf-images.py new file mode 100644 index 0000000..3482e09 --- /dev/null +++ b/scripts/migrations/2026-08-31-data-import-batches-allow-pdf-images.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +""" +One-shot PocketBase migration applied via: + + PB_URL=https://pb.mega.infra.apphub.co.za \\ + PB_ADMIN_EMAIL=admin@mega.co.za \\ + PB_ADMIN_PASSWORD=... \\ + python3 scripts/migrations/2026-08-31-data-import-batches-allow-pdf-images.py + +Extends the `original_file` field on DATA_IMPORT_BATCHES so PDF and +image uploads (used by the PO and delivery-note import flows) are +accepted. PB rejects files whose MIME type isn't in the field's +allowlist with: + + validation_invalid_mime_type: mime type must be one of: ... + +The bootstrap allowlist was scoped to the original CSV/Excel imports +(timesheets, fuel logs, plant hours). The redesigned PO upload route +posts PDFs directly to the same audit field, which crashed the +batch_create call with `batch_create_failed`. This migration unblocks +the queue for PDF/image uploads. + +Also adds application/vnd.openxmlformats-officedocument.spreadsheetml.sheet +(xlsx). The 2026-08-24 migration that should have added it was never +applied to prod (verified in PB: xlsx is missing from the live +mimeTypes); unioning it here keeps the schema at parity with the +fuel-logs/plant-hours import paths. + +Strictly additive: existing mimeTypes are kept and the new ones are +unioned in. Re-running is a no-op (early-return on identical list). + +Made reproducible by bug-report 2026-08-31 (PO upload +batch_create_failed on production). +""" +import json +import os +import sys +import urllib.error +import urllib.request + + +COLLECTION = "data_import_batches" +FIELD_NAME = "original_file" + +# MIME types to add. PB stores these as a flat array on the field's +# `mimeTypes` property (PB v0.39 shape: options live at the field top +# level, not nested under "options"). +NEW_MIME_TYPES = [ + "application/pdf", + "image/jpeg", + "image/png", + "image/webp", + # xlsx - completing the 2026-08-24 migration that was never run on prod. + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", +] + + +def admin_auth(base_url: str) -> str: + email = os.environ["PB_ADMIN_EMAIL"] + password = os.environ["PB_ADMIN_PASSWORD"] + body = json.dumps({"identity": email, "password": password}).encode() + req = urllib.request.Request( + f"{base_url}/api/collections/_superusers/auth-with-password", + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read())["token"] + + +def get_collection(base_url: str, token: str, name: str): + req = urllib.request.Request( + f"{base_url}/api/collections/{name}", + headers={"Authorization": token}, + ) + try: + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read()) + except urllib.error.HTTPError as e: + if e.code == 404: + return None + raise + + +def patch_collection(base_url: str, token: str, name: str, body: dict): + req = urllib.request.Request( + f"{base_url}/api/collections/{name}", + data=json.dumps(body).encode(), + headers={ + "Authorization": token, + "Content-Type": "application/json", + }, + method="PATCH", + ) + try: + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read()) + except urllib.error.HTTPError as e: + body_txt = e.read().decode() if e.fp else "" + raise RuntimeError(f"PATCH {name} failed: {e.code} {body_txt}") from e + + +def main() -> int: + base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/") + token = admin_auth(base_url) + coll = get_collection(base_url, token, COLLECTION) + if not coll: + print(f"ERROR: collection {COLLECTION!r} not found on {base_url}") + return 1 + + existing_fields = coll.get("fields", []) + field = next((f for f in existing_fields if f["name"] == FIELD_NAME), None) + if not field: + print(f"ERROR: field {FIELD_NAME!r} not found on {COLLECTION}") + return 1 + if field.get("type") != "file": + print(f"ERROR: {FIELD_NAME!r} is type {field.get('type')!r}, not 'file'") + return 1 + + current = list(field.get("mimeTypes", []) or []) + wanted = sorted(set(current) | set(NEW_MIME_TYPES)) + + if wanted == current: + print(f"No-op: {FIELD_NAME}.mimeTypes already includes all needed types.") + print(f" current = {current}") + return 0 + + # Build the merged fields array. Everything else passes through + # untouched; only this field gets its mimeTypes updated. + merged_fields = [] + for f in existing_fields: + if f["name"] == FIELD_NAME: + f = dict(f) + f["mimeTypes"] = wanted + merged_fields.append(f) + + print(f"Updating {FIELD_NAME}.mimeTypes:") + print(f" current ({len(current)}): {current}") + print(f" + adding: {sorted(set(wanted) - set(current))}") + patch_collection(base_url, token, COLLECTION, {"fields": merged_fields}) + + # Final sanity dump. + final = get_collection(base_url, token, COLLECTION) + final_field = next( + f for f in final.get("fields", []) if f["name"] == FIELD_NAME + ) + print(f"\nFinal {FIELD_NAME}.mimeTypes = {final_field.get('mimeTypes')}") + print("\nMigration complete.") + return 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except KeyError as e: + print(f"ERROR: missing environment variable {e}. " + "Set PB_ADMIN_EMAIL and PB_ADMIN_PASSWORD.", file=sys.stderr) + sys.exit(2)