72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
import sqlite3
|
|
from typing import Optional, Dict, Any
|
|
|
|
DB_PATH = "data/db.sqlite3"
|
|
|
|
def init_db():
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
CREATE TABLE IF NOT EXISTS players (
|
|
user_id INTEGER PRIMARY KEY,
|
|
balance REAL DEFAULT 10000,
|
|
servers INTEGER DEFAULT 1,
|
|
cables INTEGER DEFAULT 0,
|
|
coverage_customers INTEGER DEFAULT 500,
|
|
max_customers INTEGER DEFAULT 500,
|
|
speed INTEGER DEFAULT 10,
|
|
rating INTEGER DEFAULT 0,
|
|
rating_threshold INTEGER DEFAULT 100,
|
|
next_build_end_ts INTEGER DEFAULT 0,
|
|
last_login_ts INTEGER DEFAULT 0,
|
|
total_income REAL DEFAULT 0,
|
|
infrastructure_level INTEGER DEFAULT 1,
|
|
power_capacity REAL DEFAULT 5.0,
|
|
power_usage REAL DEFAULT 0.0,
|
|
outage_end_ts INTEGER DEFAULT 0,
|
|
backup_channel_active INTEGER DEFAULT 0,
|
|
network_channel_level INTEGER DEFAULT 1,
|
|
channel_capacity INTEGER DEFAULT 10
|
|
)
|
|
""")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def get_player(user_id: int) -> Optional[Dict[str, Any]]:
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT * FROM players WHERE user_id = ?", (user_id,))
|
|
row = cur.fetchone()
|
|
conn.close()
|
|
if not row:
|
|
return None
|
|
keys = [
|
|
"user_id", "balance", "servers", "cables", "coverage_customers",
|
|
"max_customers", "speed", "rating", "rating_threshold",
|
|
"next_build_end_ts", "last_login_ts", "total_income", "infrastructure_level",
|
|
"power_capacity", "power_usage", "outage_end_ts", "backup_channel_active",
|
|
"network_channel_level", "channel_capacity"
|
|
]
|
|
return dict(zip(keys, row))
|
|
|
|
def update_player(user_id: int, updates: Dict[str, Any]):
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cur = conn.cursor()
|
|
fields = ", ".join(f"{k} = ?" for k in updates.keys())
|
|
values = list(updates.values()) + [user_id]
|
|
cur.execute(f"UPDATE players SET {fields} WHERE user_id = ?", values)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def create_player_if_not_exists(user_id: int):
|
|
player = get_player(user_id)
|
|
if not player:
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
INSERT INTO players (user_id, balance, servers, cables, coverage_customers, max_customers, speed, power_capacity, channel_capacity)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (user_id, INITIAL_BALANCE, INITIAL_SERVERS, 0, INITIAL_COVERAGE_CUSTOMERS,
|
|
INITIAL_MAX_CUSTOMERS, INITIAL_SPEED, INITIAL_POWER_CAPACITY, INITIAL_NETWORK_CHANNEL_CAPACITY))
|
|
conn.commit()
|
|
conn.close() |