#!/usr/bin/env python3 """ Idempotent PocketBase schema setup for Kings & Queens monthly billing. Creates: - student_billing_profiles - billing_periods - payment_orders - payment_events """ import getpass import json import os import urllib.error import urllib.request PB_URL = os.environ.get("KQ_PB_URL", "http://localhost:8091").rstrip("/") def pb_req(method, path, data=None, token=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) try: with urllib.request.urlopen(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 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}): {data}") return data["token"] def get_collection(name, token): data, status = pb_req("GET", f"collections/{name}", token=token) return data if status == 200 else None def upsert_collection(name, definition, token): current = get_collection(name, token) if current: print(f"- collection {name}: exists") return current data, status = pb_req("POST", "collections", definition, token=token) if status not in (200, 204): raise RuntimeError(f"Error creating {name} ({status}): {data}") print(f"- collection {name}: created") return data def upsert_fields(collection_name, fields_to_add, token): col = get_collection(collection_name, token) if not col: raise RuntimeError(f"Collection {collection_name} does not exist") fields = col.get("fields") or col.get("schema") or [] existing = {field["name"]: field for field in fields} changed = False for field in fields_to_add: current = existing.get(field["name"]) if not current: fields.append(field) changed = True print(f"- field {collection_name}.{field['name']}: added") continue if field.get("type") == "select" and field.get("values"): values = list(dict.fromkeys([*(current.get("values") or []), *field["values"]])) if values != current.get("values"): current["values"] = values changed = True print(f"- field {collection_name}.{field['name']}: values extended") if changed: data, status = pb_req("PATCH", f"collections/{col['id']}", {"fields": fields}, token=token) if status not in (200, 204): raise RuntimeError(f"Error updating {collection_name} ({status}): {data}") else: print(f"- collection {collection_name}: fields already up to date") def relation_field(name, collection_id, required=False, cascade=False): return { "name": name, "type": "relation", "required": required, "collectionId": collection_id, "maxSelect": 1, "cascadeDelete": cascade, } def main(): token = auth() teacher_only = "@request.auth.id != '' && @request.auth.rol = 'teacher'" authenticated = "@request.auth.id != ''" upsert_collection("student_billing_profiles", { "name": "student_billing_profiles", "type": "base", "fields": [ relation_field("student_id", "_pb_users_auth_", True, True), {"name": "weekly_classes", "type": "number", "required": False}, {"name": "monthly_classes", "type": "number", "required": False}, {"name": "monthly_amount", "type": "number", "required": True}, {"name": "late_fee_amount", "type": "number", "required": False}, {"name": "currency", "type": "select", "required": False, "maxSelect": 1, "values": ["ARS", "USD"]}, {"name": "payment_method", "type": "select", "required": False, "maxSelect": 1, "values": ["cash", "transfer", "international"]}, {"name": "due_day", "type": "number", "required": True}, {"name": "default_provider", "type": "select", "required": False, "maxSelect": 1, "values": ["cash", "transfer", "international", "mercadopago", "psp"]}, {"name": "is_active", "type": "bool", "required": False}, {"name": "notes", "type": "text", "required": False}, ], "listRule": authenticated, "viewRule": authenticated, "createRule": teacher_only, "updateRule": teacher_only, "deleteRule": teacher_only, }, token) upsert_collection("billing_periods", { "name": "billing_periods", "type": "base", "fields": [ relation_field("student_id", "_pb_users_auth_", True, True), relation_field("billing_profile_id", "student_billing_profiles"), {"name": "period_key", "type": "text", "required": True}, {"name": "included_classes", "type": "number", "required": False}, {"name": "amount_due", "type": "number", "required": True}, {"name": "base_amount", "type": "number", "required": False}, {"name": "surcharge_amount", "type": "number", "required": False}, {"name": "late_fee_amount", "type": "number", "required": False}, {"name": "currency", "type": "select", "required": False, "maxSelect": 1, "values": ["ARS", "USD"]}, {"name": "payment_method", "type": "select", "required": False, "maxSelect": 1, "values": ["cash", "transfer", "international"]}, {"name": "due_date", "type": "text", "required": True}, {"name": "expires_at", "type": "text", "required": True}, {"name": "payment_reminder_sent_at", "type": "text", "required": False}, {"name": "due_reminder_sent_at", "type": "text", "required": False}, {"name": "late_fee_applied_at", "type": "text", "required": False}, {"name": "status", "type": "select", "required": True, "maxSelect": 1, "values": ["pending", "paid", "overdue", "closed"]}, ], "listRule": authenticated, "viewRule": authenticated, "createRule": teacher_only, "updateRule": teacher_only, "deleteRule": teacher_only, }, token) upsert_collection("payment_orders", { "name": "payment_orders", "type": "base", "fields": [ relation_field("student_id", "_pb_users_auth_", True, True), relation_field("billing_period_id", "billing_periods", True, True), {"name": "provider", "type": "select", "required": True, "maxSelect": 1, "values": ["cash", "transfer", "international", "mercadopago", "psp"]}, {"name": "status", "type": "select", "required": True, "maxSelect": 1, "values": ["pending", "approved", "rejected", "review"]}, {"name": "amount_due", "type": "number", "required": True}, {"name": "currency", "type": "select", "required": False, "maxSelect": 1, "values": ["ARS", "USD"]}, {"name": "reference", "type": "text", "required": True}, {"name": "payment_url", "type": "url", "required": False}, {"name": "instructions", "type": "text", "required": False}, {"name": "approved_at", "type": "text", "required": False}, {"name": "external_id", "type": "text", "required": False}, ], "listRule": authenticated, "viewRule": authenticated, "createRule": teacher_only, "updateRule": teacher_only, "deleteRule": teacher_only, }, token) upsert_collection("payment_events", { "name": "payment_events", "type": "base", "fields": [ relation_field("payment_order_id", "payment_orders"), {"name": "provider", "type": "select", "required": True, "maxSelect": 1, "values": ["cash", "transfer", "international", "mercadopago", "psp", "manual"]}, {"name": "provider_event_id", "type": "text", "required": False}, {"name": "amount", "type": "number", "required": False}, {"name": "reference", "type": "text", "required": False}, {"name": "status", "type": "select", "required": True, "maxSelect": 1, "values": ["received", "matched", "duplicate", "review", "rejected"]}, {"name": "raw_payload", "type": "json", "required": False}, {"name": "received_at", "type": "text", "required": True}, ], "listRule": authenticated, "viewRule": teacher_only, "createRule": teacher_only, "updateRule": teacher_only, "deleteRule": teacher_only, }, token) upsert_fields("student_billing_profiles", [ {"name": "weekly_classes", "type": "number", "required": False}, {"name": "monthly_classes", "type": "number", "required": False}, {"name": "late_fee_amount", "type": "number", "required": False}, {"name": "currency", "type": "select", "required": False, "maxSelect": 1, "values": ["ARS", "USD"]}, {"name": "payment_method", "type": "select", "required": False, "maxSelect": 1, "values": ["cash", "transfer", "international"]}, {"name": "default_provider", "type": "select", "required": False, "maxSelect": 1, "values": ["cash", "transfer", "international", "mercadopago", "psp"]}, ], token) upsert_fields("billing_periods", [ {"name": "base_amount", "type": "number", "required": False}, {"name": "surcharge_amount", "type": "number", "required": False}, {"name": "late_fee_amount", "type": "number", "required": False}, {"name": "currency", "type": "select", "required": False, "maxSelect": 1, "values": ["ARS", "USD"]}, {"name": "payment_method", "type": "select", "required": False, "maxSelect": 1, "values": ["cash", "transfer", "international"]}, {"name": "payment_reminder_sent_at", "type": "text", "required": False}, {"name": "due_reminder_sent_at", "type": "text", "required": False}, {"name": "late_fee_applied_at", "type": "text", "required": False}, ], token) upsert_fields("payment_orders", [ {"name": "provider", "type": "select", "required": True, "maxSelect": 1, "values": ["cash", "transfer", "international", "mercadopago", "psp"]}, {"name": "currency", "type": "select", "required": False, "maxSelect": 1, "values": ["ARS", "USD"]}, ], token) upsert_fields("payment_events", [ {"name": "provider", "type": "select", "required": True, "maxSelect": 1, "values": ["cash", "transfer", "international", "mercadopago", "psp", "manual"]}, ], token) print("\nBilling schema ready.") if __name__ == "__main__": main()