122 lines
6.2 KiB
Python
122 lines
6.2 KiB
Python
"""
|
|
PATCH schema fields onto the JEI_* collections if they're missing.
|
|
|
|
Workaround for a known PocketBase quirk where collections created via API
|
|
sometimes persist with empty schema. The rules PATCH trick works for empty
|
|
rules; for empty schema we explicitly PATCH the schema field.
|
|
|
|
Run: python scripts/patch_collections.py
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
import urllib.request
|
|
from urllib.error import HTTPError
|
|
|
|
PB_URL = os.environ.get("PB_URL", "https://pb.agent.apphub.co.za")
|
|
PB_TOKEN = os.environ.get("PB_TOKEN", "")
|
|
if not PB_TOKEN:
|
|
env_path = os.path.join(os.path.dirname(__file__), "..", ".env.local")
|
|
if os.path.exists(env_path):
|
|
for line in open(env_path):
|
|
if line.startswith("PB_TOKEN="):
|
|
PB_TOKEN = line.strip().split("=", 1)[1]
|
|
if not PB_TOKEN:
|
|
sys.exit("ERROR: PB_TOKEN not set")
|
|
HEADERS = {"Authorization": PB_TOKEN, "Content-Type": "application/json"}
|
|
|
|
|
|
def pb_get(path):
|
|
req = urllib.request.Request(PB_URL + path, headers=HEADERS, method="GET")
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return json.loads(r.read())
|
|
|
|
|
|
def pb_patch(path, body):
|
|
req = urllib.request.Request(PB_URL + path, data=json.dumps(body).encode("utf-8"),
|
|
headers=HEADERS, method="PATCH")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return json.loads(r.read()), None
|
|
except HTTPError as e:
|
|
return None, e.read().decode()
|
|
|
|
|
|
# Mirror of create_collections.py schemas (kept independent for clarity)
|
|
SUPERUSER_RULES = {"listRule": "", "viewRule": "", "createRule": "",
|
|
"updateRule": "", "deleteRule": ""}
|
|
|
|
|
|
def invoices_schema():
|
|
return [
|
|
{"name": "invoice_no", "type": "text", "required": True, "options": {"min": 1, "max": 50, "pattern": ""}},
|
|
{"name": "invoice_date", "type": "date", "required": True, "options": {"min": "", "max": ""}},
|
|
{"name": "document_type", "type": "select", "required": True, "options": {"maxSelect": 1, "values": ["invoice", "proforma", "quote"]}},
|
|
{"name": "customer_order_no", "type": "text", "required": False, "options": {"min": 0, "max": 100, "pattern": ""}},
|
|
{"name": "buyer_name_variant", "type": "select", "required": True, "options": {"maxSelect": 1, "values": ["t/a JEI South Africa", "JEI South Africa", "JEI South Africa Importers"]}},
|
|
{"name": "supplier_name", "type": "text", "required": True, "options": {"min": 0, "max": 200, "pattern": ""}},
|
|
{"name": "account_ref", "type": "text", "required": False, "options": {"min": 0, "max": 50, "pattern": ""}},
|
|
{"name": "total_net", "type": "number", "required": True, "options": {"min": 0, "max": None, "noDecimal": False}},
|
|
{"name": "carriage_net", "type": "number", "required": False, "options": {"min": 0, "max": None, "noDecimal": False}},
|
|
{"name": "invoice_total", "type": "number", "required": True, "options": {"min": 0, "max": None, "noDecimal": False}},
|
|
{"name": "deliver_to_name", "type": "text", "required": False, "options": {"min": 0, "max": 200, "pattern": ""}},
|
|
{"name": "deliver_to_address", "type": "text", "required": False, "options": {"min": 0, "max": 1000, "pattern": ""}},
|
|
{"name": "source_filename", "type": "text", "required": False, "options": {"min": 0, "max": 500, "pattern": ""}},
|
|
{"name": "notes", "type": "text", "required": False, "options": {"min": 0, "max": 5000, "pattern": ""}},
|
|
{"name": "pdf_file", "type": "file", "required": False, "options": {"maxSelect": 1, "maxSize": 0, "mimeTypes": ["application/pdf"], "thumbs": []}},
|
|
]
|
|
|
|
|
|
def stock_codes_schema():
|
|
return [
|
|
{"name": "code", "type": "text", "required": True, "options": {"min": 1, "max": 100, "pattern": ""}},
|
|
{"name": "description_goods", "type": "text", "required": True, "options": {"min": 1, "max": 500, "pattern": ""}},
|
|
{"name": "first_seen_date", "type": "date", "required": False, "options": {"min": "", "max": ""}},
|
|
{"name": "last_seen_date", "type": "date", "required": False, "options": {"min": "", "max": ""}},
|
|
]
|
|
|
|
|
|
def lines_schema():
|
|
# We need to resolve the relation target ids
|
|
cols = pb_get("/api/collections?perPage=200")["items"]
|
|
inv = next(c for c in cols if c["name"] == "JEI_invoices")
|
|
sc = next(c for c in cols if c["name"] == "JEI_stock_codes")
|
|
return [
|
|
{"name": "invoice", "type": "relation", "required": True, "options": {"collectionId": inv["id"], "cascadeDelete": True, "minSelect": 1, "maxSelect": 1}},
|
|
{"name": "line_no", "type": "number", "required": True, "options": {"min": 0, "max": None, "noDecimal": True}},
|
|
{"name": "stock_code", "type": "relation", "required": True, "options": {"collectionId": sc["id"], "cascadeDelete": True, "minSelect": 1, "maxSelect": 1}},
|
|
{"name": "quantity", "type": "number", "required": True, "options": {"min": 0, "max": None, "noDecimal": False}},
|
|
{"name": "unit_price", "type": "number", "required": True, "options": {"min": 0, "max": None, "noDecimal": False}},
|
|
{"name": "net_amount", "type": "number", "required": True, "options": {"min": 0, "max": None, "noDecimal": False}},
|
|
{"name": "commodity_code", "type": "text", "required": False, "options": {"min": 0, "max": 50, "pattern": ""}},
|
|
{"name": "country_of_origin", "type": "text", "required": False, "options": {"min": 0, "max": 50, "pattern": ""}},
|
|
]
|
|
|
|
|
|
PATCHES = [
|
|
("JEI_invoices", invoices_schema),
|
|
("JEI_stock_codes", stock_codes_schema),
|
|
("JEI_invoice_lines", lines_schema),
|
|
]
|
|
|
|
|
|
def main():
|
|
cols = pb_get("/api/collections?perPage=200")["items"]
|
|
for name, factory in PATCHES:
|
|
col = next((c for c in cols if c["name"] == name), None)
|
|
if not col:
|
|
print(f"SKIP: {name} does not exist")
|
|
continue
|
|
if col.get("schema") and len(col["schema"]) > 0:
|
|
print(f"OK: {name} already has {len(col['schema'])} fields")
|
|
continue
|
|
body = {"schema": factory(), **SUPERUSER_RULES}
|
|
result, err = pb_patch(f"/api/collections/{col['id']}", body)
|
|
if err:
|
|
print(f"FAIL: {name} — {err}")
|
|
sys.exit(1)
|
|
print(f"PATCHED: {name} → {len(result.get('schema', []))} fields")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|