plugins/json/openai_parser.py

"""
OpenAI 对话格式解析器

支持 OpenAI messages、content parts、tool calls/results 的结构化解析。
"""

import json
import logging
import uuid
from typing import Any, Dict, List, Optional, Tuple

logger = logging.getLogger(__name__)


class OpenAIMessageParser:
    """OpenAI 对话格式解析器"""

    def parse_json(self, json_data: Any) -> Dict[str, Any]:
        """
        解析 JSON 数据为对话视图模型

        Args:
            json_data: JSON 数据（dict 或 list）

        Returns:
            对话视图模型 {
                "conversation_id": str,
                "messages": List[dict],
                "raw_warnings": List[str],
            }
        """
        conversation_id = str(uuid.uuid4())[:8]
        messages = []
        raw_warnings = []

        # 处理不同的输入格式
        if isinstance(json_data, dict):
            # 检查是否有 messages 字段
            if "messages" in json_data:
                conversation_id = json_data.get("conversation_id", conversation_id)
                raw_messages = json_data["messages"]
                if isinstance(raw_messages, list):
                    messages, raw_warnings = self._parse_messages(raw_messages)
                else:
                    raw_warnings.append("messages 字段不是数组")
            else:
                # 尝试将整个 dict 作为单条消息解析
                msg, warnings = self._parse_single_message(json_data, 0)
                if msg:
                    messages.append(msg)
                raw_warnings.extend(warnings)

        elif isinstance(json_data, list):
            # 直接作为消息列表解析
            messages, raw_warnings = self._parse_messages(json_data)

        else:
            raw_warnings.append(f"不支持的 JSON 类型: {type(json_data).__name__}")

        return {
            "conversation_id": conversation_id,
            "messages": messages,
            "raw_warnings": raw_warnings,
        }

    def parse_jsonl(self, text: str) -> Dict[str, Any]:
        """
        解析 JSONL 格式

        Args:
            text: JSONL 文本

        Returns:
            对话视图模型
        """
        conversation_id = str(uuid.uuid4())[:8]
        raw_messages = []
        raw_warnings = []

        for line_num, line in enumerate(text.strip().split("\n"), 1):
            line = line.strip()
            if not line:
                continue
            try:
                data = json.loads(line)
                raw_messages.append(data)
            except json.JSONDecodeError as e:
                raw_warnings.append(f"第 {line_num} 行 JSON 解析失败: {e}")

        messages, warnings = self._parse_messages(raw_messages)
        raw_warnings.extend(warnings)

        return {
            "conversation_id": conversation_id,
            "messages": messages,
            "raw_warnings": raw_warnings,
        }

    def _parse_messages(self, raw_messages: List[Any]) -> Tuple[List[dict], List[str]]:
        """
        解析消息列表

        Args:
            raw_messages: 原始消息列表

        Returns:
            (解析后的消息列表, 警告列表)
        """
        messages = []
        warnings = []

        for idx, raw_msg in enumerate(raw_messages):
            if not isinstance(raw_msg, dict):
                warnings.append(f"消息 {idx} 不是对象，跳过")
                continue

            msg, msg_warnings = self._parse_single_message(raw_msg, idx)
            if msg:
                messages.append(msg)
            warnings.extend(msg_warnings)

        return messages, warnings

    def _parse_single_message(self, raw_msg: dict, index: int) -> Tuple[Optional[dict], List[str]]:
        """
        解析单条消息

        Args:
            raw_msg: 原始消息
            index: 消息索引

        Returns:
            (解析后的消息, 警告列表)
        """
        warnings = []

        # 提取基本字段
        role = raw_msg.get("role", "unknown")
        msg_id = raw_msg.get("id", f"msg_{index}")

        # 解析 content
        content_parts, content_warnings = self._parse_content(raw_msg.get("content"))
        warnings.extend(content_warnings)

        # 解析 tool_calls
        tool_calls = self._parse_tool_calls(raw_msg.get("tool_calls"))

        # 解析 tool results (从 tool_call_id 字段判断)
        tool_results = self._parse_tool_results(raw_msg)

        # 提取 metadata
        metadata = {}
        for key in ["name", "function_call", "tool_call_id", "finish_reason", "usage"]:
            if key in raw_msg:
                metadata[key] = raw_msg[key]

        return {
            "id": msg_id,
            "role": role,
            "content_parts": content_parts,
            "tool_calls": tool_calls,
            "tool_results": tool_results,
            "metadata": metadata,
            "raw_index": index,
        }, warnings

    def _parse_content(self, content: Any) -> Tuple[List[dict], List[str]]:
        """
        解析 content 字段

        Args:
            content: content 字段值

        Returns:
            (content parts 列表, 警告列表)
        """
        parts = []
        warnings = []

        if content is None:
            return parts, warnings

        if isinstance(content, str):
            # 纯文本
            if content.strip():
                parts.append({
                    "type": "text",
                    "text": content.strip(),
                })

        elif isinstance(content, list):
            # 数组格式（OpenAI 多模态）
            for item in content:
                if isinstance(item, dict):
                    part, item_warnings = self._parse_content_part(item)
                    if part:
                        parts.append(part)
                    warnings.extend(item_warnings)
                elif isinstance(item, str):
                    if item.strip():
                        parts.append({
                            "type": "text",
                            "text": item.strip(),
                        })

        else:
            warnings.append(f"不支持的 content 类型: {type(content).__name__}")

        return parts, warnings

    def _parse_content_part(self, item: dict) -> Tuple[Optional[dict], List[str]]:
        """
        解析单个 content part

        Args:
            item: content part 对象

        Returns:
            (解析后的 part, 警告列表)
        """
        warnings = []
        item_type = item.get("type", "")

        if item_type == "text":
            text = item.get("text", "")
            if text.strip():
                return {
                    "type": "text",
                    "text": text.strip(),
                }, warnings

        elif item_type == "thinking":
            thinking = item.get("thinking", "")
            if thinking.strip():
                return {
                    "type": "thinking",
                    "text": thinking.strip(),
                }, warnings

        elif item_type == "image_url":
            image_url = item.get("image_url", {})
            url = image_url.get("url", "") if isinstance(image_url, dict) else ""
            return {
                "type": "image",
                "url": url,
            }, warnings

        elif item_type == "tool_use":
            # Anthropic 格式的 tool_use
            return {
                "type": "tool_use",
                "tool_name": item.get("name", "未知工具"),
                "tool_input": item.get("input", {}),
                "tool_use_id": item.get("id", ""),
            }, warnings

        elif item_type == "tool_result":
            # Anthropic 格式的 tool_result
            result_content = item.get("content", "")
            is_error = item.get("is_error", False)
            return {
                "type": "tool_result",
                "tool_use_id": item.get("tool_use_id", ""),
                "content": result_content,
                "is_error": is_error,
            }, warnings

        else:
            warnings.append(f"不支持的 content part 类型: {item_type}")

        return None, warnings

    def _parse_tool_calls(self, tool_calls: Any) -> List[dict]:
        """
        解析 tool_calls 字段（OpenAI 格式）

        Args:
            tool_calls: tool_calls 字段值

        Returns:
            解析后的 tool calls 列表
        """
        if not tool_calls or not isinstance(tool_calls, list):
            return []

        result = []
        for tc in tool_calls:
            if not isinstance(tc, dict):
                continue

            tc_id = tc.get("id", "")
            tc_type = tc.get("type", "function")
            function = tc.get("function", {})

            if isinstance(function, dict):
                func_name = function.get("name", "")
                func_args = function.get("arguments", "")

                # 尝试解析 arguments JSON
                if isinstance(func_args, str):
                    try:
                        func_args = json.loads(func_args)
                    except json.JSONDecodeError:
                        pass

                result.append({
                    "id": tc_id,
                    "type": tc_type,
                    "function_name": func_name,
                    "function_arguments": func_args,
                })

        return result

    def _parse_tool_results(self, raw_msg: dict) -> List[dict]:
        """
        解析 tool results（从消息中提取）

        Args:
            raw_msg: 原始消息

        Returns:
            tool results 列表
        """
        results = []

        # OpenAI 格式：tool_call_id 字段
        if "tool_call_id" in raw_msg:
            results.append({
                "tool_call_id": raw_msg["tool_call_id"],
                "content": raw_msg.get("content", ""),
                "is_error": False,
            })

        return results


# 全局单例
_parser: Optional[OpenAIMessageParser] = None


def get_openai_parser() -> OpenAIMessageParser:
    """获取 OpenAI 对话解析器全局单例"""
    global _parser
    if _parser is None:
        _parser = OpenAIMessageParser()
    return _parser