import os import time import math import json import collections from typing import Optional, List, Dict from fastapi import FastAPI, HTTPException, Depends, Header, Request, status from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel import bcrypt import jwt from .database import get_connection, init_db SECRET_KEY = "grimhold_dark_fantasy_2026_super_secret_jwt" JWT_ALGORITHM = "HS256" # 16 x 10 Landscape Grid GRID_W = 16 GRID_H = 10 START_TILE = (0, 4) END_TILE = (15, 4) GATE_TILE = (14, 4) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) app = FastAPI(title="GrimHold: Dark Fantasy Tower Defense & PvP Raids", version="3.5.0") app.mount("/static", StaticFiles(directory=os.path.join(BASE_DIR, "static")), name="static") def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(plain: str, hashed: str) -> bool: return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8")) def create_access_token(user_id: int, username: str) -> str: payload = { "sub": str(user_id), "username": username, "exp": time.time() + 86400 * 14 } return jwt.encode(payload, SECRET_KEY, algorithm=JWT_ALGORITHM) def decode_access_token(token: str) -> dict: try: return jwt.decode(token, SECRET_KEY, algorithms=[JWT_ALGORITHM]) except Exception: return None def get_current_user(authorization: Optional[str] = Header(None)): if not authorization or not authorization.startswith("Bearer "): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Nicht authentifiziert.") token = authorization.split(" ")[1] payload = decode_access_token(token) if not payload: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token ungültig oder abgelaufen.") conn = get_connection() cur = conn.cursor() cur.execute("SELECT * FROM users WHERE id = ?", (int(payload["sub"]),)) user = cur.fetchone() conn.close() if not user: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Benutzer nicht gefunden.") return dict(user) # BFS Path validation def check_path_exists(occupied_tiles: set) -> bool: start = START_TILE end = END_TILE if start in occupied_tiles or end in occupied_tiles: return False queue = collections.deque([start]) visited = {start} while queue: cx, cy = queue.popleft() if (cx, cy) == end: return True for dx, dy in [(-1, 0), (1, 0), (0, -1), (0, 1)]: nx, ny = cx + dx, cy + dy if 0 <= nx < GRID_W and 0 <= ny < GRID_H: if (nx, ny) not in occupied_tiles and (nx, ny) not in visited: visited.add((nx, ny)) queue.append((nx, ny)) return False # Paced Machine definitions (Balanced economy) MACHINE_CONFIG = { "drill": { "name": "Erz-Bohrer & Schredder", "description": "Fördert kontinuierlich Eisenerz für schwere Wachtürme und Einheiten.", "icon": "anvil", "resource": "ore", "base_rate": 6.0 / 60.0, "rate_per_lvl": 4.5 / 60.0, "base_cost": {"gold": 80, "ore": 30, "energy": 20}, "base_time": 240 }, "turbine": { "name": "Dampf- & Blitz-Turbine", "description": "Erzeugt Hochspannungs-Energie für Spezial-Fähigkeiten und Magietürme.", "icon": "zap", "resource": "energy", "base_rate": 4.0 / 60.0, "rate_per_lvl": 3.0 / 60.0, "base_cost": {"gold": 90, "ore": 40, "energy": 15}, "base_time": 300 }, "press": { "name": "Königliche Münz-Presse", "description": "Prägt Goldmünzen für Wehrmauern und Truppenrekrutierung.", "icon": "coins", "resource": "gold", "base_rate": 8.0 / 60.0, "rate_per_lvl": 6.0 / 60.0, "base_cost": {"gold": 70, "ore": 50, "energy": 20}, "base_time": 240 } } # Tower definitions (Balanced for readable, strategic combat) TOWER_CONFIG = { "rock": {"name": "Felsbrocken / Mauer", "cost": 5, "type": "wall", "desc": "Blockiert Laufweg für Mazing."}, "mg": {"name": "MG-Nest / Balliste", "cost": 25, "type": "tower", "range": 3.0, "damage": 6, "rate": 0.5, "desc": "Leichtes Schnellfeuer-Geschütz."}, "cannon": {"name": "Belagerungs-Mörser", "cost": 45, "type": "tower", "range": 4.2, "damage": 35, "rate": 1.4, "splash": 1.3, "desc": "Schwerer Flächenschaden (nur Boden)."}, "ice": {"name": "Frost-Projektor", "cost": 35, "type": "tower", "range": 2.8, "damage": 4, "rate": 0.55, "slow": 0.4, "desc": "Verlangsamt Feinde um 40%."}, "tesla": {"name": "Tesla-Blitz-Spule", "cost": 65, "type": "tower", "range": 3.4, "damage": 22, "rate": 0.85, "chain": 2, "desc": "Kettenblitz auf bis zu 2 Ziele (Boden & Luft)."} } # Building slot upgrade costs (+2 slots each tier) SLOT_UPGRADE_CONFIG = { 8: {"cost": {"gold": 80, "ore": 40}, "next_slots": 10}, 10: {"cost": {"gold": 160, "ore": 80}, "next_slots": 12}, 12: {"cost": {"gold": 280, "ore": 140}, "next_slots": 14}, 14: {"cost": {"gold": 450, "ore": 220}, "next_slots": 16}, 16: {"cost": {"gold": 700, "ore": 360}, "next_slots": 18}, 18: {"cost": {"gold": 1000, "ore": 500}, "next_slots": 20} } # Training Camp Progression & Caps CAMP_CONFIG = { 1: { "name": "Baracken (Stufe 1)", "desc": "Ausbildung von Kobold-Spähern und Gepanzerten Trollen.", "pool_cap": 16, "squad_cap": 6, "defense_slots": 2, "cost": None, "duration": 0 }, 2: { "name": "Waffenmeisterei (Stufe 2)", "desc": "Schaltet den Belagerungs-Brecher frei (zertrümmert Felsen).", "pool_cap": 26, "squad_cap": 10, "defense_slots": 3, "cost": {"gold": 160, "ore": 90, "energy": 45}, "duration": 300 }, 3: { "name": "Schatten-Krypta (Stufe 3)", "desc": "Schaltet Schatten-Assassinen frei (Rauch-Tarnung).", "pool_cap": 40, "squad_cap": 14, "defense_slots": 4, "cost": {"gold": 380, "ore": 220, "energy": 120}, "duration": 900 }, 4: { "name": "Gargoyle-Horst (Stufe 4)", "desc": "Schaltet fliegende Gargoyles frei (überwindet Felsen).", "pool_cap": 60, "squad_cap": 18, "defense_slots": 5, "cost": {"gold": 800, "ore": 480, "energy": 280}, "duration": 2400 } } # 5 Unit classes with distinct combat parameters UNITS_CONFIG = { "scout": { "id": "scout", "name": "Kobold-Späher", "req_camp_level": 1, "cost": {"gold": 8, "ore": 0, "energy": 0}, "hp": 65, "speed": 2.1, "melee_dmg": 18, "icon": "🐀", "special": "Akrobat", "special_desc": "35% Chance, Geschossen komplett auszuweichen (Dodge)." }, "tank": { "id": "tank", "name": "Gepanzerter Troll", "req_camp_level": 1, "cost": {"gold": 24, "ore": 8, "energy": 0}, "hp": 380, "speed": 0.85, "melee_dmg": 30, "icon": "🛡️", "special": "Eisenhaut", "special_desc": "Nimmt 40% weniger Schaden von Türmen, beschützt die Gruppe." }, "crusher": { "id": "crusher", "name": "Belagerungs-Brecher", "req_camp_level": 2, "cost": {"gold": 45, "ore": 20, "energy": 0}, "hp": 260, "speed": 1.1, "melee_dmg": 45, "icon": "🔨", "special": "Fels-Zertrümmerer", "special_desc": "Zertrümmert blockierende Felsen und bricht neue Abkürzungen!" }, "infiltrator": { "id": "infiltrator", "name": "Schatten-Assassine", "req_camp_level": 3, "cost": {"gold": 60, "ore": 10, "energy": 18}, "hp": 110, "speed": 1.7, "melee_dmg": 50, "icon": "🥷", "special": "Rauch-Tarnung", "special_desc": "Wird beim Eintritt 5s lang von keinem Turm anvisiert." }, "gargoyle": { "id": "gargoyle", "name": "Fliegender Gargoyle", "req_camp_level": 4, "cost": {"gold": 80, "ore": 20, "energy": 30}, "hp": 150, "speed": 1.4, "melee_dmg": 35, "icon": "🦇", "special": "Luftangriff", "special_desc": "Fliegt über Mauern hinweg; immun gegen Bodengeschosse." } } # Neutral PvE training targets for 16x10 landscape grid PVE_TARGETS = [ { "id": -1, "username": "Räuber-Bande (PvE)", "fortress_name": "🏚️ Räuber-Versteck (Leicht)", "rating": 800, "camp_level": 1, "tile_count": 2, "is_npc": True, "gold": 120, "ore": 60, "energy": 40, "defense_units": {"scout": 1}, "tiles": [ {"x": 6, "y": 5, "tile_type": "mg", "level": 1}, {"x": 10, "y": 4, "tile_type": "rock", "level": 1} ] }, { "id": -2, "username": "Steinbruch-Clan (PvE)", "fortress_name": "⛏️ Granit-Schlucht (Mittel)", "rating": 1050, "camp_level": 2, "tile_count": 5, "is_npc": True, "gold": 260, "ore": 150, "energy": 80, "defense_units": {"scout": 1}, "tiles": [ {"x": 4, "y": 4, "tile_type": "rock", "level": 1}, {"x": 7, "y": 3, "tile_type": "mg", "level": 1}, {"x": 7, "y": 4, "tile_type": "rock", "level": 1}, {"x": 10, "y": 5, "tile_type": "ice", "level": 1}, {"x": 10, "y": 4, "tile_type": "rock", "level": 1} ] }, { "id": -3, "username": "Finster-Legion (PvE)", "fortress_name": "🏰 Finsterwald-Bastion (Schwer)", "rating": 1350, "camp_level": 3, "tile_count": 7, "is_npc": True, "gold": 500, "ore": 300, "energy": 200, "defense_units": {"tank": 1, "scout": 1}, "tiles": [ {"x": 3, "y": 3, "tile_type": "rock", "level": 1}, {"x": 3, "y": 5, "tile_type": "rock", "level": 1}, {"x": 6, "y": 4, "tile_type": "ice", "level": 1}, {"x": 8, "y": 2, "tile_type": "cannon", "level": 1}, {"x": 8, "y": 6, "tile_type": "mg", "level": 1}, {"x": 11, "y": 4, "tile_type": "tesla", "level": 1}, {"x": 11, "y": 5, "tile_type": "rock", "level": 1} ] } ] def ensure_user_army(conn, user_id: int): cur = conn.cursor() cur.execute("SELECT unit_type FROM user_army WHERE user_id = ?", (user_id,)) existing = set(r[0] for r in cur.fetchall()) for u_id in UNITS_CONFIG: if u_id not in existing: default_pool = 6 if u_id == "scout" else (2 if u_id == "tank" else 0) default_def = 1 if u_id == "scout" else 0 cur.execute(""" INSERT OR IGNORE INTO user_army (user_id, unit_type, pool_count, defense_count) VALUES (?, ?, ?, ?) """, (user_id, u_id, default_pool, default_def)) conn.commit() def update_user_state(conn, user_id: int): now = time.time() cur = conn.cursor() cur.execute("SELECT * FROM users WHERE id = ?", (user_id,)) u = cur.fetchone() if not u: return None # Camp upgrade check camp_lvl = u["camp_level"] if 0 < u["camp_upgrade_finishes_at"] <= now: camp_lvl += 1 cur.execute("UPDATE users SET camp_level = ?, camp_upgrade_finishes_at = 0 WHERE id = ?", (camp_lvl, user_id)) dt = max(0.0, now - u["last_tick_at"]) # Update machines (timers & production) cur.execute("SELECT * FROM machines WHERE user_id = ?", (user_id,)) machines = cur.fetchall() gold_gain = 0.0 ore_gain = 0.0 energy_gain = 0.0 rates = {"gold_min": 1.5, "ore_min": 1.0, "energy_min": 0.8} for m in machines: m_type = m["machine_type"] lvl = m["level"] m_cfg = MACHINE_CONFIG.get(m_type) if not m_cfg: continue if 0 < m["upgrade_finishes_at"] <= now: lvl += 1 cur.execute("UPDATE machines SET level = ?, upgrade_finishes_at = 0 WHERE id = ?", (lvl, m["id"])) is_sabotaged = m["sabotaged_until"] > now rate_sec = m_cfg["base_rate"] + (lvl - 1) * m_cfg["rate_per_lvl"] if is_sabotaged: rate_sec *= 0.5 rates[f"{m_cfg['resource']}_min"] += rate_sec * 60 gain = dt * rate_sec if m_cfg["resource"] == "gold": gold_gain += gain elif m_cfg["resource"] == "ore": ore_gain += gain elif m_cfg["resource"] == "energy": energy_gain += gain # Passive base tick gold_gain += dt * (1.5 / 60.0) ore_gain += dt * (1.0 / 60.0) energy_gain += dt * (0.8 / 60.0) cur.execute(""" UPDATE users SET gold = gold + ?, ore = ore + ?, energy = energy + ?, last_tick_at = ? WHERE id = ? """, (gold_gain, ore_gain, energy_gain, now, user_id)) return rates # ----------------- SCHEMAS ----------------- class RegisterReq(BaseModel): username: str password: str fortress_name: Optional[str] = '' class LoginReq(BaseModel): username: str password: str class SetTileReq(BaseModel): x: int y: int tile_type: Optional[str] = None # None to demolish class MachineUpgradeReq(BaseModel): machine_type: str class RecruitReq(BaseModel): unit_id: str count: int = 1 class SetDefenseReq(BaseModel): defense: Dict[str, int] # e.g. {"scout": 1, "tank": 1} class SabotageReportReq(BaseModel): defender_id: int creeps_leaked: int total_creeps: int units_sent: Dict[str, int] = {} # ----------------- AUTH ENDPOINTS ----------------- @app.post("/api/auth/register") def register(req: RegisterReq): uname = req.username.strip() if len(uname) < 3 or len(uname) > 20: raise HTTPException(status_code=400, detail="Benutzername muss zwischen 3 und 20 Zeichen lang sein.") if len(req.password) < 4: raise HTTPException(status_code=400, detail="Passwort muss mindestens 4 Zeichen lang sein.") fortress = (req.fortress_name or '').strip() or f"Feste {uname}" conn = get_connection() cur = conn.cursor() cur.execute("SELECT id FROM users WHERE username = ?", (uname,)) if cur.fetchone(): conn.close() raise HTTPException(status_code=400, detail="Dieser Benutzername ist bereits vergeben.") now = time.time() shield_until = now + 86400 * 2 # 48 hours newbie shield pwd_hash = hash_password(req.password) cur.execute(""" INSERT INTO users (username, password_hash, fortress_name, gold, ore, energy, rating, camp_level, camp_upgrade_finishes_at, shield_until, max_building_slots, last_tick_at, created_at) VALUES (?, ?, ?, 220.0, 120.0, 80.0, 1000, 1, 0, ?, 8, ?, ?) """, (uname, pwd_hash, fortress, shield_until, now, now)) user_id = cur.lastrowid # Create default machines for m_type in ["drill", "turbine", "press"]: cur.execute("INSERT INTO machines (user_id, machine_type, level) VALUES (?, ?, 1)", (user_id, m_type)) # Starter army ensure_user_army(conn, user_id) conn.commit() conn.close() token = create_access_token(user_id, uname) return {"token": token, "username": uname, "user_id": user_id, "fortress_name": fortress} @app.post("/api/auth/login") def login(req: LoginReq): conn = get_connection() cur = conn.cursor() cur.execute("SELECT * FROM users WHERE username = ?", (req.username.strip(),)) user = cur.fetchone() conn.close() if not user or not verify_password(req.password, user["password_hash"]): raise HTTPException(status_code=400, detail="Ungültiger Benutzername oder Passwort.") token = create_access_token(user["id"], user["username"]) return {"token": token, "username": user["username"], "user_id": user["id"], "fortress_name": user["fortress_name"]} @app.get("/api/auth/me") def get_me(user = Depends(get_current_user)): return user # ----------------- GAME STATE, MAZE & ARMY ----------------- @app.get("/api/game/state") def get_game_state(user = Depends(get_current_user)): conn = get_connection() try: ensure_user_army(conn, user["id"]) rates = update_user_state(conn, user["id"]) conn.commit() cur = conn.cursor() cur.execute("SELECT * FROM users WHERE id = ?", (user["id"],)) u = dict(cur.fetchone()) cur.execute("SELECT * FROM maze_tiles WHERE user_id = ?", (user["id"],)) tiles = [dict(r) for r in cur.fetchall()] cur.execute("SELECT * FROM machines WHERE user_id = ?", (user["id"],)) raw_machines = [dict(r) for r in cur.fetchall()] cur.execute("SELECT unit_type, pool_count, defense_count FROM user_army WHERE user_id = ?", (user["id"],)) army_rows = cur.fetchall() finally: conn.close() now = time.time() machines_dict = {} for m in raw_machines: m_type = m["machine_type"] cfg = MACHINE_CONFIG.get(m_type, {}) m["config"] = cfg m["is_sabotaged"] = m["sabotaged_until"] > now m["sabotage_remaining"] = max(0, math.ceil(m["sabotaged_until"] - now)) m["upgrade_remaining"] = max(0, math.ceil(m["upgrade_finishes_at"] - now)) machines_dict[m_type] = m camp_remaining = max(0, math.ceil(u["camp_upgrade_finishes_at"] - now)) # Army dict army = {} defense_guards = [] total_pool = 0 total_def = 0 for r in army_rows: ut = r["unit_type"] p_cnt = r["pool_count"] d_cnt = r["defense_count"] total_pool += p_cnt total_def += d_cnt army[ut] = { "pool": p_cnt, "defense": d_cnt, "available": max(0, p_cnt - d_cnt) } for _ in range(d_cnt): cfg = UNITS_CONFIG.get(ut, {}) defense_guards.append({ "unit_id": ut, "name": cfg.get("name", ut), "hp": cfg.get("hp", 100), "max_hp": cfg.get("hp", 100), "melee_dmg": cfg.get("melee_dmg", 20), "icon": cfg.get("icon", "🛡️") }) # Fill missing units if any for u_id in UNITS_CONFIG: if u_id not in army: army[u_id] = {"pool": 0, "defense": 0, "available": 0} shield_remaining = max(0, math.ceil(u.get("shield_until", 0) - now)) max_slots = u.get("max_building_slots", 8) next_slot_info = SLOT_UPGRADE_CONFIG.get(max_slots) income_rates = { "gold": round(rates.get("gold_min", 0.0), 1), "ore": round(rates.get("ore_min", 0.0), 1), "energy": round(rates.get("energy_min", 0.0), 1), "gold_min": round(rates.get("gold_min", 0.0), 1), "ore_min": round(rates.get("ore_min", 0.0), 1), "energy_min": round(rates.get("energy_min", 0.0), 1), } return { "user": { "id": u["id"], "username": u["username"], "fortress_name": u["fortress_name"], "rating": u["rating"], "camp_level": u["camp_level"], "camp_upgrade_remaining": camp_remaining, "shield_remaining": shield_remaining, "has_shield": shield_remaining > 0, "max_building_slots": max_slots, "used_building_slots": len(tiles), "next_slot_upgrade": next_slot_info }, "resources": { "gold": round(u["gold"], 1), "ore": round(u["ore"], 1), "energy": round(u["energy"], 1), "rates": income_rates }, "income_rates": income_rates, "maze": { "grid_w": GRID_W, "grid_h": GRID_H, "start": START_TILE, "end": END_TILE, "gate": GATE_TILE, "tiles": tiles }, "army": army, "total_army_pool": total_pool, "total_defense_count": total_def, "defense_guards": defense_guards, "machines": machines_dict, "camp_config": CAMP_CONFIG, "units_config": UNITS_CONFIG, "tower_config": TOWER_CONFIG, "machine_config": MACHINE_CONFIG, "slot_upgrade_config": SLOT_UPGRADE_CONFIG, "server_time": now } # ----------------- BUILDING SLOTS & MAZE ----------------- @app.post("/api/maze/buy-slot") def buy_building_slots(user = Depends(get_current_user)): conn = get_connection() update_user_state(conn, user["id"]) cur = conn.cursor() cur.execute("SELECT gold, ore, max_building_slots FROM users WHERE id = ?", (user["id"],)) u = cur.fetchone() current_slots = u["max_building_slots"] upgrade_info = SLOT_UPGRADE_CONFIG.get(current_slots) if not upgrade_info: conn.close() raise HTTPException(status_code=400, detail="Maximale Anzahl an Baufeldern bereits erreicht!") cost = upgrade_info["cost"] if u["gold"] < cost["gold"] or u["ore"] < cost["ore"]: conn.close() raise HTTPException(status_code=400, detail=f"Nicht genügend Ressourcen ({cost['gold']} Gold, {cost['ore']} Erz erforderlich).") new_slots = upgrade_info["next_slots"] cur.execute("UPDATE users SET gold = gold - ?, ore = ore - ?, max_building_slots = ? WHERE id = ?", (cost["gold"], cost["ore"], new_slots, user["id"])) conn.commit() conn.close() return {"success": True, "message": f"Festung auf {new_slots} Baufelder erweitert!", "max_building_slots": new_slots} @app.post("/api/maze/set-tile") def set_maze_tile(req: SetTileReq, user = Depends(get_current_user)): if req.x < 0 or req.x >= GRID_W or req.y < 0 or req.y >= GRID_H: raise HTTPException(status_code=400, detail="Koordinaten außerhalb des Spielfelds.") if (req.x, req.y) == START_TILE or (req.x, req.y) == END_TILE or (req.x, req.y) == GATE_TILE: raise HTTPException(status_code=400, detail="Start-, Tor- und Zielfeld dürfen nicht bebaut werden.") conn = get_connection() cur = conn.cursor() cur.execute("SELECT x, y, tile_type FROM maze_tiles WHERE user_id = ?", (user["id"],)) existing_rows = cur.fetchall() existing_map = {(r["x"], r["y"]): r["tile_type"] for r in existing_rows} existing_coords = set(existing_map.keys()) cur.execute("SELECT gold, max_building_slots FROM users WHERE id = ?", (user["id"],)) u_data = cur.fetchone() max_slots = u_data["max_building_slots"] simulated = set(existing_coords) if req.tile_type: simulated.add((req.x, req.y)) else: simulated.discard((req.x, req.y)) if not check_path_exists(simulated): conn.close() raise HTTPException(status_code=400, detail="Ungültiger Bau: Dieser Fels/Turm würde den Weg komplett versperren!") if req.tile_type: # Check building slot cap if placing a new tile is_new_tile = (req.x, req.y) not in existing_coords if is_new_tile and len(existing_coords) >= max_slots: conn.close() raise HTTPException(status_code=400, detail=f"Baufelder-Limit erreicht ({len(existing_coords)}/{max_slots} belegt)! Erweitere deine Festung für mehr Bauslots.") t_cfg = TOWER_CONFIG.get(req.tile_type) if not t_cfg: conn.close() raise HTTPException(status_code=400, detail="Unbekannter Turm-Typ.") cost = t_cfg["cost"] if u_data["gold"] < cost: conn.close() raise HTTPException(status_code=400, detail=f"Nicht genügend Gold (Kosten: {cost} Gold).") cur.execute("UPDATE users SET gold = gold - ? WHERE id = ?", (cost, user["id"])) cur.execute(""" INSERT OR REPLACE INTO maze_tiles (user_id, x, y, tile_type, level) VALUES (?, ?, ?, ?, 1) """, (user["id"], req.x, req.y, req.tile_type)) else: if (req.x, req.y) in existing_map: old_type = existing_map[(req.x, req.y)] t_cfg = TOWER_CONFIG.get(old_type, {}) refund = math.floor(t_cfg.get("cost", 0) * 0.5) cur.execute("UPDATE users SET gold = gold + ? WHERE id = ?", (refund, user["id"])) cur.execute("DELETE FROM maze_tiles WHERE user_id = ? AND x = ? AND y = ?", (user["id"], req.x, req.y)) conn.commit() conn.close() return {"success": True} # ----------------- RECRUITMENT & DEFENSE ASSIGNMENT ----------------- @app.post("/api/army/recruit") def recruit_troops(req: RecruitReq, user = Depends(get_current_user)): if req.count <= 0 or req.count > 50: raise HTTPException(status_code=400, detail="Ungültige Anzahl (1 - 50).") u_cfg = UNITS_CONFIG.get(req.unit_id) if not u_cfg: raise HTTPException(status_code=400, detail="Unbekannte Einheit.") conn = get_connection() ensure_user_army(conn, user["id"]) update_user_state(conn, user["id"]) cur = conn.cursor() cur.execute("SELECT * FROM users WHERE id = ?", (user["id"],)) u = cur.fetchone() camp_lvl = u["camp_level"] if camp_lvl < u_cfg["req_camp_level"]: conn.close() raise HTTPException(status_code=400, detail=f"Benötigt Trainingslager Stufe {u_cfg['req_camp_level']}.") camp_cfg = CAMP_CONFIG.get(camp_lvl, {}) pool_cap = camp_cfg.get("pool_cap", 20) cur.execute("SELECT SUM(pool_count) FROM user_army WHERE user_id = ?", (user["id"],)) current_pool = cur.fetchone()[0] or 0 if current_pool + req.count > pool_cap: conn.close() raise HTTPException(status_code=400, detail=f"Armee-Kapazität erschöpft ({current_pool}/{pool_cap} Einheiten). Baue das Trainingslager aus!") total_gold = u_cfg["cost"]["gold"] * req.count total_ore = u_cfg["cost"]["ore"] * req.count total_energy = u_cfg["cost"]["energy"] * req.count if u["gold"] < total_gold or u["ore"] < total_ore or u["energy"] < total_energy: conn.close() raise HTTPException(status_code=400, detail=f"Nicht genügend Ressourcen für {req.count}x {u_cfg['name']}.") cur.execute(""" UPDATE users SET gold = gold - ?, ore = ore - ?, energy = energy - ? WHERE id = ? """, (total_gold, total_ore, total_energy, user["id"])) cur.execute(""" UPDATE user_army SET pool_count = pool_count + ? WHERE user_id = ? AND unit_type = ? """, (req.count, user["id"], req.unit_id)) conn.commit() conn.close() return {"success": True, "message": f"{req.count}x {u_cfg['name']} erfolgreich ausgebildet!"} @app.post("/api/army/set-defense") def set_defense_guards(req: SetDefenseReq, user = Depends(get_current_user)): conn = get_connection() ensure_user_army(conn, user["id"]) cur = conn.cursor() cur.execute("SELECT camp_level FROM users WHERE id = ?", (user["id"],)) camp_lvl = cur.fetchone()["camp_level"] camp_cfg = CAMP_CONFIG.get(camp_lvl, {}) max_slots = camp_cfg.get("defense_slots", 2) total_def = sum(max(0, c) for c in req.defense.values()) if total_def > max_slots: conn.close() raise HTTPException(status_code=400, detail=f"Maximal {max_slots} Verteidigungs-Wächter auf Stufe {camp_lvl} erlaubt.") cur.execute("SELECT unit_type, pool_count FROM user_army WHERE user_id = ?", (user["id"],)) pools = {r["unit_type"]: r["pool_count"] for r in cur.fetchall()} for u_id, count in req.defense.items(): if count < 0: conn.close() raise HTTPException(status_code=400, detail="Ungültige Anzahl.") if count > pools.get(u_id, 0): conn.close() raise HTTPException(status_code=400, detail=f"Nicht genügend {u_id} im Vorrat für die Torwache vorhanden.") for u_id in UNITS_CONFIG: c = max(0, req.defense.get(u_id, 0)) cur.execute("UPDATE user_army SET defense_count = ? WHERE user_id = ? AND unit_type = ?", (c, user["id"], u_id)) conn.commit() conn.close() return {"success": True, "message": "Torwachen erfolgreich aufgestellt!"} # ----------------- CAMP & MACHINES UPGRADES ----------------- @app.post("/api/camp/upgrade") def upgrade_camp(user = Depends(get_current_user)): conn = get_connection() update_user_state(conn, user["id"]) cur = conn.cursor() cur.execute("SELECT * FROM users WHERE id = ?", (user["id"],)) u = cur.fetchone() now = time.time() if u["camp_upgrade_finishes_at"] > now: conn.close() raise HTTPException(status_code=400, detail="Trainingslager wird bereits ausgebaut.") next_lvl = u["camp_level"] + 1 cfg = CAMP_CONFIG.get(next_lvl) if not cfg or not cfg.get("cost"): conn.close() raise HTTPException(status_code=400, detail="Maximale Stufe bereits erreicht.") cost = cfg["cost"] if u["gold"] < cost["gold"] or u["ore"] < cost["ore"] or u["energy"] < cost["energy"]: conn.close() raise HTTPException(status_code=400, detail=f"Ressourcen unzureichend ({cost['gold']}G, {cost['ore']}E, {cost['energy']}⚡).") cur.execute(""" UPDATE users SET gold = gold - ?, ore = ore - ?, energy = energy - ?, camp_upgrade_finishes_at = ? WHERE id = ? """, (cost["gold"], cost["ore"], cost["energy"], now + cfg["duration"], user["id"])) conn.commit() conn.close() return {"success": True, "target_level": next_lvl, "duration": cfg["duration"]} @app.post("/api/machine/upgrade") def upgrade_machine(req: MachineUpgradeReq, user = Depends(get_current_user)): cfg = MACHINE_CONFIG.get(req.machine_type) if not cfg: raise HTTPException(status_code=400, detail="Ungültige Maschine.") conn = get_connection() update_user_state(conn, user["id"]) cur = conn.cursor() cur.execute("SELECT * FROM machines WHERE user_id = ? AND machine_type = ?", (user["id"], req.machine_type)) m = cur.fetchone() if not m: conn.close() raise HTTPException(status_code=404, detail="Maschine nicht gefunden.") now = time.time() if m["upgrade_finishes_at"] > now: conn.close() raise HTTPException(status_code=400, detail="Maschine wird bereits ausgebaut.") next_lvl = m["level"] + 1 multiplier = 1.75 ** (next_lvl - 2) if next_lvl > 1 else 1.0 cost_gold = math.floor(cfg["base_cost"]["gold"] * multiplier) cost_ore = math.floor(cfg["base_cost"]["ore"] * multiplier) cost_energy = math.floor(cfg["base_cost"]["energy"] * multiplier) duration = math.floor(cfg["base_time"] * (1.6 ** (next_lvl - 2))) cur.execute("SELECT gold, ore, energy FROM users WHERE id = ?", (user["id"],)) u = cur.fetchone() if u["gold"] < cost_gold or u["ore"] < cost_ore or u["energy"] < cost_energy: conn.close() raise HTTPException(status_code=400, detail=f"Nicht genügend Ressourcen ({cost_gold} Gold, {cost_ore} Erz, {cost_energy} Energie).") cur.execute(""" UPDATE users SET gold = gold - ?, ore = ore - ?, energy = energy - ? WHERE id = ? """, (cost_gold, cost_ore, cost_energy, user["id"])) cur.execute(""" UPDATE machines SET upgrade_finishes_at = ? WHERE id = ? """, (now + duration, m["id"])) conn.commit() conn.close() return {"success": True, "duration": duration, "finishes_at": now + duration} @app.post("/api/machine/repair") def repair_machine(req: MachineUpgradeReq, user = Depends(get_current_user)): conn = get_connection() cur = conn.cursor() cur.execute("SELECT * FROM machines WHERE user_id = ? AND machine_type = ?", (user["id"], req.machine_type)) m = cur.fetchone() if not m: conn.close() raise HTTPException(status_code=404, detail="Maschine nicht gefunden.") now = time.time() if m["sabotaged_until"] <= now: conn.close() return {"success": True, "message": "Maschine ist nicht beschädigt."} cur.execute("SELECT ore, energy FROM users WHERE id = ?", (user["id"],)) u = cur.fetchone() if u["ore"] < 15 or u["energy"] < 15: conn.close() raise HTTPException(status_code=400, detail="Reparatur erfordert 15 Erz und 15 Energie.") cur.execute("UPDATE users SET ore = ore - 15, energy = energy - 15 WHERE id = ?", (user["id"],)) cur.execute("UPDATE machines SET sabotaged_until = 0 WHERE id = ?", (m["id"],)) conn.commit() conn.close() return {"success": True, "message": "Maschine repariert! Volle Fördermenge wiederhergestellt."} # ----------------- PVP & PVE RAIDS ----------------- @app.get("/api/pvp/targets") def get_pvp_targets(user = Depends(get_current_user)): conn = get_connection() cur = conn.cursor() now = time.time() cur.execute("SELECT camp_level, rating FROM users WHERE id = ?", (user["id"],)) me = cur.fetchone() my_rating = me["rating"] # Real player targets (within fair range) cur.execute(""" SELECT u.id, u.username, u.fortress_name, u.gold, u.ore, u.energy, u.rating, u.camp_level, u.shield_until, (SELECT COUNT(*) FROM maze_tiles WHERE user_id = u.id) as tile_count, (SELECT SUM(defense_count) FROM user_army WHERE user_id = u.id) as defense_count FROM users u WHERE u.id != ? ORDER BY ABS(u.rating - ?) ASC LIMIT 10 """, (user["id"], my_rating)) raw_players = [dict(r) for r in cur.fetchall()] conn.close() players = [] for p in raw_players: shield_left = max(0, math.ceil(p.get("shield_until", 0) - now)) p["is_npc"] = False p["has_shield"] = shield_left > 0 p["shield_remaining"] = shield_left p["defense_count"] = p.get("defense_count") or 0 players.append(p) return { "pve_targets": PVE_TARGETS, "player_targets": players } @app.get("/api/pvp/maze/{target_id}") def get_target_maze(target_id: int, user = Depends(get_current_user)): if target_id < 0: for npc in PVE_TARGETS: if npc["id"] == target_id: guards = [] for ut, cnt in npc.get("defense_units", {}).items(): cfg = UNITS_CONFIG.get(ut, {}) for _ in range(cnt): is_starter = (npc["id"] == -1) guards.append({ "unit_id": ut, "name": "Räuber-Wächter" if is_starter else cfg.get("name", ut), "hp": 35 if is_starter else cfg.get("hp", 100), "max_hp": 35 if is_starter else cfg.get("hp", 100), "melee_dmg": 8 if is_starter else cfg.get("melee_dmg", 20), "icon": "🗡️" if is_starter else cfg.get("icon", "🛡️") }) return { "defender": npc, "grid_w": GRID_W, "grid_h": GRID_H, "start": START_TILE, "end": END_TILE, "gate": GATE_TILE, "tiles": npc["tiles"], "defense_guards": guards, "tower_config": TOWER_CONFIG } raise HTTPException(status_code=404, detail="PvE-Ziel nicht gefunden.") conn = get_connection() cur = conn.cursor() cur.execute("SELECT id, username, fortress_name, gold, ore, energy, rating, camp_level, shield_until FROM users WHERE id = ?", (target_id,)) target = cur.fetchone() if not target: conn.close() raise HTTPException(status_code=404, detail="Gegner nicht gefunden.") target_dict = dict(target) now = time.time() if target_dict.get("shield_until", 0) > now: conn.close() raise HTTPException(status_code=400, detail="Dieser Spieler steht noch unter Anfänger-Schutz!") cur.execute("SELECT x, y, tile_type, level FROM maze_tiles WHERE user_id = ?", (target_id,)) tiles = [dict(r) for r in cur.fetchall()] cur.execute("SELECT unit_type, defense_count FROM user_army WHERE user_id = ? AND defense_count > 0", (target_id,)) def_rows = cur.fetchall() guards = [] for r in def_rows: cfg = UNITS_CONFIG.get(r["unit_type"], {}) for _ in range(r["defense_count"]): guards.append({ "unit_id": r["unit_type"], "name": cfg.get("name", r["unit_type"]), "hp": cfg.get("hp", 100), "max_hp": cfg.get("hp", 100), "melee_dmg": cfg.get("melee_dmg", 20), "icon": cfg.get("icon", "🛡️") }) conn.close() return { "defender": target_dict, "grid_w": GRID_W, "grid_h": GRID_H, "start": START_TILE, "end": END_TILE, "gate": GATE_TILE, "tiles": tiles, "defense_guards": guards, "tower_config": TOWER_CONFIG } @app.post("/api/pvp/sabotage-report") def report_sabotage(req: SabotageReportReq, user = Depends(get_current_user)): if req.defender_id == user["id"]: raise HTTPException(status_code=400, detail="Man kann sich nicht selbst sabotieren.") conn = get_connection() try: ensure_user_army(conn, user["id"]) cur = conn.cursor() now = time.time() stolen = {"gold": 0, "ore": 0, "energy": 0} # Verify units sent cur.execute("SELECT unit_type, pool_count, defense_count FROM user_army WHERE user_id = ?", (user["id"],)) army_map = {r["unit_type"]: max(0, r["pool_count"] - r["defense_count"]) for r in cur.fetchall()} total_sent = 0 effective_units = {} for u_id, count in req.units_sent.items(): if count <= 0: continue available = army_map.get(u_id, 0) actual = min(count, available) if actual > 0: effective_units[u_id] = actual total_sent += actual if total_sent == 0: raise HTTPException(status_code=400, detail="Keine freien Einheiten im Vorrat für diesen Raubzug verfügbar!") # Target verification BEFORE deducting casualties is_npc = req.defender_id < 0 npc = None defender = None if is_npc: npc = next((n for n in PVE_TARGETS if n["id"] == req.defender_id), None) if not npc: raise HTTPException(status_code=404, detail="PvE-Ziel nicht gefunden.") else: cur.execute("SELECT id, username, fortress_name, gold, ore, energy, shield_until FROM users WHERE id = ?", (req.defender_id,)) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Verteidiger nicht gefunden.") defender = dict(row) if defender.get("shield_until", 0) > now: raise HTTPException(status_code=400, detail="Gegner hat aktiven Anfänger-Schutz!") # Transaction block for army casualties, loot transfer, and logging with conn: casualties = max(0, total_sent - req.creeps_leaked) if casualties > 0: for u_id, count in effective_units.items(): if total_sent > 0 and count > 0: loss = math.ceil(casualties * (count / total_sent)) loss = min(loss, count) cur.execute(""" UPDATE user_army SET pool_count = MAX(0, pool_count - ?) WHERE user_id = ? AND unit_type = ? """, (loss, user["id"], u_id)) if is_npc: if req.creeps_leaked > 0: ratio = min(1.0, req.creeps_leaked / max(1, req.total_creeps)) stolen["gold"] = math.floor(npc["gold"] * ratio) stolen["ore"] = math.floor(npc["ore"] * ratio) stolen["energy"] = math.floor(npc["energy"] * ratio) cur.execute(""" UPDATE users SET gold = gold + ?, ore = ore + ?, energy = energy + ?, rating = rating + 15 WHERE id = ? """, (stolen["gold"], stolen["ore"], stolen["energy"], user["id"])) result = "sabotage_success" else: result = "defense_held" else: if req.creeps_leaked > 0: leak_ratio = min(1.0, req.creeps_leaked / max(1, req.total_creeps)) stolen["gold"] = math.floor(defender["gold"] * 0.3 * leak_ratio) stolen["ore"] = math.floor(defender["ore"] * 0.3 * leak_ratio) stolen["energy"] = math.floor(defender["energy"] * 0.3 * leak_ratio) cur.execute(""" UPDATE users SET gold = MAX(0, gold - ?), ore = MAX(0, ore - ?), energy = MAX(0, energy - ?) WHERE id = ? """, (stolen["gold"], stolen["ore"], stolen["energy"], req.defender_id)) cur.execute(""" UPDATE users SET gold = gold + ?, ore = ore + ?, energy = energy + ?, rating = rating + 25 WHERE id = ? """, (stolen["gold"], stolen["ore"], stolen["energy"], user["id"])) cur.execute(""" UPDATE machines SET sabotaged_until = ? WHERE user_id = ? """, (now + 1200, req.defender_id)) result = "sabotage_success" else: cur.execute("UPDATE users SET rating = rating + 15 WHERE id = ?", (req.defender_id,)) result = "defense_held" cur.execute(""" INSERT INTO sabotage_logs (attacker_id, defender_id, result, creeps_leaked, total_creeps, units_sent, stolen_gold, stolen_ore, stolen_energy, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (user["id"], req.defender_id, result, req.creeps_leaked, req.total_creeps, json.dumps(req.units_sent), stolen["gold"], stolen["ore"], stolen["energy"], now)) return { "success": True, "result": result, "stolen": stolen, "creeps_leaked": req.creeps_leaked, "casualties": casualties } finally: conn.close() @app.get("/api/reports/history") def get_reports(user = Depends(get_current_user)): conn = get_connection() try: cur = conn.cursor() cur.execute(""" SELECT s.*, u_atk.username as attacker_name, u_def.username as defender_name, u_def.fortress_name as target_fortress FROM sabotage_logs s JOIN users u_atk ON s.attacker_id = u_atk.id JOIN users u_def ON s.defender_id = u_def.id WHERE s.attacker_id = ? OR s.defender_id = ? ORDER BY s.created_at DESC LIMIT 15 """, (user["id"], user["id"])) reports = [dict(r) for r in cur.fetchall()] return reports finally: conn.close() # ----------------- DEV TOOLS ----------------- @app.post("/api/dev/add-resources") def dev_add_resources(user = Depends(get_current_user)): conn = get_connection() cur = conn.cursor() cur.execute("UPDATE users SET gold = gold + 300, ore = ore + 150, energy = energy + 100 WHERE id = ?", (user["id"],)) conn.commit() conn.close() return {"success": True, "message": "+300 Gold, +150 Erz, +100 Energie erhalten!"} @app.post("/api/dev/add-army") def dev_add_army(user = Depends(get_current_user)): conn = get_connection() ensure_user_army(conn, user["id"]) cur = conn.cursor() cur.execute("UPDATE user_army SET pool_count = pool_count + 4 WHERE user_id = ? AND unit_type = 'scout'", (user["id"],)) cur.execute("UPDATE user_army SET pool_count = pool_count + 2 WHERE user_id = ? AND unit_type = 'tank'", (user["id"],)) cur.execute("UPDATE user_army SET pool_count = pool_count + 1 WHERE user_id = ? AND unit_type = 'crusher'", (user["id"],)) conn.commit() conn.close() return {"success": True, "message": "Truppen erhalten (+4 Späher, +2 Trolle, +1 Brecher)!"} @app.post("/api/dev/finish-upgrades") def dev_finish_upgrades(user = Depends(get_current_user)): conn = get_connection() cur = conn.cursor() now = time.time() - 1 cur.execute("UPDATE machines SET upgrade_finishes_at = ? WHERE user_id = ? AND upgrade_finishes_at > 0", (now, user["id"])) cur.execute("UPDATE users SET camp_upgrade_finishes_at = ? WHERE id = ? AND camp_upgrade_finishes_at > 0", (now, user["id"])) conn.commit() update_user_state(conn, user["id"]) conn.commit() conn.close() return {"success": True, "message": "Alle Maschinen- & Trainingslager-Upgrades sofort abgeschlossen!"} # ----------------- MYTHIC DEPTHS DUNGEON ENDPOINTS ----------------- ONLINE_HEROES: Dict[str, dict] = {} def ensure_dungeon_social_tables(conn): cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS dungeon_chat ( id INTEGER PRIMARY KEY AUTOINCREMENT, sender TEXT NOT NULL, hero_icon TEXT DEFAULT '⚔️', message TEXT NOT NULL, is_system INTEGER DEFAULT 0, created_at REAL NOT NULL ) """) cur.execute("SELECT COUNT(*) FROM dungeon_chat") if cur.fetchone()[0] == 0: now = time.time() cur.execute( "INSERT INTO dungeon_chat (sender, hero_icon, message, is_system, created_at) VALUES (?, ?, ?, 1, ?)", ("Wächter von Grimhold", "🏰", "Willkommen in der Stadt Grimhold-Feste! Tauscht euch hier im globalen Helden-Chat aus.", now) ) conn.commit() class FinishRunReq(BaseModel): player_name: Optional[str] = "Abenteurer" hero_name: Optional[str] = None hero_class: str = "Klingensänger" floor_reached: int = 1 hero_level: int = 1 monsters_slain: Optional[int] = None kills: Optional[int] = None gold_collected: Optional[int] = None gold: Optional[int] = None turns_count: Optional[int] = None turns: Optional[int] = None score: Optional[int] = None cause_of_death: Optional[str] = "Gefallen im Verlies" is_victory: Optional[bool] = None victory: Optional[bool] = None dungeon_name: Optional[str] = "Vergessene Katakomben" class SaveRunReq(BaseModel): hero_class: str run_state: str class PresenceReq(BaseModel): client_id: Optional[str] = "default" player_name: str = "Abenteurer" hero_class: str = "Klingensänger" hero_icon: str = "⚔️" level: int = 1 dungeon_id: str = "catacombs" dungeon_name: str = "Vergessene Katakomben" floor: int = 1 in_town: bool = False x: int = 12 y: int = 9 class ChatMessageReq(BaseModel): client_id: Optional[str] = "default" sender: str = "Abenteurer" hero_icon: str = "⚔️" message: str is_system: bool = False @app.get("/api/dungeon/highscores") def get_dungeon_highscores(): conn = get_connection() try: cur = conn.cursor() cur.execute(""" SELECT id, username, hero_class, floor_reached, hero_level, monsters_slain, gold_collected, turns_count, score, cause_of_death, is_victory, created_at FROM dungeon_runs ORDER BY score DESC, floor_reached DESC, turns_count ASC LIMIT 25 """) rows = [] for r in cur.fetchall(): d = dict(r) d["hero_name"] = d.get("username") or "Held" d["kills"] = d.get("monsters_slain", 0) d["gold"] = d.get("gold_collected", 0) d["victory"] = bool(d.get("is_victory", 0)) rows.append(d) return {"highscores": rows} finally: conn.close() @app.post("/api/dungeon/finish-run") def finish_dungeon_run(req: FinishRunReq, authorization: Optional[str] = Header(None)): user_id = 0 raw_name = req.hero_name or req.player_name or "Abenteurer" username = raw_name.strip()[:24] or "Abenteurer" if authorization and authorization.startswith("Bearer "): token = authorization.split(" ")[1] payload = decode_access_token(token) if payload: user_id = int(payload.get("sub", 0)) username = payload.get("username", username) kills = req.monsters_slain if req.monsters_slain is not None else (req.kills or 0) gold = req.gold_collected if req.gold_collected is not None else (req.gold or 0) turns = req.turns_count if req.turns_count is not None else (req.turns or 0) victory = req.is_victory if req.is_victory is not None else bool(req.victory) score = req.score if req.score is not None else (req.floor_reached * 250 + kills * 35 + gold * 2 + (2500 if victory else 0)) conn = get_connection() try: ensure_dungeon_social_tables(conn) with conn: cur = conn.cursor() now = time.time() cur.execute(""" INSERT INTO dungeon_runs ( user_id, username, hero_class, floor_reached, hero_level, monsters_slain, gold_collected, turns_count, score, cause_of_death, is_victory, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( user_id, username, req.hero_class, req.floor_reached, req.hero_level, kills, gold, turns, score, req.cause_of_death or "Gefallen im Verlies", 1 if victory else 0, now )) if user_id > 0: cur.execute("DELETE FROM dungeon_active_runs WHERE user_id = ?", (user_id,)) if victory: sys_msg = f"🏆 {username} ({req.hero_class}) hat {req.dungeon_name} auf Ebene {req.floor_reached} gemeistert! ({score} Pkt)" cur.execute("INSERT INTO dungeon_chat (sender, hero_icon, message, is_system, created_at) VALUES (?, ?, ?, 1, ?)", ("Herold", "🏆", sys_msg, now)) elif req.floor_reached >= 5: sys_msg = f"⚔️ {username} fiel auf Ebene {req.floor_reached} ({req.dungeon_name}) mit {score} Punkten." cur.execute("INSERT INTO dungeon_chat (sender, hero_icon, message, is_system, created_at) VALUES (?, ?, ?, 1, ?)", ("Chronist", "💀", sys_msg, now)) return {"success": True, "message": "Schlachtbericht der Erkundung gespeichert!"} finally: conn.close() def guest_uid_from_client_id(client_id: Optional[str]) -> int: cid = (client_id or "default").strip() h = 0 for ch in cid: h = (h * 31 + ord(ch)) & 0x7FFFFFFF return 1000000 + (h % 9000000) def resolve_dungeon_uid(client_id: Optional[str], authorization: Optional[str]) -> int: if authorization and authorization.startswith("Bearer "): token = authorization.split(" ")[1] payload = decode_access_token(token) if payload and "sub" in payload: return int(payload["sub"]) return guest_uid_from_client_id(client_id) @app.post("/api/dungeon/save-active") def save_active_dungeon_run(req: SaveRunReq, client_id: Optional[str] = "default", authorization: Optional[str] = Header(None)): uid = resolve_dungeon_uid(client_id, authorization) conn = get_connection() try: with conn: cur = conn.cursor() now = time.time() cur.execute(""" INSERT INTO dungeon_active_runs (user_id, hero_class, run_state, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET hero_class = excluded.hero_class, run_state = excluded.run_state, updated_at = excluded.updated_at """, (uid, req.hero_class, req.run_state, now)) return {"success": True, "uid": uid} finally: conn.close() @app.get("/api/dungeon/load-active") def load_active_dungeon_run(client_id: Optional[str] = "default", authorization: Optional[str] = Header(None)): uid = resolve_dungeon_uid(client_id, authorization) guest_uid = guest_uid_from_client_id(client_id) conn = get_connection() try: cur = conn.cursor() cur.execute("SELECT hero_class, run_state, updated_at FROM dungeon_active_runs WHERE user_id = ?", (uid,)) row = cur.fetchone() # Seamless Guest -> Registered Account migration if account has no saved run yet! if not row and uid != guest_uid: cur.execute("SELECT hero_class, run_state, updated_at FROM dungeon_active_runs WHERE user_id = ?", (guest_uid,)) guest_row = cur.fetchone() if guest_row: with conn: cur.execute(""" INSERT INTO dungeon_active_runs (user_id, hero_class, run_state, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET hero_class = excluded.hero_class, run_state = excluded.run_state, updated_at = excluded.updated_at """, (uid, guest_row["hero_class"], guest_row["run_state"], guest_row["updated_at"])) row = guest_row if not row: return {"has_saved_run": False} return { "has_saved_run": True, "hero_class": row["hero_class"], "run_state": row["run_state"], "updated_at": row["updated_at"] } finally: conn.close() @app.post("/api/dungeon/presence") def update_dungeon_presence(req: PresenceReq, authorization: Optional[str] = Header(None)): now = time.time() uid_key = str(resolve_dungeon_uid(req.client_id, authorization)) pname = (req.player_name or "Abenteurer").strip()[:20] or "Abenteurer" if authorization and authorization.startswith("Bearer "): token = authorization.split(" ")[1] payload = decode_access_token(token) if payload and payload.get("username"): pname = payload["username"] ONLINE_HEROES[uid_key] = { "uid": uid_key, "player_name": pname, "hero_class": req.hero_class, "hero_icon": req.hero_icon, "level": req.level, "dungeon_id": req.dungeon_id, "dungeon_name": req.dungeon_name, "floor": req.floor, "in_town": req.in_town, "x": req.x, "y": req.y, "updated_at": now } # Prune stale players (> 95s) stale = [k for k, v in ONLINE_HEROES.items() if now - v.get("updated_at", 0) > 95] for k in stale: ONLINE_HEROES.pop(k, None) conn = get_connection() try: ensure_dungeon_social_tables(conn) cur = conn.cursor() cur.execute(""" SELECT id, sender, hero_icon, message, is_system, created_at FROM dungeon_chat ORDER BY id DESC LIMIT 35 """) msgs = [dict(r) for r in reversed(cur.fetchall())] finally: conn.close() return { "online_players": list(ONLINE_HEROES.values()), "chat_messages": msgs } @app.post("/api/dungeon/chat") def post_dungeon_chat(req: ChatMessageReq, authorization: Optional[str] = Header(None)): text = (req.message or "").strip() if not text: raise HTTPException(status_code=400, detail="Leere Nachricht.") text = text[:220] sender = (req.sender or "Abenteurer").strip()[:20] or "Abenteurer" if authorization and authorization.startswith("Bearer "): token = authorization.split(" ")[1] payload = decode_access_token(token) if payload and payload.get("username"): sender = payload["username"] conn = get_connection() try: ensure_dungeon_social_tables(conn) now = time.time() with conn: cur = conn.cursor() cur.execute(""" INSERT INTO dungeon_chat (sender, hero_icon, message, is_system, created_at) VALUES (?, ?, ?, ?, ?) """, (sender, req.hero_icon or "⚔️", text, 1 if req.is_system else 0, now)) cur.execute("DELETE FROM dungeon_chat WHERE id NOT IN (SELECT id FROM dungeon_chat ORDER BY id DESC LIMIT 120)") cur.execute(""" SELECT id, sender, hero_icon, message, is_system, created_at FROM dungeon_chat ORDER BY id DESC LIMIT 35 """) msgs = [dict(r) for r in reversed(cur.fetchall())] return {"success": True, "chat_messages": msgs} finally: conn.close() # ----------------- ROOT SPA ----------------- @app.get("/", response_class=HTMLResponse) def index_page(): html_path = os.path.join(BASE_DIR, "templates", "index.html") with open(html_path, "r", encoding="utf-8") as f: return HTMLResponse(content=f.read())