from __future__ import annotations
import hashlib
import hmac
import math
import random
import re
import secrets
import sqlite3
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional
from jinja2 import Environment, FileSystemLoader, select_autoescape
APP_DIR = Path(__file__).resolve().parent
REPO_ROOT = APP_DIR.parent.parent
if (REPO_ROOT / "micropie.py").exists() and str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from micropie import App
DB_PATH = APP_DIR / "torn_city.sqlite3"
TEMPLATES_DIR = APP_DIR / "templates"
USERNAME_RE = re.compile(r"^[a-z0-9_]{3,20}$")
PASSWORD_ITERATIONS = 200_000
ITEM_SEEDS: List[Dict[str, Any]] = [
{
"slug": "rusty_knife",
"name": "Rusty Knife",
"category": "weapons",
"subcategory": "melee",
"description": "Barely held together, but still better than fists.",
"buy_price": 300,
"sell_price": 150,
"tradable": 1,
"stackable": 1,
"consumable": 0,
"weapon_damage_min": 8,
"weapon_damage_max": 16,
"dexterity_bonus": 1,
},
{
"slug": "baseball_bat",
"name": "Baseball Bat",
"category": "weapons",
"subcategory": "melee",
"description": "Reliable blunt force for back-alley disputes.",
"buy_price": 700,
"sell_price": 350,
"tradable": 1,
"stackable": 1,
"consumable": 0,
"weapon_damage_min": 12,
"weapon_damage_max": 24,
"strength_bonus": 2,
},
{
"slug": "street_pistol",
"name": "Street Pistol",
"category": "weapons",
"subcategory": "firearm",
"description": "Cheap, loud, and enough to make mugging serious.",
"buy_price": 1800,
"sell_price": 900,
"tradable": 1,
"stackable": 1,
"consumable": 0,
"weapon_damage_min": 18,
"weapon_damage_max": 32,
"speed_bonus": 1,
},
{
"slug": "leather_vest",
"name": "Leather Vest",
"category": "armor",
"subcategory": "light",
"description": "Looks cool and absorbs a little punishment.",
"buy_price": 500,
"sell_price": 250,
"tradable": 1,
"stackable": 1,
"consumable": 0,
"defense_bonus": 6,
"armor_reduction": 4,
},
{
"slug": "kevlar_vest",
"name": "Kevlar Vest",
"category": "armor",
"subcategory": "medium",
"description": "The first armor people actually respect.",
"buy_price": 1800,
"sell_price": 900,
"tradable": 1,
"stackable": 1,
"consumable": 0,
"defense_bonus": 12,
"armor_reduction": 8,
},
{
"slug": "adrenaline_shot",
"name": "Adrenaline Shot",
"category": "drugs",
"subcategory": "booster",
"description": "A dirty boost that gets you moving again.",
"buy_price": 350,
"sell_price": 100,
"tradable": 1,
"stackable": 1,
"consumable": 1,
"energy_gain": 18,
"nerve_gain": 2,
"happy_gain": 2,
},
{
"slug": "nerve_tonic",
"name": "Nerve Tonic",
"category": "drugs",
"subcategory": "booster",
"description": "Burns on the way down, sharpens you on the way up.",
"buy_price": 520,
"sell_price": 150,
"tradable": 1,
"stackable": 1,
"consumable": 1,
"nerve_gain": 5,
"happy_gain": 3,
},
{
"slug": "painkillers",
"name": "Painkillers",
"category": "drugs",
"subcategory": "booster",
"description": "Takes the edge off a rough night.",
"buy_price": 220,
"sell_price": 75,
"tradable": 1,
"stackable": 1,
"consumable": 1,
"life_gain": 10,
"happy_gain": 4,
},
{
"slug": "bandage",
"name": "Bandage",
"category": "medical",
"subcategory": "medical",
"description": "A cheap way to patch yourself back together.",
"buy_price": 120,
"sell_price": 40,
"tradable": 1,
"stackable": 1,
"consumable": 1,
"life_gain": 20,
},
{
"slug": "first_aid_kit",
"name": "First Aid Kit",
"category": "medical",
"subcategory": "medical",
"description": "A proper heal for when the mug goes badly.",
"buy_price": 420,
"sell_price": 150,
"tradable": 1,
"stackable": 1,
"consumable": 1,
"life_gain": 55,
},
{
"slug": "studio_apartment",
"name": "Studio Apartment",
"category": "homes",
"subcategory": "property",
"description": "Cramped, but at least it is yours.",
"buy_price": 2500,
"sell_price": 1500,
"tradable": 1,
"stackable": 1,
"consumable": 0,
"home_life_bonus": 10,
"home_happy_bonus": 5,
},
{
"slug": "suburban_house",
"name": "Suburban House",
"category": "homes",
"subcategory": "property",
"description": "A safer place to recover between jobs and crimes.",
"buy_price": 6500,
"sell_price": 4000,
"tradable": 1,
"stackable": 1,
"consumable": 0,
"home_life_bonus": 25,
"home_happy_bonus": 10,
},
{
"slug": "lockpick_set",
"name": "Lockpick Set",
"category": "crime_tools",
"subcategory": "tool",
"description": "Boosts confidence during low-rent break-ins.",
"buy_price": 280,
"sell_price": 120,
"tradable": 1,
"stackable": 1,
"consumable": 0,
"dexterity_bonus": 1,
},
{
"slug": "crowbar",
"name": "Crowbar",
"category": "crime_tools",
"subcategory": "tool",
"description": "Versatile, intimidating, and useful on stubborn doors.",
"buy_price": 450,
"sell_price": 180,
"tradable": 1,
"stackable": 1,
"consumable": 0,
"strength_bonus": 1,
},
{
"slug": "gold_watch",
"name": "Gold Watch",
"category": "misc",
"subcategory": "valuable",
"description": "Purely valuable. Great for fences and the player market.",
"buy_price": 0,
"sell_price": 280,
"tradable": 1,
"stackable": 1,
"consumable": 0,
},
{
"slug": "gemstones",
"name": "Loose Gemstones",
"category": "misc",
"subcategory": "valuable",
"description": "Small, rare, and easy to move for cash.",
"buy_price": 0,
"sell_price": 450,
"tradable": 1,
"stackable": 1,
"consumable": 0,
},
]
JOB_SEEDS: List[Dict[str, Any]] = [
{
"name": "Grocer",
"description": "Unloading stock and handling the till.",
"required_level": 1,
"base_pay": 140,
"cooldown_seconds": 3600,
},
{
"name": "Clerk",
"description": "Paperwork, errands, and long hours.",
"required_level": 2,
"base_pay": 240,
"cooldown_seconds": 3600,
},
{
"name": "Delivery Driver",
"description": "Fast cash if you can stay on schedule.",
"required_level": 4,
"base_pay": 380,
"cooldown_seconds": 3600,
},
{
"name": "Security Guard",
"description": "Long shifts, but the city notices the badge.",
"required_level": 6,
"base_pay": 600,
"cooldown_seconds": 3600,
},
{
"name": "Manager",
"description": "Better pay for people who survived the grind.",
"required_level": 8,
"base_pay": 900,
"cooldown_seconds": 3600,
},
]
CRIME_SEEDS: List[Dict[str, Any]] = [
{
"tier": "low",
"name": "Pickpocket a Tourist",
"description": "Easy money if your hands stay quick.",
"nerve_cost": 4,
"success_rate": 0.84,
"money_min": 80,
"money_max": 170,
"xp_reward": 18,
"jail_seconds": 300,
"drop_item_slug": "",
"drop_rate": 0.0,
},
{
"tier": "low",
"name": "Shoplift Electronics",
"description": "Grab something small before staff notices.",
"nerve_cost": 5,
"success_rate": 0.79,
"money_min": 100,
"money_max": 220,
"xp_reward": 22,
"jail_seconds": 360,
"drop_item_slug": "painkillers",
"drop_rate": 0.18,
},
{
"tier": "low",
"name": "Shed Break-In",
"description": "Tools, parts, or maybe a quick arrest.",
"nerve_cost": 6,
"success_rate": 0.75,
"money_min": 120,
"money_max": 260,
"xp_reward": 26,
"jail_seconds": 420,
"drop_item_slug": "lockpick_set",
"drop_rate": 0.14,
},
{
"tier": "mid",
"name": "Mug a Tourist Strip",
"description": "Higher stakes, louder consequences.",
"nerve_cost": 8,
"success_rate": 0.64,
"money_min": 220,
"money_max": 450,
"xp_reward": 40,
"jail_seconds": 540,
"drop_item_slug": "gold_watch",
"drop_rate": 0.14,
},
{
"tier": "mid",
"name": "Warehouse Swipe",
"description": "Slip in, grab inventory, disappear fast.",
"nerve_cost": 9,
"success_rate": 0.58,
"money_min": 280,
"money_max": 540,
"xp_reward": 46,
"jail_seconds": 660,
"drop_item_slug": "bandage",
"drop_rate": 0.22,
},
{
"tier": "mid",
"name": "Car Parts Lift",
"description": "Useful parts if you can beat the response time.",
"nerve_cost": 10,
"success_rate": 0.54,
"money_min": 320,
"money_max": 620,
"xp_reward": 52,
"jail_seconds": 720,
"drop_item_slug": "crowbar",
"drop_rate": 0.12,
},
{
"tier": "high",
"name": "Nightclub Shakeout",
"description": "Fast money, crowded room, bad odds.",
"nerve_cost": 12,
"success_rate": 0.40,
"money_min": 500,
"money_max": 950,
"xp_reward": 76,
"jail_seconds": 900,
"drop_item_slug": "adrenaline_shot",
"drop_rate": 0.16,
},
{
"tier": "high",
"name": "Electronics Heist",
"description": "Big haul if the alarm stays quiet.",
"nerve_cost": 14,
"success_rate": 0.34,
"money_min": 650,
"money_max": 1250,
"xp_reward": 90,
"jail_seconds": 1080,
"drop_item_slug": "gemstones",
"drop_rate": 0.14,
},
{
"tier": "high",
"name": "Armored Van Snatch",
"description": "Stupidly risky, exactly why people try it.",
"nerve_cost": 16,
"success_rate": 0.28,
"money_min": 900,
"money_max": 1800,
"xp_reward": 120,
"jail_seconds": 1260,
"drop_item_slug": "first_aid_kit",
"drop_rate": 0.18,
},
]
SHOP_CONFIG = {
"weapons": {"label": "Weapon Shop", "categories": ["weapons"]},
"armor": {"label": "Armor Shop", "categories": ["armor"]},
"drugs": {"label": "Drug Shop", "categories": ["drugs"]},
"medical": {"label": "Medical Shop", "categories": ["medical"]},
"homes": {"label": "Property Broker", "categories": ["homes"]},
"misc": {"label": "Pawn & Tools", "categories": ["crime_tools", "misc"]},
}
def utcnow() -> datetime:
return datetime.now(timezone.utc).replace(microsecond=0)
def iso_now() -> str:
return utcnow().isoformat()
def parse_dt(value: Optional[str]) -> Optional[datetime]:
if not value:
return None
return datetime.fromisoformat(value)
def clamp(value: int, low: int, high: int) -> int:
return max(low, min(high, value))
def level_from_xp(xp: int) -> int:
return 1 + int(math.sqrt(max(0, xp) / 100))
def format_duration(seconds: int) -> str:
seconds = max(0, int(seconds))
hours, rem = divmod(seconds, 3600)
minutes, secs = divmod(rem, 60)
parts: List[str] = []
if hours:
parts.append(f"{hours}h")
if minutes or hours:
parts.append(f"{minutes}m")
parts.append(f"{secs}s")
return " ".join(parts)
def format_ts(value: Optional[str]) -> str:
dt = parse_dt(value)
if not dt:
return "-"
return dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
def normalize_username(username: str) -> str:
return username.strip().lower()
def hash_password(password: str) -> str:
salt = secrets.token_bytes(16)
digest = hashlib.pbkdf2_hmac(
"sha256", password.encode("utf-8"), salt, PASSWORD_ITERATIONS
)
return f"{salt.hex()}${digest.hex()}"
def verify_password(password: str, encoded: str) -> bool:
try:
salt_hex, digest_hex = encoded.split("$", 1)
except ValueError:
return False
expected = bytes.fromhex(digest_hex)
actual = hashlib.pbkdf2_hmac(
"sha256", password.encode("utf-8"), bytes.fromhex(salt_hex), PASSWORD_ITERATIONS
)
return hmac.compare_digest(actual, expected)
class GameError(Exception):
pass
class GameStore:
slot_fields = {
"equipped_weapon_item_id": "weapons",
"equipped_armor_item_id": "armor",
"active_home_item_id": "homes",
}
regen_rules = {
"energy": {"timestamp": "energy_updated_at", "interval": 300, "cap": 100},
"nerve": {"timestamp": "nerve_updated_at", "interval": 300, "cap": 50},
"happy": {"timestamp": "happy_updated_at", "interval": 900, "cap": 100},
}
def __init__(self, db_path: Path) -> None:
self.db_path = str(db_path)
def connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
return conn
def init_db(self) -> None:
with self.connect() as conn:
conn.execute("PRAGMA journal_mode = WAL")
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
level INTEGER NOT NULL DEFAULT 1,
xp INTEGER NOT NULL DEFAULT 0,
wallet_money INTEGER NOT NULL DEFAULT 0,
bank_money INTEGER NOT NULL DEFAULT 0,
energy INTEGER NOT NULL DEFAULT 100,
nerve INTEGER NOT NULL DEFAULT 50,
happy INTEGER NOT NULL DEFAULT 100,
life INTEGER NOT NULL DEFAULT 100,
max_life INTEGER NOT NULL DEFAULT 100,
strength INTEGER NOT NULL DEFAULT 10,
speed INTEGER NOT NULL DEFAULT 10,
defense INTEGER NOT NULL DEFAULT 10,
dexterity INTEGER NOT NULL DEFAULT 10,
hospital_until TEXT,
jail_until TEXT,
job_id INTEGER,
job_cooldown_until TEXT,
equipped_weapon_item_id INTEGER,
equipped_armor_item_id INTEGER,
active_home_item_id INTEGER,
energy_updated_at TEXT NOT NULL,
nerve_updated_at TEXT NOT NULL,
happy_updated_at TEXT NOT NULL,
life_updated_at TEXT NOT NULL,
created_at TEXT NOT NULL,
last_login_at TEXT
);
CREATE TABLE IF NOT EXISTS item_catalog (
id INTEGER PRIMARY KEY AUTOINCREMENT,
slug TEXT NOT NULL UNIQUE,
name TEXT NOT NULL,
category TEXT NOT NULL,
subcategory TEXT NOT NULL DEFAULT '',
description TEXT NOT NULL DEFAULT '',
buy_price INTEGER NOT NULL DEFAULT 0,
sell_price INTEGER NOT NULL DEFAULT 0,
tradable INTEGER NOT NULL DEFAULT 1,
stackable INTEGER NOT NULL DEFAULT 1,
consumable INTEGER NOT NULL DEFAULT 0,
strength_bonus INTEGER NOT NULL DEFAULT 0,
speed_bonus INTEGER NOT NULL DEFAULT 0,
defense_bonus INTEGER NOT NULL DEFAULT 0,
dexterity_bonus INTEGER NOT NULL DEFAULT 0,
life_bonus INTEGER NOT NULL DEFAULT 0,
energy_gain INTEGER NOT NULL DEFAULT 0,
nerve_gain INTEGER NOT NULL DEFAULT 0,
happy_gain INTEGER NOT NULL DEFAULT 0,
life_gain INTEGER NOT NULL DEFAULT 0,
weapon_damage_min INTEGER NOT NULL DEFAULT 0,
weapon_damage_max INTEGER NOT NULL DEFAULT 0,
armor_reduction INTEGER NOT NULL DEFAULT 0,
home_life_bonus INTEGER NOT NULL DEFAULT 0,
home_happy_bonus INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS user_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
item_id INTEGER NOT NULL,
quantity INTEGER NOT NULL,
created_at TEXT NOT NULL,
UNIQUE(user_id, item_id),
FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY(item_id) REFERENCES item_catalog(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
description TEXT NOT NULL,
required_level INTEGER NOT NULL,
base_pay INTEGER NOT NULL,
cooldown_seconds INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS crime_catalog (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tier TEXT NOT NULL,
name TEXT NOT NULL UNIQUE,
description TEXT NOT NULL,
nerve_cost INTEGER NOT NULL,
success_rate REAL NOT NULL,
money_min INTEGER NOT NULL,
money_max INTEGER NOT NULL,
xp_reward INTEGER NOT NULL,
jail_seconds INTEGER NOT NULL,
drop_item_slug TEXT NOT NULL DEFAULT '',
drop_rate REAL NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS market_listings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
seller_user_id INTEGER NOT NULL,
item_id INTEGER NOT NULL,
quantity INTEGER NOT NULL,
price_each INTEGER NOT NULL,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(seller_user_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY(item_id) REFERENCES item_catalog(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS combat_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
attacker_id INTEGER NOT NULL,
defender_id INTEGER NOT NULL,
result TEXT NOT NULL,
action_type TEXT NOT NULL,
money_stolen INTEGER NOT NULL DEFAULT 0,
attacker_life_after INTEGER NOT NULL,
defender_life_after INTEGER NOT NULL,
summary TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(attacker_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY(defender_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS bank_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
action TEXT NOT NULL,
amount INTEGER NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS crime_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
crime_id INTEGER NOT NULL,
crime_name TEXT NOT NULL,
result TEXT NOT NULL,
reward_money INTEGER NOT NULL,
reward_xp INTEGER NOT NULL,
dropped_item_name TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY(crime_id) REFERENCES crime_catalog(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS job_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
job_id INTEGER NOT NULL,
job_name TEXT NOT NULL,
payout INTEGER NOT NULL,
xp_reward INTEGER NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY(job_id) REFERENCES jobs(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS item_transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_user_id INTEGER,
target_user_id INTEGER,
item_id INTEGER NOT NULL,
item_name TEXT NOT NULL,
quantity INTEGER NOT NULL,
reason TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(source_user_id) REFERENCES users(id) ON DELETE SET NULL,
FOREIGN KEY(target_user_id) REFERENCES users(id) ON DELETE SET NULL,
FOREIGN KEY(item_id) REFERENCES item_catalog(id) ON DELETE CASCADE
);
"""
)
self._seed_items(conn)
self._seed_jobs(conn)
self._seed_crimes(conn)
def _seed_items(self, conn: sqlite3.Connection) -> None:
columns = [
"slug",
"name",
"category",
"subcategory",
"description",
"buy_price",
"sell_price",
"tradable",
"stackable",
"consumable",
"strength_bonus",
"speed_bonus",
"defense_bonus",
"dexterity_bonus",
"life_bonus",
"energy_gain",
"nerve_gain",
"happy_gain",
"life_gain",
"weapon_damage_min",
"weapon_damage_max",
"armor_reduction",
"home_life_bonus",
"home_happy_bonus",
]
for item in ITEM_SEEDS:
values = {column: item.get(column, 0 if column not in {"slug", "name", "category", "subcategory", "description"} else "") for column in columns}
conn.execute(
f"""
INSERT OR IGNORE INTO item_catalog (
{", ".join(columns)}
) VALUES (
{", ".join(f":{column}" for column in columns)}
)
""",
values,
)
def _seed_jobs(self, conn: sqlite3.Connection) -> None:
for job in JOB_SEEDS:
conn.execute(
"""
INSERT OR IGNORE INTO jobs (
name, description, required_level, base_pay, cooldown_seconds
) VALUES (?, ?, ?, ?, ?)
""",
(
job["name"],
job["description"],
job["required_level"],
job["base_pay"],
job["cooldown_seconds"],
),
)
def _seed_crimes(self, conn: sqlite3.Connection) -> None:
for crime in CRIME_SEEDS:
conn.execute(
"""
INSERT OR IGNORE INTO crime_catalog (
tier, name, description, nerve_cost, success_rate,
money_min, money_max, xp_reward, jail_seconds,
drop_item_slug, drop_rate
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
crime["tier"],
crime["name"],
crime["description"],
crime["nerve_cost"],
crime["success_rate"],
crime["money_min"],
crime["money_max"],
crime["xp_reward"],
crime["jail_seconds"],
crime["drop_item_slug"],
crime["drop_rate"],
),
)
def _row_to_dict(self, row: Optional[sqlite3.Row]) -> Optional[Dict[str, Any]]:
return dict(row) if row else None
def _one(
self, conn: sqlite3.Connection, query: str, params: Iterable[Any] = ()
) -> Optional[Dict[str, Any]]:
return self._row_to_dict(conn.execute(query, tuple(params)).fetchone())
def _all(
self, conn: sqlite3.Connection, query: str, params: Iterable[Any] = ()
) -> List[Dict[str, Any]]:
return [dict(row) for row in conn.execute(query, tuple(params)).fetchall()]
def _set_user_fields(
self, conn: sqlite3.Connection, user_id: int, fields: Dict[str, Any]
) -> None:
if not fields:
return
assignments = ", ".join(f"{column} = ?" for column in fields)
values = list(fields.values()) + [user_id]
conn.execute(f"UPDATE users SET {assignments} WHERE id = ?", values)
def _inventory_quantity(
self, conn: sqlite3.Connection, user_id: int, item_id: int
) -> int:
row = conn.execute(
"SELECT quantity FROM user_items WHERE user_id = ? AND item_id = ?",
(user_id, item_id),
).fetchone()
return int(row["quantity"]) if row else 0
def _get_item(self, conn: sqlite3.Connection, item_id: int) -> Optional[Dict[str, Any]]:
return self._one(conn, "SELECT * FROM item_catalog WHERE id = ?", (item_id,))
def _get_item_by_slug(
self, conn: sqlite3.Connection, slug: str
) -> Optional[Dict[str, Any]]:
return self._one(conn, "SELECT * FROM item_catalog WHERE slug = ?", (slug,))
def _get_job(self, conn: sqlite3.Connection, job_id: int) -> Optional[Dict[str, Any]]:
return self._one(conn, "SELECT * FROM jobs WHERE id = ?", (job_id,))
def _get_crime(
self, conn: sqlite3.Connection, crime_id: int
) -> Optional[Dict[str, Any]]:
return self._one(conn, "SELECT * FROM crime_catalog WHERE id = ?", (crime_id,))
def _add_item(
self, conn: sqlite3.Connection, user_id: int, item_id: int, quantity: int
) -> None:
if quantity <= 0:
return
now = iso_now()
conn.execute(
"""
INSERT INTO user_items (user_id, item_id, quantity, created_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(user_id, item_id)
DO UPDATE SET quantity = user_items.quantity + excluded.quantity
""",
(user_id, item_id, quantity, now),
)
def _remove_item(
self, conn: sqlite3.Connection, user_id: int, item_id: int, quantity: int
) -> None:
if quantity <= 0:
return
owned = self._inventory_quantity(conn, user_id, item_id)
if owned < quantity:
raise GameError("You do not own enough of that item.")
remaining = owned - quantity
if remaining == 0:
conn.execute(
"DELETE FROM user_items WHERE user_id = ? AND item_id = ?",
(user_id, item_id),
)
else:
conn.execute(
"UPDATE user_items SET quantity = ? WHERE user_id = ? AND item_id = ?",
(remaining, user_id, item_id),
)
def _record_item_transaction(
self,
conn: sqlite3.Connection,
item: Dict[str, Any],
quantity: int,
reason: str,
source_user_id: Optional[int] = None,
target_user_id: Optional[int] = None,
) -> None:
conn.execute(
"""
INSERT INTO item_transactions (
source_user_id, target_user_id, item_id, item_name, quantity, reason, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
source_user_id,
target_user_id,
item["id"],
item["name"],
quantity,
reason,
iso_now(),
),
)
def _crime_tool_bonus(self, conn: sqlite3.Connection, user_id: int) -> float:
row = conn.execute(
"""
SELECT COUNT(*) AS tool_count
FROM user_items ui
JOIN item_catalog ic ON ic.id = ui.item_id
WHERE ui.user_id = ? AND ic.category = 'crime_tools' AND ui.quantity > 0
""",
(user_id,),
).fetchone()
return 0.05 if row and row["tool_count"] else 0.0
def _equipped_items(
self, conn: sqlite3.Connection, user: Dict[str, Any]
) -> Dict[str, Optional[Dict[str, Any]]]:
return {
"weapon": self._get_item(conn, user.get("equipped_weapon_item_id") or 0)
if user.get("equipped_weapon_item_id")
else None,
"armor": self._get_item(conn, user.get("equipped_armor_item_id") or 0)
if user.get("equipped_armor_item_id")
else None,
"home": self._get_item(conn, user.get("active_home_item_id") or 0)
if user.get("active_home_item_id")
else None,
}
def _compute_effective_stats(
self, equipped: Dict[str, Optional[Dict[str, Any]]], user: Dict[str, Any]
) -> Dict[str, Any]:
bonuses = {
"strength": 0,
"speed": 0,
"defense": 0,
"dexterity": 0,
"life_bonus": 0,
"armor_reduction": 0,
"weapon_damage_min": 3,
"weapon_damage_max": 7,
"happy_bonus": 0,
}
for item in equipped.values():
if not item:
continue
bonuses["strength"] += int(item.get("strength_bonus", 0))
bonuses["speed"] += int(item.get("speed_bonus", 0))
bonuses["defense"] += int(item.get("defense_bonus", 0))
bonuses["dexterity"] += int(item.get("dexterity_bonus", 0))
bonuses["life_bonus"] += int(item.get("life_bonus", 0))
bonuses["armor_reduction"] += int(item.get("armor_reduction", 0))
bonuses["weapon_damage_min"] = max(
bonuses["weapon_damage_min"], int(item.get("weapon_damage_min", 0))
)
bonuses["weapon_damage_max"] = max(
bonuses["weapon_damage_max"], int(item.get("weapon_damage_max", 0))
)
bonuses["life_bonus"] += int(item.get("home_life_bonus", 0))
bonuses["happy_bonus"] += int(item.get("home_happy_bonus", 0))
effective_defense = int(user["defense"]) + bonuses["defense"]
max_life = 100 + effective_defense + bonuses["life_bonus"]
return {
"effective_strength": int(user["strength"]) + bonuses["strength"],
"effective_speed": int(user["speed"]) + bonuses["speed"],
"effective_defense": effective_defense,
"effective_dexterity": int(user["dexterity"]) + bonuses["dexterity"],
"effective_armor_reduction": bonuses["armor_reduction"],
"weapon_damage_min": bonuses["weapon_damage_min"],
"weapon_damage_max": bonuses["weapon_damage_max"],
"effective_happy_cap": 100 + bonuses["happy_bonus"],
"max_life": max_life,
}
def _cleanup_equipment(
self, conn: sqlite3.Connection, user: Dict[str, Any]
) -> Dict[str, Any]:
changes: Dict[str, Any] = {}
for field, category in self.slot_fields.items():
item_id = user.get(field)
if not item_id:
continue
item = self._get_item(conn, int(item_id))
if not item or item["category"] != category:
changes[field] = None
continue
if self._inventory_quantity(conn, int(user["id"]), int(item_id)) <= 0:
changes[field] = None
if changes:
user.update(changes)
self._set_user_fields(conn, int(user["id"]), changes)
return user
def _apply_regen(
self,
user: Dict[str, Any],
value_field: str,
timestamp_field: str,
interval: int,
cap: int,
now: datetime,
updates: Dict[str, Any],
) -> None:
current = int(user[value_field])
last = parse_dt(user.get(timestamp_field)) or now
if current >= cap:
if last != now:
updates[timestamp_field] = now.isoformat()
user[timestamp_field] = now.isoformat()
return
elapsed = max(0, int((now - last).total_seconds()))
steps = elapsed // interval
if steps <= 0:
return
current = min(cap, current + steps)
remainder = elapsed % interval
new_last = now - timedelta(seconds=remainder)
updates[value_field] = current
updates[timestamp_field] = new_last.isoformat()
user[value_field] = current
user[timestamp_field] = new_last.isoformat()
def _refresh_user_locked(
self, conn: sqlite3.Connection, user_id: int
) -> Optional[Dict[str, Any]]:
user = self._one(conn, "SELECT * FROM users WHERE id = ?", (user_id,))
if not user:
return None
now = utcnow()
updates: Dict[str, Any] = {}
user = self._cleanup_equipment(conn, user)
for field in ("hospital_until", "jail_until", "job_cooldown_until"):
dt = parse_dt(user.get(field))
if dt and dt <= now:
updates[field] = None
user[field] = None
level = level_from_xp(int(user["xp"]))
if level != int(user["level"]):
updates["level"] = level
user["level"] = level
equipped = self._equipped_items(conn, user)
effective = self._compute_effective_stats(equipped, user)
for field, rule in self.regen_rules.items():
cap = int(rule["cap"])
if field == "happy":
cap = int(effective["effective_happy_cap"])
self._apply_regen(
user,
field,
rule["timestamp"],
int(rule["interval"]),
cap,
now,
updates,
)
self._apply_regen(
user,
"life",
"life_updated_at",
120,
int(effective["max_life"]),
now,
updates,
)
if int(user["life"]) > int(effective["max_life"]):
updates["life"] = int(effective["max_life"])
user["life"] = int(effective["max_life"])
if int(user["max_life"]) != int(effective["max_life"]):
updates["max_life"] = int(effective["max_life"])
user["max_life"] = int(effective["max_life"])
if updates:
self._set_user_fields(conn, user_id, updates)
jail_until = parse_dt(user.get("jail_until"))
hospital_until = parse_dt(user.get("hospital_until"))
user.update(effective)
user["weapon_item"] = equipped["weapon"]
user["armor_item"] = equipped["armor"]
user["home_item"] = equipped["home"]
user["jail_seconds_left"] = (
max(0, int((jail_until - now).total_seconds())) if jail_until else 0
)
user["hospital_seconds_left"] = (
max(0, int((hospital_until - now).total_seconds())) if hospital_until else 0
)
if user["hospital_seconds_left"]:
user["status"] = "Hospitalized"
elif user["jail_seconds_left"]:
user["status"] = "Jailed"
else:
user["status"] = "Ready"
user["can_act"] = not user["jail_seconds_left"] and not user["hospital_seconds_left"]
return user
def get_world_stats(self) -> Dict[str, int]:
with self.connect() as conn:
row = conn.execute(
"""
SELECT
(SELECT COUNT(*) FROM users) AS users_count,
(SELECT COUNT(*) FROM market_listings WHERE status = 'active') AS active_listings,
(SELECT COUNT(*) FROM combat_logs) AS combat_count
"""
).fetchone()
return dict(row)
def register_user(self, username: str, password: str) -> Dict[str, Any]:
username = normalize_username(username)
if not USERNAME_RE.fullmatch(username):
raise GameError("Username must be 3-20 characters of lowercase letters, numbers, or underscores.")
if len(password) < 6:
raise GameError("Password must be at least 6 characters.")
with self.connect() as conn:
existing = conn.execute(
"SELECT 1 FROM users WHERE username = ?", (username,)
).fetchone()
if existing:
raise GameError("That username is already taken.")
now = iso_now()
conn.execute(
"""
INSERT INTO users (
username, password_hash, level, xp, wallet_money, bank_money,
energy, nerve, happy, life, max_life,
strength, speed, defense, dexterity,
energy_updated_at, nerve_updated_at, happy_updated_at, life_updated_at,
created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
username,
hash_password(password),
1,
0,
750,
0,
100,
50,
100,
110,
110,
10,
10,
10,
10,
now,
now,
now,
now,
now,
),
)
user_id = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])
bandage = self._get_item_by_slug(conn, "bandage")
if bandage:
self._add_item(conn, user_id, int(bandage["id"]), 2)
self._record_item_transaction(
conn,
bandage,
2,
"starter_pack",
target_user_id=user_id,
)
return self._refresh_user_locked(conn, user_id) or {}
def authenticate_user(self, username: str, password: str) -> Optional[Dict[str, Any]]:
username = normalize_username(username)
with self.connect() as conn:
user = self._one(conn, "SELECT * FROM users WHERE username = ?", (username,))
if not user or not verify_password(password, user["password_hash"]):
return None
self._set_user_fields(conn, int(user["id"]), {"last_login_at": iso_now()})
return self._refresh_user_locked(conn, int(user["id"]))
def get_user(self, user_id: int) -> Optional[Dict[str, Any]]:
with self.connect() as conn:
return self._refresh_user_locked(conn, user_id)
def get_player(self, player_id: int) -> Optional[Dict[str, Any]]:
return self.get_user(player_id)
def list_players(self, current_user_id: Optional[int] = None) -> List[Dict[str, Any]]:
with self.connect() as conn:
query = "SELECT id FROM users"
params: List[Any] = []
if current_user_id:
query += " WHERE id != ?"
params.append(current_user_id)
query += " ORDER BY level DESC, xp DESC, username ASC"
ids = [int(row["id"]) for row in conn.execute(query, params).fetchall()]
return [
player
for player_id in ids
if (player := self._refresh_user_locked(conn, player_id)) is not None
]
def get_recent_combat_logs(
self, user_id: Optional[int] = None, limit: int = 8
) -> List[Dict[str, Any]]:
with self.connect() as conn:
where = ""
params: List[Any] = []
if user_id:
where = "WHERE cl.attacker_id = ? OR cl.defender_id = ?"
params.extend([user_id, user_id])
params.append(limit)
return self._all(
conn,
f"""
SELECT cl.*, a.username AS attacker_name, d.username AS defender_name
FROM combat_logs cl
JOIN users a ON a.id = cl.attacker_id
JOIN users d ON d.id = cl.defender_id
{where}
ORDER BY cl.id DESC
LIMIT ?
""",
params,
)
def get_recent_crime_logs(self, user_id: int, limit: int = 8) -> List[Dict[str, Any]]:
with self.connect() as conn:
return self._all(
conn,
"""
SELECT *
FROM crime_logs
WHERE user_id = ?
ORDER BY id DESC
LIMIT ?
""",
(user_id, limit),
)
def get_recent_bank_logs(self, user_id: int, limit: int = 8) -> List[Dict[str, Any]]:
with self.connect() as conn:
return self._all(
conn,
"""
SELECT *
FROM bank_logs
WHERE user_id = ?
ORDER BY id DESC
LIMIT ?
""",
(user_id, limit),
)
def get_recent_job_logs(self, user_id: int, limit: int = 8) -> List[Dict[str, Any]]:
with self.connect() as conn:
return self._all(
conn,
"""
SELECT *
FROM job_logs
WHERE user_id = ?
ORDER BY id DESC
LIMIT ?
""",
(user_id, limit),
)
def get_inventory(self, user_id: int) -> Dict[str, Any]:
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
if not user:
raise GameError("Player not found.")
items = self._all(
conn,
"""
SELECT ui.quantity, ic.*
FROM user_items ui
JOIN item_catalog ic ON ic.id = ui.item_id
WHERE ui.user_id = ?
ORDER BY ic.category, ic.name
""",
(user_id,),
)
for item in items:
item["is_equipped_weapon"] = (
int(user.get("equipped_weapon_item_id") or 0) == int(item["id"])
)
item["is_equipped_armor"] = (
int(user.get("equipped_armor_item_id") or 0) == int(item["id"])
)
item["is_active_home"] = (
int(user.get("active_home_item_id") or 0) == int(item["id"])
)
return {"user": user, "items": items}
def get_shop_items(self, shop_key: Optional[str]) -> Dict[str, Any]:
with self.connect() as conn:
shops = []
for slug, config in SHOP_CONFIG.items():
count = conn.execute(
f"""
SELECT COUNT(*) AS count
FROM item_catalog
WHERE category IN ({",".join("?" for _ in config["categories"])})
""",
tuple(config["categories"]),
).fetchone()
shops.append(
{
"slug": slug,
"label": config["label"],
"count": int(count["count"]) if count else 0,
}
)
if not shop_key:
return {"shop": None, "shops": shops, "items": []}
if shop_key not in SHOP_CONFIG:
raise GameError("That shop does not exist.")
config = SHOP_CONFIG[shop_key]
items = self._all(
conn,
f"""
SELECT *
FROM item_catalog
WHERE category IN ({",".join("?" for _ in config["categories"])})
ORDER BY buy_price, name
""",
tuple(config["categories"]),
)
return {"shop": {"slug": shop_key, **config}, "shops": shops, "items": items}
def get_jobs(self, user_id: int) -> Dict[str, Any]:
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
if not user:
raise GameError("Player not found.")
jobs = self._all(conn, "SELECT * FROM jobs ORDER BY required_level, id")
current_job = (
self._get_job(conn, int(user["job_id"])) if user.get("job_id") else None
)
cooldown_until = parse_dt(user.get("job_cooldown_until"))
user["job_seconds_left"] = (
max(0, int((cooldown_until - utcnow()).total_seconds()))
if cooldown_until
else 0
)
return {"user": user, "jobs": jobs, "current_job": current_job}
def get_crimes(self, user_id: int) -> Dict[str, Any]:
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
if not user:
raise GameError("Player not found.")
crimes = self._all(
conn, "SELECT * FROM crime_catalog ORDER BY tier, nerve_cost, id"
)
return {"user": user, "crimes": crimes}
def get_market(self, user_id: int) -> Dict[str, Any]:
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
if not user:
raise GameError("Player not found.")
listings = self._all(
conn,
"""
SELECT ml.*, ic.name AS item_name, ic.category, seller.username AS seller_name
FROM market_listings ml
JOIN item_catalog ic ON ic.id = ml.item_id
JOIN users seller ON seller.id = ml.seller_user_id
WHERE ml.status = 'active'
ORDER BY ml.created_at DESC, ml.id DESC
""",
)
my_listings = self._all(
conn,
"""
SELECT ml.*, ic.name AS item_name, ic.category
FROM market_listings ml
JOIN item_catalog ic ON ic.id = ml.item_id
WHERE ml.seller_user_id = ?
ORDER BY ml.id DESC
""",
(user_id,),
)
sellable = self._all(
conn,
"""
SELECT ui.quantity, ic.*
FROM user_items ui
JOIN item_catalog ic ON ic.id = ui.item_id
WHERE ui.user_id = ? AND ic.tradable = 1
ORDER BY ic.category, ic.name
""",
(user_id,),
)
return {
"user": user,
"listings": listings,
"my_listings": my_listings,
"sellable_items": sellable,
}
def buy_shop_item(self, user_id: int, item_id: int, quantity: int) -> str:
if quantity <= 0:
raise GameError("Buy quantity must be positive.")
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
item = self._get_item(conn, item_id)
if not user or not item:
raise GameError("That purchase is no longer valid.")
total = int(item["buy_price"]) * quantity
if total <= 0:
raise GameError("That item cannot be bought from an NPC shop.")
if int(user["wallet_money"]) < total:
raise GameError("You do not have enough wallet cash.")
conn.execute(
"""
UPDATE users
SET wallet_money = wallet_money - ?
WHERE id = ?
""",
(total, user_id),
)
self._add_item(conn, user_id, item_id, quantity)
self._record_item_transaction(
conn, item, quantity, "npc_buy", target_user_id=user_id
)
return f"Bought {quantity}x {item['name']} for ${total}."
def sell_item_to_npc(self, user_id: int, item_id: int, quantity: int) -> str:
if quantity <= 0:
raise GameError("Sell quantity must be positive.")
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
item = self._get_item(conn, item_id)
if not user or not item:
raise GameError("That sale is no longer valid.")
if int(item["sell_price"]) <= 0:
raise GameError("This item cannot be sold to an NPC.")
self._remove_item(conn, user_id, item_id, quantity)
payout = int(item["sell_price"]) * quantity
conn.execute(
"UPDATE users SET wallet_money = wallet_money + ? WHERE id = ?",
(payout, user_id),
)
self._cleanup_equipment(
conn, self._refresh_user_locked(conn, user_id) or dict(user)
)
self._record_item_transaction(
conn, item, quantity, "npc_sell", source_user_id=user_id
)
return f"Sold {quantity}x {item['name']} for ${payout}."
def equip_item(self, user_id: int, item_id: int) -> str:
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
item = self._get_item(conn, item_id)
if not user or not item:
raise GameError("That item is unavailable.")
if self._inventory_quantity(conn, user_id, item_id) <= 0:
raise GameError("You do not own that item.")
if item["category"] == "weapons":
self._set_user_fields(conn, user_id, {"equipped_weapon_item_id": item_id})
return f"Equipped {item['name']}."
if item["category"] == "armor":
self._set_user_fields(conn, user_id, {"equipped_armor_item_id": item_id})
return f"Equipped {item['name']}."
if item["category"] == "homes":
self._set_user_fields(conn, user_id, {"active_home_item_id": item_id})
return f"{item['name']} is now your active home."
raise GameError("That item cannot be equipped.")
def use_item(self, user_id: int, item_id: int) -> str:
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
item = self._get_item(conn, item_id)
if not user or not item:
raise GameError("That item is unavailable.")
if not int(item["consumable"]):
raise GameError("That item cannot be used directly.")
if self._inventory_quantity(conn, user_id, item_id) <= 0:
raise GameError("You do not own that item.")
energy = clamp(
int(user["energy"]) + int(item["energy_gain"]),
0,
100,
)
nerve = clamp(
int(user["nerve"]) + int(item["nerve_gain"]),
0,
50,
)
happy_cap = int(user["effective_happy_cap"])
happy = clamp(
int(user["happy"]) + int(item["happy_gain"]),
0,
happy_cap,
)
life = clamp(
int(user["life"]) + int(item["life_gain"]),
0,
int(user["max_life"]),
)
now = iso_now()
self._remove_item(conn, user_id, item_id, 1)
self._set_user_fields(
conn,
user_id,
{
"energy": energy,
"nerve": nerve,
"happy": happy,
"life": life,
"energy_updated_at": now,
"nerve_updated_at": now,
"happy_updated_at": now,
"life_updated_at": now,
},
)
self._record_item_transaction(
conn, item, 1, "consume", source_user_id=user_id
)
return f"Used {item['name']}."
def train_stat(self, user_id: int, stat_name: str) -> str:
if stat_name not in {"strength", "speed", "defense", "dexterity"}:
raise GameError("That stat does not exist.")
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
if not user:
raise GameError("Player not found.")
if not user["can_act"]:
raise GameError("You cannot train while jailed or hospitalized.")
if int(user["energy"]) < 5:
raise GameError("You need at least 5 energy to train.")
gain = random.randint(1, 3)
xp = int(user["xp"]) + 5
conn.execute(
f"""
UPDATE users
SET {stat_name} = {stat_name} + ?,
energy = energy - 5,
xp = ?,
level = ?,
energy_updated_at = ?
WHERE id = ?
""",
(gain, xp, level_from_xp(xp), iso_now(), user_id),
)
return f"Trained {stat_name}. Gain: +{gain}."
def perform_crime(self, user_id: int, crime_id: int) -> str:
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
crime = self._get_crime(conn, crime_id)
if not user or not crime:
raise GameError("That crime is unavailable.")
if not user["can_act"]:
raise GameError("You cannot commit crimes while jailed or hospitalized.")
if int(user["nerve"]) < int(crime["nerve_cost"]):
raise GameError("You do not have enough nerve.")
now = iso_now()
result = "fail"
reward_money = 0
reward_xp = 0
dropped_item_name = ""
updates: Dict[str, Any] = {
"nerve": int(user["nerve"]) - int(crime["nerve_cost"]),
"nerve_updated_at": now,
}
success_rate = min(
0.95, float(crime["success_rate"]) + self._crime_tool_bonus(conn, user_id)
)
if random.random() <= success_rate:
result = "success"
reward_money = random.randint(
int(crime["money_min"]), int(crime["money_max"])
)
reward_xp = int(crime["xp_reward"])
new_xp = int(user["xp"]) + reward_xp
updates.update(
{
"wallet_money": int(user["wallet_money"]) + reward_money,
"xp": new_xp,
"level": level_from_xp(new_xp),
}
)
if crime["drop_item_slug"] and random.random() <= float(crime["drop_rate"]):
item = self._get_item_by_slug(conn, str(crime["drop_item_slug"]))
if item:
self._add_item(conn, user_id, int(item["id"]), 1)
self._record_item_transaction(
conn, item, 1, "crime_drop", target_user_id=user_id
)
dropped_item_name = str(item["name"])
else:
jail_until = utcnow() + timedelta(seconds=int(crime["jail_seconds"]))
updates["jail_until"] = jail_until.isoformat()
self._set_user_fields(conn, user_id, updates)
conn.execute(
"""
INSERT INTO crime_logs (
user_id, crime_id, crime_name, result,
reward_money, reward_xp, dropped_item_name, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
user_id,
crime_id,
crime["name"],
result,
reward_money,
reward_xp,
dropped_item_name,
iso_now(),
),
)
if result == "success":
suffix = f" and found {dropped_item_name}" if dropped_item_name else ""
return f"Crime succeeded. You earned ${reward_money}, {reward_xp} xp{suffix}."
return f"Crime failed. You landed in jail for {format_duration(int(crime['jail_seconds']))}."
def join_job(self, user_id: int, job_id: int) -> str:
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
job = self._get_job(conn, job_id)
if not user or not job:
raise GameError("That job is unavailable.")
if not user["can_act"]:
raise GameError("You cannot switch jobs while jailed or hospitalized.")
if int(user["level"]) < int(job["required_level"]):
raise GameError("Your level is too low for that job.")
self._set_user_fields(conn, user_id, {"job_id": job_id})
return f"You joined {job['name']}."
def work_job(self, user_id: int) -> str:
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
if not user:
raise GameError("Player not found.")
if not user["can_act"]:
raise GameError("You cannot work while jailed or hospitalized.")
if not user.get("job_id"):
raise GameError("You do not have a job yet.")
job = self._get_job(conn, int(user["job_id"]))
if not job:
raise GameError("Your job no longer exists.")
cooldown = parse_dt(user.get("job_cooldown_until"))
now_dt = utcnow()
if cooldown and cooldown > now_dt:
raise GameError(f"Job cooldown remaining: {format_duration(int((cooldown - now_dt).total_seconds()))}.")
payout = int(job["base_pay"])
reward_xp = 10
new_xp = int(user["xp"]) + reward_xp
conn.execute(
"""
UPDATE users
SET wallet_money = wallet_money + ?,
xp = ?,
level = ?,
job_cooldown_until = ?
WHERE id = ?
""",
(
payout,
new_xp,
level_from_xp(new_xp),
(now_dt + timedelta(seconds=int(job["cooldown_seconds"]))).isoformat(),
user_id,
),
)
conn.execute(
"""
INSERT INTO job_logs (user_id, job_id, job_name, payout, xp_reward, created_at)
VALUES (?, ?, ?, ?, ?, ?)
""",
(user_id, job["id"], job["name"], payout, reward_xp, iso_now()),
)
return f"Shift complete. Earned ${payout} and {reward_xp} xp."
def deposit_money(self, user_id: int, amount: int) -> str:
if amount <= 0:
raise GameError("Deposit amount must be positive.")
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
if not user:
raise GameError("Player not found.")
if int(user["wallet_money"]) < amount:
raise GameError("You do not have that much wallet cash.")
conn.execute(
"""
UPDATE users
SET wallet_money = wallet_money - ?, bank_money = bank_money + ?
WHERE id = ?
""",
(amount, amount, user_id),
)
conn.execute(
"INSERT INTO bank_logs (user_id, action, amount, created_at) VALUES (?, ?, ?, ?)",
(user_id, "deposit", amount, iso_now()),
)
return f"Deposited ${amount} into the bank."
def withdraw_money(self, user_id: int, amount: int) -> str:
if amount <= 0:
raise GameError("Withdraw amount must be positive.")
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
if not user:
raise GameError("Player not found.")
if int(user["bank_money"]) < amount:
raise GameError("You do not have that much bank cash.")
conn.execute(
"""
UPDATE users
SET wallet_money = wallet_money + ?, bank_money = bank_money - ?
WHERE id = ?
""",
(amount, amount, user_id),
)
conn.execute(
"INSERT INTO bank_logs (user_id, action, amount, created_at) VALUES (?, ?, ?, ?)",
(user_id, "withdraw", amount, iso_now()),
)
return f"Withdrew ${amount} from the bank."
def create_market_listing(
self, user_id: int, item_id: int, quantity: int, price_each: int
) -> str:
if quantity <= 0 or price_each <= 0:
raise GameError("Listing quantity and price must be positive.")
with self.connect() as conn:
user = self._refresh_user_locked(conn, user_id)
item = self._get_item(conn, item_id)
if not user or not item:
raise GameError("That listing is unavailable.")
if not int(item["tradable"]):
raise GameError("That item cannot be listed on the market.")
self._remove_item(conn, user_id, item_id, quantity)
conn.execute(
"""
INSERT INTO market_listings (
seller_user_id, item_id, quantity, price_each, status, created_at
) VALUES (?, ?, ?, ?, 'active', ?)
""",
(user_id, item_id, quantity, price_each, iso_now()),
)
self._cleanup_equipment(
conn, self._refresh_user_locked(conn, user_id) or dict(user)
)
self._record_item_transaction(
conn, item, quantity, "market_list", source_user_id=user_id
)
return f"Listed {quantity}x {item['name']} for ${price_each} each."
def cancel_market_listing(self, user_id: int, listing_id: int) -> str:
with self.connect() as conn:
listing = self._one(
conn,
"SELECT * FROM market_listings WHERE id = ? AND status = 'active'",
(listing_id,),
)
if not listing or int(listing["seller_user_id"]) != user_id:
raise GameError("That listing cannot be removed.")
item = self._get_item(conn, int(listing["item_id"]))
if not item:
raise GameError("Listing item missing.")
quantity = int(listing["quantity"])
self._add_item(conn, user_id, int(item["id"]), quantity)
conn.execute(
"""
UPDATE market_listings
SET status = 'cancelled', quantity = 0
WHERE id = ?
""",
(listing_id,),
)
self._record_item_transaction(
conn, item, quantity, "market_cancel", target_user_id=user_id
)
return f"Removed listing for {item['name']}."
def buy_market_listing(self, buyer_id: int, listing_id: int, quantity: int) -> str:
if quantity <= 0:
raise GameError("Purchase quantity must be positive.")
with self.connect() as conn:
buyer = self._refresh_user_locked(conn, buyer_id)
listing = self._one(
conn,
"SELECT * FROM market_listings WHERE id = ? AND status = 'active'",
(listing_id,),
)
if not buyer or not listing:
raise GameError("That listing is no longer available.")
if int(listing["seller_user_id"]) == buyer_id:
raise GameError("You cannot buy your own listing.")
remaining = int(listing["quantity"])
if quantity > remaining:
raise GameError("Not enough quantity remains on that listing.")
item = self._get_item(conn, int(listing["item_id"]))
if not item:
raise GameError("Listing item missing.")
total = quantity * int(listing["price_each"])
if int(buyer["wallet_money"]) < total:
raise GameError("You do not have enough wallet cash.")
conn.execute(
"UPDATE users SET wallet_money = wallet_money - ? WHERE id = ?",
(total, buyer_id),
)
conn.execute(
"UPDATE users SET wallet_money = wallet_money + ? WHERE id = ?",
(total, int(listing["seller_user_id"])),
)
self._add_item(conn, buyer_id, int(item["id"]), quantity)
new_quantity = remaining - quantity
new_status = "sold" if new_quantity == 0 else "active"
conn.execute(
"""
UPDATE market_listings
SET quantity = ?, status = ?
WHERE id = ?
""",
(new_quantity, new_status, listing_id),
)
self._record_item_transaction(
conn,
item,
quantity,
"market_buy",
source_user_id=int(listing["seller_user_id"]),
target_user_id=buyer_id,
)
return f"Bought {quantity}x {item['name']} for ${total}."
def _roll_damage(self, attacker: Dict[str, Any], defender: Dict[str, Any]) -> int:
base = random.randint(
int(attacker["weapon_damage_min"]), int(attacker["weapon_damage_max"])
)
offense = int(attacker["effective_strength"]) + int(attacker["effective_speed"])
mitigation = int(defender["effective_defense"]) + int(defender["effective_dexterity"])
reduction = int(defender["effective_armor_reduction"])
damage = (offense // 6) + base + random.randint(0, 5) - (mitigation // 8) - reduction
return max(1, damage)
def perform_attack(self, attacker_id: int, defender_id: int, action_type: str) -> Dict[str, Any]:
if action_type not in {"leave", "hospitalize", "mug"}:
raise GameError("Invalid attack action.")
if attacker_id == defender_id:
raise GameError("You cannot attack yourself.")
with self.connect() as conn:
attacker = self._refresh_user_locked(conn, attacker_id)
defender = self._refresh_user_locked(conn, defender_id)
if not attacker or not defender:
raise GameError("Attack target not found.")
if not attacker["can_act"]:
raise GameError("You cannot attack while jailed or hospitalized.")
if defender["hospital_seconds_left"] or defender["jail_seconds_left"]:
raise GameError("That player is currently unavailable for attack.")
attacker_hp = int(attacker["life"])
defender_hp = int(defender["life"])
rounds: List[str] = []
for round_no in range(1, 7):
damage = self._roll_damage(attacker, defender)
defender_hp = max(0, defender_hp - damage)
rounds.append(
f"Round {round_no}: {attacker['username']} hits {defender['username']} for {damage}."
)
if defender_hp <= 0:
break
damage = self._roll_damage(defender, attacker)
attacker_hp = max(0, attacker_hp - damage)
rounds.append(
f"Round {round_no}: {defender['username']} hits back for {damage}."
)
if attacker_hp <= 0:
break
attacker_won = False
if defender_hp <= 0:
attacker_won = True
elif attacker_hp <= 0:
attacker_won = False
else:
attacker_won = attacker_hp > defender_hp
now = utcnow()
money_stolen = 0
result = "defender_win"
summary = ""
attacker_updates: Dict[str, Any] = {
"life": max(1, attacker_hp) if attacker_hp > 0 else 1,
"life_updated_at": now.isoformat(),
}
defender_updates: Dict[str, Any] = {
"life": max(1, defender_hp) if defender_hp > 0 else 1,
"life_updated_at": now.isoformat(),
}
if attacker_won:
result = "attacker_win"
attacker_new_xp = int(attacker["xp"]) + 25
attacker_updates.update(
{"xp": attacker_new_xp, "level": level_from_xp(attacker_new_xp)}
)
defender_hospitalized = defender_hp <= 0 or action_type in {
"hospitalize",
"mug",
}
if defender_hospitalized:
defender_updates["hospital_until"] = (
now + timedelta(minutes=10)
).isoformat()
defender_updates["life"] = 1
if action_type == "mug":
wallet = int(defender["wallet_money"])
money_stolen = min(max(50, (wallet * 15) // 100), 5000, wallet)
if money_stolen > 0:
attacker_updates["wallet_money"] = int(attacker["wallet_money"]) + money_stolen
defender_updates["wallet_money"] = int(defender["wallet_money"]) - money_stolen
summary = f"{attacker['username']} won a {action_type} attack against {defender['username']}."
if money_stolen:
summary += f" Mug haul: ${money_stolen}."
else:
defender_new_xp = int(defender["xp"]) + 25
defender_updates.update(
{"xp": defender_new_xp, "level": level_from_xp(defender_new_xp)}
)
if attacker_hp <= 0:
attacker_updates["hospital_until"] = (
now + timedelta(minutes=10)
).isoformat()
attacker_updates["life"] = 1
summary = f"{defender['username']} fought off {attacker['username']}."
self._set_user_fields(conn, attacker_id, attacker_updates)
self._set_user_fields(conn, defender_id, defender_updates)
conn.execute(
"""
INSERT INTO combat_logs (
attacker_id, defender_id, result, action_type, money_stolen,
attacker_life_after, defender_life_after, summary, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
attacker_id,
defender_id,
result,
action_type,
money_stolen,
int(attacker_updates["life"]),
int(defender_updates["life"]),
summary,
now.isoformat(),
),
)
refreshed_attacker = self._refresh_user_locked(conn, attacker_id)
refreshed_defender = self._refresh_user_locked(conn, defender_id)
return {
"winner": "attacker" if attacker_won else "defender",
"action_type": action_type,
"money_stolen": money_stolen,
"rounds": rounds,
"summary": summary,
"attacker": refreshed_attacker,
"defender": refreshed_defender,
}
def get_player_market_history(self, player_id: int, limit: int = 8) -> List[Dict[str, Any]]:
with self.connect() as conn:
return self._all(
conn,
"""
SELECT *
FROM item_transactions
WHERE source_user_id = ? OR target_user_id = ?
ORDER BY id DESC
LIMIT ?
""",
(player_id, player_id, limit),
)
GAME = GameStore(DB_PATH)
GAME.init_db()
class TornCityApp(App):
def __init__(self) -> None:
super().__init__()
self.env = Environment(
loader=FileSystemLoader(str(TEMPLATES_DIR)),
autoescape=select_autoescape(["html", "xml"]),
enable_async=True,
)
self.env.globals.update(
format_duration=format_duration,
format_ts=format_ts,
)
self.startup_handlers.append(self._startup)
async def _startup(self) -> None:
GAME.init_db()
def _flash(self, message: str, kind: str = "info") -> None:
self.request.session["_flash"] = {"message": message, "kind": kind}
def _consume_flash(self) -> Optional[Dict[str, str]]:
return self.request.session.pop("_flash", None)
def _session_user_id(self) -> Optional[int]:
raw = self.request.session.get("user_id")
if raw is None:
return None
try:
return int(raw)
except (TypeError, ValueError):
self.request.session.clear()
return None
def _current_user(self) -> Optional[Dict[str, Any]]:
user_id = self._session_user_id()
if not user_id:
return None
user = GAME.get_user(user_id)
if not user:
self.request.session.clear()
return None
return user
def _request_value(
self, source_attr: str, helper_name: str, name: str, default: str = ""
) -> str:
helper = getattr(self.request, helper_name, None)
if callable(helper):
value = helper(name, default)
return default if value is None else str(value)
values = getattr(self.request, source_attr, {}) or {}
value = values.get(name)
if isinstance(value, list):
return default if not value else str(value[0])
if value is None:
return default
return str(value)
def _form_value(self, name: str, default: str = "") -> str:
return self._request_value("body_params", "form", name, default)
def _query_value(self, name: str, default: str = "") -> str:
return self._request_value("query_params", "query", name, default)
def _require_login(self) -> tuple[Optional[Dict[str, Any]], Optional[Any]]:
user = self._current_user()
if user:
return user, None
self._flash("Login required.", "error")
return None, self._redirect("/login")
async def _page(self, template: str, **context: Any) -> str:
context.setdefault("current_user", self._current_user())
context.setdefault("flash", self._consume_flash())
context.setdefault("nav", "")
return await self._render_template(template, **context)
def _as_int(self, value: Optional[str], field: str) -> int:
try:
return int((value or "").strip())
except ValueError:
raise GameError(f"{field} must be a whole number.")
async def index(self) -> Any:
user = self._current_user()
if user:
return self._redirect("/city")
return await self._page(
"index.html",
nav="home",
world=GAME.get_world_stats(),
)
async def register(self) -> Any:
if self.request.method == "GET":
return await self._page("register.html", nav="register")
try:
user = GAME.register_user(
self._form_value("username"), self._form_value("password")
)
except GameError as exc:
self._flash(str(exc), "error")
return self._redirect("/register")
self.request.session["user_id"] = user["id"]
self._flash(f"Welcome to the city, {user['username']}.", "success")
return self._redirect("/city")
async def login(self) -> Any:
if self.request.method == "GET":
return await self._page("login.html", nav="login")
user = GAME.authenticate_user(
self._form_value("username"), self._form_value("password")
)
if not user:
self._flash("Invalid username or password.", "error")
return self._redirect("/login")
self.request.session["user_id"] = user["id"]
self._flash(f"Logged in as {user['username']}.", "success")
return self._redirect("/city")
async def logout(self) -> Any:
self.request.session.clear()
self._flash("Logged out.", "success")
return self._redirect("/")
async def city(self) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
return await self._page(
"city.html",
nav="city",
current_user=user,
combat_logs=GAME.get_recent_combat_logs(user["id"]),
crime_logs=GAME.get_recent_crime_logs(user["id"]),
bank_logs=GAME.get_recent_bank_logs(user["id"]),
job_logs=GAME.get_recent_job_logs(user["id"]),
)
async def profile(self) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
inventory = GAME.get_inventory(user["id"])
return await self._page(
"profile.html",
nav="profile",
current_user=user,
inventory_count=len(inventory["items"]),
combat_logs=GAME.get_recent_combat_logs(user["id"], limit=6),
market_logs=GAME.get_player_market_history(user["id"], limit=6),
)
async def players(self) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
return await self._page(
"players.html",
nav="players",
current_user=user,
players=GAME.list_players(user["id"]),
)
async def player(self, player_id: str) -> Any:
user, guard = self._require_login()
if guard:
return guard
try:
target_id = int(player_id)
except ValueError:
self._flash("Invalid player id.", "error")
return self._redirect("/players")
target = GAME.get_player(target_id)
if not target:
self._flash("Player not found.", "error")
return self._redirect("/players")
return await self._page(
"player.html",
nav="players",
current_user=user,
target=target,
combat_logs=GAME.get_recent_combat_logs(target["id"], limit=6),
)
async def gym(self) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
if self.request.method == "POST":
try:
message = GAME.train_stat(user["id"], self._form_value("stat"))
self._flash(message, "success")
except GameError as exc:
self._flash(str(exc), "error")
return self._redirect("/gym")
return await self._page("gym.html", nav="gym", current_user=GAME.get_user(user["id"]))
async def crimes(self) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
if self.request.method == "POST":
try:
message = GAME.perform_crime(
user["id"], self._as_int(self._form_value("crime_id"), "Crime")
)
self._flash(message, "success")
except GameError as exc:
self._flash(str(exc), "error")
return self._redirect("/crimes")
payload = GAME.get_crimes(user["id"])
return await self._page(
"crimes.html",
nav="crimes",
current_user=payload["user"],
crimes=payload["crimes"],
crime_logs=GAME.get_recent_crime_logs(user["id"]),
)
async def inventory(self) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
if self.request.method == "POST":
try:
item_id = self._as_int(self._form_value("item_id"), "Item")
action = self._form_value("action")
if action == "equip":
message = GAME.equip_item(user["id"], item_id)
elif action == "use":
message = GAME.use_item(user["id"], item_id)
elif action == "sell":
message = GAME.sell_item_to_npc(
user["id"],
item_id,
self._as_int(self._form_value("quantity"), "Quantity"),
)
else:
raise GameError("Invalid inventory action.")
self._flash(message, "success")
except GameError as exc:
self._flash(str(exc), "error")
return self._redirect("/inventory")
payload = GAME.get_inventory(user["id"])
grouped: Dict[str, List[Dict[str, Any]]] = {}
for item in payload["items"]:
grouped.setdefault(str(item["category"]), []).append(item)
return await self._page(
"inventory.html",
nav="inventory",
current_user=payload["user"],
grouped_items=grouped,
)
async def shops(self, category: Optional[str] = None) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
target_shop = category or None
if self.request.method == "POST":
redirect_shop = self._form_value("redirect_shop", target_shop or "")
try:
message = GAME.buy_shop_item(
user["id"],
self._as_int(self._form_value("item_id"), "Item"),
self._as_int(self._form_value("quantity"), "Quantity"),
)
self._flash(message, "success")
except GameError as exc:
self._flash(str(exc), "error")
if redirect_shop:
return self._redirect(f"/shops/{redirect_shop}")
return self._redirect("/shops")
try:
payload = GAME.get_shop_items(target_shop)
except GameError as exc:
self._flash(str(exc), "error")
return self._redirect("/shops")
return await self._page(
"shops.html",
nav="shops",
current_user=GAME.get_user(user["id"]),
shop=payload["shop"],
shops=payload["shops"],
items=payload["items"],
)
async def job(self) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
if self.request.method == "POST":
try:
action = self._form_value("action")
if action == "join":
message = GAME.join_job(
user["id"], self._as_int(self._form_value("job_id"), "Job")
)
elif action == "work":
message = GAME.work_job(user["id"])
else:
raise GameError("Invalid job action.")
self._flash(message, "success")
except GameError as exc:
self._flash(str(exc), "error")
return self._redirect("/job")
payload = GAME.get_jobs(user["id"])
return await self._page(
"job.html",
nav="job",
current_user=payload["user"],
jobs=payload["jobs"],
current_job=payload["current_job"],
job_logs=GAME.get_recent_job_logs(user["id"]),
)
async def bank(self) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
if self.request.method == "POST":
try:
amount = self._as_int(self._form_value("amount"), "Amount")
action = self._form_value("action")
if action == "deposit":
message = GAME.deposit_money(user["id"], amount)
elif action == "withdraw":
message = GAME.withdraw_money(user["id"], amount)
else:
raise GameError("Invalid bank action.")
self._flash(message, "success")
except GameError as exc:
self._flash(str(exc), "error")
return self._redirect("/bank")
return await self._page(
"bank.html",
nav="bank",
current_user=GAME.get_user(user["id"]),
bank_logs=GAME.get_recent_bank_logs(user["id"]),
)
async def market(self, section: Optional[str] = None) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
my_view = section == "my-listings"
if self.request.method == "POST":
try:
action = self._form_value("action")
if action == "list":
message = GAME.create_market_listing(
user["id"],
self._as_int(self._form_value("item_id"), "Item"),
self._as_int(self._form_value("quantity"), "Quantity"),
self._as_int(self._form_value("price_each"), "Price"),
)
target = "/market/my-listings"
elif action == "buy":
message = GAME.buy_market_listing(
user["id"],
self._as_int(self._form_value("listing_id"), "Listing"),
self._as_int(self._form_value("quantity"), "Quantity"),
)
target = "/market"
elif action == "remove":
message = GAME.cancel_market_listing(
user["id"],
self._as_int(self._form_value("listing_id"), "Listing"),
)
target = "/market/my-listings"
else:
raise GameError("Invalid market action.")
self._flash(message, "success")
except GameError as exc:
self._flash(str(exc), "error")
target = "/market/my-listings" if my_view else "/market"
return self._redirect(target)
payload = GAME.get_market(user["id"])
return await self._page(
"market.html",
nav="market",
current_user=payload["user"],
listings=payload["listings"],
my_listings=payload["my_listings"],
sellable_items=payload["sellable_items"],
my_view=my_view,
)
async def attack(self, player_id: str) -> Any:
user, guard = self._require_login()
if guard:
return guard
assert user is not None
try:
target_id = int(player_id)
except ValueError:
self._flash("Invalid player id.", "error")
return self._redirect("/players")
target = GAME.get_player(target_id)
if not target:
self._flash("Player not found.", "error")
return self._redirect("/players")
if self.request.method == "POST":
try:
result = GAME.perform_attack(
user["id"], target_id, self._form_value("action_type", "leave")
)
self._flash(result["summary"], "success")
except GameError as exc:
self._flash(str(exc), "error")
return self._redirect(f"/attack/{target_id}")
return await self._page(
"attack.html",
nav="players",
current_user=GAME.get_user(user["id"]),
target=target,
combat_logs=GAME.get_recent_combat_logs(user["id"], limit=10),
)
app = TornCityApp()
