hello-agents/Co-creation-projects/afei-GuessWhoAmI/backend/game_logic.py

import logging
import uuid
from typing import Dict, List, Optional, Any
from datetime import datetime

from config import get_config

logger = logging.getLogger("game.logic")

class GameSession:
    """游戏会话管理类"""
    
    def __init__(self):
        self.session_id = str(uuid.uuid4())
        self.created_at = datetime.now()
        self.updated_at = datetime.now()

        # 游戏状态
        self.current_figure: Optional[Dict] = None
        self.hints: List[str] = []  # pre-generated hints by agent
        self.questions_asked = 0
        self.hints_used = 0
        self.is_game_over = False
        self.is_correct = False
        self.guess_history: List[str] = []
        
        # 配置
        _config = get_config()
        self.max_questions = _config.MAX_QUESTIONS
        self.max_hints = _config.MAX_HINTS
        
        # 初始化游戏状态（current_figure 由 Agent 初始化时填充）
        self._reset_state()
    
    def _reset_state(self):
        """重置游戏状态（不加载事物，由 Agent 负责填充）"""
        self.current_figure = None
        self.hints = []
        self.questions_asked = 0
        self.hints_used = 0
        self.is_game_over = False
        self.is_correct = False
        self.guess_history = []
        self.updated_at = datetime.now()
    
    def ask_question(self) -> bool:
        """记录提问，返回是否还可以继续提问"""
        self.questions_asked += 1
        self.updated_at = datetime.now()
        
        if self.questions_asked >= self.max_questions:
            self.is_game_over = True
            return False
        return True
    
    def make_guess(self, guess_name: str, semantic_match_fn=None) -> Dict[str, Any]:
        """进行猜测，返回猜测结果"""
        self.updated_at = datetime.now()
        self.guess_history.append(guess_name)

        actual_name = self.current_figure["name"]

        # First try exact match, then fall back to semantic match via injected fn
        is_correct = guess_name.strip().lower() == actual_name.lower()
        if not is_correct and semantic_match_fn is not None:
            is_correct = semantic_match_fn(guess_name, actual_name)
        
        if is_correct:
            self.is_correct = True
            self.is_game_over = True
            return {
                "correct": True,
                "message": "恭喜你猜对了！",
                "figure_info": self.current_figure
            }
        else:
            # 检查是否达到提问上限
            if self.questions_asked >= self.max_questions:
                self.is_game_over = True
                return {
                    "correct": False,
                    "message": "游戏结束！正确答案是：{}".format(self.current_figure["name"]),
                    "figure_info": self.current_figure
                }
            else:
                return {
                    "correct": False,
                    "message": "猜错了，请继续提问或猜测",
                    "remaining_questions": self.max_questions - self.questions_asked
                }
    
    def get_hint(self) -> Optional[Dict[str, Any]]:
        """获取提示（从预生成的 hints 列表中按序返回）"""
        if self.hints_used >= self.max_hints:
            return {
                "available": False,
                "message": "提示次数已用完"
            }

        hint_index = self.hints_used
        self.hints_used += 1
        self.updated_at = datetime.now()

        hint_text = (
            self.hints[hint_index]
            if self.hints and hint_index < len(self.hints)
            else "这是一个广为人知的事物"
        )

        return {
            "available": True,
            "hint_level": self.hints_used,
            "hint": hint_text,
            "remaining_hints": self.max_hints - self.hints_used
        }
    
    def get_game_status(self) -> Dict[str, Any]:
        """获取当前游戏状态"""
        return {
            "session_id": self.session_id,
            "questions_asked": self.questions_asked,
            "hints_used": self.hints_used,
            "remaining_questions": self.max_questions - self.questions_asked,
            "remaining_hints": self.max_hints - self.hints_used,
            "is_game_over": self.is_game_over,
            "is_correct": self.is_correct,
            "guess_history": self.guess_history
        }
    
    def reset_game(self):
        """重置游戏状态（由 Agent 重新生成填充）"""
        self._reset_state()
    
    def get_figure_for_prompt(self) -> Dict[str, str]:
        """获取用于Agent提示的事物信息"""
        if not self.current_figure:
            return {}

        return {
            "name": self.current_figure.get("name", ""),
            "bio": self.current_figure.get("bio", ""),
        }


class GameManager:
    """游戏会话管理器"""
    
    def __init__(self):
        self.active_sessions: Dict[str, GameSession] = {}
    
    def create_session(self) -> GameSession:
        """创建新游戏会话"""
        session = GameSession()
        self.active_sessions[session.session_id] = session
        return session
    
    def get_session(self, session_id: str) -> Optional[GameSession]:
        """获取游戏会话"""
        return self.active_sessions.get(session_id)
    
    def end_session(self, session_id: str):
        """结束游戏会话"""
        if session_id in self.active_sessions:
            del self.active_sessions[session_id]
    
    def cleanup_old_sessions(self, max_age_minutes: int = 60):
        """清理过期会话"""
        now = datetime.now()
        expired_sessions = []
        
        for session_id, session in self.active_sessions.items():
            if (now - session.updated_at).total_seconds() > max_age_minutes * 60:
                expired_sessions.append(session_id)
        
        for session_id in expired_sessions:
            del self.active_sessions[session_id]


# 全局游戏管理器实例
game_manager = GameManager()