494 lines
19 KiB
Python
494 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
SQLite Database Manager for Plants vs. Zombies Game (ai.qpid.pro)
|
|
Zero external dependencies. Manages users, game progress profiles, leaderboards, sessions, and bug reports.
|
|
"""
|
|
|
|
import sqlite3
|
|
import json
|
|
import os
|
|
import secrets
|
|
import hashlib
|
|
import hmac
|
|
from datetime import datetime, timedelta
|
|
|
|
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
DATA_DIR = os.path.join(BASE_DIR, "data")
|
|
DB_PATH = os.path.join(DATA_DIR, "pvz.db")
|
|
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
|
|
def get_connection():
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def hash_password(password, salt=None):
|
|
"""Secure server-side salted PBKDF2 password hashing."""
|
|
if not password:
|
|
return ""
|
|
salt = salt or secrets.token_hex(16)
|
|
iterations = 100_000
|
|
derived = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), iterations).hex()
|
|
return f"pbkdf2:sha256:{iterations}:{salt}:{derived}"
|
|
|
|
def verify_password(password, stored_hash):
|
|
"""Verifies a password against stored hash with backward compatibility for legacy hashes."""
|
|
if not password or not stored_hash:
|
|
return False
|
|
if stored_hash.startswith("pbkdf2:sha256:"):
|
|
try:
|
|
parts = stored_hash.split(":")
|
|
if len(parts) == 5:
|
|
iterations = int(parts[2])
|
|
salt = parts[3]
|
|
expected = parts[4]
|
|
derived = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), iterations).hex()
|
|
return hmac.compare_digest(derived, expected)
|
|
except Exception:
|
|
return False
|
|
# Backward compatibility for plain/legacy hashes
|
|
return hmac.compare_digest(str(password), str(stored_hash))
|
|
|
|
def init_db():
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
username TEXT UNIQUE NOT NULL COLLATE NOCASE,
|
|
password_hash TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
last_login TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS sessions (
|
|
token TEXT PRIMARY KEY,
|
|
user_id INTEGER NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
expires_at TIMESTAMP NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
)
|
|
""")
|
|
cursor.execute("CREATE INDEX IF NOT EXISTS idx_sessions_user_id ON sessions(user_id)")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS profiles (
|
|
user_id INTEGER PRIMARY KEY,
|
|
coins INTEGER DEFAULT 150,
|
|
level INTEGER DEFAULT 1,
|
|
max_unlocked_level INTEGER DEFAULT 1,
|
|
high_score INTEGER DEFAULT 0,
|
|
theme TEXT DEFAULT 'classic',
|
|
upgrades TEXT DEFAULT '{}',
|
|
materials TEXT DEFAULT '{}',
|
|
plant_levels TEXT DEFAULT '{}',
|
|
perks TEXT DEFAULT '{}',
|
|
zen_garden TEXT DEFAULT '{}',
|
|
tree_of_wisdom TEXT DEFAULT '{}',
|
|
achievements TEXT DEFAULT '{}',
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS leaderboards (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER,
|
|
player_name TEXT NOT NULL,
|
|
mode TEXT NOT NULL,
|
|
score INTEGER NOT NULL,
|
|
detail TEXT,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS bugs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
author TEXT NOT NULL,
|
|
category TEXT NOT NULL,
|
|
text TEXT NOT NULL,
|
|
status TEXT DEFAULT 'новый',
|
|
timestamp TEXT NOT NULL,
|
|
admin_reply TEXT
|
|
)
|
|
""")
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def create_session(user_id, duration_days=30):
|
|
"""Generates a secure 256-bit token and creates a session record."""
|
|
token = secrets.token_hex(32)
|
|
now = datetime.now()
|
|
created_at = now.strftime("%Y-%m-%d %H:%M:%S")
|
|
expires_at = (now + timedelta(days=duration_days)).strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"INSERT INTO sessions (token, user_id, created_at, expires_at) VALUES (?, ?, ?, ?)",
|
|
(token, user_id, created_at, expires_at)
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return token
|
|
|
|
def validate_session(token):
|
|
"""Validates session token. Returns user dict if valid, else None."""
|
|
if not token or not isinstance(token, str):
|
|
return None
|
|
token = token.strip()
|
|
if not token:
|
|
return None
|
|
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT u.id, u.username, u.created_at, u.last_login, s.expires_at, s.token
|
|
FROM sessions s
|
|
JOIN users u ON s.user_id = u.id
|
|
WHERE s.token = ?
|
|
""", (token,))
|
|
row = cursor.fetchone()
|
|
|
|
if not row:
|
|
conn.close()
|
|
return None
|
|
|
|
# Check expiration
|
|
try:
|
|
expires_at = datetime.strptime(row["expires_at"], "%Y-%m-%d %H:%M:%S")
|
|
if datetime.now() > expires_at:
|
|
cursor.execute("DELETE FROM sessions WHERE token = ?", (token,))
|
|
conn.commit()
|
|
conn.close()
|
|
return None
|
|
except Exception:
|
|
conn.close()
|
|
return None
|
|
|
|
user_info = {
|
|
"id": row["id"],
|
|
"user_id": row["id"],
|
|
"username": row["username"],
|
|
"createdAt": row["created_at"],
|
|
"lastLogin": row["last_login"],
|
|
"token": row["token"]
|
|
}
|
|
conn.close()
|
|
return user_info
|
|
|
|
def delete_session(token):
|
|
"""Deletes/invalidates a session by token."""
|
|
if not token:
|
|
return True
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM sessions WHERE token = ?", (token.strip(),))
|
|
conn.commit()
|
|
conn.close()
|
|
return True
|
|
|
|
def get_default_profile_dict():
|
|
return {
|
|
"coins": 150,
|
|
"level": 1,
|
|
"maxUnlockedLevel": 1,
|
|
"highScore": 0,
|
|
"theme": "classic",
|
|
"upgrades": {
|
|
"extraSlot": False,
|
|
"hasRake": False,
|
|
"unlockedPlants": ["sunflower", "peashooter", "wallnut"],
|
|
"selectedDeck": ["sunflower", "peashooter", "wallnut"]
|
|
},
|
|
"unlockedPlants": ["sunflower", "peashooter", "wallnut"],
|
|
"selectedDeck": ["sunflower", "peashooter", "wallnut"],
|
|
"materials": {"bioHumus": 10, "scrapMetal": 5, "sunEssence": 20, "crystals": 0},
|
|
"plantLevels": {"peashooter": 1, "sunflower": 1, "wallnut": 1, "potatomine": 1, "repeater": 1, "snowpea": 1, "chomper": 1},
|
|
"perks": {"magnetism": 0, "startSun": 0, "fastCool": 0, "hardyRoots": 0, "luckyDrop": 0},
|
|
"zenGarden": {
|
|
"pots": [
|
|
{"id": 1, "plantType": "sunflower", "stage": 2, "watered": True, "fertilized": True, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0},
|
|
{"id": 2, "plantType": "peashooter", "stage": 1, "watered": True, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0},
|
|
{"id": 3, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0},
|
|
{"id": 4, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0},
|
|
{"id": 5, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0},
|
|
{"id": 6, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0},
|
|
{"id": 7, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0},
|
|
{"id": 8, "plantType": None, "stage": 0, "watered": False, "fertilized": False, "needWaterTime": 0, "matureTime": 0, "dropsCollected": 0}
|
|
],
|
|
"fertilizerCount": 5,
|
|
"chocolateCount": 2,
|
|
"snail": {"name": "Стинки", "active": False, "speed": 60, "boostTimer": 0, "x": 50, "direction": 1},
|
|
"lastVisit": int(datetime.now().timestamp() * 1000)
|
|
},
|
|
"treeOfWisdom": {"height": 0, "cheats": {}},
|
|
"achievements": {}
|
|
}
|
|
|
|
def get_user_by_username(username):
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM users WHERE username = ? COLLATE NOCASE", (username.strip(),))
|
|
user = cursor.fetchone()
|
|
conn.close()
|
|
return user
|
|
|
|
def register_user(username, password_hash, initial_data=None):
|
|
username = username.strip()
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
try:
|
|
now = datetime.now()
|
|
now_str = now.strftime("%Y-%m-%d %H:%M:%S")
|
|
secure_hash = hash_password(password_hash)
|
|
cursor.execute(
|
|
"INSERT INTO users (username, password_hash, created_at, last_login) VALUES (?, ?, ?, ?)",
|
|
(username, secure_hash, now_str, now_str)
|
|
)
|
|
user_id = cursor.lastrowid
|
|
|
|
profile_data = get_default_profile_dict()
|
|
if initial_data and isinstance(initial_data, dict):
|
|
for k, v in initial_data.items():
|
|
if v is not None:
|
|
profile_data[k] = v
|
|
|
|
cursor.execute("""
|
|
INSERT INTO profiles (
|
|
user_id, coins, level, max_unlocked_level, high_score, theme,
|
|
upgrades, materials, plant_levels, perks, zen_garden, tree_of_wisdom, achievements, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
user_id,
|
|
profile_data.get("coins", 150),
|
|
profile_data.get("level", 1),
|
|
profile_data.get("maxUnlockedLevel", 1),
|
|
profile_data.get("highScore", 0),
|
|
profile_data.get("theme", "classic"),
|
|
json.dumps(profile_data.get("upgrades", {})),
|
|
json.dumps(profile_data.get("materials", {})),
|
|
json.dumps(profile_data.get("plantLevels", {})),
|
|
json.dumps(profile_data.get("perks", {})),
|
|
json.dumps(profile_data.get("zenGarden", {})),
|
|
json.dumps(profile_data.get("treeOfWisdom", {})),
|
|
json.dumps(profile_data.get("achievements", {})),
|
|
now_str
|
|
))
|
|
|
|
# Create session token
|
|
token = secrets.token_hex(32)
|
|
expires_at = (now + timedelta(days=30)).strftime("%Y-%m-%d %H:%M:%S")
|
|
cursor.execute(
|
|
"INSERT INTO sessions (token, user_id, created_at, expires_at) VALUES (?, ?, ?, ?)",
|
|
(token, user_id, now_str, expires_at)
|
|
)
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
return {"success": True, "token": token, "user_id": user_id, "username": username, "data": profile_data}
|
|
except sqlite3.IntegrityError:
|
|
conn.close()
|
|
return {"success": False, "message": "Пользователь с таким именем уже существует"}
|
|
except Exception as e:
|
|
conn.close()
|
|
return {"success": False, "message": str(e)}
|
|
|
|
def authenticate_user(username, password_hash):
|
|
username = username.strip()
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("SELECT * FROM users WHERE username = ? COLLATE NOCASE", (username,))
|
|
user = cursor.fetchone()
|
|
if not user:
|
|
conn.close()
|
|
return None
|
|
|
|
if not verify_password(password_hash, user["password_hash"]):
|
|
conn.close()
|
|
return {"success": False, "message": "Неверный пароль"}
|
|
|
|
now = datetime.now()
|
|
now_str = now.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
# Seamless upgrade from legacy plaintext/djb2 hash to salted PBKDF2
|
|
if not user["password_hash"].startswith("pbkdf2:"):
|
|
new_secure_hash = hash_password(password_hash)
|
|
cursor.execute("UPDATE users SET password_hash = ?, last_login = ? WHERE id = ?", (new_secure_hash, now_str, user["id"]))
|
|
else:
|
|
cursor.execute("UPDATE users SET last_login = ? WHERE id = ?", (now_str, user["id"]))
|
|
|
|
# Create session token
|
|
token = secrets.token_hex(32)
|
|
expires_at = (now + timedelta(days=30)).strftime("%Y-%m-%d %H:%M:%S")
|
|
cursor.execute(
|
|
"INSERT INTO sessions (token, user_id, created_at, expires_at) VALUES (?, ?, ?, ?)",
|
|
(token, user["id"], now_str, expires_at)
|
|
)
|
|
conn.commit()
|
|
|
|
# Fetch profile
|
|
cursor.execute("SELECT * FROM profiles WHERE user_id = ?", (user["id"],))
|
|
p = cursor.fetchone()
|
|
conn.close()
|
|
|
|
if not p:
|
|
profile_dict = get_default_profile_dict()
|
|
else:
|
|
upg_data = json.loads(p["upgrades"] or "{}")
|
|
unlocked_plants = upg_data.get("unlockedPlants", ["sunflower", "peashooter", "wallnut"])
|
|
selected_deck = upg_data.get("selectedDeck", ["sunflower", "peashooter", "wallnut"])
|
|
|
|
profile_dict = {
|
|
"coins": p["coins"],
|
|
"level": p["level"],
|
|
"maxUnlockedLevel": p["max_unlocked_level"],
|
|
"highScore": p["high_score"],
|
|
"theme": p["theme"],
|
|
"upgrades": upg_data,
|
|
"unlockedPlants": unlocked_plants,
|
|
"selectedDeck": selected_deck,
|
|
"materials": json.loads(p["materials"] or "{}"),
|
|
"plantLevels": json.loads(p["plant_levels"] or "{}"),
|
|
"perks": json.loads(p["perks"] or "{}"),
|
|
"zenGarden": json.loads(p["zen_garden"] or "{}"),
|
|
"treeOfWisdom": json.loads(p["tree_of_wisdom"] or "{}"),
|
|
"achievements": json.loads(p["achievements"] or "{}")
|
|
}
|
|
|
|
return {"success": True, "token": token, "user_id": user["id"], "username": user["username"], "data": profile_dict}
|
|
|
|
def save_user_profile(username, data, token=None):
|
|
username = username.strip()
|
|
|
|
# If token is provided, verify it belongs to username
|
|
if token is not None:
|
|
session_user = validate_session(token)
|
|
if not session_user or session_user["username"].lower() != username.lower():
|
|
return {"success": False, "message": "Недействительный токен авторизации или доступ запрещен"}
|
|
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("SELECT id FROM users WHERE username = ? COLLATE NOCASE", (username,))
|
|
user = cursor.fetchone()
|
|
if not user:
|
|
conn.close()
|
|
return {"success": False, "message": "Пользователь не найден"}
|
|
|
|
user_id = user["id"]
|
|
cursor.execute("SELECT * FROM profiles WHERE user_id = ?", (user_id,))
|
|
current = cursor.fetchone()
|
|
|
|
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
coins = data.get("coins", current["coins"] if current else 150)
|
|
level = data.get("level", current["level"] if current else 1)
|
|
max_unlocked = data.get("maxUnlockedLevel", current["max_unlocked_level"] if current else 1)
|
|
high_score = data.get("highScore", current["high_score"] if current else 0)
|
|
theme = data.get("theme", current["theme"] if current else "classic")
|
|
|
|
# Merge JSON dictionaries
|
|
def merge_json(key, default):
|
|
cur_obj = json.loads(current[key] or "{}") if current and current[key] else default
|
|
new_val = data.get(key)
|
|
if new_val is not None and isinstance(new_val, dict):
|
|
cur_obj.update(new_val)
|
|
return json.dumps(cur_obj)
|
|
|
|
upg_obj = json.loads(current["upgrades"] or "{}") if current and current["upgrades"] else {}
|
|
if "upgrades" in data and isinstance(data["upgrades"], dict):
|
|
upg_obj.update(data["upgrades"])
|
|
if "unlockedPlants" in data and isinstance(data["unlockedPlants"], list):
|
|
upg_obj["unlockedPlants"] = data["unlockedPlants"]
|
|
if "selectedDeck" in data and isinstance(data["selectedDeck"], list):
|
|
upg_obj["selectedDeck"] = data["selectedDeck"]
|
|
upgrades_json = json.dumps(upg_obj)
|
|
|
|
materials_json = merge_json("materials", {})
|
|
plant_levels_json = merge_json("plant_levels", {}) if "plant_levels" in data else (json.dumps(data["plantLevels"]) if "plantLevels" in data else (current["plant_levels"] if current else "{}"))
|
|
perks_json = merge_json("perks", {})
|
|
zen_garden_json = json.dumps(data["zenGarden"]) if "zenGarden" in data else (current["zen_garden"] if current else "{}")
|
|
tree_json = json.dumps(data["treeOfWisdom"]) if "treeOfWisdom" in data else (current["tree_of_wisdom"] if current else "{}")
|
|
achievements_json = json.dumps(data["achievements"]) if "achievements" in data else (current["achievements"] if current else "{}")
|
|
|
|
if current:
|
|
cursor.execute("""
|
|
UPDATE profiles SET
|
|
coins = ?, level = ?, max_unlocked_level = ?, high_score = ?, theme = ?,
|
|
upgrades = ?, materials = ?, plant_levels = ?, perks = ?,
|
|
zen_garden = ?, tree_of_wisdom = ?, achievements = ?, updated_at = ?
|
|
WHERE user_id = ?
|
|
""", (
|
|
coins, level, max_unlocked, high_score, theme,
|
|
upgrades_json, materials_json, plant_levels_json, perks_json,
|
|
zen_garden_json, tree_json, achievements_json, now_str,
|
|
user_id
|
|
))
|
|
else:
|
|
cursor.execute("""
|
|
INSERT INTO profiles (
|
|
user_id, coins, level, max_unlocked_level, high_score, theme,
|
|
upgrades, materials, plant_levels, perks, zen_garden, tree_of_wisdom, achievements, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
user_id, coins, level, max_unlocked, high_score, theme,
|
|
upgrades_json, materials_json, plant_levels_json, perks_json,
|
|
zen_garden_json, tree_json, achievements_json, now_str
|
|
))
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
return {"success": True}
|
|
|
|
def list_all_users():
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT
|
|
u.id, u.username, u.created_at, u.last_login,
|
|
p.coins, p.level, p.max_unlocked_level, p.high_score, p.theme,
|
|
p.materials, p.plant_levels, p.perks, p.upgrades, p.zen_garden, p.tree_of_wisdom, p.achievements
|
|
FROM users u
|
|
LEFT JOIN profiles p ON u.id = p.user_id
|
|
ORDER BY u.id ASC
|
|
""")
|
|
rows = cursor.fetchall()
|
|
conn.close()
|
|
|
|
result = []
|
|
for r in rows:
|
|
result.append({
|
|
"id": r["id"],
|
|
"username": r["username"],
|
|
"createdAt": r["created_at"],
|
|
"lastLogin": r["last_login"],
|
|
"coins": r["coins"] if r["coins"] is not None else 150,
|
|
"level": r["level"] if r["level"] is not None else 1,
|
|
"maxUnlockedLevel": r["max_unlocked_level"] if r["max_unlocked_level"] is not None else 1,
|
|
"highScore": r["high_score"] if r["high_score"] is not None else 0,
|
|
"theme": r["theme"] or "classic",
|
|
"materials": json.loads(r["materials"] or "{}"),
|
|
"plantLevels": json.loads(r["plant_levels"] or "{}"),
|
|
"perks": json.loads(r["perks"] or "{}"),
|
|
"upgrades": json.loads(r["upgrades"] or "{}"),
|
|
"zenGarden": json.loads(r["zen_garden"] or "{}"),
|
|
"treeOfWisdom": json.loads(r["tree_of_wisdom"] or "{}"),
|
|
"achievements": json.loads(r["achievements"] or "{}")
|
|
})
|
|
return result
|
|
|
|
if __name__ == "__main__":
|
|
init_db()
|
|
users = list_all_users()
|
|
print(f"Database initialized. Total registered users: {len(users)}")
|