"""Destructively recreate TipiBOT PocketBase collections. Usage: python scripts/reset_pb_collections.py --confirm This will DELETE and recreate the collections configured by: - PB_ECONOMY_COLLECTION_DEV - PB_ECONOMY_COLLECTION_ECONOMY - PB_ECONOMY_COLLECTION_LAN - PB_FIENTA_COLLECTION_LAN """ from __future__ import annotations import argparse 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 def _text_field(name: str, required: bool = False) -> dict: return { "name": name, "type": "text", "required": required, "options": {"min": None, "max": None, "pattern": ""}, } def _number_field(name: str) -> dict: return { "name": name, "type": "number", "required": False, "options": {"min": None, "max": None, "noDecimal": False}, } def _bool_field(name: str) -> dict: return {"name": name, "type": "bool", "required": False} def _json_field(name: str) -> dict: return {"name": name, "type": "json", "required": False} def _collection_payload(name: str) -> dict: fields = [ _text_field("user_id", required=True), _number_field("balance"), _number_field("exp"), _number_field("daily_streak"), _text_field("last_daily"), _text_field("last_work"), _text_field("last_beg"), _text_field("last_crime"), _text_field("last_rob"), _text_field("last_heist"), _text_field("last_streak_date"), _text_field("jailed_until"), _text_field("last_fish"), _json_field("items"), _json_field("item_uses"), _json_field("reminders"), _json_field("prestige_upgrades"), _json_field("fish_book"), _json_field("fish_inventory"), _bool_field("eco_banned"), _bool_field("jailbreak_used"), _number_field("heist_global_cd_until"), _number_field("peak_balance"), _number_field("lifetime_earned"), _number_field("lifetime_lost"), _number_field("work_count"), _number_field("beg_count"), _number_field("total_wagered"), _number_field("biggest_win"), _number_field("biggest_loss"), _number_field("slots_jackpots"), _number_field("crimes_attempted"), _number_field("crimes_succeeded"), _number_field("times_jailed"), _number_field("total_bail_paid"), _number_field("heists_joined"), _number_field("heists_won"), _number_field("total_given"), _number_field("total_received"), _number_field("best_daily_streak"), _number_field("prestige_level"), _number_field("prestige_points"), _number_field("season_total_exp"), _number_field("total_fish_caught"), ] return { "name": name, "type": "base", "fields": fields, "listRule": None, "viewRule": None, "createRule": None, "updateRule": None, "deleteRule": None, } def _fienta_collection_payload(name: str) -> dict: fields = [ _text_field("registration_key", required=True), _text_field("order_id"), _text_field("ticket_code"), _text_field("order_status"), _text_field("order_url"), _text_field("payment_time"), _text_field("game"), _text_field("kind"), _text_field("ticket_type_id"), _text_field("ticket_title"), _text_field("ticket_group_title"), _text_field("team_name"), _text_field("discord_username"), _text_field("nickname"), _text_field("country"), _text_field("country_code"), _text_field("riot_id"), _text_field("steam64_id"), _text_field("vrs_ranking"), _bool_field("is_main"), _bool_field("is_reserve"), _bool_field("is_manager"), _bool_field("is_captain"), _bool_field("sheet_public"), _bool_field("blocked_country"), _bool_field("active"), _bool_field("roles_synced"), _text_field("last_sync_error"), _text_field("updated_at"), ] return { "name": name, "type": "base", "fields": fields, "listRule": None, "viewRule": None, "createRule": None, "updateRule": None, "deleteRule": None, } async def _auth_token(session: aiohttp.ClientSession) -> str: 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: raise RuntimeError(f"Auth failed ({resp.status}): {await resp.text()}") return (await resp.json())["token"] async def _delete_if_exists(session: aiohttp.ClientSession, headers: dict[str, str], name: str) -> None: async with session.get(f"{PB_URL}/api/collections/{name}", headers=headers) as resp: if resp.status == 404: print(f"[SKIP] {name} does not exist") return if resp.status != 200: raise RuntimeError(f"Could not fetch {name} ({resp.status}): {await resp.text()}") async with session.delete(f"{PB_URL}/api/collections/{name}", headers=headers) as resp: if resp.status not in (200, 204): raise RuntimeError(f"Delete failed for {name} ({resp.status}): {await resp.text()}") print(f"[DELETE] {name}") async def _create_collection( session: aiohttp.ClientSession, headers: dict[str, str], name: str, payload: dict, ) -> None: async with session.post(f"{PB_URL}/api/collections", json=payload, headers=headers) as resp: if resp.status not in (200, 201): raise RuntimeError(f"Create failed for {name} ({resp.status}): {await resp.text()}") print(f"[CREATE] {name}") async def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--confirm", action="store_true", help="Required flag to run destructive reset") args = parser.parse_args() if not args.confirm: raise SystemExit("Refusing to run without --confirm (this operation deletes collections).") targets: list[tuple[str, dict]] = [] for name in [ config.PB_ECONOMY_COLLECTION_DEV, config.PB_ECONOMY_COLLECTION_ECONOMY, config.PB_ECONOMY_COLLECTION_LAN, ]: if name and all(existing != name for existing, _ in targets): targets.append((name, _collection_payload(name))) fienta_name = config.PB_FIENTA_COLLECTION_LAN if fienta_name and all(name != fienta_name for name, _ in targets): targets.append((fienta_name, _fienta_collection_payload(fienta_name))) if not targets: raise SystemExit("No target collections configured.") timeout = aiohttp.ClientTimeout(total=20) async with aiohttp.ClientSession(timeout=timeout) as session: token = await _auth_token(session) headers = {"Authorization": token} for name, payload in targets: await _delete_if_exists(session, headers, name) await _create_collection(session, headers, name, payload) print("\nDone. Collections recreated:") for name, _ in targets: print(f" - {name}") if __name__ == "__main__": asyncio.run(main())