init_db.py•3.3 kB
#!/usr/bin/env python3
"""Initialize the Chess MXCP database tables."""
import duckdb
# Connect to the database
conn = duckdb.connect('data/db-default.duckdb')
# Define all tables
tables = [
"""CREATE TABLE IF NOT EXISTS chess_api_cache (
cache_key VARCHAR PRIMARY KEY,
endpoint VARCHAR NOT NULL,
response_data JSON NOT NULL,
cached_at TIMESTAMP NOT NULL,
expires_at TIMESTAMP NOT NULL
)""",
"""CREATE TABLE IF NOT EXISTS chess_games_cache (
game_id VARCHAR PRIMARY KEY,
username VARCHAR NOT NULL,
game_data JSON NOT NULL,
pgn_data TEXT,
cached_at TIMESTAMP NOT NULL,
year INTEGER,
month INTEGER
)""",
"""CREATE TABLE IF NOT EXISTS player_profile_views (
id INTEGER PRIMARY KEY,
username VARCHAR NOT NULL,
viewed_at TIMESTAMP NOT NULL,
follower_count INTEGER,
country VARCHAR
)""",
"""CREATE TABLE IF NOT EXISTS rating_history (
id INTEGER PRIMARY KEY,
username VARCHAR NOT NULL,
game_type VARCHAR NOT NULL,
rating INTEGER NOT NULL,
games_played INTEGER,
recorded_at TIMESTAMP NOT NULL
)""",
"""CREATE TABLE IF NOT EXISTS player_activity (
id INTEGER PRIMARY KEY,
username VARCHAR NOT NULL,
game_count INTEGER NOT NULL,
checked_at TIMESTAMP NOT NULL
)""",
"""CREATE TABLE IF NOT EXISTS titled_players_snapshot (
title VARCHAR NOT NULL,
player_count INTEGER NOT NULL,
players_list JSON,
snapshot_date DATE NOT NULL,
PRIMARY KEY (title, snapshot_date)
)""",
"""CREATE TABLE IF NOT EXISTS club_statistics (
id INTEGER PRIMARY KEY,
club_id VARCHAR NOT NULL,
club_name VARCHAR,
member_count INTEGER,
checked_at TIMESTAMP NOT NULL
)""",
"""CREATE TABLE IF NOT EXISTS club_member_snapshots (
club_id VARCHAR NOT NULL,
member_count INTEGER NOT NULL,
members_list JSON,
snapshot_date DATE NOT NULL,
PRIMARY KEY (club_id, snapshot_date)
)""",
"""CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY,
timestamp TIMESTAMP NOT NULL,
event_type VARCHAR NOT NULL,
endpoint VARCHAR,
status_code INTEGER,
username VARCHAR
)"""
]
# Create tables
for table_sql in tables:
table_name = table_sql.split()[5] # Extract table name
print(f"Creating table: {table_name}")
conn.execute(table_sql)
# Create indexes
indexes = [
"CREATE INDEX IF NOT EXISTS idx_expires ON chess_api_cache(expires_at)",
"CREATE INDEX IF NOT EXISTS idx_player_date ON chess_games_cache(username, year, month)",
"CREATE INDEX IF NOT EXISTS idx_cached ON chess_games_cache(cached_at)"
]
for index_sql in indexes:
index_name = index_sql.split()[5] # Extract index name
print(f"Creating index: {index_name}")
try:
conn.execute(index_sql)
except Exception as e:
print(f" Warning: {e}")
# Show tables
print("\nCreated tables:")
tables = conn.execute("SHOW TABLES").fetchall()
for table in tables:
print(f" - {table[0]}")
conn.close()
print("\nDatabase initialization complete!")