plugins/json/core.py

"""
JSON内容提取核心逻辑
"""

import json
import logging
import os
import glob
from typing import List, Any

logger = logging.getLogger(__name__)


class JsonContentExtractorCore:
    """从JSON文件中提取content字段的核心逻辑"""

    def _is_conversation_format(self, json_data: Any) -> bool:
        """
        检测JSON是否为OpenAI对话格式

        Args:
            json_data: JSON数据

        Returns:
            是否为对话格式
        """
        if not isinstance(json_data, dict):
            return False

        # 检查是否有messages字段
        if "messages" not in json_data:
            return False

        messages = json_data["messages"]
        if not isinstance(messages, list) or len(messages) == 0:
            return False

        # 检查第一条消息是否有role和content字段
        first_msg = messages[0]
        if isinstance(first_msg, dict) and "role" in first_msg and "content" in first_msg:
            return True

        return False

    def _extract_conversation(self, json_data: dict) -> List[dict]:
        """
        从OpenAI对话格式中提取消息

        Args:
            json_data: 对话格式的JSON数据

        Returns:
            消息列表，每个元素包含role和content
        """
        messages = json_data.get("messages", [])
        result = []

        for msg in messages:
            if not isinstance(msg, dict):
                continue

            role = msg.get("role", "unknown")
            content = msg.get("content", "")

            # 处理content可能是数组的情况（OpenAI多模态格式）
            if isinstance(content, list):
                formatted_parts = []
                for item in content:
                    if not isinstance(item, dict):
                        if isinstance(item, str):
                            formatted_parts.append(item)
                        continue

                    item_type = item.get("type", "")

                    # 处理文本内容
                    if item_type == "text":
                        text = item.get("text", "")
                        if text.strip():
                            formatted_parts.append(text.strip())

                    # 处理思考内容
                    elif item_type == "thinking":
                        thinking = item.get("thinking", "")
                        if thinking.strip():
                            formatted_parts.append(f"[思考] {thinking.strip()}")

                    # 处理工具调用
                    elif item_type == "tool_use":
                        tool_name = item.get("name", "未知工具")
                        tool_input = item.get("input", {})
                        # 格式化工具调用
                        tool_str = f"[调用工具: {tool_name}]"
                        if tool_input and isinstance(tool_input, dict):
                            # 将输入参数格式化为可读形式
                            params = []
                            for k, v in tool_input.items():
                                if isinstance(v, str):
                                    # 多行字符串单独显示
                                    if "\n" in v or len(v) > 80:
                                        params.append(f"{k}:\n    {v}")
                                    else:
                                        params.append(f"{k}={v}")
                                else:
                                    params.append(f"{k}={v}")
                            tool_str += "\n" + "\n".join(params)
                        formatted_parts.append(tool_str)

                    # 处理工具结果
                    elif item_type == "tool_result":
                        tool_result_content = item.get("content", "")
                        is_error = item.get("is_error", False)
                        if isinstance(tool_result_content, str):
                            if is_error:
                                formatted_parts.append(f"[工具错误] {tool_result_content}")
                            else:
                                # 截断过长的结果
                                if len(tool_result_content) > 500:
                                    tool_result_content = tool_result_content[:500] + "..."
                                formatted_parts.append(f"[工具结果] {tool_result_content}")

                content = "\n".join(formatted_parts)
            elif isinstance(content, str):
                content = content.strip()
            else:
                content = ""

            if content:
                result.append({
                    "role": role,
                    "content": content
                })

        return result

    def extract_content_from_json(self, json_data: Any) -> List[str]:
        """
        从JSON数据中递归提取所有content字段的内容

        Args:
            json_data: JSON数据

        Returns:
            提取的content内容列表
        """
        contents = []

        if isinstance(json_data, dict):
            # 如果当前字典有content字段
            if "content" in json_data:
                content = json_data["content"]
                if isinstance(content, str):
                    contents.append(content)
                elif isinstance(content, list):
                    # 处理content是数组的情况（如OpenAI API格式）
                    for item in content:
                        if isinstance(item, dict):
                            # 提取text字段
                            if "text" in item:
                                contents.append(item["text"])
                            # 递归处理其他字段
                            contents.extend(self.extract_content_from_json(item))
                        elif isinstance(item, str):
                            contents.append(item)

            # 递归处理所有值
            for key, value in json_data.items():
                if key != "content":  # 避免重复处理
                    contents.extend(self.extract_content_from_json(value))

        elif isinstance(json_data, list):
            for item in json_data:
                contents.extend(self.extract_content_from_json(item))

        return contents

    def format_content(self, content: str) -> str:
        """
        格式化content内容，处理换行等转义字符
        
        Args:
            content: 原始content内容
            
        Returns:
            格式化后的内容
        """
        # 处理转义的换行符
        formatted = content.replace("\n", "\n")
        # 处理其他常见的转义字符
        formatted = formatted.replace("\t", "\t")
        formatted = formatted.replace('\\"', '"')
        return formatted

    def extract_content_from_file(self, file_path: str) -> List[str]:
        """
        从JSON文件中提取所有content字段的内容

        Args:
            file_path: JSON文件路径

        Returns:
            提取的content内容列表
        """
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                json_data = json.load(f)

            return self.extract_content_from_json(json_data)

        except Exception as e:
            logger.error(f"读取文件 {file_path} 时出错: {str(e)}")
            return []

    def process_json_file(self, file_path: str) -> str:
        """
        处理单个JSON文件，返回格式化的内容

        Args:
            file_path: JSON文件路径

        Returns:
            格式化的内容字符串
        """
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                json_data = json.load(f)

            # 检测是否为对话格式
            if self._is_conversation_format(json_data):
                return self._format_conversation(json_data)

            # 回退到通用content提取
            contents = self.extract_content_from_json(json_data)

            if not contents:
                return f"文件 {file_path} 中未找到content内容\n"

            result = []
            for i, content in enumerate(contents, 1):
                formatted = self.format_content(content)
                result.append(f"=== Content {i} ===\n{formatted}\n")

            return "\n".join(result)

        except Exception as e:
            logger.error(f"处理文件 {file_path} 时出错: {str(e)}")
            return f"处理文件 {file_path} 时出错: {str(e)}\n"

    def _format_conversation(self, json_data: dict) -> str:
        """
        将对话格式的JSON格式化为可读的对话记录

        Args:
            json_data: 对话格式的JSON数据

        Returns:
            格式化的对话记录
        """
        messages = self._extract_conversation(json_data)

        if not messages:
            return "未找到对话消息\n"

        result = []
        for msg in messages:
            role = msg["role"]
            content = msg["content"]

            # 使用中文角色名
            role_display = {
                "system": "系统",
                "user": "用户",
                "assistant": "助手"
            }.get(role, role)

            # 格式化内容，处理多行缩进
            lines = content.split("\n")
            formatted_lines = []
            for i, line in enumerate(lines):
                if i == 0:
                    formatted_lines.append(line)
                else:
                    # 后续行添加缩进
                    formatted_lines.append("  " + line)
            formatted_content = "\n".join(formatted_lines)

            result.append(f"[{role_display}] {formatted_content}\n")

        return "\n".join(result)

    def process_directory(self, dir_path: str, output_file: str = None) -> str:
        """
        处理目录中的所有JSON文件
        
        Args:
            dir_path: 目录路径
            output_file: 输出文件路径（可选）
            
        Returns:
            所有文件的提取结果
        """
        if not os.path.exists(dir_path):
            return f"目录 {dir_path} 不存在\n"

        # 查找所有JSON文件
        json_files = glob.glob(os.path.join(dir_path, "*.json"))

        if not json_files:
            return f"目录 {dir_path} 中未找到JSON文件\n"

        all_results = []
        for json_file in sorted(json_files):
            file_name = os.path.basename(json_file)
            all_results.append(f"\n{'='*60}")
            all_results.append(f"文件: {file_name}")
            all_results.append(f"{'='*60}\n")

            result = self.process_json_file(json_file)
            all_results.append(result)

        final_output = "\n".join(all_results)

        # 如果指定了输出文件，保存到文件
        if output_file:
            try:
                with open(output_file, 'w', encoding='utf-8') as f:
                    f.write(final_output)
                logger.info(f"结果已保存到: {output_file}")
            except Exception as e:
                logger.error(f"保存文件时出错: {str(e)}")

        return final_output

    def process_single_file(self, file_path: str, output_file: str = None) -> str:
        """
        处理单个文件

        Args:
            file_path: 文件路径
            output_file: 输出文件路径（可选）

        Returns:
            提取结果
        """
        if not os.path.exists(file_path):
            return f"文件 {file_path} 不存在\n"

        result = self.process_json_file(file_path)

        # 如果指定了输出文件，保存到文件
        if output_file:
            try:
                with open(output_file, 'w', encoding='utf-8') as f:
                    f.write(result)
                logger.info(f"结果已保存到: {output_file}")
            except Exception as e:
                logger.error(f"保存文件时出错: {str(e)}")

        return result

    def extract_conversation_from_json(self, json_data: Any) -> List[dict]:
        """
        从JSON数据中提取对话消息（结构化）

        Args:
            json_data: JSON数据

        Returns:
            对话消息列表，每个元素包含role和content
        """
        if self._is_conversation_format(json_data):
            return self._extract_conversation(json_data)
        return []

    def extract_conversation_from_file(self, file_path: str) -> List[dict]:
        """
        从文件中提取对话消息（结构化）

        Args:
            file_path: JSON文件路径

        Returns:
            对话消息列表
        """
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                json_data = json.load(f)
            return self.extract_conversation_from_json(json_data)
        except Exception as e:
            logger.error(f"读取文件 {file_path} 时出错: {str(e)}")
            return []