mirror of
https://github.com/snailyp/gemini-balance.git
synced 2026-09-04 15:11:15 +08:00
snaily
4becc8d4d4
本次提交主要包含以下更新：
- **错误日志页面增强**:
- 重构了 [`app/static/js/error_logs.js`](app/static/js/error_logs.js) 中的分页逻辑，将样式控制移至 CSS，简化了 JavaScript 代码。
- 更新了 [`app/templates/error_logs.html`](app/templates/error_logs.html) 中的分页样式，使其与 `keys_status.html` 保持一致，提升了视觉统一性。
- 在错误日志页面新增了“清空全部”按钮，方便用户一键清除所有错误记录。
- 调整了错误日志表格头部的文本颜色为白色，以改善深色主题下的可读性。
- **应用初始化与配置优化**:
- 调整了 [`app/config/config.py`](app/config/config.py) 中日志记录器的获取方式，确保在配置加载早期即可用。
- 在 [`app/core/application.py`](app/core/application.py) 中引入了更明确的数据库连接管理（连接、断开、初始化）逻辑。
- 优化了 [`app/utils/helpers.py`](app/utils/helpers.py) 中项目路径和版本文件路径的定义方式，使其在模块级别初始化。
- **依赖清理**:
- 从 [`requirements.txt`](requirements.txt) 中移除了不必要的注释。
这些更改旨在提升错误日志模块的用户体验和功能性，并优化应用程序的启动和配置管理流程。
350 lines
14 KiB
Python
350 lines
14 KiB
Python
import base64
|
|
import json
|
|
import re
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import requests
|
|
|
|
from app.core.constants import (
|
|
AUDIO_FORMAT_TO_MIMETYPE,
|
|
DATA_URL_PATTERN,
|
|
IMAGE_URL_PATTERN,
|
|
MAX_AUDIO_SIZE_BYTES,
|
|
MAX_VIDEO_SIZE_BYTES,
|
|
SUPPORTED_AUDIO_FORMATS,
|
|
SUPPORTED_ROLES,
|
|
SUPPORTED_VIDEO_FORMATS,
|
|
VIDEO_FORMAT_TO_MIMETYPE,
|
|
)
|
|
from app.log.logger import get_message_converter_logger
|
|
|
|
logger = get_message_converter_logger()
|
|
|
|
|
|
class MessageConverter(ABC):
|
|
"""消息转换器基类"""
|
|
|
|
@abstractmethod
|
|
def convert(
|
|
self, messages: List[Dict[str, Any]]
|
|
) -> tuple[List[Dict[str, Any]], Optional[Dict[str, Any]]]:
|
|
pass
|
|
|
|
|
|
def _get_mime_type_and_data(base64_string):
|
|
"""
|
|
从 base64 字符串中提取 MIME 类型和数据。
|
|
|
|
参数:
|
|
base64_string (str): 可能包含 MIME 类型信息的 base64 字符串
|
|
|
|
返回:
|
|
tuple: (mime_type, encoded_data)
|
|
"""
|
|
# 检查字符串是否以 "data:" 格式开始
|
|
if base64_string.startswith("data:"):
|
|
# 提取 MIME 类型和数据
|
|
pattern = DATA_URL_PATTERN
|
|
match = re.match(pattern, base64_string)
|
|
if match:
|
|
mime_type = (
|
|
"image/jpeg" if match.group(1) == "image/jpg" else match.group(1)
|
|
)
|
|
encoded_data = match.group(2)
|
|
return mime_type, encoded_data
|
|
|
|
# 如果不是预期格式，假定它只是数据部分
|
|
return None, base64_string
|
|
|
|
|
|
def _convert_image(image_url: str) -> Dict[str, Any]:
|
|
if image_url.startswith("data:image"):
|
|
mime_type, encoded_data = _get_mime_type_and_data(image_url)
|
|
return {"inline_data": {"mime_type": mime_type, "data": encoded_data}}
|
|
else:
|
|
encoded_data = _convert_image_to_base64(image_url)
|
|
return {"inline_data": {"mime_type": "image/png", "data": encoded_data}}
|
|
|
|
|
|
def _convert_image_to_base64(url: str) -> str:
|
|
"""
|
|
将图片URL转换为base64编码
|
|
Args:
|
|
url: 图片URL
|
|
Returns:
|
|
str: base64编码的图片数据
|
|
"""
|
|
response = requests.get(url)
|
|
if response.status_code == 200:
|
|
# 将图片内容转换为base64
|
|
img_data = base64.b64encode(response.content).decode("utf-8")
|
|
return img_data
|
|
else:
|
|
raise Exception(f"Failed to fetch image: {response.status_code}")
|
|
|
|
|
|
def _process_text_with_image(text: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
处理可能包含图片URL的文本，提取图片并转换为base64
|
|
|
|
Args:
|
|
text: 可能包含图片URL的文本
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: 包含文本和图片的部分列表
|
|
"""
|
|
parts = []
|
|
img_url_match = re.search(IMAGE_URL_PATTERN, text)
|
|
if img_url_match:
|
|
# 提取URL
|
|
img_url = img_url_match.group(2)
|
|
# 将URL对应的图片转换为base64
|
|
try:
|
|
base64_data = _convert_image_to_base64(img_url)
|
|
parts.append(
|
|
{"inline_data": {"mimeType": "image/png", "data": base64_data}}
|
|
)
|
|
except Exception:
|
|
# 如果转换失败，回退到文本模式
|
|
parts.append({"text": text})
|
|
else:
|
|
# 没有图片URL，作为纯文本处理
|
|
parts.append({"text": text})
|
|
return parts
|
|
|
|
|
|
class OpenAIMessageConverter(MessageConverter):
|
|
"""OpenAI消息格式转换器"""
|
|
|
|
def _validate_media_data(
|
|
self, format: str, data: str, supported_formats: List[str], max_size: int
|
|
) -> tuple[Optional[str], Optional[str]]:
|
|
"""Validates format and size of Base64 media data."""
|
|
if format.lower() not in supported_formats:
|
|
logger.error(
|
|
f"Unsupported media format: {format}. Supported: {supported_formats}"
|
|
)
|
|
raise ValueError(f"Unsupported media format: {format}")
|
|
|
|
try:
|
|
decoded_data = base64.b64decode(data, validate=True)
|
|
if len(decoded_data) > max_size:
|
|
logger.error(
|
|
f"Media data size ({len(decoded_data)} bytes) exceeds limit ({max_size} bytes)."
|
|
)
|
|
raise ValueError(
|
|
f"Media data size exceeds limit of {max_size // 1024 // 1024}MB"
|
|
)
|
|
return data
|
|
except base64.binascii.Error as e:
|
|
logger.error(f"Invalid Base64 data provided: {e}")
|
|
raise ValueError("Invalid Base64 data")
|
|
except Exception as e:
|
|
logger.error(f"Error validating media data: {e}")
|
|
raise
|
|
|
|
def convert(
|
|
self, messages: List[Dict[str, Any]]
|
|
) -> tuple[List[Dict[str, Any]], Optional[Dict[str, Any]]]:
|
|
converted_messages = []
|
|
system_instruction_parts = []
|
|
|
|
for idx, msg in enumerate(messages):
|
|
role = msg.get("role", "")
|
|
parts = []
|
|
|
|
if "content" in msg and isinstance(msg["content"], list):
|
|
for content_item in msg["content"]:
|
|
if not isinstance(content_item, dict):
|
|
logger.warning(
|
|
f"Skipping unexpected content item format: {type(content_item)}"
|
|
)
|
|
continue
|
|
|
|
content_type = content_item.get("type")
|
|
|
|
if content_type == "text" and content_item.get("text"):
|
|
parts.append({"text": content_item["text"]})
|
|
elif content_type == "image_url" and content_item.get(
|
|
"image_url", {}
|
|
).get("url"):
|
|
try:
|
|
parts.append(
|
|
_convert_image(content_item["image_url"]["url"])
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Failed to convert image URL {content_item['image_url']['url']}: {e}"
|
|
)
|
|
parts.append(
|
|
{
|
|
"text": f"[Error processing image: {content_item['image_url']['url']}]"
|
|
}
|
|
)
|
|
elif content_type == "input_audio" and content_item.get(
|
|
"input_audio"
|
|
):
|
|
audio_info = content_item["input_audio"]
|
|
audio_data = audio_info.get("data")
|
|
audio_format = audio_info.get("format", "").lower()
|
|
|
|
if not audio_data or not audio_format:
|
|
logger.warning(
|
|
"Skipping audio part due to missing data or format."
|
|
)
|
|
continue
|
|
|
|
try:
|
|
validated_data = self._validate_media_data(
|
|
audio_format,
|
|
audio_data,
|
|
SUPPORTED_AUDIO_FORMATS,
|
|
MAX_AUDIO_SIZE_BYTES,
|
|
)
|
|
|
|
# Get MIME type
|
|
mime_type = AUDIO_FORMAT_TO_MIMETYPE.get(audio_format)
|
|
if not mime_type:
|
|
# Should not happen if format validation passed, but double-check
|
|
logger.error(
|
|
f"Could not find MIME type for supported format: {audio_format}"
|
|
)
|
|
raise ValueError(
|
|
f"Internal error: MIME type mapping missing for {audio_format}"
|
|
)
|
|
|
|
parts.append(
|
|
{
|
|
"inline_data": {
|
|
"mimeType": mime_type,
|
|
"data": validated_data, # Use the validated Base64 data
|
|
}
|
|
}
|
|
)
|
|
logger.debug(
|
|
f"Successfully added audio part (format: {audio_format})"
|
|
)
|
|
|
|
except ValueError as e:
|
|
logger.error(
|
|
f"Skipping audio part due to validation error: {e}"
|
|
)
|
|
parts.append({"text": f"[Error processing audio: {e}]"})
|
|
except Exception:
|
|
logger.exception("Unexpected error processing audio part.")
|
|
parts.append(
|
|
{"text": "[Unexpected error processing audio]"}
|
|
)
|
|
|
|
elif content_type == "input_video" and content_item.get(
|
|
"input_video"
|
|
):
|
|
video_info = content_item["input_video"]
|
|
video_data = video_info.get("data")
|
|
video_format = video_info.get("format", "").lower()
|
|
|
|
if not video_data or not video_format:
|
|
logger.warning(
|
|
"Skipping video part due to missing data or format."
|
|
)
|
|
continue
|
|
|
|
try:
|
|
validated_data = self._validate_media_data(
|
|
video_format,
|
|
video_data,
|
|
SUPPORTED_VIDEO_FORMATS,
|
|
MAX_VIDEO_SIZE_BYTES,
|
|
)
|
|
mime_type = VIDEO_FORMAT_TO_MIMETYPE.get(video_format)
|
|
if not mime_type:
|
|
raise ValueError(
|
|
f"Internal error: MIME type mapping missing for {video_format}"
|
|
)
|
|
|
|
parts.append(
|
|
{
|
|
"inline_data": {
|
|
"mimeType": mime_type,
|
|
"data": validated_data,
|
|
}
|
|
}
|
|
)
|
|
logger.debug(
|
|
f"Successfully added video part (format: {video_format})"
|
|
)
|
|
|
|
except ValueError as e:
|
|
logger.error(
|
|
f"Skipping video part due to validation error: {e}"
|
|
)
|
|
parts.append({"text": f"[Error processing video: {e}]"})
|
|
except Exception:
|
|
logger.exception("Unexpected error processing video part.")
|
|
parts.append(
|
|
{"text": "[Unexpected error processing video]"}
|
|
)
|
|
|
|
else:
|
|
# Log unrecognized but present types
|
|
if content_type:
|
|
logger.warning(
|
|
f"Unsupported content type or missing data in structured content: {content_type}"
|
|
)
|
|
|
|
elif (
|
|
"content" in msg and isinstance(msg["content"], str) and msg["content"]
|
|
):
|
|
parts.extend(_process_text_with_image(msg["content"]))
|
|
elif "tool_calls" in msg and isinstance(msg["tool_calls"], list):
|
|
# Keep existing tool call processing
|
|
for tool_call in msg["tool_calls"]:
|
|
function_call = tool_call.get("function", {})
|
|
# Sanitize arguments loading
|
|
arguments_str = function_call.get("arguments", "{}")
|
|
try:
|
|
function_call["args"] = json.loads(arguments_str)
|
|
except json.JSONDecodeError:
|
|
logger.warning(
|
|
f"Failed to decode tool call arguments: {arguments_str}"
|
|
)
|
|
function_call["args"] = {}
|
|
if "arguments" in function_call:
|
|
if "arguments" in function_call:
|
|
del function_call["arguments"]
|
|
|
|
parts.append({"functionCall": function_call})
|
|
|
|
if role not in SUPPORTED_ROLES:
|
|
if role == "tool":
|
|
role = "user"
|
|
else:
|
|
# 如果是最后一条消息，则认为是用户消息
|
|
if idx == len(messages) - 1:
|
|
role = "user"
|
|
else:
|
|
role = "model"
|
|
if parts:
|
|
if role == "system":
|
|
text_only_parts = [p for p in parts if "text" in p]
|
|
if len(text_only_parts) != len(parts):
|
|
logger.warning(
|
|
"Non-text parts found in system message; discarding them."
|
|
)
|
|
if text_only_parts:
|
|
system_instruction_parts.extend(text_only_parts)
|
|
|
|
else:
|
|
converted_messages.append({"role": role, "parts": parts})
|
|
|
|
system_instruction = (
|
|
None
|
|
if not system_instruction_parts
|
|
else {
|
|
"role": "system",
|
|
"parts": system_instruction_parts,
|
|
}
|
|
)
|
|
return converted_messages, system_instruction
|