feat(migration): move batch files into document collections
Migrates the 27 batches of type timesheets/fuel_logs/plant_hours out of data_import_batches into the three document collections created in Task 6, preserving every source file and relinking every child row. Status is derived from actual child count, not copied from the batch row. Production proved the stored status cannot be trusted: every one of these batches was written status=processed at creation time, before any insert ran. The bug is worse than the plan's own audit found. It predicted 18 phantom-processed batches, all plant_hours. Production has 21 -- three timesheets batches are also phantom, including two byte-identical re-uploads of the same wage sheet that both reported success and created nothing. This hit payroll, not just plant-hour smoke tests. Also: computed file_hash on migrated documents while the bytes were already in hand (closes the same dedup gap Task 1 closed on delivery notes and purchase orders), and split the schema-additive step (--apply-fields) from the row migration (--apply) so the reversible part can run independently of the irreversible one. The plan claimed idempotency via a migrated_from_batch field that doesn't exist; replaced with an explicit refuse-if-already-populated guard rather than pretend resumability that isn't there. 27/27 migrated. 6/6 post-migration invariants verified: 3 fuel_sheets, 19 plant_hours_sheets, 5 timesheets, 1367/247/1 children relinked. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
DEls
parent
3a71e7e884
commit
539ec93d2a
1 changed files with 623 additions and 0 deletions
623
scripts/migrations/2026-09-12-migrate-batches-to-documents.py
Normal file
623
scripts/migrations/2026-09-12-migrate-batches-to-documents.py
Normal file
|
|
@ -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)
|
||||
Loading…
Reference in a new issue