func.py

from io import BytesIO
import os
from flask import jsonify
import logging
import json
import re
logger = logging.getLogger(__name__)

ProxyPasswords = os.environ.get('ProxyPasswords').split(',')

def authenticate_request(request):
    auth_header = request.headers.get('Authorization')

    if not auth_header:
        return False, jsonify({'error': 'Authorization header is missing'}), 401
    
    try:
        auth_type, api_key = auth_header.split(' ', 1)
    except ValueError:
        return False, jsonify({'error': 'Invalid Authorization header format'}), 401

    if auth_type.lower() != 'bearer':
        return False, jsonify({'error': 'Authorization type must be Bearer'}), 401

    if api_key not in ProxyPasswords:
        return False, jsonify({'error': 'Unauthorized'}), 401

    return True, None, None

def sanitize_request_data(request_data):
    """

    从请求数据中删除base64编码的数据。



    Args:

        request_data: 包含可能存在base64数据的字典。



    Returns:

        清理后的字典，其中base64数据被替换为"[Base64 Data Omitted]"。

    """
    
    
    def replace_base64(match):
        # 替换base64数据为提示信息
        return '"[Base64 Data Omitted]"'

    
    request_data_str = json.dumps(request_data)

    # 使用正则表达式匹配base64数据，并替换为提示信息
    sanitized_request_data_str = re.sub(
        r'"(data:[^;]+;base64,)[^"]+"',
        replace_base64,
        request_data_str
    )

    return json.loads(sanitized_request_data_str)

def process_messages_for_gemini(messages):
    """

    将通用的对话消息格式转换为Gemini API所需的格式

    这个函数处理消息列表并将其转换为Gemini API兼容的格式。它支持文本、图片和文件内容的处理。

    参数:

        messages (list): 包含对话消息的列表。每条消息应该是一个字典，包含'role'和'content'字段。

            - role: 可以是 'system', 'user' 或 'assistant'

            - content: 可以是字符串或包含多个内容项的列表

    返回:

        tuple: 包含三个元素:

            - gemini_history (list): 转换后的历史消息列表

            - user_message (dict): 最新的用户消息

            - error (tuple or None): 如果有错误，返回错误响应；否则返回None

    错误处理:

        - 检查角色是否有效

        - 验证图片URL格式

        - 验证文件URL格式

    示例消息格式:

        文本消息:

        {

            'role': 'user',

            'content': '你好'

        }

        多模态消息:

        {

            'role': 'user',

            'content': [

                {'type': 'text', 'text': '这是什么图片?'},

                {'type': 'image_url', 'image_url': {'url': 'data:image/jpeg;base64,...'}}

            ]

        }

    """
    gemini_history = []
    errors = []
    for message in messages:
        role = message.get('role')
        content = message.get('content')

        if isinstance(content, str):
            if role == 'system':
                gemini_history.append({"role": "user", "parts": [content]})
            elif role == 'user':
                gemini_history.append({"role": "user", "parts": [content]})
            elif role == 'assistant':
                gemini_history.append({"role": "model", "parts": [content]})
            else:
                errors.append(f"Invalid role: {role}")
        elif isinstance(content, list):
            parts = []
            for item in content:
                if item.get('type') == 'text':
                    parts.append({"text": item.get('text')})  
                elif item.get('type') == 'image_url':
                    image_data = item.get('image_url', {}).get('url', '')
                    if image_data.startswith('data:image/'):

                        try:
                            mime_type, base64_data = image_data.split(';')[0].split(':')[1], image_data.split(',')[1]
                            parts.append({
                                "inline_data": {
                                    "mime_type": mime_type,
                                    "data": base64_data
                                }
                            })
                        except (IndexError, ValueError):
                            errors.append(f"Invalid data URI for image: {image_data}")
                    else:
                        errors.append(f"Invalid image URL format for item: {item}")
                elif item.get('type') == 'file_url':
                    file_data = item.get('file_url', {}).get('url', '')
                    if file_data.startswith('data:'):

                        try:
                            mime_type, base64_data = file_data.split(';')[0].split(':')[1], file_data.split(',')[1]
                            parts.append({
                                "inline_data": {
                                    "mime_type": mime_type,
                                    "data": base64_data
                                }
                            })
                        except (IndexError, ValueError):
                            errors.append(f"Invalid data URI for file: {file_data}")
                    else:
                        errors.append(f"Invalid file URL format for item: {item}")

            if parts: 
                if role in ['user', 'system']:
                    gemini_history.append({"role": "user", "parts": parts})
                elif role == 'assistant':
                    gemini_history.append({"role": "model", "parts": parts})
                else:
                    errors.append(f"Invalid role: {role}")

    if gemini_history:
        user_message = gemini_history[-1]
        gemini_history = gemini_history[:-1]
    else:
        user_message = {"role": "user", "parts": [""]}

    if errors:
        return gemini_history, user_message, (jsonify({'error': errors}), 400)
    else:
        return gemini_history, user_message, None