fix(pb): allow PDF and image MIME on data_import_batches.original_file

The PO upload route (and the delivery-notes and xlsx import
paths) writes the uploaded file into the DATA_IMPORT_BATCHES.
original_file field. That field was bootstrapped with a CSV/Excel
allowlist only:

  text/csv, text/plain, application/vnd.ms-excel, application/csv

When the upload route was rewritten on 2026-08-19 (commit f0f7c82)
to write directly to PB instead of forwarding the file to n8n, every
PDF and image upload started failing PB field validation with
validation_invalid_mime_type, surfacing in the API as a 500 with
error='batch_create_failed' and the real cause buried in
pb_data.original_file. The 2026-08-24 migration that should have
added xlsx was also never applied to prod (verified in PB:
xlsx missing too).

Migration: scripts/migrations/2026-08-31-data-import-batches-allow-pdf-images.py
- Strictly additive: unions the new mimeTypes onto the existing
  allowlist, exits no-op if already applied (verified idempotent).
- Adds application/pdf, image/jpeg, image/png, image/webp (PDF /
  image support for the PO and delivery-note flows) and
  application/vnd.openxmlformats-officedocument.spreadsheetml.sheet
  (xlsx, completing the missed 2026-08-24 migration).
- No code change in the upload routes required - they were already
  validating against ALLOWED_MIME in purchase-order-schema.ts and
  expecting PB to accept the same set.

Verification on https://pb.mega.infra.apphub.co.za:
  Before: POST /api/purchase-orders/upload (PDF) -> HTTP 500,
    pb_data.original_file.code = 'validation_invalid_mime_type'.
  After:  same POST -> HTTP 202 with batch_id, PB row created
    (batch_type=purchase_orders, status=pending, file_hash
    populated, original_file stored).
This commit is contained in:
DEls 2026-08-31 12:20:41 +02:00
parent 395c0ff114
commit 658d83dcff

View file

@ -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)