#!/usr/bin/env python3 """ Idempotent PocketBase schema setup for Kings & Queens support (English tutoring) calendar and booking. Mirrors the environment-driven client pattern used by setup_auth_family.py / setup_billing.py. """ import copy import getpass import json import os import re import urllib.error import urllib.request PB_URL = os.environ.get("KQ_PB_URL", "http://127.0.0.1:8091").rstrip("/") # Order of the per-collection ``rules`` arrays returned by build_schema(). RULE_KEYS = ("listRule", "viewRule", "createRule", "updateRule", "deleteRule") TEACHER_RULE = "@request.auth.id != '' && @request.auth.rol = 'teacher'" def pb_req(method, path, data=None, token=None, request_client=None): url = f"{PB_URL}/api/{path}" body = json.dumps(data).encode("utf-8") if data is not None else None req = urllib.request.Request(url, data=body, method=method) req.add_header("Content-Type", "application/json") if token: req.add_header("Authorization", token) client = request_client or urllib.request.urlopen try: with client(req) as resp: raw = resp.read().decode("utf-8") return (json.loads(raw) if raw else {}, resp.status) except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8") try: return json.loads(raw), exc.code except json.JSONDecodeError: return {"raw": raw}, exc.code except urllib.error.URLError: return {"error": "network"}, 0 def auth(): email = os.environ.get("KQ_PB_ADMIN_EMAIL") or input("Superuser email: ").strip() password = os.environ.get("KQ_PB_ADMIN_PASSWORD") or getpass.getpass("Superuser password: ") data, status = pb_req( "POST", "collections/_superusers/auth-with-password", {"identity": email, "password": password}, ) if status != 200: raise SystemExit(f"Auth failed ({status})") return data["token"] def _unique_values(values): return list(dict.fromkeys(values)) def merge_field(existing, desired): merged = copy.deepcopy(existing) changed = False for key, value in desired.items(): if key == "values": desired_values = desired.get("values") or [] merged_values = _unique_values([*(merged.get("values") or []), *desired_values]) if merged_values != merged.get("values"): merged["values"] = merged_values changed = True continue if merged.get(key) != value: merged[key] = copy.deepcopy(value) changed = True return merged, changed def merge_fields(existing_fields, desired_fields): merged = [copy.deepcopy(field) for field in existing_fields] by_name = {field["name"]: idx for idx, field in enumerate(merged)} changed = False for desired in desired_fields: current_index = by_name.get(desired["name"]) if current_index is None: merged.append(copy.deepcopy(desired)) by_name[desired["name"]] = len(merged) - 1 changed = True continue updated, field_changed = merge_field(merged[current_index], desired) if field_changed: merged[current_index] = updated changed = True return merged, changed def merge_indexes(existing_indexes, desired_indexes): merged = list(existing_indexes or []) canonical_existing = {re.sub(r"\s+", "", index).lower(): index for index in merged} changed = False for index in desired_indexes or []: canonical = re.sub(r"\s+", "", index).lower() if canonical not in canonical_existing: merged.append(index) canonical_existing[canonical] = index changed = True return merged, changed def default_settings(): return { "default_capacity": 2, "default_deposit": 5000, "hold_hours": 12, "cancel_credit_hours": 24, "transfer_alias": "", "transfer_instructions": "Transferí la seña y enviá el comprobante por este chat.", } def build_schema(): teacher = TEACHER_RULE schema = { "support_settings": {"fields": [ {"name": "default_capacity", "type": "number", "required": True}, {"name": "default_deposit", "type": "number", "required": True}, {"name": "hold_hours", "type": "number", "required": True}, {"name": "cancel_credit_hours", "type": "number", "required": True}, {"name": "transfer_alias", "type": "text", "required": False}, {"name": "transfer_instructions", "type": "text", "required": False}, ], "rules": [teacher, teacher, None, teacher, None]}, "support_slots": {"fields": [ {"name": "date", "type": "text", "required": True}, {"name": "time_start", "type": "text", "required": True}, {"name": "time_end", "type": "text", "required": True}, {"name": "status", "type": "select", "required": True, "maxSelect": 1, "values": ["open", "suspended"]}, {"name": "capacity", "type": "number", "required": True}, {"name": "deposit_amount", "type": "number", "required": True}, {"name": "notes", "type": "text", "required": False}, ], "rules": [teacher, teacher, None, None, None], "indexes": [ "CREATE INDEX idx_support_slots_date ON support_slots (date, time_start)", ]}, "support_bookings": {"fields": [ {"name": "slot_id", "type": "relation", "required": True, "collectionId": "support_slots", "maxSelect": 1}, {"name": "student_name", "type": "text", "required": True}, {"name": "contact_phone", "type": "text", "required": True}, {"name": "guardian_name", "type": "text", "required": False}, {"name": "education_level", "type": "select", "required": True, "maxSelect": 1, "values": ["primary", "secondary"]}, {"name": "school_year", "type": "text", "required": True}, {"name": "topic", "type": "text", "required": True}, {"name": "conversation_id", "type": "number", "required": True}, {"name": "status", "type": "select", "required": True, "maxSelect": 1, "values": ["held", "payment_review", "confirmed", "rejected", "cancelled", "expired", "credit"]}, {"name": "hold_expires_at", "type": "date", "required": False}, {"name": "deposit_amount", "type": "number", "required": True}, {"name": "receipt", "type": "file", "required": False, "maxSelect": 1, "maxSize": 5242880, "mimeTypes": ["image/jpeg", "image/png", "application/pdf"]}, {"name": "idempotency_key", "type": "text", "required": True}, {"name": "seat_no", "type": "number", "required": False}, {"name": "credit_used", "type": "bool", "required": False}, {"name": "confirmed_at", "type": "date", "required": False}, {"name": "confirmation_sent_at", "type": "date", "required": False}, ], "rules": [teacher, teacher, None, None, None], "indexes": [ "CREATE UNIQUE INDEX idx_support_booking_key ON support_bookings (idempotency_key)", # Atomic capacity guard: at most one seat-consuming booking per (slot, seat_no). # The partial WHERE means a booking frees its seat simply by leaving these # statuses, so two racing holds for the same seat collide deterministically. "CREATE UNIQUE INDEX `idx_support_seat` ON `support_bookings` (`slot_id`, `seat_no`) WHERE `status` = 'held' OR `status` = 'payment_review' OR `status` = 'confirmed'", "CREATE INDEX idx_support_bookings_slot ON support_bookings (slot_id)", "CREATE INDEX idx_support_bookings_conversation ON support_bookings (conversation_id)", ]}, } # PocketBase base collections created via the API have no created/updated # fields unless declared; the hooks sort by created, so add them everywhere. autodate = [ {"name": "created", "type": "autodate", "onCreate": True, "onUpdate": False}, {"name": "updated", "type": "autodate", "onCreate": True, "onUpdate": True}, ] for entry in schema.values(): entry["fields"].extend({**field} for field in autodate) return schema def rules_payload(rules): return {key: rules[idx] for idx, key in enumerate(RULE_KEYS) if idx < len(rules)} def get_collection(name, token, request_client=None): return pb_req("GET", f"collections/{name}", token=token, request_client=request_client) def create_collection(name, definition, token, request_client=None): payload = {"name": name, "type": "base", **definition} data, status = pb_req("POST", "collections", payload, token=token, request_client=request_client) if status not in (200, 201, 204): raise RuntimeError(f"Could not create collection {name} ({status}): {data}") print(name) return data def update_collection(name, definition, token, request_client=None): current, status = get_collection(name, token, request_client=request_client) if status != 200: raise RuntimeError(f"Cannot update collection {name} (status {status})") current_fields = current.get("fields") or current.get("schema") or [] merged_fields, fields_changed = merge_fields(current_fields, definition.get("fields") or []) merged_indexes, indexes_changed = merge_indexes(current.get("indexes") or [], definition.get("indexes") or []) payload = {} for key in RULE_KEYS: if key in definition and current.get(key) != definition.get(key): payload[key] = definition.get(key) if fields_changed: payload["fields"] = merged_fields if indexes_changed: payload["indexes"] = merged_indexes if not payload: return current data, status = pb_req("PATCH", f"collections/{current['id']}", payload, token=token, request_client=request_client) if status not in (200, 204): raise RuntimeError(f"Could not update collection {name} ({status}): {data}") print(name) return data def ensure_collection(name, definition, token, request_client=None): current, status = get_collection(name, token, request_client=request_client) if status == 200: return update_collection(name, definition, token, request_client=request_client) if status == 404: return create_collection(name, definition, token, request_client=request_client) raise RuntimeError(f"Cannot load collection {name} (status {status})") def definition_for(name, raw, slot_collection_id=None): """Convert a build_schema() entry into a PocketBase collection payload.""" definition = {"fields": copy.deepcopy(raw["fields"]), **rules_payload(raw.get("rules", []))} if raw.get("indexes"): definition["indexes"] = list(raw["indexes"]) # Resolve the slot_id relation to the real collection id once it exists. if slot_collection_id: for field in definition["fields"]: if field.get("type") == "relation" and field.get("collectionId") == "support_slots": field["collectionId"] = slot_collection_id return definition def seed_settings(token, request_client=None): existing, status = pb_req( "GET", "collections/support_settings/records?perPage=1", token=token, request_client=request_client) if status == 200 and existing.get("totalItems", 0) > 0: return data, status = pb_req( "POST", "collections/support_settings/records", default_settings(), token=token, request_client=request_client) if status not in (200, 201): raise RuntimeError(f"Could not seed support_settings ({status}): {data}") print("support_settings (seeded)") def main(): token = auth() schema = build_schema() ensure_collection("support_settings", definition_for("support_settings", schema["support_settings"]), token) slots = ensure_collection("support_slots", definition_for("support_slots", schema["support_slots"]), token) slot_id = slots.get("id") if not slot_id: slots, _ = get_collection("support_slots", token) slot_id = slots.get("id") ensure_collection( "support_bookings", definition_for("support_bookings", schema["support_bookings"], slot_collection_id=slot_id), token, ) seed_settings(token) if __name__ == "__main__": main()