"""Who this phone belongs to, and who is still on the crew. Two company-level files, beside the jobs rather than inside one, because a bloke moves between jobs and his phone does not stop being his phone: /_signon/devices.csv append-only - a phone asked to be someone, a foreman said yes or no /_signon/people.csv one row each - the crew, their role, their number, active or not A device holds a random 128-bit token. Only the SHA-256 of it is ever written down, so the file on Drive cannot be read back into a working phone. Approving a token binds it to a name; from then on that phone IS that bloke and there is no name box to type someone else's into. Nothing here overwrites a device row. A change of mind is a NEW row, and the latest row for a token is the one that counts - same rule as the attendance register. """ from __future__ import annotations import hashlib import re import secrets from dataclasses import dataclass from datetime import datetime DEVICES = "devices.csv" DEVICE_HEADER = ["timestamp", "token_hash", "name", "status", "by", "note"] PEOPLE = "people.csv" PEOPLE_HEADER = ["name", "status", "role", "phone", "since", "by"] COOKIE = "dev" # transport only; localStorage is where the phone keeps it COOKIE_MAX_AGE = 10 * 365 * 24 * 3600 ROLES = ("worker", "foreman", "admin") BOSS_ROLES = ("foreman", "admin") # Who the crew rings on a given job is a per-JOB fact, so it lives with the job in Foreman.txt, # next to Crew.txt and Hazards.txt the foreman already keeps (see storage.parse_foreman). people.csv # is company-wide and a man can be foreman on one job and a hand on another, so a role column there # cannot answer "who do I ring on THIS job". def new_token() -> str: """128 bits. Never stored anywhere on our side - only its hash.""" return secrets.token_hex(16) def token_hash(token: str) -> str: return hashlib.sha256((token or "").encode("utf-8")).hexdigest() def norm(name: str) -> str: return re.sub(r"\s+", " ", (name or "").strip()).lower() def tel_href(phone: str) -> str: """0412 345 678 -> tel:0412345678. Keeps a leading + for internationals.""" p = re.sub(r"[^\d+]", "", phone or "") return f"tel:{p}" if p else "" @dataclass class Identity: """What the server knows about the phone in front of it.""" token: str = "" thash: str = "" name: str = "" status: str = "none" # none | pending | approved | revoked role: str = "" person_status: str = "" # active | inactive | "" (not on people.csv at all) @property def bound(self) -> bool: return self.status == "approved" and self.person_status != "inactive" @property def is_boss(self) -> bool: return self.bound and self.role in BOSS_ROLES @property def inactive(self) -> bool: return self.person_status == "inactive" class People: """Reads and writes the two company files through whichever storage backend is in play.""" def __init__(self, store): self.store = store # -- devices ----------------------------------------------------------- def devices(self) -> list[dict]: return self.store.read_company_csv(DEVICES) def latest_by_token(self) -> dict[str, dict]: """Last row wins, per token. The file is a history; this is the state it adds up to.""" out: dict[str, dict] = {} for r in self.devices(): th = (r.get("token_hash") or "").strip() if th: out[th] = r return out def append_device(self, thash: str, name: str, status: str, by: str, note: str = "") -> None: self.store.append_company_csv(DEVICES, DEVICE_HEADER, { "timestamp": datetime.now().astimezone().isoformat(timespec="seconds"), "token_hash": thash, "name": name, "status": status, "by": by, "note": note, }) def pending(self) -> list[dict]: return [r for r in self.latest_by_token().values() if (r.get("status") or "") == "pending"] def approved_names(self) -> set[str]: return {norm(r.get("name", "")) for r in self.latest_by_token().values() if (r.get("status") or "") == "approved"} def devices_for(self, name: str) -> list[dict]: n = norm(name) return [r for r in self.latest_by_token().values() if norm(r.get("name", "")) == n] OFF_NOTE = "marked inactive" def revoke_all_for(self, name: str, by: str, note: str) -> int: """Marked inactive = every one of his phones is dead. New row each, never an edit.""" n = 0 for r in self.devices_for(name): if (r.get("status") or "") != "revoked": self.append_device(r.get("token_hash", ""), r.get("name", ""), "revoked", by, note) n += 1 return n def restore_all_for(self, name: str, by: str) -> int: """Back on the crew = the phones the deactivation killed come back. One tap off, one tap on. A phone knocked back with NOT HIM stays dead — that was a different decision and putting a man back on the crew must not quietly undo it. """ n = 0 for r in self.devices_for(name): if (r.get("status") or "") == "revoked" and (r.get("note") or "") == self.OFF_NOTE: self.append_device(r.get("token_hash", ""), r.get("name", ""), "approved", by, "back on the crew") n += 1 return n # -- people ------------------------------------------------------------ def people(self) -> list[dict]: return self.store.read_company_csv(PEOPLE) def person(self, name: str) -> dict | None: n = norm(name) for r in self.people(): if norm(r.get("name", "")) == n: return r return None def upsert_person(self, name: str, role: str = "worker", phone: str = "", status: str = "active", by: str = "") -> None: """people.csv is a state file, one row per bloke - so this rewrites, unlike devices.csv.""" name = re.sub(r"\s+", " ", (name or "").strip()) if not name: return if role not in ROLES: role = "worker" rows = self.people() today = datetime.now().astimezone().date().isoformat() for r in rows: if norm(r.get("name", "")) == norm(name): r["status"] = status if role: r["role"] = role if phone: r["phone"] = phone r["by"] = by or r.get("by", "") break else: rows.append({"name": name, "status": status, "role": role, "phone": phone, "since": today, "by": by}) self.store.write_company_csv(PEOPLE, PEOPLE_HEADER, rows) def set_status(self, name: str, status: str, by: str) -> None: p = self.person(name) self.upsert_person(name, role=(p or {}).get("role", "worker"), phone=(p or {}).get("phone", ""), status=status, by=by) if status == "inactive": self.revoke_all_for(name, by, self.OFF_NOTE) else: self.restore_all_for(name, by) def phone_of(self, name: str) -> str: return ((self.person(name) or {}).get("phone") or "").strip() # -- the questions main.py actually asks -------------------------------- def identify(self, token: str) -> Identity: if not token: return Identity() th = token_hash(token) row = self.latest_by_token().get(th) if not row: return Identity(token=token, thash=th) name = (row.get("name") or "").strip() p = self.person(name) or {} return Identity(token=token, thash=th, name=name, status=(row.get("status") or "").strip(), role=(p.get("role") or "worker").strip().lower(), person_status=(p.get("status") or "").strip().lower()) def enforced_for(self, name: str) -> bool: """Is binding live for this bloke yet? Only once he has an approved phone, or the office has switched him off. Until then the old name-memory keeps working - otherwise the day this ships every man on site is locked out waiting on a foreman who has not been told he needs to tap anything. """ p = self.person(name) if p and (p.get("status") or "").strip().lower() == "inactive": return True return norm(name) in self.approved_names() def boss_bound_anywhere(self) -> bool: """Has anyone been approved as a foreman/admin yet? Until someone has, the foreman screen stays open - otherwise nobody can approve the first phone, including the phone that would do the approving. """ approved = [r for r in self.latest_by_token().values() if (r.get("status") or "") == "approved"] for r in approved: p = self.person(r.get("name", "")) or {} if (p.get("role") or "").strip().lower() in BOSS_ROLES and \ (p.get("status") or "active").strip().lower() != "inactive": return True return False