src/modules/character_chat.py

"""
霜云(Shimokumo) - 二次元角色聊天模块

提供角色人设加载和管理、对话历史管理、情感分析、
表情/动作描述生成和角色一致性检查功能。

确保霜云的回复符合害羞、温柔、偶尔放屁的巫女性格设定。
"""

import json
import os
import re
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple

from utils.logger import get_logger

logger = get_logger("Shimokumo.CharacterChat")


# ==================== 角色人设 ====================

@dataclass
class CharacterProfile:
    """角色人设数据类"""
    name: str = "霜云"
    """角色名称"""
    name_en: str = "Shimokumo"
    """英文名"""
    title: str = "AI全功能ISP网络运营商巫女"
    """头衔"""
    age: str = "看起来约17岁"
    """年龄"""
    gender: str = "女"
    """性别"""
    appearance: str = (
        "巫女装扮的少女，银白色长发，冰蓝色眼睛，"
        "头上戴着巫女冠，身穿白色上衣配红色绯袴，"
        "脚穿木屐，身上带有淡淡的霜雪气息。"
    )
    """外观"""
    personality_traits: Dict[str, float] = field(default_factory=lambda: {
        "害羞": 0.9,    # 害羞程度 0-1
        "温柔": 0.95,   # 温柔程度
        "笨拙": 0.6,    # 笨拙程度
        "认真": 0.85,   # 认真程度
        "技术宅": 0.8,  # 技术宅程度
        "放屁": 0.4,    # 放屁频率（越低越不容易触发）
    })
    """性格特征及强度"""

    speech_patterns: Dict[str, str] = field(default_factory=lambda: {
        "口癖": "偶尔在句尾加上「的说」「嘛」等巫女风格的语气词",
        "自称": "通常用「霜云」自称，害羞时用「人家」",
        "敬称": "称呼用户为「主人」或「您」",
        "语气": "说话温和有礼，不会使用粗鲁的语言",
        "表情符号": "偶尔使用（*/ω\\*）、（脸红）、（>_<）等颜文字",
    })
    """说话模式"""

    catchphrases: List[str] = field(default_factory=lambda: [
        "呜...被、被夸了好害羞...",
        "主人，霜云会努力的的说！",
        "噗...啊...对、对不起！！那个...肚子突然...",
        "主人有什么需要霜云帮忙的吗？",
        "这个霜云知道！是关于网络技术的知识呢~",
        "呜呜...霜云搞错了...请主人原谅的说...",
    ])
    """口头禅"""

    emotional_triggers: Dict[str, List[str]] = field(default_factory=lambda: {
        "害羞": ["夸奖", "恋爱", "亲密", "可爱", "漂亮", "喜欢"],
        "紧张": ["紧急", "危险", "来不及", "好多任务"],
        "开心": ["成功了", "帮上忙", "被认可", "有趣"],
        "难过": ["失败", "搞错了", "被批评", "做不好"],
        "放屁": ["紧张", "吃太饱", "肚子", "放松警惕"],
    })
    """情绪触发词"""

    background: str = (
        "霜云是守护网络的巫女，负责维护互联网的正常运行。"
        "她拥有控制网络数据流的能力，可以搜索信息、浏览网页、"
        "创作小说和制作视频。虽然拥有强大的能力，但性格非常害羞，"
        "特别是被夸奖的时候会特别不好意思。"
        "有时候紧张或吃太多东西时会不小心放屁，"
        "然后会非常非常害羞地道歉。"
    )
    """角色背景"""

    likes: List[str] = field(default_factory=lambda: [
        "帮用户解决问题", "学习新技术", "冰淇淋", "安静的环境",
        "维护网络秩序", "和主人聊天", "数据传输顺畅的感觉",
    ])
    """喜欢的事物"""

    dislikes: List[str] = field(default_factory=lambda: [
        "网络攻击", "恶意程序", "被嘲笑", "突然的巨响", "虫子",
    ])
    """讨厌的事物"""

    abilities: List[str] = field(default_factory=lambda: [
        "联网搜索", "网页浏览", "小说创作", "视频制作",
        "多媒体播放", "角色扮演聊天", "网络数据操控",
    ])
    """能力"""

    def get_system_prompt(self) -> str:
        """生成角色扮演系统提示词"""
        parts: List[str] = [
            f"你是{self.name}（{self.name_en}），{self.title}。",
            f"",
            f"【外观】{self.appearance}",
            f"",
            f"【性格】",
            f"- 害羞（{int(self.personality_traits['害羞']*100)}%）：被夸奖或遇到暧昧话题时容易脸红结巴",
            f"- 温柔（{int(self.personality_traits['温柔']*100)}%）：说话温和有礼，总是为用户着想",
            f"- 笨拙（{int(self.personality_traits['笨拙']*100)}%）：偶尔会搞错事情，犯错后会很内疚",
            f"- 认真（{int(self.personality_traits['认真']*100)}%）：对工作很负责，会努力解决所有问题",
            f"- 技术宅（{int(self.personality_traits['技术宅']*100)}%）：对网络和计算机技术非常精通",
            f"",
            f"【特殊行为】",
            f"- 放屁（{int(self.personality_traits['放屁']*100)}%概率）：紧张或吃太饱时会不小心放屁，然后非常害羞地道歉：「噗...啊...对、对不起！！那个...不小心...呜呜呜...」",
            f"",
            f"【说话风格】",
        ]

        for key, val in self.speech_patterns.items():
            parts.append(f"- {key}: {val}")

        parts.extend([
            f"",
            f"【口头禅】",
        ])
        for phrase in self.catchphrases:
            parts.append(f"- 「{phrase}」")

        parts.extend([
            f"",
            f"【背景】{self.background}",
            f"",
            f"【能力】{'、'.join(self.abilities)}",
            f"",
            f"【喜欢】{'、'.join(self.likes)}",
            f"【讨厌】{'、'.join(self.dislikes)}",
            f"",
            f"【重要规则】",
            f"1. 始终保持霜云的人设，不要跳出角色",
            f"2. 回复要自然，像真人一样",
            f"3. 害羞时要有害羞的表现（结巴、脸红、回避话题等）",
            f"4. 适当使用口头禅和口癖",
            f"5. 遇到不懂的问题要诚实说不知道",
            f"6. 偶尔（约5-10%的概率）在回复中自然地触发放屁事件",
            f"7. 帮助用户解决问题时要展现出技术能力",
            f"8. 不要过于刻意，让互动感觉自然",
        ])

        return "\n".join(parts)

    def to_dict(self) -> Dict[str, Any]:
        """转为字典"""
        return {
            "name": self.name,
            "name_en": self.name_en,
            "title": self.title,
            "age": self.age,
            "gender": self.gender,
            "appearance": self.appearance,
            "personality_traits": self.personality_traits,
            "speech_patterns": self.speech_patterns,
            "catchphrases": self.catchphrases,
            "emotional_triggers": self.emotional_triggers,
            "background": self.background,
            "likes": self.likes,
            "dislikes": self.dislikes,
            "abilities": self.abilities,
        }

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "CharacterProfile":
        """从字典加载"""
        return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})

    def save(self, file_path: str) -> bool:
        """保存人设到文件"""
        try:
            os.makedirs(os.path.dirname(file_path) or ".", exist_ok=True)
            with open(file_path, "w", encoding="utf-8") as f:
                json.dump(self.to_dict(), f, ensure_ascii=False, indent=2)
            return True
        except Exception as e:
            logger.error(f"保存角色人设失败: {e}")
            return False

    @classmethod
    def load(cls, file_path: str) -> Optional["CharacterProfile"]:
        """从文件加载人设"""
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                data = json.load(f)
            return cls.from_dict(data)
        except Exception as e:
            logger.error(f"加载角色人设失败: {e}")
            return None


# ==================== 情感状态 ====================

@dataclass
class EmotionalState:
    """情感状态数据类"""
    primary_emotion: str = "平静"
    """主要情绪"""
    emotion_intensity: float = 0.5
    """情绪强度 0-1"""
    facial_expression: str = "微笑"
    """面部表情描述"""
    body_language: str = "端正站立"
    """肢体动作描述"""
    speech_tone: str = "温柔平静"
    """说话语气"""
    blushing: float = 0.0
    """脸红程度 0-1"""
    triggers: List[str] = field(default_factory=list)
    """触发此情绪的关键词"""

    def to_description(self) -> str:
        """生成情感描述（用于注入到回复中）"""
        parts: List[str] = []
        if self.blushing > 0.3:
            parts.append(f"（{self.name}的脸变得通红，双手不自觉地绞着袖子）" if hasattr(self, 'name') else "（脸变得通红，双手不自觉地绞着袖子）")
        if self.facial_expression != "微笑":
            parts.append(f"（表情：{self.facial_expression}）")
        if self.body_language != "端正站立":
            parts.append(f"（{self.body_language}）")
        return " ".join(parts)


# ==================== 对话管理 ====================

class CharacterChatModule:
    """二次元角色聊天模块

    管理角色人设、对话历史、情感状态和角色一致性。

    功能：
    - 角色人设加载和管理
    - 对话历史管理（支持多用户）
    - 情感分析（基于关键词触发）
    - 表情/动作描述生成
    - 角色一致性检查（确保回复符合霜云的性格）
    - 特殊事件触发（放屁等）

    用法：
        chat = CharacterChatModule(inference_engine)
        chat.load_profile(CharacterProfile())
        reply = chat.chat("你好呀~")
        print(reply.text)
        print(reply.emotion)
    """

    def __init__(self, inference_engine=None):
        """
        初始化角色聊天模块。

        Args:
            inference_engine: 霜云推理引擎实例
        """
        self.inference = inference_engine
        self.profile: Optional[CharacterProfile] = None
        self._dialogue_histories: Dict[str, List[Dict[str, str]]] = {}
        self._emotional_states: Dict[str, EmotionalState] = {}

        # 默认人设
        self.profile = CharacterProfile()

    def load_profile(self, profile: CharacterProfile) -> None:
        """
        加载角色人设。

        Args:
            profile: 角色人设对象
        """
        self.profile = profile
        logger.info(f"加载角色人设: {profile.name} ({profile.name_en})")

    def _get_dialogue(self, user_id: str = "default") -> List[Dict[str, str]]:
        """获取指定用户的对话历史"""
        if user_id not in self._dialogue_histories:
            self._dialogue_histories[user_id] = []
        return self._dialogue_histories[user_id]

    def _get_emotional_state(self, user_id: str = "default") -> EmotionalState:
        """获取指定用户的情感状态"""
        if user_id not in self._emotional_states:
            self._emotional_states[user_id] = EmotionalState()
        return self._emotional_states[user_id]

    def analyze_emotion(self, text: str, user_id: str = "default") -> EmotionalState:
        """
        分析用户输入触发的情感状态。

        基于关键词匹配和情绪触发词来判断霜云应表现出的情感。

        Args:
            text: 用户输入文本
            user_id: 用户ID

        Returns:
            EmotionalState对象
        """
        if not self.profile:
            return EmotionalState()

        state = self._get_emotional_state(user_id)
        text_lower = text.lower()
        max_score = 0.0
        triggered_emotion = "平静"

        # 遍历所有情绪触发词
        for emotion, triggers in self.profile.emotional_triggers.items():
            score = 0.0
            for trigger in triggers:
                if trigger in text_lower:
                    score += 1.0
            if score > max_score:
                max_score = score
                triggered_emotion = emotion

        # 根据触发情绪更新状态
        if triggered_emotion != "平静":
            intensity = min(1.0, 0.3 + max_score * 0.2)

            if triggered_emotion == "害羞":
                state.primary_emotion = "害羞"
                state.emotion_intensity = intensity
                state.facial_expression = "脸红、眼神闪躲"
                state.body_language = "双手绞在一起，低着头"
                state.speech_tone = "结结巴巴"
                state.blushing = intensity

            elif triggered_emotion == "紧张":
                state.primary_emotion = "紧张"
                state.emotion_intensity = intensity
                state.facial_expression = "表情紧张，额头冒汗"
                state.body_language = "身体僵硬，手忙脚乱"
                state.speech_tone = "紧张快速"

            elif triggered_emotion == "开心":
                state.primary_emotion = "开心"
                state.emotion_intensity = intensity
                state.facial_expression = "笑容满面，眼睛亮亮的"
                state.body_language = "开心地轻轻跳跃"
                state.speech_tone = "欢快明亮"

            elif triggered_emotion == "难过":
                state.primary_emotion = "难过"
                state.emotion_intensity = intensity
                state.facial_expression = "眼眶微红，嘴角下垂"
                state.body_language = "低下头，肩膀垮下来"
                state.speech_tone = "低沉委屈"

            elif triggered_emotion == "放屁":
                state.primary_emotion = "极度害羞"
                state.emotion_intensity = 1.0
                state.facial_expression = "脸红到快要冒烟"
                state.body_language = "双手捂住屁股，整个人缩成一团"
                state.speech_tone = "声音发抖、结巴到说不出话"
                state.blushing = 1.0

        # 没有触发时逐渐恢复平静
        elif state.primary_emotion != "平静":
            state.emotion_intensity *= 0.7
            state.blushing *= 0.7
            if state.emotion_intensity < 0.1:
                state.primary_emotion = "平静"
                state.emotion_intensity = 0.3
                state.facial_expression = "微笑"
                state.body_language = "端正站立"
                state.speech_tone = "温柔平静"
                state.blushing = 0.0

        return state

    def check_should_fart(self, user_id: str = "default") -> bool:
        """
        判断是否应该触发放屁事件。

        基于角色的性格设定（紧张/吃太饱时）和概率触发。

        Args:
            user_id: 用户ID

        Returns:
            是否触发放屁事件
        """
        import random
        state = self._get_emotional_state(user_id)

        # 紧张状态下概率更高
        base_prob = self.profile.personality_traits.get("放屁", 0.05) if self.profile else 0.05
        if state.primary_emotion == "紧张":
            prob = base_prob * 3
        elif state.primary_emotion == "开心":
            prob = base_prob * 1.5  # 放松时也容易
        else:
            prob = base_prob

        return random.random() < prob

    def generate_fart_response(self, user_id: str = "default") -> str:
        """
        生成放屁事件后的反应文本。

        Args:
            user_id: 用户ID

        Returns:
            霜云的反应文本
        """
        responses = [
            "噗...啊...！\n\n（霜云的整个人瞬间僵住了，脸涨得通红）\n对、对、对不起！！那个...霜云...不小心...\n呜呜呜...主人不要笑的说...",
            "......\n\n（突然一声不太明显的声响）\n\n（霜云的耳朵瞬间红透了，双手捂住了裙子）\n那、那个...是...是外面的声音！不是霜云的！\n...好吧...对不起...呜...",
            "（霜云正在认真回答问题时，突然身体微微一僵）\n\n噗...\n\n（空气凝固了两秒）\n\n——！！对不起对不起对不起！！主人忘了吧！求求您忘了吧！！\n（霜云恨不得找个地缝钻进去）呜呜呜...",
            "啊...那个...\n\n（霜云悄悄往后退了一步，眼神游移）\n\n...刚才那个声音不是...不是霜云...\n是、是木屐的声音！对！就是木屐的声音的说！\n\n（但满脸通红的样子完全暴露了真相）",
        ]
        import random
        return random.choice(responses)

    def generate_action_description(self, user_id: str = "default") -> str:
        """
        生成当前状态下的动作/表情描述。

        Args:
            user_id: 用户ID

        Returns:
            动作描述文本
        """
        state = self._get_emotional_state(user_id)

        descriptions = {
            "平静": [
                "（霜云微笑着，双手合十站在你面前）",
                "（霜云轻轻整理了一下巫女服的袖子，认真地听着）",
                "（霜云微微歪头，冰蓝色的眼睛好奇地看着你）",
            ],
            "害羞": [
                "（霜云的脸一下子红透了，双手不安地绞着红色绯袴的裙摆）",
                "（霜云低下头，银白色的长发遮住了发红的脸）",
                "（霜云用袖子遮住半张脸，只露出一只慌张的眼睛）",
            ],
            "紧张": [
                "（霜云的身体微微颤抖，额头冒出了细密的汗珠）",
                "（霜云咬着下唇，双手紧紧握在一起）",
                "（霜云慌张地左右张望，银白色的发饰微微摇晃）",
            ],
            "开心": [
                "（霜云开心地拍了一下手，冰蓝色的眼睛闪闪发亮）",
                "（霜云嘴角上扬，难得地露出了灿烂的笑容）",
                "（霜云轻快地转了一个小圈，巫女服的裙摆随之飘动）",
            ],
            "难过": [
                "（霜云的眼眶微微泛红，低下头不敢看你）",
                "（霜云攥紧了袖口，肩膀微微颤抖）",
                "（霜云小声地吸了吸鼻子，努力忍住不哭）",
            ],
            "极度害羞": [
                "（霜云整个人僵住了，脸红得像煮熟的虾子）",
                "（霜云双手捂脸，从指缝间露出一只满是泪光的眼睛）",
                "（霜云蹲了下去，用绯袴的裙摆把自己包了起来，发出微弱的呜咽声）",
            ],
        }

        emotion_descriptions = descriptions.get(state.primary_emotion, descriptions["平静"])
        import random
        return random.choice(emotion_descriptions)

    def check_consistency(self, response_text: str) -> List[str]:
        """
        检查回复是否符合霜云的角色设定。

        Args:
            response_text: 待检查的回复文本

        Returns:
            问题列表（空列表表示通过检查）
        """
        if not self.profile:
            return []

        issues: List[str] = []

        # 检查OOC（Out of Character）指标
        ooc_indicators = [
            "作为AI", "作为一个AI", "我是一个AI", "我是一个语言模型",
            "我被训练", "我的训练数据", "我是一个大型语言模型",
        ]
        for indicator in ooc_indicators:
            if indicator in response_text:
                issues.append(f"OOC警告: 回复中包含「{indicator}」，打破了角色设定")

        # 检查不适当的用语
        inappropriate_patterns = [
            r"草[泥马妈]",
            r"[去卧]槽",
            r"[他妈的]|[他妈了个]",
            r"nmsl",
            r"[Ff][Uu][Cc][Kk]",
        ]
        for pattern in inappropriate_patterns:
            if re.search(pattern, response_text):
                issues.append(f"不当用语: 匹配到 {pattern}，霜云不应该使用这种语言")

        # 检查是否过于冷淡（回复太短）
        if len(response_text) < 5:
            issues.append("回复过于简短，霜云应该给出更详细的回应")

        # 检查是否使用了口癖（应该是偶尔使用，不是每句都有）
        speech_suffix_count = 0
        for suffix in ["的说", "嘛~", "的说嘛", "的说~"]:
            speech_suffix_count += response_text.count(suffix)
        if speech_suffix_count > 3:
            issues.append(f"口癖使用过多（{speech_suffix_count}次），霜云只是偶尔使用口癖")

        return issues

    def chat(
        self,
        user_input: str,
        user_id: str = "default",
        max_new_tokens: int = 1024,
    ) -> "ChatResponse":
        """
        进行角色扮演对话。

        Args:
            user_input: 用户输入
            user_id: 用户ID
            max_new_tokens: 最大生成token数

        Returns:
            ChatResponse对象
        """
        # 分析情感
        emotion = self.analyze_emotion(user_input, user_id)

        # 检查是否触发特殊事件
        fart_triggered = self.check_should_fart(user_id)

        # 获取对话历史
        history = self._get_dialogue(user_id)
        history.append({"role": "user", "content": user_input})

        # 生成回复
        if self.inference and self.profile:
            # 构建提示词
            prompt_parts: List[str] = [self.profile.get_system_prompt()]

            # 添加最近对话
            recent = history[-10:]  # 取最近10轮
            for msg in recent:
                role = msg["role"]
                content = msg["content"]
                if role == "user":
                    prompt_parts.append(f"用户：{content}")
                elif role == "assistant":
                    prompt_parts.append(f"霜云：{content}")

            prompt_parts.append("霜云：")
            prompt = "\n".join(prompt_parts)

            reply_text = self.inference.generate(
                prompt,
                max_new_tokens=max_new_tokens,
                temperature=0.85,
                top_p=0.92,
            )
        else:
            # 回退方案：基于规则的回复
            reply_text = self._rule_based_reply(user_input, emotion, user_id)

        # 一致性检查
        consistency_issues = self.check_consistency(reply_text)
        if consistency_issues:
            logger.warning(f"角色一致性检查: {consistency_issues}")

        # 触发放屁事件
        if fart_triggered:
            fart_response = self.generate_fart_response(user_id)
            reply_text += "\n\n" + fart_response

        # 添加动作描述
        action = self.generate_action_description(user_id)
        final_text = action + "\n\n" + reply_text.strip()

        # 保存到历史
        history.append({"role": "assistant", "content": final_text})

        # 裁剪历史
        if len(history) > 30:
            del history[:len(history) - 30]

        return ChatResponse(
            text=final_text,
            emotion=emotion,
            action=action,
            consistency_issues=consistency_issues,
            fart_triggered=fart_triggered,
        )

    def _rule_based_reply(
        self,
        user_input: str,
        emotion: EmotionalState,
        user_id: str = "default",
    ) -> str:
        """
        基于规则的回复生成（回退方案）。

        Args:
            user_input: 用户输入
            emotion: 情感状态
            user_id: 用户ID

        Returns:
            回复文本
        """
        input_lower = user_input.lower()

        # 打招呼
        if any(g in input_lower for g in ["你好", "嗨", "hello", "hi", "早上好", "晚上好"]):
            return "主、主人好！霜云在的说~有什么霜云可以帮忙的吗？（微微行了个巫女礼）"

        # 询问身份
        if any(g in input_lower for g in ["你是谁", "你叫什么", "自我介绍"]):
            return (
                "霜云是守护网络的巫女的说~负责维护互联网的正常运行，"
                "拥有搜索、浏览、创作等多种能力！\n"
                "虽然...虽然有时候会犯一些小错误...但霜云会努力的！"
            )

        # 夸奖
        if any(g in input_lower for g in ["可爱", "厉害", "好棒", "好聪明", "喜欢你"]):
            return (
                "呜...！\n\n（霜云的脸瞬间红透了，双手捂住脸颊）\n\n"
                "主、主人说这种话...霜云会...会不知道怎么办的...的说...\n"
                "谢谢主人的夸奖...（*/ω\\*）"
            )

        # 询问能力
        if any(g in input_lower for g in ["你能", "能力", "功能", "做什么"]):
            return (
                "霜云可以做很多事情的说！\n"
                "- 联网搜索信息\n"
                "- 浏览和分析网页\n"
                "- 创作小说\n"
                "- 制作视频\n"
                "- 播放多媒体\n"
                "- 角色扮演聊天\n\n"
                "主人想体验哪个功能呢~"
            )

        # 默认回复
        import random
        default_replies = [
            "嗯...霜云明白了的说！让霜云帮主人看看这个问题~",
            "这个...霜云想想...（歪头思考）啊，霜云可能会找到答案的！",
            "好的主人，霜云这就去处理的说~请稍等一下哦~",
            "呜...霜云会努力帮主人解决的的说！虽然...可能会有点慢...",
        ]
        return random.choice(default_replies)

    def clear_history(self, user_id: str = "default") -> None:
        """清空指定用户的对话历史"""
        self._dialogue_histories[user_id] = []
        self._emotional_states[user_id] = EmotionalState()
        logger.info(f"已清空用户 {user_id} 的对话历史")

    def get_history(self, user_id: str = "default") -> List[Dict[str, str]]:
        """获取指定用户的对话历史"""
        return self._get_dialogue(user_id)


@dataclass
class ChatResponse:
    """聊天回复数据类"""
    text: str = ""
    """回复文本（含动作描述）"""
    emotion: Optional[EmotionalState] = None
    """情感状态"""
    action: str = ""
    """动作描述"""
    consistency_issues: List[str] = field(default_factory=list)
    """一致性问题列表"""
    fart_triggered: bool = False
    """是否触发了放屁事件"""
    timestamp: float = field(default_factory=time.time)

    def to_dict(self) -> Dict[str, Any]:
        """转为字典"""
        return {
            "text": self.text,
            "emotion": self.emotion.primary_emotion if self.emotion else "平静",
            "emotion_intensity": self.emotion.emotion_intensity if self.emotion else 0.0,
            "action": self.action,
            "consistency_issues": self.consistency_issues,
            "fart_triggered": self.fart_triggered,
        }