David/MEGAINFRA-WEBAPP
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(clock): dry-run historical shift pairing

Browse source
This commit is contained in:
DEls 2026-09-25 03:16:08 +02:00
parent 5529c3f53d
commit e01c45765b
2 changed files with 308 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

232
scripts/migrations/2026-09-25-backfill-clock-shifts.py Normal file
View file

@ -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)

76
scripts/migrations/test_2026_09_25_backfill_clock_shifts.py Normal file
View file

@ -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()