"""MaiSaka 实时监控事件广播模块。

通过统一 WebSocket 将 MaiSaka 推理引擎各阶段状态实时推送给前端监控界面。
"""

from datetime import datetime
import time
from typing import Any, Dict, List, Optional

from src.common.logger import get_logger

logger = get_logger("maisaka_monitor")

MONITOR_DOMAIN = "maisaka_monitor"
MONITOR_TOPIC = "main"


def _normalize_payload_value(value: Any) -> Any:
    """将事件载荷中的任意值规范化为可序列化结构。"""

    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    if isinstance(value, datetime):
        return value.isoformat()
    if isinstance(value, dict):
        normalized_dict: Dict[str, Any] = {}
        for key, item in value.items():
            normalized_dict[str(key)] = _normalize_payload_value(item)
        return normalized_dict
    if isinstance(value, (list, tuple, set)):
        return [_normalize_payload_value(item) for item in value]
    if hasattr(value, "model_dump"):
        try:
            return _normalize_payload_value(value.model_dump())
        except Exception:
            return str(value)
    if hasattr(value, "__dict__"):
        try:
            return _normalize_payload_value(dict(value.__dict__))
        except Exception:
            return str(value)
    return str(value)


def _extract_text_content(content: Any) -> Optional[str]:
    """从消息内容中提取纯文本表示。"""

    if content is None:
        return None
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        text_parts: List[str] = []
        for block in content:
            if isinstance(block, dict):
                block_type = block.get("type", "")
                if block_type == "text":
                    text_parts.append(str(block.get("text", "")))
                elif block_type == "image_url":
                    text_parts.append("[图片]")
                else:
                    text_parts.append(f"[{block_type}]")
            elif isinstance(block, str):
                text_parts.append(block)
        return "\n".join(text_parts) if text_parts else None
    return str(content)


def _serialize_tool_calls_from_objects(tool_calls: List[Any]) -> List[Dict[str, Any]]:
    """将工具调用对象列表序列化为字典列表。"""

    result: List[Dict[str, Any]] = []
    for tool_call in tool_calls:
        serialized: Dict[str, Any] = {
            "id": getattr(tool_call, "id", None) or getattr(tool_call, "call_id", ""),
            "name": getattr(tool_call, "func_name", None) or getattr(tool_call, "name", "unknown"),
        }
        args = getattr(tool_call, "args", None) or getattr(tool_call, "arguments", None)
        if isinstance(args, dict):
            serialized["arguments"] = _normalize_payload_value(args)
        elif isinstance(args, str):
            serialized["arguments_raw"] = args
        result.append(serialized)
    return result


def _serialize_tool_calls_from_dicts(tool_calls: List[Any]) -> List[Dict[str, Any]]:
    """将工具调用字典列表标准化为可传输格式。"""

    result: List[Dict[str, Any]] = []
    for tool_call in tool_calls:
        if isinstance(tool_call, dict):
            result.append({
                "id": str(tool_call.get("id", "")),
                "name": str(tool_call.get("name", tool_call.get("func_name", "unknown"))),
                "arguments": _normalize_payload_value(tool_call.get("arguments", tool_call.get("args", {}))),
            })
            continue

        result.append({
            "id": str(getattr(tool_call, "id", getattr(tool_call, "call_id", ""))),
            "name": str(getattr(tool_call, "func_name", getattr(tool_call, "name", "unknown"))),
            "arguments": _normalize_payload_value(getattr(tool_call, "args", getattr(tool_call, "arguments", {}))),
        })
    return result


def _serialize_message(message: Any) -> Dict[str, Any]:
    """将单条消息序列化为可通过 WebSocket 传输的字典。"""

    if isinstance(message, dict):
        serialized: Dict[str, Any] = {
            "role": str(message.get("role", "unknown")),
            "content": _extract_text_content(message.get("content")),
        }
        if message.get("tool_call_id"):
            serialized["tool_call_id"] = str(message["tool_call_id"])
        if message.get("tool_calls"):
            serialized["tool_calls"] = _serialize_tool_calls_from_dicts(message["tool_calls"])
        return serialized

    raw_role = getattr(message, "role", "unknown")
    role_str = raw_role.value if hasattr(raw_role, "value") else str(raw_role)

    serialized = {
        "role": role_str,
        "content": _extract_text_content(getattr(message, "content", None)),
    }
    tool_call_id = getattr(message, "tool_call_id", None)
    if tool_call_id:
        serialized["tool_call_id"] = str(tool_call_id)

    tool_calls = getattr(message, "tool_calls", None)
    if tool_calls:
        serialized["tool_calls"] = _serialize_tool_calls_from_objects(tool_calls)

    return serialized


def _serialize_messages(messages: List[Any]) -> List[Dict[str, Any]]:
    """批量序列化消息列表。"""

    return [_serialize_message(message) for message in messages]


def _serialize_tool_results(tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """标准化最终 planner 卡中的工具结果列表。"""

    serialized_tools: List[Dict[str, Any]] = []
    for tool in tools:
        serialized_tool = {
            "tool_call_id": str(tool.get("tool_call_id", "")),
            "tool_name": str(tool.get("tool_name", "")),
            "tool_args": _normalize_payload_value(tool.get("tool_args", {})),
            "success": bool(tool.get("success", False)),
            "duration_ms": float(tool.get("duration_ms", 0.0) or 0.0),
            "summary": str(tool.get("summary", "")),
        }
        detail = tool.get("detail")
        if detail is not None:
            serialized_tool["detail"] = _normalize_payload_value(detail)
        serialized_tools.append(serialized_tool)
    return serialized_tools


def _serialize_request_block(
    messages: Optional[List[Any]],
    selected_history_count: Optional[int],
    tool_count: Optional[int],
) -> Optional[Dict[str, Any]]:
    """标准化请求区块。"""

    if messages is None and selected_history_count is None and tool_count is None:
        return None

    return {
        "messages": _serialize_messages(list(messages or [])),
        "selected_history_count": int(selected_history_count or 0),
        "tool_count": int(tool_count or 0),
    }


def _serialize_planner_block(
    content: Optional[str],
    tool_calls: Optional[List[Any]],
    prompt_tokens: Optional[int],
    completion_tokens: Optional[int],
    total_tokens: Optional[int],
    duration_ms: Optional[float],
) -> Optional[Dict[str, Any]]:
    """标准化 planner 结果区块。"""

    if (
        content is None
        and tool_calls is None
        and prompt_tokens is None
        and completion_tokens is None
        and total_tokens is None
        and duration_ms is None
    ):
        return None

    return {
        "content": content,
        "tool_calls": _serialize_tool_calls_from_objects(list(tool_calls or [])),
        "prompt_tokens": int(prompt_tokens or 0),
        "completion_tokens": int(completion_tokens or 0),
        "total_tokens": int(total_tokens or 0),
        "duration_ms": float(duration_ms or 0.0),
    }


def _serialize_timing_gate_block(
    *,
    request_messages: Optional[List[Any]],
    selected_history_count: Optional[int],
    tool_count: Optional[int],
    action: Optional[str],
    content: Optional[str],
    tool_calls: Optional[List[Any]],
    tool_results: Optional[List[str]],
    prompt_tokens: Optional[int],
    completion_tokens: Optional[int],
    total_tokens: Optional[int],
    duration_ms: Optional[float],
) -> Optional[Dict[str, Any]]:
    """标准化 Timing Gate 结果区块。"""

    if (
        request_messages is None
        and selected_history_count is None
        and tool_count is None
        and action is None
        and content is None
        and tool_calls is None
        and tool_results is None
        and prompt_tokens is None
        and completion_tokens is None
        and total_tokens is None
        and duration_ms is None
    ):
        return None

    return {
        "request": _serialize_request_block(
            request_messages,
            selected_history_count,
            tool_count,
        ),
        "result": {
            "action": action,
            "content": content,
            "tool_calls": _serialize_tool_calls_from_objects(list(tool_calls or [])),
            "tool_results": _normalize_payload_value(list(tool_results or [])),
            "prompt_tokens": int(prompt_tokens or 0),
            "completion_tokens": int(completion_tokens or 0),
            "total_tokens": int(total_tokens or 0),
            "duration_ms": float(duration_ms or 0.0),
        },
    }


async def _broadcast(event: str, data: Dict[str, Any]) -> None:
    """通过统一 WebSocket 管理器向监控主题广播事件。"""

    try:
        from src.webui.routers.websocket.manager import websocket_manager

        subscription_key = f"{MONITOR_DOMAIN}:{MONITOR_TOPIC}"
        total_connections = len(websocket_manager.connections)
        subscriber_count = sum(
            1
            for connection in websocket_manager.connections.values()
            if subscription_key in connection.subscriptions
        )
# The above code is using the Python logging module to log a diagnostic message. It is logging
# information about the `_broadcast` function, including the `manager_id`, `total_connections`,
# `subscriber_count`, and `event` variables. The `logger.info()` function is used to log the message
# at the INFO level.
        # logger.info(
        #     f"[诊断] _broadcast: manager_id={id(websocket_manager)} "
        #     f"总连接={total_connections} 订阅者={subscriber_count} event={event}"
        # )
        await websocket_manager.broadcast_to_topic(
            domain=MONITOR_DOMAIN,
            topic=MONITOR_TOPIC,
            event=event,
            data=data,
        )
    except Exception as exc:
        logger.warning(f"MaiSaka 监控事件广播失败: {exc}", exc_info=True)


async def emit_session_start(session_id: str, session_name: str) -> None:
    """广播会话开始事件。"""

    await _broadcast("session.start", {
        "session_id": session_id,
        "session_name": session_name,
        "timestamp": time.time(),
    })


async def emit_message_ingested(
    session_id: str,
    speaker_name: str,
    content: str,
    message_id: str,
    timestamp: float,
) -> None:
    """广播新消息注入事件。"""

    await _broadcast("message.ingested", {
        "session_id": session_id,
        "speaker_name": speaker_name,
        "content": content,
        "message_id": message_id,
        "timestamp": timestamp,
    })


async def emit_cycle_start(
    session_id: str,
    cycle_id: int,
    round_index: int,
    max_rounds: int,
    history_count: int,
) -> None:
    """广播推理循环开始事件。"""

    await _broadcast("cycle.start", {
        "session_id": session_id,
        "cycle_id": cycle_id,
        "round_index": round_index,
        "max_rounds": max_rounds,
        "history_count": history_count,
        "timestamp": time.time(),
    })


async def emit_timing_gate_result(
    session_id: str,
    cycle_id: int,
    action: str,
    content: Optional[str],
    tool_calls: List[Any],
    messages: List[Any],
    prompt_tokens: int,
    selected_history_count: int,
    duration_ms: float,
) -> None:
    """广播 Timing Gate 结果事件。"""

    await _broadcast("timing_gate.result", {
        "session_id": session_id,
        "cycle_id": cycle_id,
        "action": action,
        "content": content,
        "tool_calls": _serialize_tool_calls_from_objects(tool_calls),
        "messages": _serialize_messages(messages),
        "prompt_tokens": prompt_tokens,
        "selected_history_count": selected_history_count,
        "duration_ms": duration_ms,
        "timestamp": time.time(),
    })


async def emit_planner_finalized(
    *,
    session_id: str,
    cycle_id: int,
    timing_request_messages: Optional[List[Any]],
    timing_selected_history_count: Optional[int],
    timing_tool_count: Optional[int],
    timing_action: Optional[str],
    timing_content: Optional[str],
    timing_tool_calls: Optional[List[Any]],
    timing_tool_results: Optional[List[str]],
    timing_prompt_tokens: Optional[int],
    timing_completion_tokens: Optional[int],
    timing_total_tokens: Optional[int],
    timing_duration_ms: Optional[float],
    planner_request_messages: Optional[List[Any]],
    planner_selected_history_count: Optional[int],
    planner_tool_count: Optional[int],
    planner_content: Optional[str],
    planner_tool_calls: Optional[List[Any]],
    planner_prompt_tokens: Optional[int],
    planner_completion_tokens: Optional[int],
    planner_total_tokens: Optional[int],
    planner_duration_ms: Optional[float],
    tools: Optional[List[Dict[str, Any]]],
    time_records: Dict[str, float],
    agent_state: str,
) -> None:
    """广播一轮 planner 结束后的最终聚合事件。"""

    await _broadcast("planner.finalized", {
        "session_id": session_id,
        "cycle_id": cycle_id,
        "timestamp": time.time(),
        "timing_gate": _serialize_timing_gate_block(
            request_messages=timing_request_messages,
            selected_history_count=timing_selected_history_count,
            tool_count=timing_tool_count,
            action=timing_action,
            content=timing_content,
            tool_calls=timing_tool_calls,
            tool_results=timing_tool_results,
            prompt_tokens=timing_prompt_tokens,
            completion_tokens=timing_completion_tokens,
            total_tokens=timing_total_tokens,
            duration_ms=timing_duration_ms,
        ),
        "request": _serialize_request_block(
            planner_request_messages,
            planner_selected_history_count,
            planner_tool_count,
        ),
        "planner": _serialize_planner_block(
            planner_content,
            planner_tool_calls,
            planner_prompt_tokens,
            planner_completion_tokens,
            planner_total_tokens,
            planner_duration_ms,
        ),
        "tools": _serialize_tool_results(list(tools or [])),
        "final_state": {
            "time_records": _normalize_payload_value(time_records),
            "agent_state": agent_state,
        },
    })
