"""
统一历史对话管理器
整合所有分散的 parse_history / _parse_history / build_history_str / filter_image_urls 逻辑
"""

import json
from typing import Any, Dict, List, Optional


class HistoryManager:
    """统一的历史对话管理器，提供解析、截断、格式化、过滤等功能"""

    def __init__(
        self,
        max_messages: int = 10,
        max_assistant_chars: int = 300,
    ):
        """
        Args:
            max_messages: 保留的最近消息条数
            max_assistant_chars: assistant 角色内容截断字符数
        """
        self.max_messages = max_messages
        self.max_assistant_chars = max_assistant_chars

    # ==================== 解析 ====================

    def parse(self, history_message: Any) -> List[Dict[str, Any]]:
        """
        解析历史消息为列表，保留最近 max_messages 条。
        兼容字符串和列表输入，替代所有 parse_history / _parse_history。
        """
        if not history_message:
            return []
        try:
            data = json.loads(history_message) if isinstance(history_message, str) else history_message
            return data[-self.max_messages:] if isinstance(data, list) else []
        except Exception:
            return []

    # ==================== 截断 ====================

    def truncate(self, messages: List[Dict[str, Any]], keep_recent: Optional[int] = None) -> List[Dict[str, Any]]:
        """按数量截断，保留最近 keep_recent 条"""
        if not messages:
            return []
        n = keep_recent if keep_recent is not None else self.max_messages
        return messages[-n:]

    # ==================== 格式化 ====================

    def to_prompt_str(
        self,
        messages: List[Dict[str, Any]],
        recent_count: Optional[int] = None,
        assistant_truncate: Optional[int] = None,
    ) -> str:
        """
        格式化为 "用户: xxx\\n助手: xxx" 纯文本字符串，用于 prompt 注入。
        替代 build_history_str。
        """
        if not messages:
            return ""
        n = recent_count if recent_count is not None else self.max_messages
        trunc = assistant_truncate if assistant_truncate is not None else self.max_assistant_chars
        recent = messages[-n:]
        parts = []
        for m in recent:
            if isinstance(m, dict):
                role = m.get("role", "")
                content = (m.get("content") or "").strip()
                if role == "user":
                    parts.append(f"用户: {content}")
                elif role == "assistant":
                    parts.append(f"助手: {content[:trunc]}")
        return "\n".join(parts) if parts else ""

    def to_openai_messages(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """
        格式化为 OpenAI messages 列表，直接传入 LLM。
        用于百科问答等需要完整消息格式的场景。
        """
        if not messages:
            return []
        result = []
        for m in messages:
            if isinstance(m, dict) and m.get("role") and m.get("content"):
                result.append({"role": m["role"], "content": m["content"]})
        return result

    # ==================== 请求预处理 ====================

    def preprocess_from_request(self, history_message: Any, exclude_last: bool = True) -> str:
        """
        处理前端传入的历史消息：解析、移除当前用户消息、序列化。
        对应 app.py 中 history_message 预处理逻辑。

        Args:
            history_message: 前端传入的历史消息，支持 str 或 list
            exclude_last: 是否移除最后一条（当前用户消息），默认 True

        Returns:
            处理后的 JSON 字符串
        """
        if not history_message:
            return ""
        try:
            data = json.loads(history_message) if isinstance(history_message, str) else history_message
        except (json.JSONDecodeError, TypeError):
            return ""
        if not isinstance(data, list):
            return ""
        if exclude_last and data:
            data = data[:-1]
        return json.dumps(data, ensure_ascii=False)

    # ==================== 状态初始化 ====================

    def init_history(self, history_message: Any) -> Dict[str, Any]:
        """
        一步完成历史消息解析，返回 history_message(str) + history_messages(list) 双字段。
        用于各工作流入口初始化 state，消除重复的 parse_history 调用。

        Args:
            history_message: 历史消息，支持 str 或 list

        Returns:
            {"history_message": str, "history_messages": list}
        """
        parsed = self.parse(history_message)
        raw = json.dumps(parsed, ensure_ascii=False) if parsed else ""
        return {"history_message": raw, "history_messages": parsed}

    # ==================== 过滤 ====================

    @staticmethod
    def filter_image_urls(history_message: Any) -> Any:
        """
        过滤历史消息中的图片URL，将多模态内容转为纯文本。
        替代 main_agent.filter_image_urls。
        """
        if not history_message:
            return history_message

        try:
            history_messages = json.loads(history_message) if isinstance(history_message, str) else history_message
        except (json.JSONDecodeError, TypeError):
            return history_message

        if not isinstance(history_messages, list):
            return history_messages

        filtered_messages = []
        for msg in history_messages:
            if not isinstance(msg, dict):
                filtered_messages.append(msg)
                continue

            new_msg = msg.copy()
            content = msg.get("content")

            if isinstance(content, list):
                text_parts = []
                for content_item in content:
                    if isinstance(content_item, dict):
                        if content_item.get("type") == "text":
                            text_value = content_item.get("text", "")
                            if text_value:
                                text_parts.append(text_value)
                    elif isinstance(content_item, str):
                        text_parts.append(content_item)

                if text_parts:
                    new_msg["content"] = "".join(text_parts)
                    filtered_messages.append(new_msg)
            else:
                filtered_messages.append(new_msg)

        return filtered_messages


# ==================== 模块级便捷实例 ====================
# 提供与旧函数签名兼容的模块级函数，方便渐进式迁移

_default_manager = HistoryManager()


def parse_history(history_message: Any) -> List[Dict[str, Any]]:
    """兼容旧调用的模块级函数"""
    return _default_manager.parse(history_message)


def build_history_str(
    history_messages: List[Dict[str, Any]],
    recent_count: int = 10,
    assistant_truncate: int = 300,
) -> str:
    """兼容旧调用的模块级函数"""
    mgr = HistoryManager(max_messages=recent_count, max_assistant_chars=assistant_truncate)
    return mgr.to_prompt_str(history_messages, recent_count=recent_count, assistant_truncate=assistant_truncate)


def filter_image_urls(history_message: Any) -> Any:
    """兼容旧调用的模块级函数"""
    return HistoryManager.filter_image_urls(history_message)


def preprocess_from_request(history_message: Any, exclude_last: bool = True) -> str:
    """兼容旧调用的模块级函数"""
    return _default_manager.preprocess_from_request(history_message, exclude_last=exclude_last)


def init_history(history_message: Any) -> Dict[str, Any]:
    """兼容旧调用的模块级函数"""
    return _default_manager.init_history(history_message)