Spaces:
Sleeping
Sleeping
File size: 4,135 Bytes
80a4a65 04fc815 80a4a65 04fc815 80a4a65 04fc815 80a4a65 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | """Read the live Supabase state and print a one-page health report.
Usage:
python scripts/check_state.py
What it checks:
* each per-type challenges table — row count per team
* 1v1 tables — room / match / submission counts
* certificates — issued count
* user_completions — total + breakdown by category
"""
import os
import sys
from pathlib import Path
try:
from dotenv import load_dotenv
except ImportError:
print("Missing python-dotenv.")
sys.exit(1)
import httpx
ROOT = Path(__file__).resolve().parent.parent
ENV_PATH = ROOT / ".env"
def _load_env() -> None:
if ENV_PATH.exists():
load_dotenv(ENV_PATH)
def _headers() -> dict:
key = os.environ.get("SUPABASE_ANON_KEY", "")
return {
"apikey": key,
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
}
def _count(client: httpx.Client, table: str, team: str | None = None) -> int:
base = os.environ.get("SUPABASE_URL", "").rstrip("/")
q = f"select=id"
if team:
q += f"&team_role=eq.{team}"
r = client.get(f"{base}/rest/v1/{table}?{q}", headers=_headers())
if r.status_code != 200:
return -1
return len(r.json())
def _count_per_team(client: httpx.Client, table: str) -> dict[str, int]:
return {"red": _count(client, table, "red"), "blue": _count(client, table, "blue")}
def main() -> int:
_load_env()
base = os.environ.get("SUPABASE_URL", "").rstrip("/")
if not base:
sys.exit("SUPABASE_URL is not set.")
print(f"Supabase: {base}\n")
with httpx.Client(timeout=15) as client:
print("=== Challenge pools ===")
for table in [
"encryption_challenges",
"code_fixing_challenges",
"log_analysis_challenges",
"vulnerability_hunter_challenges",
]:
counts = _count_per_team(client, table)
print(f" {table:40s} red={counts['red']:4d} blue={counts['blue']:4d}")
# Migration 011 consistency check: every row's `module` must
# be the canonical challenge type for its table. The DB view
# `v_challenge_type_consistency` returns one row per table
# with `bad_rows = 0` when the contract is honoured.
print("\n=== Migration 011 — challenge type consistency ===")
try:
r = client.get(
f"{base}/rest/v1/v_challenge_type_consistency?select=table_name,bad_rows",
headers={k: v for k, v in _headers().items() if k != "Content-Type"},
)
if r.status_code == 200:
total_bad = 0
for row in r.json():
bad = row.get("bad_rows", 0)
total_bad += bad
marker = "OK " if bad == 0 else "FAIL"
print(f" [{marker}] {row['table_name']:35s} bad_rows={bad}")
if total_bad:
print(
"\n >>> run CyberArena/db/schema/011_challenge_type_normalization.sql"
)
else:
print(f" (view not accessible: HTTP {r.status_code})")
except Exception as e:
print(f" (skip: {e})")
print("\n=== 1v1 mode ===")
for table in ["onevone_rooms", "onevone_matches", "onevone_submissions", "onevone_players"]:
print(f" {table:40s} total={_count(client, table):4d}")
print("\n=== Certificates ===")
print(f" {'certificates':40s} total={_count(client, 'certificates'):4d}")
print("\n=== User completions ===")
print(f" {'user_completions':40s} total={_count(client, 'user_completions'):4d}")
# Per-category breakdown
for cat in ["crypto", "code-fixing", "log-analysis", "vulnerability-hunter"]:
r = client.get(
f"{base}/rest/v1/user_completions?category=eq.{cat}&select=id",
headers=_headers(),
)
if r.status_code == 200:
print(f" {cat:38s} {len(r.json()):4d}")
return 0
if __name__ == "__main__":
sys.exit(main())
|