147 lines
6.6 KiB
Python
147 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Dry-run/add one locked collection for missed QR scan-out corrections.
|
|
|
|
Set PB_URL, PB_ADMIN_EMAIL and PB_ADMIN_PASSWORD to compare the live schema.
|
|
Only --apply --backup-confirmed PATH can write the collection schema.
|
|
No worker event or timesheet record is changed by this script.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
def relation(name, collection_id, required=False):
|
|
return {"name": name, "type": "relation", "required": required,
|
|
"collectionId": collection_id, "cascadeDelete": False,
|
|
"maxSelect": 1, **({"minSelect": 1} if required else {})}
|
|
|
|
|
|
def correction_definition(ids):
|
|
return {
|
|
"name": "clock_missed_out_corrections", "type": "base",
|
|
"fields": [
|
|
relation("clock_in_event", ids["clock_events"], True),
|
|
relation("employee", ids["employees"], True),
|
|
relation("site", ids["sites"], True),
|
|
{"name": "actual_out_at", "type": "date", "required": True},
|
|
{"name": "reason", "type": "text", "required": True, "min": 3, "max": 2000},
|
|
{"name": "recorded_at", "type": "date", "required": True},
|
|
relation("timesheet_entry", ids["timesheet_entries"], True),
|
|
],
|
|
"indexes": [
|
|
"CREATE UNIQUE INDEX `idx_missed_out_in_event` ON `clock_missed_out_corrections` (`clock_in_event`)",
|
|
],
|
|
"listRule": None, "viewRule": None, "createRule": None,
|
|
"updateRule": None, "deleteRule": None,
|
|
}
|
|
|
|
|
|
def lock_definition(ids):
|
|
return {
|
|
"name": "clock_correction_locks", "type": "base",
|
|
"fields": [
|
|
relation("employee", ids["employees"], True),
|
|
{"name": "owner", "type": "text", "required": True, "min": 1, "max": 128},
|
|
],
|
|
"indexes": [
|
|
"CREATE UNIQUE INDEX `idx_clock_correction_lock_employee` ON `clock_correction_locks` (`employee`)",
|
|
],
|
|
"listRule": None, "viewRule": None, "createRule": None,
|
|
"updateRule": None, "deleteRule": None,
|
|
}
|
|
|
|
|
|
def request(base, token, method, path, body=None):
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(base + path, data=data, method=method,
|
|
headers={"Authorization": token,
|
|
"Content-Type": "application/json"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=20) as response:
|
|
content = response.read()
|
|
return json.loads(content) if content else None
|
|
except urllib.error.HTTPError as error:
|
|
if method == "GET" and error.code == 404:
|
|
return None
|
|
raise RuntimeError(f"{method} {path} failed ({error.code})") from None
|
|
|
|
|
|
def authenticate(base):
|
|
req = urllib.request.Request(
|
|
base + "/api/collections/_superusers/auth-with-password",
|
|
data=json.dumps({"identity": os.environ["PB_ADMIN_EMAIL"],
|
|
"password": os.environ["PB_ADMIN_PASSWORD"]}).encode(),
|
|
method="POST", headers={"Content-Type": "application/json"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=20) as response:
|
|
return json.load(response)["token"]
|
|
except urllib.error.HTTPError as error:
|
|
raise RuntimeError(f"authentication failed ({error.code})") from None
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--apply", action="store_true")
|
|
parser.add_argument("--backup-confirmed", type=Path)
|
|
args = parser.parse_args()
|
|
names = ("PB_URL", "PB_ADMIN_EMAIL", "PB_ADMIN_PASSWORD")
|
|
supplied = [name for name in names if os.environ.get(name)]
|
|
if args.apply and (len(supplied) != 3 or not args.backup_confirmed or
|
|
not args.backup_confirmed.is_file()):
|
|
parser.error("--apply requires PB_URL, PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD and an existing --backup-confirmed file")
|
|
if supplied and len(supplied) != 3:
|
|
parser.error("set all three PB variables for a live dry run")
|
|
print(f"Mode: {'APPLY' if args.apply else 'DRY RUN'}")
|
|
if not supplied:
|
|
ids = {name: f"__{name.upper()}_ID__" for name in
|
|
("clock_events", "employees", "sites", "timesheet_entries")}
|
|
definition = correction_definition(ids)
|
|
print(f"CREATE {definition['name']}: {len(definition['fields'])} fields, unique IN event, locked rules")
|
|
print("CREATE clock_correction_locks: unique employee, locked rules")
|
|
print("Offline plan only. No writes.")
|
|
return 0
|
|
|
|
base = os.environ["PB_URL"].rstrip("/")
|
|
token = authenticate(base)
|
|
ids = {}
|
|
for name in ("clock_events", "employees", "sites", "timesheet_entries"):
|
|
collection = request(base, token, "GET", f"/api/collections/{name}")
|
|
if collection is None:
|
|
raise RuntimeError(f"required collection missing: {name}")
|
|
ids[name] = collection["id"]
|
|
for definition, unique_field in ((correction_definition(ids), "clock_in_event"),
|
|
(lock_definition(ids), "employee")):
|
|
existing = request(base, token, "GET", f"/api/collections/{definition['name']}")
|
|
if existing:
|
|
fields = {field["name"]: field for field in existing.get("fields", [])}
|
|
missing = [field["name"] for field in definition["fields"] if field["name"] not in fields]
|
|
locked = all(existing.get(rule) is None for rule in
|
|
("listRule", "viewRule", "createRule", "updateRule", "deleteRule"))
|
|
unique = any("unique index" in item.lower() and unique_field in item.lower()
|
|
for item in existing.get("indexes", []))
|
|
print(f"EXISTS {definition['name']}: missing fields {missing}; locked={locked}; unique {unique_field}={unique}")
|
|
if args.apply and (missing or not locked or not unique):
|
|
raise RuntimeError("existing collection differs; manual schema review required")
|
|
else:
|
|
print(f"CREATE {definition['name']}: {len(definition['fields'])} fields, unique {unique_field}, locked rules")
|
|
if args.apply:
|
|
created = request(base, token, "POST", "/api/collections", definition)
|
|
if not created or not created.get("id"):
|
|
raise RuntimeError("collection creation could not be verified")
|
|
print(f"Created {definition['name']}")
|
|
if not args.apply:
|
|
print("No writes.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except RuntimeError as error:
|
|
print(f"ERROR: {error}", file=sys.stderr)
|
|
sys.exit(1)
|