#!/usr/bin/env python
"""
FastAPI web server for Korean Learning MUD game.
Provides REST API endpoints for the HTML frontend.
"""
import os
import sys
import logging
from typing import Dict, Any, List
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
import uvicorn
# Add src to Python path
sys.path.append(os.path.join(os.path.dirname(__file__), 'src'))
try:
from korean_cpc_agents.mud_game import KoreanLearningMUD
except ImportError as e:
print(f"Error importing MUD game: {e}")
print("Make sure you're running from the project root directory")
sys.exit(1)
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# FastAPI app
app = FastAPI(title="Korean Learning MUD Game", version="1.0.0")
# CORS middleware for frontend
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Mount static files for assets
app.mount("/assets", StaticFiles(directory="assets"), name="assets")
# Global game instance
game_instance = None
# Pydantic models for API
class CommandRequest(BaseModel):
command: str
class TalkRequest(BaseModel):
message: str
class GameResponse(BaseModel):
success: bool
message: str
data: Dict[str, Any] = {}
class GameState(BaseModel):
current_room: str
discovered_words: List[str]
game_progress: Dict[str, Any]
@app.on_event("startup")
async def startup_event():
"""Initialize the game on server startup"""
global game_instance
try:
logger.info("Initializing Korean Learning MUD game...")
game_instance = KoreanLearningMUD(web_mode=True) # Enable web mode to avoid terminal encoding
logger.info("Game initialized successfully!")
except Exception as e:
logger.error(f"Failed to initialize game: {e}")
raise
@app.get("/")
async def read_root():
"""Serve the main game HTML page"""
try:
import os
logger.info(f"Current working directory: {os.getcwd()}")
logger.info(f"Looking for file: korean_mud_game.html")
logger.info(f"File exists: {os.path.exists('korean_mud_game.html')}")
with open("korean_mud_game.html", "r", encoding="utf-8") as f:
html_content = f.read()
logger.info(f"Successfully read {len(html_content)} characters from HTML file")
return HTMLResponse(content=html_content)
except FileNotFoundError as e:
logger.error(f"FileNotFoundError: {e}")
return HTMLResponse(content="
Game file not found. Please run the server from the project root.
")
except Exception as e:
logger.error(f"Unexpected error reading HTML file: {e}")
return HTMLResponse(content=f"Error loading game: {str(e)}
")
@app.get("/api/game/state")
async def get_game_state() -> GameState:
"""Get current game state"""
if not game_instance:
raise HTTPException(status_code=500, detail="Game not initialized")
try:
return GameState(
current_room=game_instance.current_room,
discovered_words=list(game_instance.discovered_words),
game_progress=game_instance.game_progress
)
except Exception as e:
logger.error(f"Error getting game state: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/game/room-info")
async def get_room_info() -> GameResponse:
"""Get current room information"""
if not game_instance:
raise HTTPException(status_code=500, detail="Game not initialized")
try:
room_info = game_instance.look_around()
# Get room mapping and NPC info
current_room = game_instance.current_room
agent_name = game_instance.room_mapping.get(current_room)
npc_info = {}
if agent_name:
npc_info = game_instance._get_npc_info(agent_name)
# Get objects in current room
room_objects = game_instance.room_objects.get(current_room, {})
objects_list = [{"name": obj_data["name"], "description": obj_data["description"]}
for obj_data in room_objects.values()]
return GameResponse(
success=True,
message=room_info,
data={
"current_room": current_room,
"npc_info": npc_info,
"room_mapping": game_instance.room_mapping,
"discovered_words": list(game_instance.discovered_words),
"objects": objects_list
}
)
except Exception as e:
logger.error(f"Error getting room info: {e}")
return GameResponse(success=False, message=f"Error: {str(e)}")
@app.post("/api/game/move")
async def move_to_room(request: CommandRequest) -> GameResponse:
"""Move to a different room"""
if not game_instance:
raise HTTPException(status_code=500, detail="Game not initialized")
try:
# Extract room name from command (e.g., "go kitchen" -> "kitchen")
parts = request.command.strip().split()
if len(parts) < 2:
return GameResponse(success=False, message="Please specify which room to go to")
room_name = parts[1].lower()
result = game_instance.go_to_room(room_name)
# Get NPC info for the new room
current_room = game_instance.current_room
agent_name = game_instance.room_mapping.get(current_room)
npc_info = {}
if agent_name:
npc_info = game_instance._get_npc_info(agent_name)
return GameResponse(
success=True,
message=result,
data={
"current_room": current_room,
"discovered_words": list(game_instance.discovered_words),
"room_objects": [obj['name'] for obj in game_instance.room_objects.get(current_room, {}).values()],
"room_mapping": game_instance.room_mapping,
"npc_info": npc_info
}
)
except Exception as e:
logger.error(f"Error moving to room: {e}")
return GameResponse(success=False, message=f"Error: {str(e)}")
@app.post("/api/game/talk")
async def talk_to_npc(request: TalkRequest) -> GameResponse:
"""Talk to the NPC in current room"""
if not game_instance:
raise HTTPException(status_code=500, detail="Game not initialized")
try:
logger.info(f"Player message: {request.message}")
response = game_instance.talk_to_npc(request.message)
return GameResponse(
success=True,
message=response,
data={
"current_room": game_instance.current_room,
"discovered_words": list(game_instance.discovered_words),
"room_objects": [obj['name'] for obj in game_instance.room_objects.get(game_instance.current_room, {}).values()]
}
)
except Exception as e:
logger.error(f"Error talking to NPC: {e}")
return GameResponse(success=False, message=f"NPC couldn't respond: {str(e)}")
@app.get("/api/game/rooms")
async def list_rooms() -> GameResponse:
"""Get list of available rooms"""
if not game_instance:
raise HTTPException(status_code=500, detail="Game not initialized")
try:
rooms = list(game_instance.room_mapping.keys())
return GameResponse(
success=True,
message="Available rooms",
data={"rooms": rooms, "room_mapping": game_instance.room_mapping}
)
except Exception as e:
logger.error(f"Error listing rooms: {e}")
return GameResponse(success=False, message=f"Error: {str(e)}")
@app.get("/api/game/help")
async def get_help() -> GameResponse:
"""Get help information with proper formatting"""
help_text = """๐ฎ KOREAN LEARNING MUD GAME HELP
๐ BASIC COMMANDS:
โข look / l โ Look around your current room
โข examine [object] โ Examine an object in detail
โข take [object] โ Try to take an object (learn cultural context)
โข go [room] โ Move to another room (hall, kitchen, garden, bedroom, study, classroom)
โข talk [message] โ Chat with Korean family member in your room
โข rooms / map โ See all available rooms and who's there
โข help / ? โ Show this help menu
๐ QUICK SHORTCUTS:
โข n/s/e/w โ Go north/south/east/west
โข Just type what you want to say directly (no "talk" needed!)
๐ KOREAN FAMILY HOUSE LAYOUT:
๐ธ Garden (์ ์)
๐ด Grandpa Park
|
๐ณ Kitchen ---- ๐ Hall ---- ๐ Study
๐ต Grandma ๐ Brother Jung
|
โ๏ธ Classroom
๐ฉโ๐ซ Teacher Choi
|
๐ต Bedroom
๐ฉ Sister Lee
๐ฏ GAME OBJECTIVE:
Learn Korean by talking to different family members! Each has their own personality:
โข ๐ต Grandma Kim (Kitchen): Teaches honorifics and formal speech
โข ๐ด Grandpa Park (Garden): Shares traditional culture and wisdom
โข ๐ฉ Sister Lee (Bedroom): Teaches K-pop slang and modern Korean
โข ๐ Brother Jung (Study): Grammar expert and linguistic genius
โข ๐ฉโ๐ซ Teacher Choi (Classroom): General Korean lessons and practical phrases
๐ฌ HOW TO PLAY:
Just type naturally! Say "Hello", "Teach me Korean", "What's your favorite food?"
Examine objects in each room to learn about Korean culture!
The game will understand most commands. When in doubt, just talk to the NPCs!
๐ INTERACTIVE OBJECTS:
Each room has objects you can examine. Try 'examine shoes' or 'look at family photo'.
Some objects trigger special conversations with family members!"""
return GameResponse(
success=True,
message=help_text
)
@app.post("/api/game/examine")
async def examine_object(request: CommandRequest) -> GameResponse:
"""Examine an object in the current room"""
if not game_instance:
raise HTTPException(status_code=500, detail="Game not initialized")
try:
# Extract object name from command
parts = request.command.strip().split()
if len(parts) < 2:
return GameResponse(success=False, message="Please specify which object to examine")
object_name = " ".join(parts[1:])
result = game_instance.examine_object(object_name)
return GameResponse(
success=True,
message=result,
data={
"current_room": game_instance.current_room,
"discovered_words": list(game_instance.discovered_words),
"room_objects": [obj['name'] for obj in game_instance.room_objects.get(game_instance.current_room, {}).values()],
"player_inventory": list(game_instance.player_inventory)
}
)
except Exception as e:
logger.error(f"Error examining object: {e}")
return GameResponse(success=False, message=f"Error: {str(e)}")
@app.post("/api/game/take")
async def take_object(request: CommandRequest) -> GameResponse:
"""Try to take an object in the current room"""
if not game_instance:
raise HTTPException(status_code=500, detail="Game not initialized")
try:
# Extract object name from command
parts = request.command.strip().split()
if len(parts) < 2:
return GameResponse(success=False, message="Please specify which object to take")
object_name = " ".join(parts[1:])
result = game_instance.take_object(object_name)
return GameResponse(
success=True,
message=result,
data={
"current_room": game_instance.current_room,
"discovered_words": list(game_instance.discovered_words),
"room_objects": [obj['name'] for obj in game_instance.room_objects.get(game_instance.current_room, {}).values()],
"player_inventory": list(game_instance.player_inventory)
}
)
except Exception as e:
logger.error(f"Error taking object: {e}")
return GameResponse(success=False, message=f"Error: {str(e)}")
@app.post("/api/game/command")
async def process_command(request: CommandRequest) -> GameResponse:
"""Process a general game command with flexible parsing"""
if not game_instance:
raise HTTPException(status_code=500, detail="Game not initialized")
try:
command = request.command.strip()
command_lower = command.lower()
# More flexible command parsing
if any(cmd in command_lower for cmd in ['look', 'see', 'observe', '๋ณด๊ธฐ', 'l']):
result = game_instance.look_around()
return GameResponse(success=True, message=result)
elif any(cmd in command_lower for cmd in ['go', 'move', 'travel', 'enter', '์ด๋']):
return await move_to_room(request)
elif any(cmd in command_lower for cmd in ['rooms', 'list', 'map', '๋ฐฉ๋ชฉ๋ก', 'where']):
# Return formatted room list with map
rooms_data = await list_rooms()
if rooms_data.success:
room_list = "\n".join([f"โข {room.title()} - {game_instance._get_npc_info(game_instance.room_mapping[room])['name'] if game_instance.room_mapping[room] else 'No NPC'}" for room in rooms_data.data['rooms']])
map_text = f"๐ KOREAN FAMILY HOUSE MAP ๐ฐ๐ท\n\n" + \
f" ๐ธ Garden (์ ์)\n" + \
f" ๐ด Grandpa Park\n" + \
f" |\n" + \
f" ๐ณ Kitchen ---- ๐ Hall ---- ๐ Study\n" + \
f" ๐ต Grandma Kim ๐ Brother Jung\n" + \
f" |\n" + \
f" โ๏ธ Classroom (๊ต์ค)\n" + \
f" ๐ฉโ๐ซ Teacher Choi\n" + \
f" |\n" + \
f" ๐ต Bedroom (์นจ์ค)\n" + \
f" ๐ฉ Sister Lee\n\n" + \
f"๐ช Available Rooms:\n{room_list}\n\nUse 'go [room]' to move around!"
return GameResponse(success=True, message=map_text)
return rooms_data
elif any(cmd in command_lower for cmd in ['help', 'commands', '๋์', '?', 'h']):
return await get_help()
elif any(cmd in command_lower for cmd in ['talk', 'say', 'speak', 'tell', '๋ํ', 'chat']):
# Extract message after talk command
talk_words = ['talk', 'say', 'speak', 'tell', '๋ํ', 'chat']
message = command
for word in talk_words:
if word in command_lower:
parts = command.split(word, 1)
if len(parts) > 1:
message = parts[1].strip()
if message:
break
if not message or message == command:
message = "Hello!"
# Use talk endpoint
talk_request = TalkRequest(message=message)
return await talk_to_npc(talk_request)
elif any(cmd in command_lower for cmd in ['north', 'south', 'east', 'west', 'n', 's', 'e', 'w']):
# Direction shortcuts from hall
directions = {
'north': 'garden', 'n': 'garden',
'south': 'classroom', 's': 'classroom',
'east': 'study', 'e': 'study',
'west': 'kitchen', 'w': 'kitchen'
}
for direction, room in directions.items():
if direction in command_lower:
move_request = CommandRequest(command=f"go {room}")
return await move_to_room(move_request)
else:
# Try to interpret as a talk message if it doesn't match commands
if len(command.split()) > 1 and not any(cmd in command_lower for cmd in ['go', 'move', 'travel', 'examine', 'take']):
talk_request = TalkRequest(message=command)
return await talk_to_npc(talk_request)
return GameResponse(
success=False,
message=f"๐ค I don't understand '{command}'. Try 'help' for commands, or just type what you want to say to the NPC!"
)
except Exception as e:
logger.error(f"Error processing command: {e}")
return GameResponse(success=False, message=f"Error: {str(e)}")
if __name__ == "__main__":
print("Starting Korean Learning MUD Web Server...")
print("Make sure you have created 'korean_mud_game.html' in the project root")
print("Server will be available at: http://localhost:7860")
# Use 0.0.0.0 for Docker compatibility, 127.0.0.1 for local dev
host = "0.0.0.0" if os.getenv("DOCKER_ENV") else "127.0.0.1"
uvicorn.run(
"web_server:app",
host=host,
port=7860,
reload=False if os.getenv("DOCKER_ENV") else True,
log_level="info"
)