diff --git a/scripts/migrations/2026-09-12-migrate-batches-to-documents.py b/scripts/migrations/2026-09-12-migrate-batches-to-documents.py new file mode 100644 index 0000000..bf5980b --- /dev/null +++ b/scripts/migrations/2026-09-12-migrate-batches-to-documents.py @@ -0,0 +1,623 @@ +#!/usr/bin/env python3 +""" +Migrate DATA_IMPORT_BATCHES rows into the new document collections. + + PB_URL=... PB_ADMIN_EMAIL=... PB_ADMIN_PASSWORD=... \\ + python3 scripts/migrations/2026-09-12-migrate-batches-to-documents.py \\ + [--apply-fields | --apply] [--skip-file-check] + +DRY RUN BY DEFAULT. + +For each batch of type timesheets / fuel_logs / plant_hours: + 1. download its original_file from PB + 2. create a document row in the matching collection, re-uploading the + file so the source spreadsheet survives the batch collection's + deletion in Phase 4 + 3. relink every child row from batch_reference to the new relation + +Status is NOT carried across, which is the point of the exercise. +The old rows were all written status='processed' at creation time, before +any data row was inserted - production has 19 plant_hours batches marked +processed against a single plant_hours_log row. So we do not trust the +stored status: we count the actual children and derive it. + + children > 0 -> processed, rows_created = len(children) + children == 0 -> failed, last_error = 'migrated: no rows found' + +rows_total is deliberately left unset. rows_created is a fact we can +count; how many rows the source sheet actually contained is not, and a +sheet that produced nothing certainly was not empty. + +TWO PHASES, TWO FLAGS +--------------------- +Step 1 (the three child relation fields) is additive and reversible. +Step 2 (creating document rows and rewriting relations) is neither. +The plan put both behind one --apply; they are split here so the schema +prerequisite can be applied and verified on its own, without arming the +row migration in the same command. + + --apply-fields add + verify the three relation fields, then stop + --apply run the row migration (requires the fields to exist) + +NOT IDEMPOTENT - see the guard in migrate(). The plan's docstring claimed +a batch already named in a document's `migrated_from_batch` field is +skipped, but no such field exists on any of the three collections and +nothing here writes one. Rather than pretend, --apply refuses to start if +any target collection already holds rows. +""" +import hashlib +import json +import os +import sys +import urllib.error +import urllib.parse +import urllib.request +import uuid + +APPLY = "--apply" in sys.argv +APPLY_FIELDS = "--apply-fields" in sys.argv +SKIP_FILE_CHECK = "--skip-file-check" in sys.argv + +# batch_type -> (document collection, child collection, child relation field) +MAPPING = { + "timesheets": ("timesheets", "timesheet_entries", "timesheet"), + "fuel_logs": ("fuel_sheets", "fuel_logs", "fuel_sheet"), + "plant_hours": ("plant_hours_sheets", "plant_hours_log", "plant_hours_sheet"), +} + +# child collection -> (relation field, target document collection) +# All three are optional and cascadeDelete=False, deliberately: +# - timesheet_entries.timesheet stays optional forever: empty means the +# entry was captured by QR scan, not uploaded on a sheet +# - fuel_logs / plant_hours_log keep pre-migration rows valid +# - a cascade would mean deleting an uploaded sheet destroys the +# payroll / fuel / plant rows it produced +RELATION_FIELDS = [ + ("timesheet_entries", "timesheet", "timesheets"), + ("fuel_logs", "fuel_sheet", "fuel_sheets"), + ("plant_hours_log", "plant_hours_sheet", "plant_hours_sheets"), +] + + +def admin_auth(base_url): + body = json.dumps( + { + "identity": os.environ["PB_ADMIN_EMAIL"], + "password": os.environ["PB_ADMIN_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 api(base_url, token, method, path, body=None): + """ + A 404 is only soft for GET - "this record/collection is absent" is a + legitimate answer to a read. On a write it is not: a POST that 404s + returning None would be read downstream as a created document, and a + PATCH that 404s returning None would be read as a relinked child. + That is how a migration reports success while losing rows. + """ + data = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request( + f"{base_url}{path}", data=data, method=method, + headers={"Authorization": token, "Content-Type": "application/json"}, + ) + try: + with urllib.request.urlopen(req) as resp: + txt = resp.read().decode() + return json.loads(txt) if txt else None + except urllib.error.HTTPError as e: + if e.code == 404 and method == "GET": + return None + raise RuntimeError( + f"{method} {path}: {e.code} {e.read().decode() if e.fp else ''}" + ) from e + + +def get_all(base_url, token, coll, fields, flt=""): + out, page = [], 1 + while True: + path = (f"/api/collections/{coll}/records" + f"?perPage=500&page={page}&fields={urllib.parse.quote(fields)}") + if flt: + path += f"&filter={urllib.parse.quote(flt)}" + data = api(base_url, token, "GET", path) + if not isinstance(data, dict) or "items" not in data: + raise RuntimeError( + f"listing {coll} returned no item list (collection missing?): " + f"{data!r}" + ) + out.extend(data["items"]) + if page >= data.get("totalPages", 0): + return out + page += 1 + + +def count(base_url, token, coll, flt=""): + path = f"/api/collections/{coll}/records?perPage=1&fields=id" + if flt: + path += f"&filter={urllib.parse.quote(flt)}" + data = api(base_url, token, "GET", path) + if not isinstance(data, dict) or "totalItems" not in data: + raise RuntimeError(f"counting {coll} failed: {data!r}") + return data["totalItems"] + + +def download_file(base_url, token, coll, rec_id, filename): + url = (f"{base_url}/api/files/{coll}/{rec_id}/" + f"{urllib.parse.quote(filename)}") + req = urllib.request.Request(url, headers={"Authorization": token}) + try: + with urllib.request.urlopen(req) as resp: + content = resp.read() + except urllib.error.HTTPError as e: + raise RuntimeError( + f"download {coll}/{rec_id}/{filename}: {e.code} " + f"{e.read().decode() if e.fp else ''}" + ) from e + if not content: + raise RuntimeError( + f"download {coll}/{rec_id}/{filename}: served 0 bytes; refusing " + f"to migrate an empty file over the only copy" + ) + return content + + +def sniff(content): + """Coarse family check. Enough to catch a file that is not a sheet.""" + if content[:4] == b"PK\x03\x04": + return "zip/ooxml" + if content[:8] == b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1": + return "ole2/xls" + if content[:4] == b"%PDF": + return "pdf" + if content[:3] in (b"\xff\xd8\xff",) or content[:8].startswith(b"\x89PNG"): + return "image" + try: + content[:4096].decode("utf-8") + return "text" + except UnicodeDecodeError: + return "binary" + + +# Families the document collections' source_file mimeTypes can accept. +# PocketBase sniffs the bytes rather than trusting the declared part +# header, so this is the check that actually predicts acceptance. +OK_FAMILIES = {"zip/ooxml", "ole2/xls", "text"} + + +def _hdr_safe(value, what): + """ + A multipart header value cannot contain a quote or a bare CR/LF. Today + every name here is a literal and every filename is PB-slugified, so + none can - but the thing this body carries is the only remaining copy + of a source spreadsheet, so it is checked rather than assumed. + """ + s = str(value) + if '"' in s or "\r" in s or "\n" in s: + raise RuntimeError(f"unsafe character in {what}: {s!r}") + return s + + +def post_multipart(base_url, token, coll, fields, file_field, + filename, content, content_type): + """PB create with a file attached. Hand-rolled multipart: no deps. + + Ordering is not significant - Go's multipart parser builds a map, so + PB reads the file part wherever it sits. Everything else here is. + """ + if not content: + raise RuntimeError(f"refusing to POST {coll} with an empty file body") + # An empty Content-Type emits a malformed `Content-Type: \r\n` header + # and leans on the parser's goodwill. PB detects the real type from + # the bytes anyway, so a correct placeholder costs nothing. + content_type = _hdr_safe(content_type or "application/octet-stream", + "content_type") + filename = _hdr_safe(filename, "filename") + file_field = _hdr_safe(file_field, "file field name") + + boundary = f"----megainfra{uuid.uuid4().hex}" + # If the delimiter occurred in the payload the body would be silently + # truncated mid-file and PB would store the truncated prefix. uuid4 + # makes that effectively impossible; this makes it actually impossible. + if boundary.encode() in content: + raise RuntimeError("multipart boundary collides with file content") + + parts = [] + for k, v in fields.items(): + if v is None: + continue + k = _hdr_safe(k, "field name") + parts.append( + (f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="{k}"\r\n\r\n').encode() + + str(v).encode() + + b"\r\n" + ) + parts.append( + (f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="{file_field}"; ' + f'filename="{filename}"\r\n' + f"Content-Type: {content_type}\r\n\r\n").encode() + + content + + b"\r\n" + ) + parts.append(f"--{boundary}--\r\n".encode()) + body = b"".join(parts) + + req = urllib.request.Request( + f"{base_url}/api/collections/{coll}/records", + data=body, method="POST", + headers={ + "Authorization": token, + "Content-Type": f"multipart/form-data; boundary={boundary}", + "Content-Length": str(len(body)), + }, + ) + try: + with urllib.request.urlopen(req) as resp: + created = json.loads(resp.read()) + except urllib.error.HTTPError as e: + raise RuntimeError( + f"POST {coll}: {e.code} {e.read().decode() if e.fp else ''}" + ) from e + + # A 200 proves a record exists, not that the file landed on it. That + # distinction is the whole point of this task. + if not created.get("id"): + raise RuntimeError(f"POST {coll}: response carried no id: {created!r}") + if not created.get(file_field): + raise RuntimeError( + f"POST {coll}/{created['id']}: record created but {file_field} is " + f"empty - the source file did not attach. Stopping before the " + f"batch pool is trusted as disposable." + ) + return created + + +# -------------------------------------------------------------------------- +# Step 1: the child relation fields +# -------------------------------------------------------------------------- + +def relation_spec(target_collection_id): + return { + "type": "relation", + "required": False, + "collectionId": target_collection_id, + "maxSelect": 1, + "minSelect": 0, + "cascadeDelete": False, + } + + +def verify_relation(field, target_collection_id, label): + problems = [] + if field.get("type") != "relation": + problems.append(f"{label}.type is {field.get('type')!r}, expected 'relation'") + if field.get("cascadeDelete") is not False: + problems.append( + f"{label}.cascadeDelete is {field.get('cascadeDelete')!r}, must be " + f"False - deleting a sheet would destroy the rows it produced" + ) + if bool(field.get("required")) is not False: + problems.append(f"{label}.required is {field.get('required')!r}, expected False") + if field.get("maxSelect") != 1: + problems.append(f"{label}.maxSelect is {field.get('maxSelect')!r}, expected 1") + if field.get("collectionId") != target_collection_id: + problems.append( + f"{label}.collectionId is {field.get('collectionId')!r}, " + f"expected {target_collection_id!r}" + ) + return problems + + +def ensure_relation_fields(base_url, token, do_apply): + """Returns (all_present_and_correct, added_count).""" + print("\n--- Step 1: child relation fields ---") + ok, added = True, 0 + for child_coll, field_name, doc_coll in RELATION_FIELDS: + label = f"{child_coll}.{field_name}" + target = api(base_url, token, "GET", f"/api/collections/{doc_coll}") + if target is None: + print(f" FAILED: target collection {doc_coll} not found") + return False, added + child = api(base_url, token, "GET", f"/api/collections/{child_coll}") + if child is None: + print(f" FAILED: collection {child_coll} not found") + return False, added + + existing = child.get("fields", []) + found = next((f for f in existing if f["name"] == field_name), None) + if found is not None: + problems = verify_relation(found, target["id"], label) + for p in problems: + print(f" FAILED: {p}") + if problems: + ok = False + else: + print(f" OK {label} -> {doc_coll} ({target['id']}) " + f"required=False maxSelect=1 cascadeDelete=False") + continue + + if not do_apply: + print(f" ADD {label} -> {doc_coll} ({target['id']}) " + f"required=False maxSelect=1 cascadeDelete=False") + ok = False + continue + + print(f" adding {label} -> {doc_coll} ({target['id']})") + api(base_url, token, "PATCH", f"/api/collections/{child_coll}", + {"fields": existing + [{"name": field_name, + **relation_spec(target["id"])}]}) + + # Read back: what landed, not what was asked for. + after = api(base_url, token, "GET", f"/api/collections/{child_coll}") + if after is None: + print(f" FAILED: {child_coll} vanished after PATCH") + return False, added + landed = next((f for f in after.get("fields", []) + if f["name"] == field_name), None) + if landed is None: + print(f" FAILED: {label} absent after PATCH") + return False, added + problems = verify_relation(landed, target["id"], label) + if problems: + for p in problems: + print(f" FAILED: {p}") + return False, added + print(f" OK {label} required=False maxSelect=1 " + f"cascadeDelete=False -> {doc_coll}") + added += 1 + return ok, added + + +# -------------------------------------------------------------------------- +# Step 2: the row migration +# -------------------------------------------------------------------------- + +def plan_migration(base_url, token): + batches = get_all( + base_url, token, "data_import_batches", + "id,batch_type,original_file,content_type,date_uploaded," + "uploaded_by,status,processed_at", + ) + planned = [] + for b in batches: + m = MAPPING.get(b.get("batch_type")) + if not m: + continue # purchase_orders / delivery_notes handled in Phase 3 + doc_coll, child_coll, rel_field = m + n = count(base_url, token, child_coll, + flt=f'batch_reference="{b["id"]}"') + planned.append((b, doc_coll, child_coll, rel_field, n)) + return planned + + +def report(base_url, token, planned): + print(f"\n--- Step 2: batches to migrate: {len(planned)} ---") + by_type = {} + for b, doc_coll, _, _, n in planned: + by_type[b["batch_type"]] = by_type.get(b["batch_type"], 0) + 1 + derived = "processed" if n > 0 else "failed" + print(f" {b['id']} {b['batch_type']:<12} -> {doc_coll:<20} " + f"children={n:<5} stored={b.get('status')} derived={derived} " + f"file={'Y' if b.get('original_file') else 'N'}") + print("\n by type: " + " ".join(f"{k}={v}" for k, v in sorted(by_type.items()))) + + mismatched = [p for p in planned + if p[0].get("status") == "processed" and p[4] == 0] + print(f"\nbatches stored 'processed' with zero children: {len(mismatched)}") + print(" (these become 'failed' - that is the bug this restructure fixes)") + mm_by_type = {} + for p in mismatched: + mm_by_type[p[0]["batch_type"]] = mm_by_type.get(p[0]["batch_type"], 0) + 1 + for k, v in sorted(mm_by_type.items()): + total = by_type.get(k, 0) + print(f" {k:<12} {v} of {total}") + return mismatched + + +def check_files(base_url, token, planned): + """ + Read every source file before anything is written. Task 18 drops the + batch pool; a file that cannot be fetched now can never be fetched. + """ + print("\n--- source file preflight ---") + problems, nofile, total_bytes = [], [], 0 + for b, doc_coll, _, _, _ in planned: + fname = b.get("original_file") + if not fname: + nofile.append(b["id"]) + continue + try: + content = download_file(base_url, token, "data_import_batches", + b["id"], fname) + except RuntimeError as e: + problems.append(str(e)) + continue + total_bytes += len(content) + family = sniff(content) + digest = hashlib.sha256(content).hexdigest()[:12] + flag = "" if family in OK_FAMILIES else " <-- NOT A SPREADSHEET" + print(f" {b['id']} {len(content):>8}B {family:<10} sha256={digest} " + f"{fname}{flag}") + if family not in OK_FAMILIES: + problems.append( + f"{b['id']}: {fname} sniffs as {family}; the document " + f"collections' source_file only accepts csv/xls/xlsx" + ) + print(f"\n {len(planned) - len(nofile)} files readable, " + f"{total_bytes} bytes total") + if nofile: + print(f" {len(nofile)} batches carry no file at all (nothing to " + f"preserve): {', '.join(nofile)}") + for p in problems: + print(f" PROBLEM: {p}") + return problems + + +def migrate(base_url, token, planned, default_site): + if default_site is None: + print("\nABORT: document site cannot be inferred (expected exactly one " + "site). Extend this script to resolve site per batch.") + return 1 + + # The plan claimed this script was idempotent via a `migrated_from_batch` + # field. No such field exists on any of the three collections and nothing + # here writes one, so a second run would create a second set of documents + # and repoint every child at it, orphaning the first set's files. Refuse. + for doc_coll in sorted({p[1] for p in planned}): + n = count(base_url, token, doc_coll) + if n: + print(f"\nABORT: {doc_coll} already holds {n} rows. This migration " + f"is not idempotent - re-running would duplicate every " + f"document and repoint children at the copy. Inspect what is " + f"there and resolve by hand.") + return 1 + + for b, doc_coll, child_coll, rel_field, n_children in planned: + fname = b.get("original_file") + fields = { + "site": default_site, + "content_type": b.get("content_type") or "", + "date_uploaded": b.get("date_uploaded") or "", + "uploaded_by": b.get("uploaded_by") or "web-import", + "processing_status": "processed" if n_children > 0 else "failed", + "processed_at": b.get("processed_at") or "", + "attempts": 0, + "rows_created": n_children, + "rows_failed": 0, + "last_error": "" if n_children > 0 else "migrated: no rows found", + } + + if fname: + content = download_file( + base_url, token, "data_import_batches", b["id"], fname + ) + # Hash here, not later: we already have the bytes in hand, and + # this is the same file_hash pattern Task 1 backfilled onto + # delivery_notes/purchase_orders. Populating it now means the + # three new collections start with working dedup rather than a + # gap that needs a fourth migration to close. + fields["file_hash"] = hashlib.sha256(content).hexdigest() + doc = post_multipart( + base_url, token, doc_coll, fields, "source_file", fname, + content, b.get("content_type") or "application/octet-stream", + ) + else: + doc = api(base_url, token, "POST", + f"/api/collections/{doc_coll}/records", fields) + if not doc or not doc.get("id"): + print(f" FAILED: creating {doc_coll} for {b['id']} returned " + f"no record") + return 1 + + relinked = 0 + for child in get_all(base_url, token, child_coll, "id", + flt=f'batch_reference="{b["id"]}"'): + api(base_url, token, "PATCH", + f"/api/collections/{child_coll}/records/{child['id']}", + {rel_field: doc["id"]}) + relinked += 1 + + if relinked != n_children: + print(f" FAILED: {b['id']} planned {n_children} children but " + f"relinked {relinked}") + return 1 + + print(f" migrated {b['id']} -> {doc_coll}/{doc['id']} " + f"({relinked} children relinked, " + f"file={'Y' if doc.get('source_file') else 'N'})") + + print(f"\nmigrated {len(planned)} batches.") + return verify_after(base_url, token, planned) + + +def verify_after(base_url, token, planned): + print("\n--- verify ---") + ok = True + for doc_coll in sorted({p[1] for p in planned}): + want = sum(1 for p in planned if p[1] == doc_coll) + got = count(base_url, token, doc_coll) + mark = "OK " if got == want else "BAD" + if got != want: + ok = False + print(f" {mark} {doc_coll:<22} {got} rows (expected {want})") + for child_coll, field_name, _ in RELATION_FIELDS: + want = sum(p[4] for p in planned if p[2] == child_coll) + got = count(base_url, token, child_coll, flt=f'{field_name}!=""') + mark = "OK " if got == want else "BAD" + if got != want: + ok = False + print(f" {mark} {child_coll}.{field_name:<20} {got} linked " + f"(expected {want})") + return 0 if ok else 1 + + +def main(): + base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/") + token = admin_auth(base_url) + mode = ("APPLY (rows)" if APPLY else + "APPLY (fields only)" if APPLY_FIELDS else "DRY RUN") + print(f"target: {base_url} mode: {mode}") + + if APPLY and APPLY_FIELDS: + print("ABORT: pass --apply-fields or --apply, not both.") + return 2 + + fields_ok, added = ensure_relation_fields(base_url, token, + do_apply=APPLY_FIELDS) + if APPLY_FIELDS: + if not fields_ok: + print("\nFAILED: relation fields are not as specified.") + return 1 + print(f"\nrelation fields OK ({added} added this run). " + f"Re-run without --apply-fields for the migration dry run.") + return 0 + + sites = get_all(base_url, token, "sites", "id,project_code") + default_site = sites[0]["id"] if len(sites) == 1 else None + print(f"\nsites: {len(sites)} " + f"document site -> {default_site or 'UNRESOLVABLE'}") + + planned = plan_migration(base_url, token) + report(base_url, token, planned) + + if not SKIP_FILE_CHECK: + problems = check_files(base_url, token, planned) + else: + problems = [] + print("\n--- source file preflight SKIPPED (--skip-file-check) ---") + + if not APPLY: + if not fields_ok: + print("\nNOTE: the child relation fields are not all in place. " + "Run with --apply-fields first.") + print("\nDRY RUN. Re-run with --apply to migrate.") + return 0 + + if not fields_ok: + print("\nABORT: child relation fields are not in place; children would " + "have nowhere to point. Run with --apply-fields first.") + return 1 + if problems: + print("\nABORT: source file preflight found problems (above).") + return 1 + + return migrate(base_url, token, planned, default_site) + + +if __name__ == "__main__": + try: + sys.exit(main()) + except KeyError as e: + print(f"ERROR: missing environment variable {e}", file=sys.stderr) + sys.exit(2) + except RuntimeError as e: + print(f"ERROR: {e}", file=sys.stderr) + sys.exit(1)