app/core/context/converter.py

# -*- coding: utf-8 -*-
"""
消息格式转换器。
主要负责将 OpenAI API 格式的消息列表转换为 Gemini API 兼容的 'contents' 列表格式。
支持处理文本内容、图像内容 (通过 Base64 编码的 Data URI)，以及可选的系统指令提取。
"""
import re # 导入正则表达式模块，用于解析 Data URI
import logging # 导入日志模块
from typing import List, Dict, Any, Tuple, Union, Set # 导入类型提示

# 导入 Message Pydantic 模型，用于类型检查和访问消息属性
from app.api.models import Message # (新路径)

logger = logging.getLogger('my_logger') # 获取日志记录器实例

# 定义 Gemini API 支持的图像 MIME 类型集合
SUPPORTED_IMAGE_MIME_TYPES: Set[str] = {
    "image/jpeg",
    "image/png",
    "image/webp",
    "image/heic",
    "image/heif",
}

# 编译用于解析和验证图像 Data URI 的正则表达式
# 模式: data:<MIME 类型>;base64,<Base64 编码的数据>
# - `data:`: 固定前缀
# - `(` + `|`.join(...) + `)`: 捕获组 1，匹配 SUPPORTED_IMAGE_MIME_TYPES 中的任意一个
#   - `re.escape(m)`: 转义 MIME 类型字符串中的特殊字符
# - `;base64,`: 固定分隔符
# - `(.+)`: 捕获组 2，匹配 Base64 编码的数据部分 (至少一个字符)
DATA_URI_REGEX = re.compile(r"^data:(" + "|".join(re.escape(m) for m in SUPPORTED_IMAGE_MIME_TYPES) + r");base64,(.+)$")

def _process_text_content(
    content: str, # 纯文本内容
    role: str, # OpenAI 格式的角色 ('system', 'user', 'assistant')
    gemini_history: List[Dict[str, Any]], # 正在构建的 Gemini contents 列表
    is_system_phase: bool, # 当前是否正在处理系统指令阶段
    system_instruction_text: str, # 已累积的系统指令文本
    errors: List[str] # 用于收集错误的列表
) -> Tuple[bool, str]:
    """
    (内部辅助函数) 处理纯文本类型的消息内容。
    - 如果处于系统指令处理阶段且角色为 'system'，则累加文本到系统指令。
    - 否则，将文本内容添加到 Gemini contents 列表中，处理角色映射和消息合并。

    Args:
        content (str): 消息的文本内容。
        role (str): 消息的角色。
        gemini_history (List[Dict[str, Any]]): 当前已转换的 Gemini contents 列表。
        is_system_phase (bool): 是否处于系统指令处理阶段。
        system_instruction_text (str): 当前累积的系统指令文本。
        errors (List[str]): 用于记录错误的列表。

    Returns:
        Tuple[bool, str]:
            - 第一个元素 (bool): 更新后的 is_system_phase 状态。
            - 第二个元素 (str): 更新后的 system_instruction_text。
    """
    if is_system_phase and role == 'system': # 如果在系统指令阶段且角色是 system
        # 将当前系统消息内容追加到已有的系统指令文本后面
        if system_instruction_text: # 如果已有系统指令
            system_instruction_text += "\n" + content # 用换行符分隔追加
        else: # 如果是第一条系统指令
            system_instruction_text = content # 直接赋值
        return True, system_instruction_text # 保持在系统指令阶段，返回更新后的文本
    else: # 如果不是系统指令阶段，或者角色不是 system
        # 遇到非系统消息或非 system 角色的消息，则退出系统指令处理阶段
        is_system_phase = False

        # --- 映射 OpenAI 角色到 Gemini 角色 ---
        # Gemini API 只接受 'user' 和 'model' 两种角色
        if role in ['user', 'system']:  # 将 'system' (如果不是第一条) 也视为 'user'
            role_to_use = 'user'
        elif role == 'assistant': # OpenAI 的 'assistant' 对应 Gemini 的 'model'
            role_to_use = 'model'
        else: # 如果遇到无法识别的角色
            # 记录错误并跳过此消息
            errors.append(f"无效的角色 '{role}'") # 添加错误信息
            # 返回非系统指令阶段，系统指令文本不变
            return False, system_instruction_text

        # --- 合并连续相同角色的消息 ---
        # Gemini API 要求 'user' 和 'model' 角色交替出现
        # 如果当前消息的角色与 gemini_history 中最后一条消息的角色相同
        if gemini_history and gemini_history[-1]['role'] == role_to_use:
            # 将当前文本内容追加到上一条消息的 'parts' 列表中
            # 首先确保上一条消息的 'parts' 是一个列表
            if isinstance(gemini_history[-1].get('parts'), list):
                gemini_history[-1]['parts'].append({"text": content}) # 追加新的 text part
            else:
                # 如果 parts 不是列表（异常情况），则创建一个新的 parts 列表
                gemini_history[-1]['parts'] = [{"text": content}]
                logger.warning(f"发现非列表类型的 parts，已重新初始化。") # 记录警告
        else: # 如果角色不同，或者 gemini_history 为空
            # 添加一个新的消息条目到 gemini_history
            gemini_history.append(
                {"role": role_to_use, "parts": [{"text": content}]} # 创建新的 content 字典
            )

        # 返回非系统指令阶段，系统指令文本不变
        return False, system_instruction_text

def _process_multi_part_content(
    content: List[Dict[str, Any]], # OpenAI 格式的多部分内容列表
    role: str, # OpenAI 格式的角色
    gemini_history: List[Dict[str, Any]], # 正在构建的 Gemini contents 列表
    errors: List[str], # 用于收集错误的列表
    message_index: int # 当前消息在原始列表中的索引 (用于错误报告)
) -> bool:
    """
    (内部辅助函数) 处理包含多部分内容（例如文本和图像）的 OpenAI 消息。
    将 OpenAI 的多部分 content 列表转换为 Gemini 的 parts 列表。
    支持 text 和 image_url (Data URI 格式) 类型。
    处理角色映射和消息合并。

    Args:
        content (List[Dict[str, Any]]): OpenAI 格式的多部分内容列表。
        role (str): 消息的角色。
        gemini_history (List[Dict[str, Any]]): 当前已转换的 Gemini contents 列表。
        errors (List[str]): 用于记录错误的列表。
        message_index (int): 当前消息的索引，用于生成更清晰的错误信息。

    Returns:
        bool: 如果处理过程中发生错误，返回 True；否则返回 False。
    """
    parts = []  # 初始化用于存储转换后的 Gemini 'parts' 的列表
    has_error_in_item = False # 标记当前消息的 content 列表处理中是否出现错误

    # 遍历 OpenAI content 列表中的每个部分 (item)
    for item_index, item in enumerate(content):
        item_type = item.get('type') # 获取部分的类型 ('text' 或 'image_url')

        # --- 特殊处理：兼容某些客户端可能发送的无类型文本部分 ---
        # 检查 item 是否为字典，是否包含 'text' 键，且值是字符串，并且没有 'type' 键
        if item_type is None and isinstance(item, dict) and 'text' in item and isinstance(item.get('text'), str):
            logger.debug(f"消息 {message_index} 项目 {item_index}: 检测到无类型的文本部分，直接处理。") # 记录调试信息
            parts.append({"text": item['text']}) # 直接添加文本 part
            continue # 处理下一个项目
        # --- 特殊处理结束 ---

        if item_type == 'text': # --- 处理文本部分 ---
            # 直接添加 text part，使用 get 获取文本内容，提供默认空字符串以防万一
            parts.append({"text": item.get('text', '')})
        elif item_type == 'image_url': # --- 处理图像 URL 部分 ---
            # 获取 image_url 字典
            image_url_dict = item.get('image_url', {})
            # 验证 image_url 是否为字典
            if not isinstance(image_url_dict, dict):
                 errors.append(f"消息 {message_index} 项目 {item_index}: 'image_url' 必须是字典，但得到 {type(image_url_dict)}") # 记录错误
                 has_error_in_item = True # 标记错误
                 continue # 跳过这个损坏的项目

            image_data = image_url_dict.get('url', '') # 获取图像的 URL (期望是 Data URI)
            # 使用预编译的正则表达式解析和验证 Data URI
            match = DATA_URI_REGEX.match(image_data)
            if match: # 如果 Data URI 格式有效且 MIME 类型受支持
                mime_type = match.group(1) # 提取 MIME 类型
                base64_data = match.group(2) # 提取 Base64 编码的数据
                # 创建 Gemini 的 inline_data part
                parts.append({
                    "inline_data": {
                        "mime_type": mime_type,
                        "data": base64_data # Base64 数据字符串
                    }
                })
            else: # 如果 Data URI 格式无效或 MIME 类型不支持
                # 构造详细的错误消息
                error_msg = f"消息 {message_index} 项目 {item_index}: 无效或不支持的图像 Data URI。"
                if image_data.startswith('data:image/'): # 如果是图像 Data URI 但格式或类型错误
                     error_msg += f" MIME 类型必须是 {SUPPORTED_IMAGE_MIME_TYPES} 之一且格式正确。"
                else: # 如果根本不是 Data URI
                     error_msg += f" 仅接受 Base64 编码的 Data URI，支持的 MIME 类型为: {', '.join(SUPPORTED_IMAGE_MIME_TYPES)}。"
                errors.append(error_msg) # 添加错误信息到列表
                has_error_in_item = True # 标记错误
        else: # --- 处理不支持的内容类型 ---
            # 如果 item 的 type 不是 'text' 或 'image_url'
            errors.append(f"消息 {message_index} 项目 {item_index}: 不支持的内容类型 '{item_type}'") # 记录错误
            has_error_in_item = True # 标记错误

    # --- 处理转换后的 parts ---
    # 只有在成功解析出 parts 并且当前消息没有发生错误时才进行后续处理
    if parts and not has_error_in_item:
        # --- 合并多个文本 parts ---
        # Gemini API 可能对单个 content 中的多个 text part 处理不佳，尝试合并它们
        text_parts = [p['text'] for p in parts if 'text' in p and len(p) == 1] # 提取所有纯文本 part 的内容
        non_text_parts = [p for p in parts if 'text' not in p or len(p) > 1] # 保留所有非文本 part (如图像)

        if len(text_parts) > 1: # 如果存在多个文本 part
            merged_text = "\n".join(text_parts) # 使用换行符将它们合并成一个字符串
            logger.debug(f"消息 {message_index}: 检测到 {len(text_parts)} 个文本 parts，合并为一个。") # 记录合并操作
            # 构建新的 parts 列表：合并后的文本在前，非文本部分在后
            merged_parts = [{"text": merged_text}] + non_text_parts
            parts = merged_parts # 使用合并后的 parts 列表
        # --- 合并结束 ---

        # --- 映射角色 ---
        if role in ['user', 'system']: # OpenAI 'user'/'system' -> Gemini 'user'
            role_to_use = 'user'
        elif role == 'assistant': # OpenAI 'assistant' -> Gemini 'model'
            role_to_use = 'model'
        else: # 无效角色
            errors.append(f"消息 {message_index}: 无效的角色 '{role}'") # 记录错误
            return True # 返回 True 表示此消息处理出错

        # --- 合并连续相同角色的消息 ---
        if gemini_history and gemini_history[-1]['role'] == role_to_use: # 如果历史不为空且最后一条消息角色相同
            # 将当前消息的 parts 追加到上一条消息的 parts 列表中
            if isinstance(gemini_history[-1].get('parts'), list): # 确保上一条的 parts 是列表
                gemini_history[-1]['parts'].extend(parts) # 追加 parts
            else: # 处理异常情况
                gemini_history[-1]['parts'] = parts # 直接替换为新的 parts
                logger.warning(f"消息 {message_index}: 发现非列表类型的 parts，已重新初始化。") # 记录警告
        else: # 如果角色不同或历史为空
            # 添加新的消息条目到 gemini_history
            gemini_history.append(
                {"role": role_to_use, "parts": parts} # 创建新的 content 字典
            )
        return False # 返回 False 表示此消息处理成功
    elif not parts and not has_error_in_item: # 如果 parts 为空但没有错误 (例如，内容列表为空或所有项都无效但被跳过)
         logger.warning(f"消息 {message_index}: 内容列表为空或所有项目均无效，已跳过。") # 记录警告
         return False # 返回 False 表示此消息未产生错误（虽然也没添加内容）
    else: # 如果 has_error_in_item 为 True
        # 错误信息已记录在 errors 列表中，直接返回 True 表示此消息处理出错
        return True

def convert_messages(messages: List[Message], use_system_prompt=False) -> Union[Tuple[List[Dict[str, Any]], Dict[str, Any]], List[str]]:
    """
    将 OpenAI 格式的消息列表转换为 Gemini API 格式的 'contents' 列表和 system_instruction。

    处理逻辑：
    1. 遍历 OpenAI 消息列表。
    2. 如果 `use_system_prompt` 为 True，将第一个 'system' 角色的消息内容提取为系统指令。
    3. 处理后续消息：
       - 映射角色 ('user'/'system' -> 'user', 'assistant' -> 'model')。
       - 处理文本内容。
       - 处理多部分内容（文本和图像 Data URI）。
       - 合并连续相同角色的消息。
    4. 如果在转换过程中遇到任何错误，返回包含错误信息的列表。
    5. 如果转换成功，返回包含 Gemini 'contents' 列表和 'system_instruction' 字典的元组。

    Args:
        messages (List[Message]): OpenAI 格式的消息列表 (Pydantic 模型对象列表)。
        use_system_prompt (bool): 是否启用系统指令提取功能。默认为 False。

    Returns:
        Union[Tuple[List[Dict[str, Any]], Dict[str, Any]], List[str]]:
            - 成功时: 返回一个元组 `(gemini_history, system_instruction_dict)`。
              `gemini_history` 是转换后的 Gemini contents 列表。
              `system_instruction_dict` 是包含系统指令的字典 (如果提取到)，否则为空字典。
            - 失败时: 返回一个包含描述性错误信息的字符串列表。
    """
    gemini_history: List[Dict[str, Any]] = []  # 初始化 Gemini contents 列表
    errors: List[str] = []          # 初始化错误信息列表
    system_instruction_text = ""  # 初始化系统指令文本
    is_system_phase = use_system_prompt  # 根据参数设置初始是否处于系统指令处理阶段

    # 遍历输入的 OpenAI 消息列表
    for i, message in enumerate(messages):
        role = message.role # 获取角色
        content = message.content # 获取内容

        # 记录正在处理的消息（用于调试）
        logger.debug(f"正在处理消息 {i}: role={role}, content_type={type(content)}") # 记录调试日志

        # --- 根据内容类型调用不同的处理函数 ---
        if isinstance(content, str): # 如果内容是纯字符串
            # 调用处理纯文本内容的辅助函数
            is_system_phase, system_instruction_text = _process_text_content(
                content, role, gemini_history, is_system_phase, system_instruction_text, errors
            )
        elif isinstance(content, list): # 如果内容是列表 (表示多部分内容)
            # 遇到多部分内容时，强制退出系统指令处理阶段
            is_system_phase = False
            # 调用处理多部分内容的辅助函数
            has_error = _process_multi_part_content(
                content, role, gemini_history, errors, i
            )
            # 如果处理多部分内容时发生错误，跳过当前消息，继续处理下一条
            if has_error:
                continue
        else: # 如果内容类型既不是字符串也不是列表
            # 记录不支持的内容类型错误
            errors.append(f"消息 {i}: 不支持的内容类型 '{type(content)}'")


    # --- 处理转换结果 ---
    if errors: # 如果在转换过程中收集到了错误
        logger.error(f"消息转换失败: {'; '.join(errors)}") # 记录整体转换失败的错误日志
        return errors # 返回错误信息列表
    else: # 如果没有错误
        # 准备 system_instruction 字典
        system_instruction_dict = {} # 初始化为空字典
        if system_instruction_text: # 如果成功提取到了系统指令文本
            # Gemini API 要求 system_instruction 是一个包含 'parts' 列表的字典
            system_instruction_dict = {"parts": [{"text": system_instruction_text}]}

        # 返回转换成功的 Gemini contents 列表和 system_instruction 字典
        return gemini_history, system_instruction_dict