1006 lines
38 KiB
Python
1006 lines
38 KiB
Python
import os
|
|
import json
|
|
import asyncio
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Set
|
|
from dataclasses import dataclass, field
|
|
import random
|
|
import hashlib
|
|
from fastapi import FastAPI, WebSocket, HTTPException, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
# Configuration
|
|
DEBUG_MODE = os.getenv("DEBUG", "0").strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
GAME_DURATION = 30 * 60 # 30 minutes in seconds
|
|
GRID_SIZE = 200 # Grid size in tiles
|
|
PLAYER_START_SPAWN_RANGE = 50
|
|
BOSS_SPAWN_DISTANCE = 70
|
|
ACTION_POINTS_MAX = 30
|
|
# In debug mode AP regenerates much faster for rapid testing.
|
|
ACTION_POINTS_REGEN_INTERVAL = float(os.getenv("ACTION_REGEN_INTERVAL", "1" if DEBUG_MODE else "60"))
|
|
HP_REGEN_INTERVAL = float(os.getenv("HP_REGEN_INTERVAL", "2" if DEBUG_MODE else "15")) # seconds per 1 HP
|
|
HP_REGEN_AMOUNT = 1 # HP healed per interval
|
|
MONSTER_SPAWN_RATE = 0.05 # Probability per game tick
|
|
MAX_MONSTERS = 50
|
|
BOSS_FIXED_HEALTH = 6000
|
|
MIN_LEVEL_FOR_BOSS = 10
|
|
BOSS_STRUCTURE_BONUS_MULTIPLIER = 0.1
|
|
RESOURCE_NODE_COUNT = 540
|
|
CHEST_SPAWN_RATE = 0.015
|
|
MAX_CHESTS = 15
|
|
CHEST_LOOT_RANGE = 3
|
|
ATTACK_RANGE_BASE = 4
|
|
WEAPON_SLOT_COUNT = 5
|
|
BLOCK_BUILD_COST_WOOD = 2
|
|
BLOCK_HEALTH = 5
|
|
PLAYER_BASE_ATTACK = 5
|
|
PLAYER_BASE_DEFENSE = 2
|
|
PLAYER_BASE_MAX_HEALTH = 100
|
|
PLAYER_BASE_MOVE_CAPACITY = 1
|
|
|
|
# Data Models
|
|
@dataclass
|
|
class Position:
|
|
x: float
|
|
y: float
|
|
z: float = 0
|
|
|
|
def to_dict(self):
|
|
return {"x": self.x, "y": self.y, "z": self.z}
|
|
|
|
@classmethod
|
|
def from_dict(cls, d):
|
|
return cls(x=d["x"], y=d["y"], z=d.get("z", 0))
|
|
|
|
@dataclass
|
|
class Player:
|
|
id: str
|
|
username: str
|
|
color: str
|
|
level: int = 1
|
|
exp: int = 0
|
|
position: Position = field(default_factory=lambda: Position(0, 0, 0))
|
|
health: int = 100
|
|
max_health: int = 100
|
|
action_points: int = ACTION_POINTS_MAX
|
|
max_action_points: int = ACTION_POINTS_MAX
|
|
last_action_regen: float = 0
|
|
last_hp_regen: float = 0
|
|
inventory: Dict[str, int] = field(default_factory=lambda: {"wood": 0, "stone": 0})
|
|
attack: int = 5
|
|
defense: int = 2
|
|
movement_capacity: int = 1
|
|
gathering_capacity: int = 10
|
|
attack_range: float = ATTACK_RANGE_BASE
|
|
weapon_slots: List[Dict] = field(default_factory=lambda: [None] * WEAPON_SLOT_COUNT)
|
|
equipped_weapon_slot: int = -1
|
|
active: bool = True
|
|
session_id: str = ""
|
|
|
|
def to_dict(self):
|
|
equipped_weapon = None
|
|
if 0 <= self.equipped_weapon_slot < len(self.weapon_slots):
|
|
equipped_weapon = self.weapon_slots[self.equipped_weapon_slot]
|
|
attack_bonus = equipped_weapon.get("attack_bonus", 0) if equipped_weapon else 0
|
|
range_bonus = equipped_weapon.get("range_bonus", 0) if equipped_weapon else 0
|
|
return {
|
|
"id": self.id,
|
|
"username": self.username,
|
|
"color": self.color,
|
|
"level": self.level,
|
|
"exp": self.exp,
|
|
"position": self.position.to_dict(),
|
|
"health": self.health,
|
|
"max_health": self.max_health,
|
|
"action_points": self.action_points,
|
|
"max_action_points": self.max_action_points,
|
|
"inventory": self.inventory,
|
|
"attack": self.attack,
|
|
"effective_attack": self.attack + attack_bonus,
|
|
"attack_range": self.attack_range + range_bonus,
|
|
"defense": self.defense,
|
|
"movement_capacity": self.movement_capacity,
|
|
"gathering_capacity": self.gathering_capacity,
|
|
"weapon_slots": self.weapon_slots,
|
|
"equipped_weapon_slot": self.equipped_weapon_slot,
|
|
"active": self.active,
|
|
}
|
|
|
|
@dataclass
|
|
class Monster:
|
|
id: str
|
|
position: Position
|
|
health: int
|
|
max_health: int
|
|
level: int
|
|
attack: int
|
|
defense: int = 1
|
|
is_boss: bool = False
|
|
boss_progress: float = 0.0 # Percentage of damage done
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.id,
|
|
"position": self.position.to_dict(),
|
|
"health": self.health,
|
|
"max_health": self.max_health,
|
|
"level": self.level,
|
|
"attack": self.attack,
|
|
"defense": self.defense,
|
|
"is_boss": self.is_boss,
|
|
"boss_progress": self.boss_progress,
|
|
}
|
|
|
|
@dataclass
|
|
class Chest:
|
|
id: str
|
|
position: Position
|
|
opened: bool = False
|
|
rarity: str = "rare"
|
|
weapon: Dict = field(default_factory=dict)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.id,
|
|
"position": self.position.to_dict(),
|
|
"opened": self.opened,
|
|
"rarity": self.rarity,
|
|
"weapon": self.weapon,
|
|
}
|
|
|
|
@dataclass
|
|
class Structure:
|
|
id: str
|
|
position: Position
|
|
structure_type: str # "house", "farm", "guard_tower"
|
|
owner_id: str
|
|
health: int
|
|
bonuses: Dict = field(default_factory=dict)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.id,
|
|
"position": self.position.to_dict(),
|
|
"structure_type": self.structure_type,
|
|
"owner_id": self.owner_id,
|
|
"health": self.health,
|
|
"bonuses": self.bonuses,
|
|
}
|
|
|
|
@dataclass
|
|
class Resource:
|
|
id: str
|
|
position: Position
|
|
resource_type: str # "tree", "mountain"
|
|
amount: int
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.id,
|
|
"position": self.position.to_dict(),
|
|
"resource_type": self.resource_type,
|
|
"amount": self.amount,
|
|
}
|
|
|
|
class GameWorld:
|
|
def __init__(self):
|
|
self.players: Dict[str, Player] = {}
|
|
self.monsters: Dict[str, Monster] = {}
|
|
self.structures: Dict[str, Structure] = {}
|
|
self.resources: Dict[str, Resource] = {}
|
|
self.chests: Dict[str, Chest] = {}
|
|
self.boss: Monster = None
|
|
self.boss_spawned: bool = False
|
|
self.game_start_time: float = 0
|
|
self.is_game_active: bool = False
|
|
self.connected_sessions: Dict[str, str] = {} # session_id -> player_id
|
|
self.generation = 0
|
|
self._init_resources()
|
|
self._init_chests()
|
|
# Auto-start a game session immediately so the timer runs from launch.
|
|
self.game_start_time = datetime.now().timestamp()
|
|
self.is_game_active = True
|
|
self._spawn_boss()
|
|
self.boss_spawned = True
|
|
|
|
def _init_resources(self):
|
|
"""Initialize static resources on the map"""
|
|
for _ in range(RESOURCE_NODE_COUNT):
|
|
resource_id = f"resource_{len(self.resources)}"
|
|
x = random.randint(-GRID_SIZE // 2, GRID_SIZE // 2)
|
|
y = random.randint(-GRID_SIZE // 2, GRID_SIZE // 2)
|
|
resource_type = random.choice(["tree", "mountain"])
|
|
self.resources[resource_id] = Resource(
|
|
id=resource_id,
|
|
position=Position(x, y, 0),
|
|
resource_type=resource_type,
|
|
amount=random.randint(50, 200),
|
|
)
|
|
|
|
def _init_chests(self):
|
|
"""Spawn rare chests with weapon loot."""
|
|
for _ in range(8):
|
|
self._spawn_chest()
|
|
|
|
def _spawn_chest(self):
|
|
chest_id = f"chest_{len(self.chests)}_{self.generation}_{random.randint(100, 999)}"
|
|
x = random.randint(-GRID_SIZE // 2, GRID_SIZE // 2)
|
|
y = random.randint(-GRID_SIZE // 2, GRID_SIZE // 2)
|
|
weapon_templates = [
|
|
{"name": "Oversized Shrimp", "attack_bonus": 2, "range_bonus": 2},
|
|
{"name": "Hunting Bow", "attack_bonus": 1, "range_bonus": 5},
|
|
{"name": "Big Ass Axe", "attack_bonus": 3, "range_bonus": 1},
|
|
{
|
|
"name": "Yaoi Paddle",
|
|
"attack_bonus": 1,
|
|
"range_bonus": 1,
|
|
"aura_radius": 6,
|
|
"aura_bonuses": {"action_regen": 1, "hp_regen": 1},
|
|
},
|
|
{
|
|
"name": "Warden Guitar",
|
|
"attack_bonus": 0,
|
|
"range_bonus": 1,
|
|
"aura_radius": 6,
|
|
"aura_bonuses": {"defense": 2},
|
|
},
|
|
{
|
|
"name": "Windstep Boots",
|
|
"attack_bonus": 0,
|
|
"range_bonus": 0,
|
|
"aura_radius": 5,
|
|
"aura_bonuses": {"free_move_chance": 0.7},
|
|
},
|
|
{
|
|
"name": "Traveler Ring",
|
|
"attack_bonus": 0,
|
|
"range_bonus": 0,
|
|
"movement_bonus": 1,
|
|
},
|
|
]
|
|
self.chests[chest_id] = Chest(
|
|
id=chest_id,
|
|
position=Position(x, y, 0),
|
|
weapon=random.choice(weapon_templates),
|
|
)
|
|
|
|
def start_game(self):
|
|
"""Start or restart the game"""
|
|
self.game_start_time = datetime.now().timestamp()
|
|
self.is_game_active = True
|
|
self.boss = None
|
|
self.boss_spawned = False
|
|
self.monsters.clear()
|
|
self.chests.clear()
|
|
self.resources.clear()
|
|
self.structures.clear()
|
|
self._init_resources()
|
|
self._init_chests()
|
|
self.generation += 1
|
|
# Keep players but reset their state
|
|
for player in self.players.values():
|
|
if not player.active:
|
|
continue
|
|
player.level = 1
|
|
player.exp = 0
|
|
player.attack = PLAYER_BASE_ATTACK
|
|
player.defense = PLAYER_BASE_DEFENSE
|
|
player.max_health = PLAYER_BASE_MAX_HEALTH
|
|
player.max_action_points = ACTION_POINTS_MAX
|
|
player.movement_capacity = PLAYER_BASE_MOVE_CAPACITY
|
|
player.health = player.max_health
|
|
player.action_points = player.max_action_points
|
|
player.inventory = {"wood": 0, "stone": 0}
|
|
player.weapon_slots = [None] * WEAPON_SLOT_COUNT
|
|
player.equipped_weapon_slot = -1
|
|
player.attack_range = ATTACK_RANGE_BASE
|
|
player.position = Position(
|
|
random.randint(-PLAYER_START_SPAWN_RANGE, PLAYER_START_SPAWN_RANGE),
|
|
random.randint(-PLAYER_START_SPAWN_RANGE, PLAYER_START_SPAWN_RANGE),
|
|
0,
|
|
)
|
|
|
|
self._spawn_boss()
|
|
self.boss_spawned = True
|
|
|
|
def get_elapsed_time(self) -> float:
|
|
if not self.is_game_active:
|
|
return 0
|
|
return datetime.now().timestamp() - self.game_start_time
|
|
|
|
def check_game_over(self) -> bool:
|
|
if not self.is_game_active:
|
|
return False
|
|
elapsed = self.get_elapsed_time()
|
|
if elapsed >= GAME_DURATION:
|
|
self.is_game_active = False
|
|
return True
|
|
return False
|
|
|
|
def update_tick(self):
|
|
"""Called periodically to update game state"""
|
|
current_time = datetime.now().timestamp()
|
|
|
|
# Spawn boss once at the beginning of each session.
|
|
if self.boss is None and not self.boss_spawned:
|
|
self._spawn_boss()
|
|
self.boss_spawned = True
|
|
|
|
# Spawn monsters
|
|
if random.random() < MONSTER_SPAWN_RATE and len(self.monsters) < MAX_MONSTERS:
|
|
self._spawn_monster()
|
|
|
|
# Spawn rare chests over time
|
|
if random.random() < CHEST_SPAWN_RATE and len(self.chests) < MAX_CHESTS:
|
|
self._spawn_chest()
|
|
|
|
# Regenerate player action points
|
|
for player in self.players.values():
|
|
if not player.active:
|
|
continue
|
|
if current_time - player.last_action_regen >= ACTION_POINTS_REGEN_INTERVAL:
|
|
regen_amount = 1
|
|
nearby_bonuses = _player_nearby_bonuses(player)
|
|
if "action_regen" in nearby_bonuses:
|
|
regen_amount += int(nearby_bonuses["action_regen"])
|
|
player.action_points = min(
|
|
player.max_action_points,
|
|
player.action_points + regen_amount,
|
|
)
|
|
player.last_action_regen = current_time
|
|
|
|
# HP regeneration
|
|
if current_time - player.last_hp_regen >= HP_REGEN_INTERVAL:
|
|
effective_max_health = _player_effective_max_health(player)
|
|
hp_regen_amount = HP_REGEN_AMOUNT + int(_player_nearby_bonuses(player).get("hp_regen", 0))
|
|
if player.health < effective_max_health:
|
|
player.health = min(effective_max_health, player.health + hp_regen_amount)
|
|
player.last_hp_regen = current_time
|
|
|
|
def _spawn_monster(self):
|
|
"""Spawn a random monster on the map with level 1-10"""
|
|
monster_id = f"monster_{len(self.monsters)}_{self.generation}"
|
|
level = random.randint(1, 10)
|
|
x = random.randint(-GRID_SIZE // 2, GRID_SIZE // 2)
|
|
y = random.randint(-GRID_SIZE // 2, GRID_SIZE // 2)
|
|
# HP scales with level: base 15 + 5 per level = 20-65 for levels 1-10
|
|
health = 15 + level * 5
|
|
self.monsters[monster_id] = Monster(
|
|
id=monster_id,
|
|
position=Position(x, y, 0),
|
|
health=health,
|
|
max_health=health,
|
|
level=level,
|
|
attack=2 + level,
|
|
defense=1 + level // 3,
|
|
is_boss=False,
|
|
)
|
|
|
|
def _spawn_boss(self):
|
|
"""Spawn the boss monster."""
|
|
boss_id = f"boss_{self.generation}"
|
|
active_players = [p for p in self.players.values() if p.active]
|
|
avg_level = (sum(p.level for p in active_players) / len(active_players)) if active_players else 1
|
|
|
|
# Spawn boss at a distance from players if possible, otherwise at map center.
|
|
if active_players:
|
|
player_pos = active_players[0].position
|
|
angle = random.random() * 2 * 3.14159
|
|
x = player_pos.x + BOSS_SPAWN_DISTANCE * (angle ** 0.5) * (1 if random.random() > 0.5 else -1)
|
|
y = player_pos.y + BOSS_SPAWN_DISTANCE * (1 - angle ** 0.5) * (1 if random.random() > 0.5 else -1)
|
|
else:
|
|
x = y = 0
|
|
|
|
self.boss = Monster(
|
|
id=boss_id,
|
|
position=Position(x, y, 0),
|
|
health=BOSS_FIXED_HEALTH,
|
|
max_health=BOSS_FIXED_HEALTH,
|
|
level=max(10, int(avg_level) + 5),
|
|
attack=15 + int(avg_level),
|
|
defense=4 + int(avg_level // 2),
|
|
is_boss=True,
|
|
)
|
|
|
|
def _player_effective_attack(player: Player) -> int:
|
|
if 0 <= player.equipped_weapon_slot < len(player.weapon_slots):
|
|
w = player.weapon_slots[player.equipped_weapon_slot]
|
|
if w:
|
|
return player.attack + int(w.get("attack_bonus", 0))
|
|
return player.attack
|
|
|
|
def _player_effective_range(player: Player) -> float:
|
|
if 0 <= player.equipped_weapon_slot < len(player.weapon_slots):
|
|
w = player.weapon_slots[player.equipped_weapon_slot]
|
|
if w:
|
|
return player.attack_range + float(w.get("range_bonus", 0))
|
|
return player.attack_range
|
|
|
|
def _player_effective_movement(player: Player) -> int:
|
|
if 0 <= player.equipped_weapon_slot < len(player.weapon_slots):
|
|
w = player.weapon_slots[player.equipped_weapon_slot]
|
|
if w:
|
|
return max(1, player.movement_capacity + int(w.get("movement_bonus", 0)))
|
|
return player.movement_capacity
|
|
|
|
def _recompute_player_stats_for_level(player: Player) -> None:
|
|
# Slow progression so levels matter without runaway scaling.
|
|
level = max(1, int(player.level))
|
|
old_max_health = player.max_health
|
|
old_max_ap = player.max_action_points
|
|
|
|
player.attack = PLAYER_BASE_ATTACK + (level - 1) // 3
|
|
player.defense = PLAYER_BASE_DEFENSE + (level - 1) // 4
|
|
player.max_health = PLAYER_BASE_MAX_HEALTH + ((level - 1) // 2) * 5
|
|
player.max_action_points = ACTION_POINTS_MAX + (level - 1) // 4
|
|
|
|
if player.max_health > old_max_health:
|
|
player.health = min(player.max_health, player.health + (player.max_health - old_max_health))
|
|
else:
|
|
player.health = min(player.max_health, player.health)
|
|
|
|
if player.max_action_points > old_max_ap:
|
|
player.action_points = min(player.max_action_points, player.action_points + (player.max_action_points - old_max_ap))
|
|
else:
|
|
player.action_points = min(player.max_action_points, player.action_points)
|
|
|
|
def _recompute_player_level_from_exp(player: Player) -> None:
|
|
level = 1
|
|
xp_needed = 30
|
|
while player.exp >= xp_needed and level < 100:
|
|
level += 1
|
|
xp_needed += 20
|
|
player.level = level
|
|
_recompute_player_stats_for_level(player)
|
|
|
|
def _player_nearby_bonuses(player: Player) -> Dict:
|
|
if "game_world" not in globals():
|
|
return {}
|
|
bonuses = dict(game_world._get_nearby_structure_bonuses(player.position))
|
|
|
|
for other_player in game_world.players.values():
|
|
if not other_player.active:
|
|
continue
|
|
if not (0 <= other_player.equipped_weapon_slot < len(other_player.weapon_slots)):
|
|
continue
|
|
item = other_player.weapon_slots[other_player.equipped_weapon_slot]
|
|
if not item:
|
|
continue
|
|
aura_bonuses = item.get("aura_bonuses") or {}
|
|
aura_radius = float(item.get("aura_radius", 0))
|
|
if not aura_bonuses or aura_radius <= 0:
|
|
continue
|
|
|
|
dist = ((other_player.position.x - player.position.x) ** 2 + (other_player.position.y - player.position.y) ** 2) ** 0.5
|
|
if dist > aura_radius:
|
|
continue
|
|
|
|
for key, value in aura_bonuses.items():
|
|
bonuses[key] = bonuses.get(key, 0) + value
|
|
|
|
return bonuses
|
|
|
|
def _player_effective_defense(player: Player) -> int:
|
|
bonuses = _player_nearby_bonuses(player)
|
|
return player.defense + int(bonuses.get("defense", 0))
|
|
|
|
def _player_effective_max_health(player: Player) -> int:
|
|
bonuses = _player_nearby_bonuses(player)
|
|
return player.max_health + int(bonuses.get("max_health", 0))
|
|
|
|
def _serialize_player(player: Player) -> Dict:
|
|
data = player.to_dict()
|
|
bonuses = _player_nearby_bonuses(player)
|
|
effective_defense = _player_effective_defense(player)
|
|
effective_max_health = _player_effective_max_health(player)
|
|
effective_movement = _player_effective_movement(player)
|
|
|
|
if player.health > effective_max_health:
|
|
player.health = effective_max_health
|
|
|
|
data["base_defense"] = player.defense
|
|
data["defense"] = effective_defense
|
|
data["base_max_health"] = player.max_health
|
|
data["max_health"] = effective_max_health
|
|
data["health"] = player.health
|
|
data["base_movement_capacity"] = player.movement_capacity
|
|
data["movement_capacity"] = effective_movement
|
|
data["nearby_bonuses"] = bonuses
|
|
return data
|
|
|
|
def _get_nearby_structure_bonuses_impl(self, position: Position, radius: int = 10) -> Dict:
|
|
bonuses = {}
|
|
for structure in self.structures.values():
|
|
dist = ((structure.position.x - position.x) ** 2 + (structure.position.y - position.y) ** 2) ** 0.5
|
|
if dist <= radius:
|
|
for key, value in structure.bonuses.items():
|
|
bonuses[key] = bonuses.get(key, 0) + value
|
|
return bonuses
|
|
|
|
GameWorld._get_nearby_structure_bonuses = _get_nearby_structure_bonuses_impl
|
|
|
|
# Use plain dicts instead of Pydantic models
|
|
|
|
# Global game state
|
|
game_world = GameWorld()
|
|
print("DEBUG: GameWorld initialized")
|
|
|
|
app = FastAPI()
|
|
print("DEBUG: FastAPI app created")
|
|
|
|
# CORS
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# ==================== API ENDPOINTS ====================
|
|
|
|
@app.post("/api/login")
|
|
async def login(request: Request):
|
|
"""Login user and create/get player"""
|
|
print("===== LOGIN START =====", flush=True)
|
|
try:
|
|
print("[1] Parsing JSON...", flush=True)
|
|
body = await request.json()
|
|
print(f"[2] Got body: {body}", flush=True)
|
|
except Exception as e:
|
|
print(f"[X] Failed to parse JSON: {e}", flush=True)
|
|
raise HTTPException(status_code=400, detail="Invalid JSON")
|
|
|
|
print("[3] Extracting fields...", flush=True)
|
|
username = body.get("username", "").strip()
|
|
color = body.get("color", "")
|
|
print(f"[4] Got username={username}, color={color}", flush=True)
|
|
|
|
if len(username) < 1 or len(username) > 30:
|
|
print("[5a] Invalid username length", flush=True)
|
|
raise HTTPException(status_code=400, detail="Invalid username length")
|
|
|
|
print("[5b] Generating player ID...", flush=True)
|
|
player_id = hashlib.md5(f"{username}_{game_world.generation}".encode()).hexdigest()[:12]
|
|
print(f"[6] Player_id={player_id}", flush=True)
|
|
|
|
print("[7] Checking if player exists...", flush=True)
|
|
if player_id not in game_world.players:
|
|
print("[8] Creating new player...", flush=True)
|
|
player = Player(
|
|
id=player_id,
|
|
username=username,
|
|
color=color,
|
|
position=Position(
|
|
random.randint(-PLAYER_START_SPAWN_RANGE, PLAYER_START_SPAWN_RANGE),
|
|
random.randint(-PLAYER_START_SPAWN_RANGE, PLAYER_START_SPAWN_RANGE),
|
|
0,
|
|
),
|
|
)
|
|
print("[9] Adding player to world...", flush=True)
|
|
game_world.players[player_id] = player
|
|
print("[10] Player added", flush=True)
|
|
|
|
print("[11] Getting player from world...", flush=True)
|
|
player = game_world.players[player_id]
|
|
player.active = True
|
|
print("[12] Creating session...", flush=True)
|
|
session_id = hashlib.md5(f"{player_id}_{datetime.now().timestamp()}".encode()).hexdigest()[:16]
|
|
player.session_id = session_id
|
|
game_world.connected_sessions[session_id] = player_id
|
|
print("[13] Session created", flush=True)
|
|
|
|
print("[14] Converting player to dict...", flush=True)
|
|
player_dict = _serialize_player(player)
|
|
print(f"[15] Player dict has {len(player_dict)} keys", flush=True)
|
|
|
|
print("[16] Building response...", flush=True)
|
|
response = {
|
|
"player_id": player_id,
|
|
"session_id": session_id,
|
|
"player": player_dict,
|
|
"game_active": game_world.is_game_active,
|
|
"time_remaining": max(0, GAME_DURATION - game_world.get_elapsed_time()),
|
|
}
|
|
print("[17] Response built, returning...", flush=True)
|
|
return response
|
|
|
|
@app.get("/api/game/state")
|
|
async def get_game_state():
|
|
"""Get full game state"""
|
|
return {
|
|
"grid_size": GRID_SIZE,
|
|
"players": {pid: _serialize_player(p) for pid, p in game_world.players.items()},
|
|
"monsters": {mid: m.to_dict() for mid, m in game_world.monsters.items()},
|
|
"boss": game_world.boss.to_dict() if game_world.boss else None,
|
|
"structures": {sid: s.to_dict() for sid, s in game_world.structures.items()},
|
|
"resources": {rid: r.to_dict() for rid, r in game_world.resources.items()},
|
|
"chests": {cid: c.to_dict() for cid, c in game_world.chests.items()},
|
|
"game_active": game_world.is_game_active,
|
|
"time_remaining": max(0, GAME_DURATION - game_world.get_elapsed_time()),
|
|
}
|
|
|
|
@app.post("/api/player/{player_id}/move")
|
|
async def move_player(player_id: str, request: Request):
|
|
"""Move player"""
|
|
if player_id not in game_world.players:
|
|
raise HTTPException(status_code=404, detail="Player not found")
|
|
|
|
body = await request.json()
|
|
dx = body.get("dx", 0)
|
|
dy = body.get("dy", 0)
|
|
|
|
player = game_world.players[player_id]
|
|
if not player.active:
|
|
raise HTTPException(status_code=400, detail="Player not active")
|
|
|
|
nearby_bonuses = _player_nearby_bonuses(player)
|
|
free_move_chance = max(0.0, min(0.95, float(nearby_bonuses.get("free_move_chance", 0))))
|
|
ap_cost = 0 if random.random() < free_move_chance else 1
|
|
|
|
if player.action_points < ap_cost:
|
|
raise HTTPException(status_code=400, detail="Insufficient action points")
|
|
|
|
# Calculate distance
|
|
distance = (dx ** 2 + dy ** 2) ** 0.5
|
|
effective_movement = _player_effective_movement(player)
|
|
if distance > effective_movement:
|
|
raise HTTPException(status_code=400, detail="Movement exceeds capacity")
|
|
|
|
next_x = max(-GRID_SIZE // 2, min(GRID_SIZE // 2, player.position.x + dx))
|
|
next_y = max(-GRID_SIZE // 2, min(GRID_SIZE // 2, player.position.y + dy))
|
|
|
|
# Blocks are impassable.
|
|
for s in game_world.structures.values():
|
|
if s.structure_type != "block":
|
|
continue
|
|
bdist = ((s.position.x - next_x) ** 2 + (s.position.y - next_y) ** 2) ** 0.5
|
|
if bdist < 0.6:
|
|
raise HTTPException(status_code=400, detail="Blocked by structure")
|
|
|
|
player.position.x = next_x
|
|
player.position.y = next_y
|
|
player.action_points -= ap_cost
|
|
|
|
return {
|
|
"position": player.position.to_dict(),
|
|
"action_points": player.action_points,
|
|
"ap_cost": ap_cost,
|
|
"free_move_triggered": ap_cost == 0,
|
|
}
|
|
|
|
@app.post("/api/player/{player_id}/action")
|
|
async def player_action(player_id: str, request: Request):
|
|
"""Player performs an action"""
|
|
if player_id not in game_world.players:
|
|
raise HTTPException(status_code=404, detail="Player not found")
|
|
|
|
body = await request.json()
|
|
player = game_world.players[player_id]
|
|
action_type = body.get("action_type")
|
|
|
|
# Equip is free — no AP required
|
|
if action_type == "equip":
|
|
result = _handle_equip(player, body.get("slot_index"))
|
|
return result
|
|
|
|
if player.action_points < 1:
|
|
raise HTTPException(status_code=400, detail="Insufficient action points")
|
|
|
|
if action_type == "attack":
|
|
result = _handle_attack(player, body.get("target_id"))
|
|
elif action_type == "gather":
|
|
result = _handle_gather(player, body.get("target_id"))
|
|
elif action_type == "build":
|
|
result = _handle_build(player, body.get("tx"), body.get("ty"), body.get("structure_type"))
|
|
elif action_type == "loot":
|
|
result = _handle_loot(player)
|
|
elif action_type == "equip":
|
|
result = _handle_equip(player, body.get("slot_index"))
|
|
else:
|
|
return {"action": action_type, "success": False, "reason": "Unknown action"}
|
|
|
|
# Only consume AP if the action actually succeeded
|
|
if result.get("success"):
|
|
player.action_points -= 1
|
|
|
|
return result
|
|
|
|
def _handle_attack(player: Player, target_id: str) -> Dict:
|
|
"""Handle player attack"""
|
|
attack_range = _player_effective_range(player)
|
|
attack_power = _player_effective_attack(player)
|
|
effective_defense = _player_effective_defense(player)
|
|
|
|
def max_retaliation_damage(monster_attack: int) -> int:
|
|
# Retaliation roll is random in [-1, +2], so +2 is the worst-case outcome.
|
|
return max(0, monster_attack + 2 - effective_defense)
|
|
|
|
def apply_retaliation(monster_attack: int) -> Dict:
|
|
retaliation_damage = max(0, monster_attack + random.randint(-1, 2) - effective_defense)
|
|
if retaliation_damage <= 0:
|
|
return {"retaliation_damage": 0, "player_health": player.health, "player_defeated": False}
|
|
|
|
player.health = max(0, player.health - retaliation_damage)
|
|
defeated = player.health <= 0
|
|
respawn_position = None
|
|
|
|
if defeated:
|
|
player.health = player.max_health
|
|
player.position = Position(
|
|
random.randint(-PLAYER_START_SPAWN_RANGE, PLAYER_START_SPAWN_RANGE),
|
|
random.randint(-PLAYER_START_SPAWN_RANGE, PLAYER_START_SPAWN_RANGE),
|
|
0,
|
|
)
|
|
respawn_position = player.position.to_dict()
|
|
|
|
return {
|
|
"retaliation_damage": retaliation_damage,
|
|
"player_health": player.health,
|
|
"player_defeated": defeated,
|
|
"respawn_position": respawn_position,
|
|
}
|
|
|
|
if target_id in game_world.monsters:
|
|
monster = game_world.monsters[target_id]
|
|
dist = ((monster.position.x - player.position.x) ** 2 + (monster.position.y - player.position.y) ** 2) ** 0.5
|
|
|
|
if dist > attack_range:
|
|
return {"success": False, "reason": "Target too far"}
|
|
|
|
min_damage = max(1, attack_power - 2 - monster.defense)
|
|
# If monster certainly survives this hit and guaranteed max retaliation is lethal,
|
|
# block the attack before consuming AP.
|
|
if monster.health > min_damage and player.health <= max_retaliation_damage(monster.attack):
|
|
return {"success": False, "reason": "Attack blocked: retaliation would be lethal"}
|
|
|
|
damage = max(1, attack_power + random.randint(-2, 2) - monster.defense)
|
|
monster.health -= damage
|
|
|
|
if monster.health <= 0:
|
|
del game_world.monsters[target_id]
|
|
player.exp += monster.level * 10
|
|
_recompute_player_level_from_exp(player)
|
|
return {"success": True, "damage": damage, "exp": monster.level * 10, "monster_killed": True}
|
|
|
|
retaliation = apply_retaliation(monster.attack)
|
|
return {
|
|
"success": True,
|
|
"damage": damage,
|
|
"monster_health_remaining": monster.health,
|
|
**retaliation,
|
|
}
|
|
|
|
elif game_world.boss and target_id == game_world.boss.id:
|
|
monster = game_world.boss
|
|
dist = ((monster.position.x - player.position.x) ** 2 + (monster.position.y - player.position.y) ** 2) ** 0.5
|
|
|
|
if dist > attack_range:
|
|
return {"success": False, "reason": "Boss too far"}
|
|
|
|
min_damage = max(1, attack_power - 2 - monster.defense)
|
|
if monster.health > min_damage and player.health <= max_retaliation_damage(monster.attack + 2):
|
|
return {"success": False, "reason": "Attack blocked: boss retaliation would be lethal"}
|
|
|
|
damage = max(1, attack_power + random.randint(-2, 2) - monster.defense)
|
|
monster.health -= damage
|
|
old_progress = monster.boss_progress
|
|
monster.boss_progress = (monster.max_health - monster.health) / monster.max_health * 100
|
|
|
|
if monster.health <= 0:
|
|
# Victory!
|
|
for p in game_world.players.values():
|
|
if p.active:
|
|
p.level += 5
|
|
p.exp += 500
|
|
_recompute_player_stats_for_level(p)
|
|
game_world.boss = None
|
|
return {"success": True, "damage": damage, "boss_killed": True}
|
|
|
|
retaliation = apply_retaliation(monster.attack + 2)
|
|
return {
|
|
"success": True,
|
|
"damage": damage,
|
|
"boss_health_remaining": monster.health,
|
|
**retaliation,
|
|
}
|
|
|
|
if target_id in game_world.structures:
|
|
structure = game_world.structures[target_id]
|
|
if structure.structure_type != "block":
|
|
return {"success": False, "reason": "Only block structures are destructible"}
|
|
dist = ((structure.position.x - player.position.x) ** 2 + (structure.position.y - player.position.y) ** 2) ** 0.5
|
|
if dist > attack_range:
|
|
return {"success": False, "reason": "Block too far"}
|
|
structure.health -= 1
|
|
if structure.health <= 0:
|
|
del game_world.structures[target_id]
|
|
return {"success": True, "damage": 1, "block_destroyed": True}
|
|
return {"success": True, "damage": 1, "block_health_remaining": structure.health}
|
|
|
|
return {"success": False, "reason": "Target not found"}
|
|
|
|
def _handle_gather(player: Player, target_id: str) -> Dict:
|
|
"""Gather from all resources within GATHER_RADIUS of the player (target_id is ignored)."""
|
|
GATHER_RADIUS = 5
|
|
# Map resource_type -> inventory key
|
|
RESOURCE_MAP = {"tree": "wood", "mountain": "stone"}
|
|
gathered_total = {"wood": 0, "stone": 0}
|
|
depleted = []
|
|
|
|
for rid, resource in list(game_world.resources.items()):
|
|
dist = ((resource.position.x - player.position.x) ** 2 +
|
|
(resource.position.y - player.position.y) ** 2) ** 0.5
|
|
if dist <= GATHER_RADIUS:
|
|
inv_key = RESOURCE_MAP.get(resource.resource_type)
|
|
if not inv_key:
|
|
continue
|
|
# Gather from every node in range (per-node cap = gathering_capacity).
|
|
amount = min(player.gathering_capacity, resource.amount)
|
|
resource.amount -= amount
|
|
gathered_total[inv_key] = gathered_total.get(inv_key, 0) + amount
|
|
if resource.amount <= 0:
|
|
depleted.append(rid)
|
|
|
|
for rid in depleted:
|
|
del game_world.resources[rid]
|
|
|
|
total_gathered = sum(gathered_total.values())
|
|
if total_gathered == 0:
|
|
return {"success": False, "reason": "No resources within reach (radius 5)"}
|
|
|
|
for inv_key, amt in gathered_total.items():
|
|
player.inventory[inv_key] = player.inventory.get(inv_key, 0) + amt
|
|
|
|
return {"success": True, "gathered": gathered_total, "inventory": player.inventory}
|
|
|
|
def _handle_build(player: Player, tx: float, ty: float, structure_type: str) -> Dict:
|
|
"""Handle structure building"""
|
|
if structure_type not in ["house", "farm", "guard_tower", "block"]:
|
|
return {"success": False, "reason": "Invalid structure type"}
|
|
|
|
costs = {
|
|
"house": {"wood": 20, "stone": 10},
|
|
"farm": {"wood": 15, "stone": 5},
|
|
"guard_tower": {"wood": 30, "stone": 20},
|
|
"block": {"wood": BLOCK_BUILD_COST_WOOD, "stone": 0},
|
|
}
|
|
cost = costs[structure_type]
|
|
|
|
# Create structure (block is placed 1 tile north of player)
|
|
structure_id = f"struct_{len(game_world.structures)}"
|
|
bonuses = {}
|
|
px, py = tx, ty
|
|
if structure_type == "block":
|
|
px, py = player.position.x, player.position.y - 1
|
|
# Cannot place on top of players.
|
|
for p in game_world.players.values():
|
|
if ((p.position.x - px) ** 2 + (p.position.y - py) ** 2) ** 0.5 < 0.6:
|
|
return {"success": False, "reason": "Cannot place block on player"}
|
|
|
|
# No structure can be placed on an existing structure.
|
|
for s in game_world.structures.values():
|
|
if ((s.position.x - px) ** 2 + (s.position.y - py) ** 2) ** 0.5 < 0.6:
|
|
return {"success": False, "reason": "Tile already occupied by a structure"}
|
|
|
|
for material, amount in cost.items():
|
|
if player.inventory.get(material, 0) < amount:
|
|
return {"success": False, "reason": f"Insufficient {material}"}
|
|
|
|
# Deduct cost only after placement is validated.
|
|
for material, amount in cost.items():
|
|
player.inventory[material] -= amount
|
|
|
|
if structure_type == "farm":
|
|
bonuses["action_regen"] = 2
|
|
elif structure_type == "guard_tower":
|
|
bonuses["defense"] = 2
|
|
elif structure_type == "house":
|
|
bonuses["max_health"] = 20
|
|
|
|
game_world.structures[structure_id] = Structure(
|
|
id=structure_id,
|
|
position=Position(px, py, 0),
|
|
structure_type=structure_type,
|
|
owner_id=player.id,
|
|
health=BLOCK_HEALTH if structure_type == "block" else 100,
|
|
bonuses=bonuses,
|
|
)
|
|
|
|
return {"success": True, "structure_id": structure_id, "inventory": player.inventory}
|
|
|
|
def _handle_loot(player: Player) -> Dict:
|
|
nearby = []
|
|
for chest in game_world.chests.values():
|
|
if chest.opened:
|
|
continue
|
|
dist = ((chest.position.x - player.position.x) ** 2 + (chest.position.y - player.position.y) ** 2) ** 0.5
|
|
if dist <= CHEST_LOOT_RANGE:
|
|
nearby.append((dist, chest))
|
|
|
|
if not nearby:
|
|
return {"success": False, "reason": "No chest nearby"}
|
|
|
|
nearby.sort(key=lambda x: x[0])
|
|
chest = nearby[0][1]
|
|
empty_slot = next((i for i, item in enumerate(player.weapon_slots) if item is None), None)
|
|
if empty_slot is None:
|
|
return {"success": False, "reason": "Weapon slots full"}
|
|
|
|
player.weapon_slots[empty_slot] = chest.weapon
|
|
chest.opened = True
|
|
return {"success": True, "weapon": chest.weapon, "slot_index": empty_slot, "weapon_slots": player.weapon_slots}
|
|
|
|
def _handle_equip(player: Player, slot_index) -> Dict:
|
|
if slot_index is None:
|
|
return {"success": False, "reason": "slot_index is required"}
|
|
if not isinstance(slot_index, int) or slot_index < 0 or slot_index >= len(player.weapon_slots):
|
|
return {"success": False, "reason": "Invalid slot index"}
|
|
if player.weapon_slots[slot_index] is None:
|
|
return {"success": False, "reason": "No weapon in that slot"}
|
|
player.equipped_weapon_slot = slot_index
|
|
return {"success": True, "equipped_weapon_slot": slot_index, "weapon": player.weapon_slots[slot_index]}
|
|
|
|
@app.websocket("/ws/{player_id}")
|
|
async def websocket_endpoint(websocket: WebSocket, player_id: str):
|
|
"""WebSocket for real-time game updates"""
|
|
if player_id not in game_world.players:
|
|
await websocket.close(code=4004, reason="Player not found")
|
|
return
|
|
|
|
player = game_world.players[player_id]
|
|
# Capture the session at the time this WS connection is established so that
|
|
# if the player re-logs (new session) while this socket is still open, the
|
|
# finally block doesn't incorrectly mark the re-logged player as inactive.
|
|
ws_session_id = player.session_id
|
|
await websocket.accept()
|
|
|
|
try:
|
|
while True:
|
|
# Send game state updates every 100ms
|
|
await asyncio.sleep(0.1)
|
|
game_world.update_tick()
|
|
|
|
if game_world.check_game_over():
|
|
await websocket.send_json({"type": "game_over", "time_remaining": 0})
|
|
break
|
|
|
|
state = {
|
|
"type": "state_update",
|
|
"players": {pid: _serialize_player(p) for pid, p in game_world.players.items()},
|
|
"monsters": {mid: m.to_dict() for mid, m in game_world.monsters.items()},
|
|
"boss": game_world.boss.to_dict() if game_world.boss else None,
|
|
"structures": {sid: s.to_dict() for sid, s in game_world.structures.items()},
|
|
"resources": {rid: r.to_dict() for rid, r in game_world.resources.items()},
|
|
"chests": {cid: c.to_dict() for cid, c in game_world.chests.items()},
|
|
"time_remaining": max(0, GAME_DURATION - game_world.get_elapsed_time()),
|
|
"elapsed_time": game_world.get_elapsed_time(),
|
|
}
|
|
await websocket.send_json(state)
|
|
except Exception as e:
|
|
print(f"WebSocket error: {e}")
|
|
finally:
|
|
# Only deactivate if this WS belongs to the current session (not superseded by a re-login).
|
|
if player.session_id == ws_session_id:
|
|
player.active = False
|
|
|
|
@app.post("/api/game/start")
|
|
async def start_game():
|
|
"""Start a new game"""
|
|
game_world.start_game()
|
|
return {"game_active": True, "time_remaining": GAME_DURATION}
|
|
|
|
@app.get("/api/health")
|
|
async def health():
|
|
"""Health check"""
|
|
return {"status": "ok"}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|