102 lines
3.7 KiB
Python
102 lines
3.7 KiB
Python
"""Add the quest-system JSON fields to the economy_users PocketBase collection.
|
|
|
|
Run once after pulling the quest changes:
|
|
python scripts/add_quest_fields.py
|
|
|
|
Requirements:
|
|
- PocketBase running and reachable at PB_URL
|
|
- PB_ADMIN_EMAIL / PB_ADMIN_PASSWORD set in .env
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import aiohttp
|
|
from dotenv import load_dotenv
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
load_dotenv()
|
|
|
|
import config # noqa: E402
|
|
|
|
PB_URL = config.PB_URL
|
|
PB_ADMIN_EMAIL = config.PB_ADMIN_EMAIL
|
|
PB_ADMIN_PASSWORD = config.PB_ADMIN_PASSWORD
|
|
# Patch both profiles' collections so the result doesn't depend on which
|
|
# BOT_PROFILE happened to be set when the script was run.
|
|
COLLECTIONS = sorted({config.PB_ECONOMY_COLLECTION_DEV, config.PB_ECONOMY_COLLECTION_ECONOMY})
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# New fields to add
|
|
# ---------------------------------------------------------------------------
|
|
_NEW_JSON_FIELDS = [
|
|
"quest_daily",
|
|
"quest_weekly",
|
|
]
|
|
|
|
|
|
def _json_field(name: str) -> dict:
|
|
return {"name": name, "type": "json", "required": False}
|
|
|
|
|
|
async def main() -> None:
|
|
timeout = aiohttp.ClientTimeout(total=15)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
# ── Authenticate ────────────────────────────────────────────────────
|
|
async with session.post(
|
|
f"{PB_URL}/api/collections/_superusers/auth-with-password",
|
|
json={"identity": PB_ADMIN_EMAIL, "password": PB_ADMIN_PASSWORD},
|
|
) as resp:
|
|
if resp.status != 200:
|
|
print(f"Auth failed ({resp.status}): {await resp.text()}")
|
|
return
|
|
token = (await resp.json())["token"]
|
|
|
|
hdrs = {"Authorization": token}
|
|
|
|
for collection in COLLECTIONS:
|
|
print(f"── {collection} ──")
|
|
|
|
# ── Fetch current collection ─────────────────────────────────────
|
|
async with session.get(
|
|
f"{PB_URL}/api/collections/{collection}", headers=hdrs
|
|
) as resp:
|
|
if resp.status != 200:
|
|
print(f"Could not fetch collection ({resp.status}): {await resp.text()}\n")
|
|
continue
|
|
col = await resp.json()
|
|
|
|
existing = {f["name"] for f in col.get("fields", [])}
|
|
print(f"Existing fields ({len(existing)}): {sorted(existing)}\n")
|
|
|
|
new_fields = []
|
|
for name in _NEW_JSON_FIELDS:
|
|
if name not in existing:
|
|
new_fields.append(_json_field(name))
|
|
print(f" + {name} (json)")
|
|
else:
|
|
print(f" = {name} (already exists)")
|
|
|
|
if not new_fields:
|
|
print("Nothing to add - schema already up to date.\n")
|
|
continue
|
|
|
|
# ── Patch collection schema ──────────────────────────────────────
|
|
updated_fields = col.get("fields", []) + new_fields
|
|
async with session.patch(
|
|
f"{PB_URL}/api/collections/{collection}",
|
|
json={"fields": updated_fields},
|
|
headers=hdrs,
|
|
) as resp:
|
|
if resp.status != 200:
|
|
print(f"Schema update failed ({resp.status}): {await resp.text()}\n")
|
|
continue
|
|
print(f"✅ Added {len(new_fields)} field(s) successfully.\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|