MEGAINFRA-WEBAPP/scripts/pb-add-management.py

#!/usr/bin/env python3
"""Create the management planning collections in PocketBase.

The Management pages read the planning workbook out of PocketBase so admins
can enter actual income, change a project's status and keep an audit trail
(see docs/superpowers/plans/2026-10-19-management-pocketbase.md).

Run: python3 scripts/pb-add-management.py [--dry-run]
     --dry-run only reads and prints what would be created; it writes nothing.

The script is idempotent: a missing collection is created, an existing one is
verified (fields, indexes, server-only rules) and never modified. No record is
touched and nothing is ever deleted.
"""
import sys
from pb_env import http, superuser_token

# The planning workbook arrays and the history blob are far bigger than
# PocketBase's 1MB json default.
JSON_MAX_SIZE = 2000000


def main():
    dry_run = "--dry-run" in sys.argv[1:]
    if any(arg not in ("--dry-run", "--help") for arg in sys.argv[1:]):
        sys.exit("Unknown option. Usage: python3 scripts/pb-add-management.py [--dry-run]")
    if "--help" in sys.argv[1:]:
        print(__doc__.strip())
        return

    token = superuser_token()
    created = []

    def read(name):
        code, result = http("GET", f"/api/collections/{name}", token=token)
        if code != 200:
            raise RuntimeError(f"Cannot read {name}: HTTP {code}")
        return result

    def relation(name, collection, required=True, cascade_delete=False):
        return {"name": name, "type": "relation", "required": required,
                "collectionId": collection["id"], "maxSelect": 1, "cascadeDelete": cascade_delete}

    def json_field(name):
        return {"name": name, "type": "json", "required": False, "maxSize": JSON_MAX_SIZE}

    def json_number_field(name):
        # A json field keeps a missing number null; a number field hands back
        # 0 for an empty value, which is not the same as "not supplied".
        return {"name": name, "type": "json", "required": False, "maxSize": 1000}

    def ensure(name, fields, indexes):
        wanted = {"name": name, "type": "base", "fields": fields + [
            {"name": "created", "type": "autodate", "onCreate": True, "onUpdate": False},
            {"name": "updated", "type": "autodate", "onCreate": True, "onUpdate": True},
        ], "indexes": indexes, **{rule: None for rule in (
            "listRule", "viewRule", "createRule", "updateRule", "deleteRule")}}
        code, actual = http("GET", f"/api/collections/{name}", token=token)
        if code == 404:
            if dry_run:
                print(f"Would create {name}")
                created.append(name)
                # The relations below need a collection id to point at; the
                # placeholder is never written, it only keeps the dry run going.
                return {"id": "000000000000000", "name": name}
            code, actual = http("POST", "/api/collections", wanted, token)
            if code != 200:
                raise RuntimeError(f"Cannot create {name}: HTTP {code}: {actual}")
            print(f"Created {name}")
            created.append(name)
        elif code != 200:
            raise RuntimeError(f"Cannot inspect {name}: HTTP {code}")
        actual = read(name)
        by_name = {f["name"]: f for f in actual["fields"]}
        for field in wanted["fields"]:
            if any(by_name.get(field["name"], {}).get(k) != v for k, v in field.items()):
                raise RuntimeError(f"Incompatible {name}.{field['name']}")
        for index in indexes:
            if index not in actual.get("indexes", []):
                raise RuntimeError(f"Missing index on {name}")
        for rule in ("listRule", "viewRule", "createRule", "updateRule", "deleteRule"):
            if actual.get(rule) is not None:
                raise RuntimeError(f"Expected server-only access on {name}")
        if not dry_run:
            print(f"Verified {name}")
        return actual

    sites = read("sites")

    plans = ensure("management_plans", [
        {"name": "plan_year", "type": "number", "required": True, "onlyInt": True},
        {"name": "source_file", "type": "text", "required": False},
        {"name": "source_sheet", "type": "text", "required": False},
        {"name": "source_sha256", "type": "text", "required": False},
        {"name": "period_label", "type": "text", "required": False},
        {"name": "cost_detail_available", "type": "bool", "required": False},
        {"name": "opening_retention", "type": "number", "required": False},
        {"name": "reported_cost_annual", "type": "number", "required": False},
        json_field("months"),
        json_field("reported_income"),
        json_field("cashflow_income_factors"),
        json_field("forecast_cost_rates"),
        json_field("reported_costs"),
        json_field("reported_retention"),
        json_field("reported_retention_closing"),
        json_field("reported_cumulative_balance"),
        json_field("retention_payouts"),
        json_field("head_office_costs"),
        # The retention periods of the source schedule, including the periods
        # that carry no release.
        json_field("retention_months"),
        json_field("history"),
        {"name": "revision", "type": "number", "required": False, "min": 1, "onlyInt": True},
    ], ["CREATE UNIQUE INDEX `idx_management_plans_plan_year` ON `management_plans` (`plan_year`)"])

    projects = ensure("management_projects", [
        {"name": "key", "type": "text", "required": True},
        # The canonical display name, the one seen first across the years. The
        # per-year client and engineer live on the project plan.
        {"name": "name", "type": "text", "required": False},
        relation("site", sites, False),
    ], ["CREATE UNIQUE INDEX `idx_management_projects_key` ON `management_projects` (`key`)"])

    project_plans = ensure("management_project_plans", [
        relation("plan", plans, True, True),
        relation("project", projects, True),
        {"name": "status", "type": "select", "required": False, "maxSelect": 1,
         "values": ["active", "awarded"]},
        {"name": "sort_order", "type": "number", "required": False},
        # The project details of this year: they differ from year to year.
        {"name": "name", "type": "text", "required": True},
        {"name": "client", "type": "text", "required": False},
        {"name": "engineer", "type": "text", "required": False},
        {"name": "manager", "type": "text", "required": False},
        json_number_field("stated_value"),
        json_number_field("remaining_value"),
        {"name": "retention_eligible", "type": "bool", "required": False},
        json_field("income"),
        json_field("forecast_income"),
        json_field("actual_income"),
        json_field("future_income"),
        json_field("forecast_baseline"),
        json_field("retention_rates"),
        json_field("costs"),
        {"name": "revision", "type": "number", "required": False, "min": 1, "onlyInt": True},
    ], ["CREATE UNIQUE INDEX `idx_management_project_plans_plan_project` ON `management_project_plans` (`plan`, `project`)"])

    ensure("management_retention", [
        relation("plan", plans, True, True),
        {"name": "project_name", "type": "text", "required": False},
        relation("project", projects, False),
        {"name": "debtor", "type": "text", "required": False},
        {"name": "amount", "type": "number", "required": False},
        {"name": "release_date", "type": "date", "required": False},
        {"name": "source_id", "type": "text", "required": False},
    ], ["CREATE INDEX `idx_management_retention_plan_source` ON `management_retention` (`plan`, `source_id`)"])

    ensure("management_edits", [
        {"name": "plan_year", "type": "number", "required": False, "onlyInt": True},
        # The audit trail outlives the project plan it points at.
        relation("project_plan", project_plans, False),
        {"name": "project_name", "type": "text", "required": False},
        {"name": "field", "type": "text", "required": False},
        {"name": "month_index", "type": "number", "required": False},
        json_field("old_value"),
        json_field("new_value"),
        {"name": "user_label", "type": "text", "required": False},
    ], ["CREATE INDEX `idx_management_edits_project_plan` ON `management_edits` (`project_plan`, `plan_year`)"])

    if dry_run:
        print(f"Dry run: {len(created)} collection(s) would be created: {', '.join(created) or 'none'}")
        print("Nothing was written.")
        return
    print("Verified the management planning schema. No plan, project or audit record changed.")


if __name__ == "__main__":
    try:
        main()
    except Exception as exc:
        sys.exit(str(exc))