Spaces:
Running
Running
File size: 4,226 Bytes
7e8a8d2 913b6ab 7e8a8d2 913b6ab 7e8a8d2 913b6ab 0af7a49 913b6ab 0af7a49 913b6ab 0af7a49 913b6ab 0af7a49 913b6ab 0af7a49 913b6ab 0af7a49 7e8a8d2 0d399a4 7e8a8d2 0d399a4 7e8a8d2 43a10de 7e8a8d2 0d399a4 7e8a8d2 0d399a4 7e8a8d2 0d399a4 7e8a8d2 0d399a4 7e8a8d2 0d399a4 7e8a8d2 0d399a4 43a10de 0d399a4 43a10de 0d399a4 7e8a8d2 0d399a4 7e8a8d2 0d399a4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 |
from io import BytesIO
import base64
from PIL import Image
from flask import jsonify
import logging
import json
import re
import os
import requests
import google.generativeai as genai
logger = logging.getLogger(__name__)
request_counts = {}
password = os.environ['password']
def authenticate_request(request):
auth_header = request.headers.get('Authorization')
if not auth_header:
return False, jsonify({'error': '缺少Authorization请求头'}), 401
try:
auth_type, pass_word = auth_header.split(' ', 1)
except ValueError:
return False, jsonify({'error': 'Authorization请求头格式错误'}), 401
if auth_type.lower() != 'bearer':
return False, jsonify({'error': 'Authorization类型必须为Bearer'}), 401
if pass_word != password:
return False, jsonify({'error': '未授权'}), 401
return True, None, None
def process_messages_for_gemini(messages):
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 in ['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 |