08ed89d9c8
Updated the operations dashboard to include a rain outlook feature, providing users with a 7-day forecast based on site coordinates. Added latitude and longitude fields to the site form for better weather data integration. Enhanced the fuel log and site management functionalities to utilize dynamic fuel pricing based on historical data. Improved overall data handling and user experience across various dashboard components.
71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
"""Shared PocketBase connection setup for the scripts in this directory.
|
|
|
|
Reads .env.local from the repo root the same way `npm run mcp` does
|
|
(`tsx --env-file=.env.local`), so the scripts run without splicing
|
|
credentials onto the command line. Real environment variables win over
|
|
the file, so CI or a one-off override still works.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def load_env_local(path: Path = REPO_ROOT / ".env.local") -> None:
|
|
if not path.exists():
|
|
return
|
|
for raw in path.read_text().splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
key = key.strip()
|
|
value = value.strip().strip('"').strip("'")
|
|
# Never clobber a variable the caller set explicitly.
|
|
if key and key not in os.environ:
|
|
os.environ[key] = value
|
|
|
|
|
|
load_env_local()
|
|
|
|
PB = os.environ.get("POCKETBASE_URL") or os.environ.get(
|
|
"NEXT_PUBLIC_POCKETBASE_URL", "https://pb.mega.infra.apphub.co.za"
|
|
)
|
|
|
|
|
|
def http(method, path, body=None, token=None):
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(f"{PB}{path}", data=data, method=method)
|
|
if body is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
if token:
|
|
req.add_header("Authorization", token)
|
|
try:
|
|
with urllib.request.urlopen(req) as r:
|
|
txt = r.read().decode()
|
|
return r.status, json.loads(txt) if txt else None
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, json.loads(e.read().decode() or "{}")
|
|
|
|
|
|
def superuser_token() -> str:
|
|
email = os.environ.get("POCKETBASE_ADMIN_EMAIL")
|
|
password = os.environ.get("POCKETBASE_ADMIN_PASSWORD")
|
|
if not email or not password:
|
|
sys.exit(
|
|
"POCKETBASE_ADMIN_EMAIL / POCKETBASE_ADMIN_PASSWORD not found in "
|
|
"the environment or .env.local."
|
|
)
|
|
code, auth = http(
|
|
"POST",
|
|
"/api/collections/_superusers/auth-with-password",
|
|
{"identity": email, "password": password},
|
|
)
|
|
if code != 200:
|
|
sys.exit(f"AUTH FAILED: {code} {auth}")
|
|
print(f"auth ok on {PB}")
|
|
return auth["token"]
|