103 lines
4.5 KiB
Python
103 lines
4.5 KiB
Python
"""Add Office user section and site access to PocketBase. Dry-run by default."""
|
|
|
|
import argparse
|
|
import copy
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
MODULES = ["overview", "sites", "workforce", "plant", "financial"]
|
|
|
|
|
|
def prepare_fields(collection):
|
|
if collection.get("type") != "auth":
|
|
raise RuntimeError("users must be an auth collection")
|
|
fields = copy.deepcopy(collection["fields"])
|
|
by_name = {field["name"]: field for field in fields}
|
|
role = by_name.get("role", {})
|
|
if role.get("type") != "select" or role.get("maxSelect") != 1:
|
|
raise RuntimeError("users.role must be the existing single-select role field")
|
|
if not all(value in role.get("values", []) for value in ["admin", "site_user", "clock_only"]):
|
|
raise RuntimeError("users.role is missing existing account types")
|
|
if "office_user" not in role["values"]:
|
|
role["values"].append("office_user")
|
|
additions = [
|
|
{"name": "modules", "type": "select", "required": False, "maxSelect": len(MODULES), "values": MODULES},
|
|
{"name": "all_sites", "type": "bool", "required": False},
|
|
]
|
|
for wanted in additions:
|
|
old = by_name.get(wanted["name"])
|
|
if old:
|
|
if any(old.get(key) != value for key, value in wanted.items()):
|
|
raise RuntimeError(f"Existing users.{wanted['name']} has an incompatible schema")
|
|
else:
|
|
fields.append(wanted)
|
|
return fields
|
|
|
|
|
|
def environment(path):
|
|
values = {}
|
|
if path.is_file():
|
|
for raw in path.read_text().splitlines():
|
|
line = raw.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
key, value = line.split("=", 1)
|
|
values[key.strip()] = value.strip().strip('"').strip("'")
|
|
return {**values, **os.environ}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--env-file", type=Path, default=Path(__file__).resolve().parents[2] / ".env.local")
|
|
parser.add_argument("--apply", action="store_true")
|
|
parser.add_argument("--confirm-host", help="Exact PocketBase base URL for applying the additive schema update")
|
|
args = parser.parse_args()
|
|
env = environment(args.env_file)
|
|
base = (env.get("POCKETBASE_URL") or env.get("NEXT_PUBLIC_POCKETBASE_URL") or "").rstrip("/")
|
|
email, password = env.get("POCKETBASE_ADMIN_EMAIL"), env.get("POCKETBASE_ADMIN_PASSWORD")
|
|
if not base or not email or not password:
|
|
parser.error("PocketBase URL and superuser credentials are required")
|
|
if args.apply and args.confirm_host != base:
|
|
parser.error("--apply requires --confirm-host matching the configured PocketBase URL")
|
|
|
|
def api(method, path, token=None, body=None):
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
request = urllib.request.Request(f"{base}{path}", data=data, method=method)
|
|
if data is not None:
|
|
request.add_header("Content-Type", "application/json")
|
|
if token:
|
|
request.add_header("Authorization", token)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
return json.load(response)
|
|
except urllib.error.HTTPError as error:
|
|
raise RuntimeError(f"{method} {path} failed (HTTP {error.code})") from None
|
|
|
|
print(f"Target: {base} | {'APPLY' if args.apply else 'DRY RUN'}")
|
|
auth = api("POST", "/api/collections/_superusers/auth-with-password", body={"identity": email, "password": password})
|
|
token = auth["token"]
|
|
users = api("GET", "/api/collections/users", token)
|
|
fields = prepare_fields(users)
|
|
if fields == users["fields"]:
|
|
print("Office user schema already matches. No changes needed.")
|
|
return
|
|
print("Add office_user role, section selection, and all-sites option; preserve current users and collection rules.")
|
|
if not args.apply:
|
|
print("Dry run complete. Apply this schema before deploying the Office user settings.")
|
|
return
|
|
api("PATCH", "/api/collections/users", token, {"fields": fields})
|
|
verified = api("GET", "/api/collections/users", token)
|
|
if prepare_fields(verified) != verified["fields"]:
|
|
raise RuntimeError("Office access schema verification failed")
|
|
print("Office access schema verified. No user accounts or passwords were changed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (RuntimeError, urllib.error.URLError) as error:
|
|
print(f"ERROR: {error}", file=sys.stderr)
|
|
sys.exit(1)
|