Spaces:
Paused
Paused
File size: 6,047 Bytes
f1db6b2 |
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 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 |
from flask import jsonify
import logging
import os
formatter = logging.Formatter('%(message)s')
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
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, use_system_prompt=False):
gemini_history = []
errors = []
system_instruction_text = ""
is_system_phase = use_system_prompt
for i, message in enumerate(messages):
role = message.get('role')
content = message.get('content')
if isinstance(content, str):
if is_system_phase and role == 'system':
if system_instruction_text:
system_instruction_text += "\n" + content
else:
system_instruction_text = content
else:
is_system_phase = False
if role in ['user', 'system']:
role_to_use = 'user'
elif role == 'assistant':
role_to_use = 'model'
else:
errors.append(f"Invalid role: {role}")
logger.error(f"Invalid role: {role}")
continue
if gemini_history and gemini_history[-1]['role'] == role_to_use:
gemini_history[-1]['parts'].append({"text": content})
else:
gemini_history.append({"role": role_to_use, "parts": [{"text": content}]})
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) as e:
error_message = f"Invalid data URI for image: {image_data}. Error: {e}"
errors.append(error_message)
logger.error(error_message)
else:
error_message = f"Invalid image URL format for item: {item}. URL should start with 'data:image/' for inline images."
errors.append(error_message)
logger.error(error_message)
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) as e:
error_message = f"Invalid data URI for file: {file_data}. Error: {e}"
errors.append(error_message)
logger.error(error_message)
else:
error_message = f"Invalid file URL format for item: {item}. URL should start with 'data:' for inline files."
errors.append(error_message)
logger.error(error_message)
if parts:
if role in ['user', 'system']:
role_to_use = 'user'
elif role == 'assistant':
role_to_use = 'model'
else:
errors.append(f"Invalid role: {role}")
logger.error(f"Invalid role: {role}")
continue
if gemini_history and gemini_history[-1]['role'] == role_to_use:
gemini_history[-1]['parts'].extend(parts)
else:
gemini_history.append({"role": role_to_use, "parts": parts})
if errors:
logger.warning(f"Errors encountered during message processing: {errors}")
return gemini_history, {"parts": [{"text": system_instruction_text}]}, (jsonify({'error': errors}), 400)
else:
return gemini_history, {"parts": [{"text": system_instruction_text}]}, None |