diff --git a/scripts/migrations/2026-09-25-backfill-clock-shifts.py b/scripts/migrations/2026-09-25-backfill-clock-shifts.py new file mode 100644 index 0000000..d98bf97 --- /dev/null +++ b/scripts/migrations/2026-09-25-backfill-clock-shifts.py @@ -0,0 +1,232 @@ +#!/usr/bin/env python3 +"""Pair historical QR scans into shifts without rewriting events or payroll. + +Dry-run is the default. Supply PB_URL, PB_ADMIN_EMAIL and PB_ADMIN_PASSWORD +for a read-only live report. --apply additionally requires --backup-confirmed +pointing to an existing backup file; apply aborts if any ambiguity remains. +""" + +import argparse +from collections import Counter, defaultdict +from datetime import datetime, timezone +from hashlib import sha256 +import json +import os +from pathlib import Path +import sys +import urllib.error +import urllib.parse +import urllib.request +from zoneinfo import ZoneInfo + + +LOCAL_ZONE = ZoneInfo("Africa/Johannesburg") + + +def instant(value): + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + raise ValueError("scan instant has no timezone") + return parsed.astimezone(timezone.utc) + + +def local_date(value): + return instant(value).astimezone(LOCAL_ZONE).date().isoformat() + + +def anomaly(kind, *record_ids): + return {"kind": kind, "record_ids": list(record_ids)} + + +def pair_events(events): + by_employee = defaultdict(list) + for row in events: + by_employee[row["employee"]].append(row) + shifts = [] + anomalies = [] + for employee, rows in sorted(by_employee.items()): + rows = sorted(rows, key=lambda row: (instant(row["scanned_at"]), row["id"])) + active = None + employee_shifts = [] + employee_anomalies = [] + for row in rows: + direction = row["direction"] + if direction == "in": + if active is not None: + employee_anomalies.append(anomaly("consecutive_in", active["id"], row["id"])) + active = row + elif direction == "out": + if active is None: + employee_anomalies.append(anomaly("orphan_out", row["id"])) + continue + if instant(row["scanned_at"]) <= instant(active["scanned_at"]): + employee_anomalies.append(anomaly("overlap", active["id"], row["id"])) + active = None + continue + if row.get("site") != active.get("site"): + employee_anomalies.append(anomaly("site_mismatch", active["id"], row["id"])) + shift = { + "employee": employee, "site": active["site"], + "clock_in_event": active["id"], + "clock_in_at": instant(active["scanned_at"]).isoformat(), + "local_work_date": local_date(active["scanned_at"]), + "clock_out_event": row["id"], + "clock_out_at": instant(row["scanned_at"]).isoformat(), + "close_source": "scan", + "payroll_status": "synced" if row.get("timesheet_entry") else "conflict", + } + if row.get("timesheet_entry"): + shift["timesheet_entry"] = row["timesheet_entry"] + employee_shifts.append(shift) + active = None + else: + employee_anomalies.append(anomaly("invalid_direction", row["id"])) + if instant(row["scanned_at"]).date().isoformat() != local_date(row["scanned_at"]): + employee_anomalies.append(anomaly("utc_local_date_disagreement", row["id"])) + if active is not None: + employee_shifts.append({ + "employee": employee, "site": active["site"], + "clock_in_event": active["id"], + "clock_in_at": instant(active["scanned_at"]).isoformat(), + "local_work_date": local_date(active["scanned_at"]), + "payroll_status": "open", + }) + # Date differences are reported but the local work date is unambiguous. + severe = [item for item in employee_anomalies + if item["kind"] != "utc_local_date_disagreement"] + if not severe: + shifts.extend(employee_shifts) + anomalies.extend(employee_anomalies) + return shifts, anomalies + + +def audit_timesheets(shifts, entries): + anomalies = [] + by_day = defaultdict(list) + by_id = {row["id"]: row for row in entries} + for row in entries: + if row.get("employee") and not row.get("timesheet"): + by_day[(row["employee"], local_date(row["date_logged"]))].append(row) + for key, rows in by_day.items(): + if len(rows) > 1: + anomalies.append(anomaly("duplicate_direct_entries", *(row["id"] for row in rows))) + for shift in shifts: + if shift["payroll_status"] == "open": + continue + linked = shift.get("timesheet_entry") + if not linked: + anomalies.append(anomaly("ambiguous_timesheet_link", shift["clock_in_event"])) + continue + entry = by_id.get(linked) + if not entry or entry.get("timesheet") or entry.get("employee") != shift["employee"]: + anomalies.append(anomaly("ambiguous_timesheet_link", shift["clock_in_event"], linked)) + elif local_date(entry["date_logged"]) != shift["local_work_date"]: + anomalies.append(anomaly("timesheet_date_mismatch", shift["clock_in_event"], linked)) + return anomalies + + +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=30) as response: + content = response.read() + return json.loads(content) if content else None + except urllib.error.HTTPError as error: + if error.code == 404 and method == "GET": + 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=30) as response: + return json.load(response)["token"] + except urllib.error.HTTPError as error: + raise RuntimeError(f"authentication failed ({error.code})") from None + + +def all_records(base, token, collection): + rows = [] + page = 1 + while True: + query = urllib.parse.urlencode({"perPage": 200, "page": page}) + response = request(base, token, "GET", f"/api/collections/{collection}/records?{query}") + rows.extend(response["items"]) + if page >= response["totalPages"]: + return rows + page += 1 + + +def shift_id(in_event_id): + return sha256(("clock-shift:" + in_event_id).encode()).hexdigest()[:15] + + +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: + if len(supplied) != 3: + parser.error("--apply requires all three PB environment variables") + if not args.backup_confirmed or not args.backup_confirmed.is_file(): + parser.error("--apply requires --backup-confirmed with an existing backup file") + if supplied and len(supplied) != 3: + parser.error("set all three PB variables for a live read-only report") + if not supplied: + print("DRY RUN: configure PB_URL, PB_ADMIN_EMAIL and PB_ADMIN_PASSWORD for read-only pairing report") + return 0 + + base = os.environ["PB_URL"].rstrip("/") + token = authenticate(base) + events = all_records(base, token, "clock_events") + entries = all_records(base, token, "timesheet_entries") + employees = {row["id"]: row for row in all_records(base, token, "employees")} + shifts, anomalies = pair_events(events) + anomalies += audit_timesheets(shifts, entries) + counts = Counter(row["direction"] for row in events) + statuses = Counter(row["payroll_status"] for row in shifts) + print(f"{'APPLY' if args.apply else 'DRY RUN'}: {len(events)} events; {counts['in']} ins; {counts['out']} outs") + print(f"Paired {len(shifts)} shifts; {statuses['open']} open; {statuses['synced']} linked closed; {statuses['conflict']} unlinked closed") + existing_collection = request(base, token, "GET", "/api/collections/clock_shifts") + existing_shifts = all_records(base, token, "clock_shifts") if existing_collection else [] + existing_ins = {row["clock_in_event"] for row in existing_shifts} + new_shifts = [row for row in shifts if row["clock_in_event"] not in existing_ins] + print(f"Existing shifts {len(existing_shifts)}; new shifts {len(new_shifts)}") + old_open = [row for row in shifts if row["payroll_status"] == "open" and + row["local_work_date"] == "2026-09-23"] + if old_open: + print("23 September unresolved badges: " + ", ".join(sorted( + str(employees.get(row["employee"], {}).get("employee_no") or row["employee"]) + for row in old_open))) + print(f"Anomalies: {len(anomalies)}") + for item in anomalies: + print(f" {item['kind']}: {', '.join(item['record_ids'])}") + if not args.apply: + print("No writes. Historical totals remain unchanged.") + return 0 + if not existing_collection or anomalies: + raise RuntimeError("apply stopped: missing shift schema or unresolved anomalies") + for row in new_shifts: + request(base, token, "POST", "/api/collections/clock_shifts/records", + {"id": shift_id(row["clock_in_event"]), **row}) + print(f"Created {len(new_shifts)} shifts. No clock events or timesheet entries changed.") + return 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except (RuntimeError, ValueError) as error: + print(f"ERROR: {error}", file=sys.stderr) + sys.exit(1) diff --git a/scripts/migrations/test_2026_09_25_backfill_clock_shifts.py b/scripts/migrations/test_2026_09_25_backfill_clock_shifts.py new file mode 100644 index 0000000..7098d09 --- /dev/null +++ b/scripts/migrations/test_2026_09_25_backfill_clock_shifts.py @@ -0,0 +1,76 @@ +"""Fixtures for historical QR pairing; no live PocketBase access.""" +import importlib.util +from pathlib import Path +import unittest + + +SCRIPT = Path(__file__).with_name("2026-09-25-backfill-clock-shifts.py") +spec = importlib.util.spec_from_file_location("clock_shift_backfill", SCRIPT) +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) + + +def event(id_, employee, direction, time, site="site1", entry=""): + return {"id": id_, "employee": employee, "direction": direction, + "scanned_at": time, "site": site, "timesheet_entry": entry} + + +class PairingTests(unittest.TestCase): + def test_normal_and_split_sessions_copy_only_real_out_link(self): + rows = [ + event("i1", "e1", "in", "2026-09-23T05:00:00Z"), + event("o1", "e1", "out", "2026-09-23T09:00:00Z", entry="t1"), + event("i2", "e1", "in", "2026-09-23T10:00:00Z"), + event("o2", "e1", "out", "2026-09-23T13:00:00Z", entry="t1"), + ] + shifts, anomalies = module.pair_events(rows) + self.assertEqual(anomalies, []) + self.assertEqual(len(shifts), 2) + self.assertEqual(shifts[0]["timesheet_entry"], "t1") + self.assertEqual(shifts[1]["payroll_status"], "synced") + self.assertEqual(shifts[0]["local_work_date"], "2026-09-23") + + def test_overnight_historical_pair_is_attributed_to_start_date(self): + shifts, anomalies = module.pair_events([ + event("i1", "e1", "in", "2026-09-23T20:30:00Z"), + event("o1", "e1", "out", "2026-09-23T23:30:00Z"), + ]) + self.assertEqual(len(shifts), 1) + self.assertEqual(shifts[0]["local_work_date"], "2026-09-23") + self.assertEqual(shifts[0]["clock_out_event"], "o1") + self.assertTrue(any(a["kind"] == "utc_local_date_disagreement" for a in anomalies)) + + def test_consecutive_in_and_orphan_out_are_not_guessed(self): + shifts, anomalies = module.pair_events([ + event("i1", "e1", "in", "2026-09-23T05:00:00Z"), + event("i2", "e1", "in", "2026-09-23T06:00:00Z"), + event("o2", "e1", "out", "2026-09-23T14:00:00Z"), + event("orphan", "e2", "out", "2026-09-23T14:00:00Z"), + ]) + self.assertEqual(shifts, []) + self.assertEqual({a["kind"] for a in anomalies}, {"consecutive_in", "orphan_out"}) + + def test_seven_latest_ins_remain_open_without_synthetic_hours(self): + rows = [event(f"in{n}", f"e{n}", "in", "2026-09-23T05:00:00Z") + for n in range(7)] + shifts, anomalies = module.pair_events(rows) + self.assertEqual(anomalies, []) + self.assertEqual(len(shifts), 7) + self.assertTrue(all(s["payroll_status"] == "open" for s in shifts)) + self.assertTrue(all("clock_out_at" not in s for s in shifts)) + self.assertTrue(all("timesheet_entry" not in s for s in shifts)) + + def test_duplicate_direct_entries_flagged_without_changing_totals(self): + shifts, _ = module.pair_events([ + event("i1", "e1", "in", "2026-09-23T05:00:00Z"), + event("o1", "e1", "out", "2026-09-23T14:00:00Z"), + ]) + anomalies = module.audit_timesheets(shifts, [ + {"id": "t1", "employee": "e1", "date_logged": "2026-09-23T00:00:00Z", "timesheet": ""}, + {"id": "t2", "employee": "e1", "date_logged": "2026-09-23T00:00:00Z", "timesheet": ""}, + ]) + self.assertTrue(any(a["kind"] == "duplicate_direct_entries" for a in anomalies)) + + +if __name__ == "__main__": + unittest.main()