MEGAINFRA-WEBAPP/scripts/migrations/2026-08-24-data-import-batches-allow-xlsx.py

#!/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-24-data-import-batches-allow-xlsx.py

Extends the `original_file` field on DATA_IMPORT_BATCHES so that
.xlsx uploads 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 original CSV-only import was fine because `text/csv` was already
on the list. The new xlsx-capable import sends
`application/vnd.openxmlformats-officedocument.spreadsheetml.sheet`
(real xlsx MIME), which wasn't. Adding it here lets both formats
share the same `original_file` audit-trail field.

Idempotent: re-running is safe; the MIME list is rebuilt from the
existing set union the new ones, so a re-run is a no-op once applied.
"""
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/vnd.openxmlformats-officedocument.spreadsheetml.sheet",  # .xlsx
]


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"  + {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)