93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
# database.py — работа с SQLite
|
||
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 player:
|
||
return
|
||
|
||
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, network_channel_level, channel_capacity)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
user_id,
|
||
10_000, # balance
|
||
1, # servers
|
||
0, # cables
|
||
500, # coverage_customers
|
||
500, # max_customers
|
||
10, # speed
|
||
5.0, # power_capacity
|
||
1, # network_channel_level
|
||
10 # channel_capacity
|
||
))
|
||
conn.commit()
|
||
conn.close() |