cb463c0faa
Add per-site CSV and PDF matrix reports, editable sick-note date ranges, and weekend calendar styling. Preserve verified duplicate uploads for audit while excluding their entries from hours and costs, and prevent repeat files and overlapping daily imports.
70 lines
2.8 KiB
Python
70 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Add inclusive sick-note end dates, preserving existing photos and dates."""
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from pb_env import http, superuser_token
|
|
|
|
|
|
def main():
|
|
token = superuser_token()
|
|
|
|
def get(path):
|
|
code, result = http("GET", path, token=token)
|
|
if code != 200:
|
|
raise RuntimeError(f"Read failed: HTTP {code}")
|
|
return result
|
|
|
|
def notes():
|
|
rows, page = [], 1
|
|
while True:
|
|
result = get(f"/api/collections/employee_sick_notes/records?perPage=500&page={page}&sort=id")
|
|
rows.extend(result["items"])
|
|
if page >= result["totalPages"]:
|
|
return rows
|
|
page += 1
|
|
|
|
schema = get("/api/collections/employee_sick_notes")
|
|
before = notes()
|
|
fd, backup = tempfile.mkstemp(prefix="mega-before-sick-note-ranges-", suffix=".json")
|
|
with os.fdopen(fd, "w") as file:
|
|
json.dump({"schema": schema, "notes": before}, file)
|
|
print("Private backup:", backup)
|
|
wanted = {"name": "end_date", "type": "date", "required": False}
|
|
fields = schema["fields"]
|
|
existing = next((f for f in fields if f["name"] == "end_date"), None)
|
|
if existing:
|
|
assert all(existing.get(k) == v for k, v in wanted.items()), "Incompatible end-date field"
|
|
else:
|
|
code, _ = http("PATCH", f"/api/collections/{schema['id']}", {"fields": fields + [wanted]}, token)
|
|
if code != 200:
|
|
raise RuntimeError(f"Schema update failed: HTTP {code}")
|
|
changed = 0
|
|
for row in before:
|
|
if row.get("end_date"):
|
|
continue
|
|
code, _ = http("PATCH", f"/api/collections/employee_sick_notes/records/{row['id']}", {"end_date": row["note_date"]}, token)
|
|
if code != 200:
|
|
raise RuntimeError(f"Note update failed: HTTP {code}")
|
|
changed += 1
|
|
after = notes()
|
|
by_id = {row["id"]: row for row in after}
|
|
assert len(after) == len(before)
|
|
for row in before:
|
|
current = by_id[row["id"]]
|
|
assert current["end_date"] == (row.get("end_date") or row["note_date"])
|
|
for key, value in row.items():
|
|
if key not in ("end_date", "updated"):
|
|
assert current.get(key) == value, f"Unexpected change: {key}"
|
|
actual = get("/api/collections/employee_sick_notes")
|
|
for key in ("listRule", "viewRule", "createRule", "updateRule", "deleteRule", "indexes"):
|
|
assert actual.get(key) == schema.get(key), f"Unexpected schema change: {key}"
|
|
by_name = {f["name"]: f for f in actual["fields"]}
|
|
assert all(by_name["end_date"].get(k) == v for k, v in wanted.items())
|
|
for field in fields:
|
|
assert by_name[field["name"]] == field, "Original field changed"
|
|
print(f"Verified {len(after)} notes; {changed} single-day end dates filled; photos and original fields preserved.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|