603 lines
21 KiB
Python
603 lines
21 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 = 500 # Grid size in tiles
|
|
PLAYER_START_SPAWN_RANGE = 50
|
|
BOSS_SPAWN_DISTANCE = 150
|
|
ACTION_POINTS_MAX = 20
|
|
# 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"))
|
|
MONSTER_SPAWN_RATE = 0.1 # Probability per game tick
|
|
MAX_MONSTERS = 50
|
|
BOSS_HEALTH_BASE = 1000
|
|
MIN_LEVEL_FOR_BOSS = 10
|
|
BOSS_STRUCTURE_BONUS_MULTIPLIER = 0.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
|
|
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
|
|
active: bool = True
|
|
session_id: str = ""
|
|
|
|
def to_dict(self):
|
|
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,
|
|
"defense": self.defense,
|
|
"movement_capacity": self.movement_capacity,
|
|
"gathering_capacity": self.gathering_capacity,
|
|
"active": self.active,
|
|
}
|
|
|
|
@dataclass
|
|
class Monster:
|
|
id: str
|
|
position: Position
|
|
health: int
|
|
max_health: int
|
|
level: int
|
|
attack: int
|
|
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,
|
|
"is_boss": self.is_boss,
|
|
"boss_progress": self.boss_progress,
|
|
}
|
|
|
|
@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.boss: Monster = None
|
|
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()
|
|
|
|
def _init_resources(self):
|
|
"""Initialize static resources on the map"""
|
|
for _ in range(50):
|
|
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 start_game(self):
|
|
"""Start or restart the game"""
|
|
self.game_start_time = datetime.now().timestamp()
|
|
self.is_game_active = True
|
|
self.boss = None
|
|
self.monsters.clear()
|
|
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.health = player.max_health
|
|
player.action_points = ACTION_POINTS_MAX
|
|
player.inventory = {"wood": 0, "stone": 0}
|
|
player.position = Position(
|
|
random.randint(-PLAYER_START_SPAWN_RANGE, PLAYER_START_SPAWN_RANGE),
|
|
random.randint(-PLAYER_START_SPAWN_RANGE, PLAYER_START_SPAWN_RANGE),
|
|
0,
|
|
)
|
|
|
|
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 if conditions are met
|
|
if self.boss is None and len(self.players) > 0:
|
|
active_players = [p for p in self.players.values() if p.active]
|
|
if active_players and any(p.level >= MIN_LEVEL_FOR_BOSS for p in active_players):
|
|
self._spawn_boss()
|
|
|
|
# Spawn monsters
|
|
if random.random() < MONSTER_SPAWN_RATE and len(self.monsters) < MAX_MONSTERS:
|
|
self._spawn_monster()
|
|
|
|
# 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 = self._get_nearby_structure_bonuses(player.position)
|
|
if "action_regen" in nearby_bonuses:
|
|
regen_amount = nearby_bonuses["action_regen"]
|
|
player.action_points = min(
|
|
player.max_action_points,
|
|
player.action_points + regen_amount,
|
|
)
|
|
player.last_action_regen = current_time
|
|
|
|
def _spawn_monster(self):
|
|
"""Spawn a random monster on the map"""
|
|
monster_id = f"monster_{len(self.monsters)}_{self.generation}"
|
|
level = random.randint(1, 5)
|
|
x = random.randint(-GRID_SIZE // 2, GRID_SIZE // 2)
|
|
y = random.randint(-GRID_SIZE // 2, GRID_SIZE // 2)
|
|
health = 20 + level * 10
|
|
self.monsters[monster_id] = Monster(
|
|
id=monster_id,
|
|
position=Position(x, y, 0),
|
|
health=health,
|
|
max_health=health,
|
|
level=level,
|
|
attack=3 + level,
|
|
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)
|
|
health = int(BOSS_HEALTH_BASE + avg_level * 500)
|
|
|
|
# Spawn boss at a distance from players
|
|
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=health,
|
|
max_health=health,
|
|
level=int(avg_level) + 5,
|
|
attack=15 + int(avg_level),
|
|
is_boss=True,
|
|
)
|
|
|
|
def _get_nearby_structure_bonuses(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
|
|
|
|
# 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 = player.to_dict()
|
|
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: p.to_dict() 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()},
|
|
"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")
|
|
|
|
if player.action_points < 1:
|
|
raise HTTPException(status_code=400, detail="Insufficient action points")
|
|
|
|
# Calculate distance
|
|
distance = (dx ** 2 + dy ** 2) ** 0.5
|
|
if distance > player.movement_capacity:
|
|
raise HTTPException(status_code=400, detail="Movement exceeds capacity")
|
|
|
|
player.position.x += dx
|
|
player.position.y += dy
|
|
player.position.x = max(-GRID_SIZE // 2, min(GRID_SIZE // 2, player.position.x))
|
|
player.position.y = max(-GRID_SIZE // 2, min(GRID_SIZE // 2, player.position.y))
|
|
player.action_points -= 1
|
|
|
|
return {"position": player.position.to_dict(), "action_points": player.action_points}
|
|
|
|
@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]
|
|
if player.action_points < 1:
|
|
raise HTTPException(status_code=400, detail="Insufficient action points")
|
|
|
|
action_type = body.get("action_type")
|
|
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"))
|
|
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"""
|
|
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 > 5:
|
|
return {"success": False, "reason": "Target too far"}
|
|
|
|
damage = max(1, player.attack + random.randint(-2, 2) - monster.defense)
|
|
monster.health -= damage
|
|
|
|
if monster.health <= 0:
|
|
del game_world.monsters[target_id]
|
|
player.exp += monster.level * 10
|
|
player.level = 1 + int(player.exp / 100)
|
|
return {"success": True, "damage": damage, "exp": monster.level * 10, "monster_killed": True}
|
|
|
|
return {"success": True, "damage": damage, "monster_health_remaining": monster.health}
|
|
|
|
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 > 5:
|
|
return {"success": False, "reason": "Boss too far"}
|
|
|
|
damage = max(1, player.attack + 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
|
|
game_world.boss = None
|
|
return {"success": True, "damage": damage, "boss_killed": True}
|
|
|
|
return {"success": True, "damage": damage, "boss_health_remaining": monster.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
|
|
gathered_total = {"wood": 0, "stone": 0}
|
|
depleted = []
|
|
remaining_capacity = player.gathering_capacity
|
|
|
|
for rid, resource in list(game_world.resources.items()):
|
|
if remaining_capacity <= 0:
|
|
break
|
|
dist = ((resource.position.x - player.position.x) ** 2 +
|
|
(resource.position.y - player.position.y) ** 2) ** 0.5
|
|
if dist <= GATHER_RADIUS:
|
|
amount = min(remaining_capacity, resource.amount)
|
|
resource.amount -= amount
|
|
gathered_total[resource.resource_type] = gathered_total.get(resource.resource_type, 0) + amount
|
|
remaining_capacity -= 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 rtype, amt in gathered_total.items():
|
|
player.inventory[rtype] = player.inventory.get(rtype, 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"]:
|
|
return {"success": False, "reason": "Invalid structure type"}
|
|
|
|
costs = {"house": {"wood": 20, "stone": 10}, "farm": {"wood": 15, "stone": 5}, "guard_tower": {"wood": 30, "stone": 20}}
|
|
cost = costs[structure_type]
|
|
|
|
for material, amount in cost.items():
|
|
if player.inventory.get(material, 0) < amount:
|
|
return {"success": False, "reason": f"Insufficient {material}"}
|
|
|
|
# Deduct cost
|
|
for material, amount in cost.items():
|
|
player.inventory[material] -= amount
|
|
|
|
# Create structure
|
|
structure_id = f"struct_{len(game_world.structures)}"
|
|
bonuses = {}
|
|
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(tx, ty, 0),
|
|
structure_type=structure_type,
|
|
owner_id=player.id,
|
|
health=100,
|
|
bonuses=bonuses,
|
|
)
|
|
|
|
return {"success": True, "structure_id": structure_id, "inventory": player.inventory}
|
|
|
|
@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]
|
|
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: p.to_dict() 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()},
|
|
"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:
|
|
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"}
|
|
|
|
|