mirror of
https://github.com/snailyp/gemini-balance.git
synced 2026-05-14 02:49:42 +08:00
主要变更:
1. **数据库集成**:
* 引入 MySQL 数据库支持,使用 SQLAlchemy 和 `databases` 库持久化存储应用程序设置。
* 添加了 `app/database` 目录,包含数据库连接、模型和初始化逻辑。
* 更新 `requirements.txt` 添加数据库相关依赖 (`pymysql`, `sqlalchemy`, `aiomysql`, `databases`, `python-dotenv`)。
2. **配置管理重构**:
* 重构 `ConfigService` (`app/service/config/config_service.py`),使其从数据库加载和保存设置,并支持从 `.env` 文件同步初始配置到数据库。
* 修改 `Settings` 模型 (`app/config/config.py`) 以包含数据库连接信息,并添加了从数据库加载/同步配置的逻辑。
* 配置相关的路由 (`app/router/config_routes.py`) 更新为异步,并调用新的 `ConfigService` 方法。
* `KeyManager` (`app/service/key/key_manager.py`) 现在可以在配置更新后重置和重新初始化。
3. **错误日志查看器**:
* 新增 `/logs` 页面 (`app/templates/error_logs.html`) 用于展示应用程序错误日志。
* 添加了相应的路由 (`app/router/log_routes.py`)、静态资源 (`app/static/css/error_logs.css`, `app/static/js/error_logs.js`) 和日志记录器 (`app/log/logger.py`)。
* 在配置页面和密钥管理页面的导航栏中添加了指向日志页面的链接。
4. **异步操作**:
* 将配置服务和相关路由转换为异步 (`async def`) 以支持异步数据库操作。
5. **其他**:
* 更新了应用程序初始化逻辑 (`app/core/application.py`, `app/core/initialization.py`) 以包含数据库连接的建立和关闭。
149 lines
3.8 KiB
Python
149 lines
3.8 KiB
Python
"""
|
|
通用工具函数模块
|
|
"""
|
|
import json
|
|
import re
|
|
import base64
|
|
import requests
|
|
from typing import Dict, Any, List, Optional, Tuple
|
|
|
|
from app.core.constants import DATA_URL_PATTERN, IMAGE_URL_PATTERN, VALID_IMAGE_RATIOS
|
|
|
|
|
|
def extract_mime_type_and_data(base64_string: str) -> Tuple[Optional[str], str]:
|
|
"""
|
|
从 base64 字符串中提取 MIME 类型和数据
|
|
|
|
Args:
|
|
base64_string: 可能包含 MIME 类型信息的 base64 字符串
|
|
|
|
Returns:
|
|
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_to_base64(url: str) -> str:
|
|
"""
|
|
将图片URL转换为base64编码
|
|
|
|
Args:
|
|
url: 图片URL
|
|
|
|
Returns:
|
|
str: base64编码的图片数据
|
|
|
|
Raises:
|
|
Exception: 如果获取图片失败
|
|
"""
|
|
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 format_json_response(data: Dict[str, Any], indent: int = 2) -> str:
|
|
"""
|
|
格式化JSON响应
|
|
|
|
Args:
|
|
data: 要格式化的数据
|
|
indent: 缩进空格数
|
|
|
|
Returns:
|
|
str: 格式化后的JSON字符串
|
|
"""
|
|
return json.dumps(data, indent=indent, ensure_ascii=False)
|
|
|
|
|
|
def parse_prompt_parameters(prompt: str, default_ratio: str = "1:1") -> Tuple[str, int, str]:
|
|
"""
|
|
从prompt中解析参数
|
|
|
|
支持的格式:
|
|
- {n:数量} 例如: {n:2} 生成2张图片
|
|
- {ratio:比例} 例如: {ratio:16:9} 使用16:9比例
|
|
|
|
Args:
|
|
prompt: 提示文本
|
|
default_ratio: 默认比例
|
|
|
|
Returns:
|
|
tuple: (清理后的提示文本, 图片数量, 比例)
|
|
"""
|
|
# 默认值
|
|
n = 1
|
|
aspect_ratio = default_ratio
|
|
|
|
# 解析n参数
|
|
n_match = re.search(r'{n:(\d+)}', prompt)
|
|
if n_match:
|
|
n = int(n_match.group(1))
|
|
if n < 1 or n > 4:
|
|
raise ValueError(f"Invalid n value: {n}. Must be between 1 and 4.")
|
|
prompt = prompt.replace(n_match.group(0), '').strip()
|
|
|
|
# 解析ratio参数
|
|
ratio_match = re.search(r'{ratio:(\d+:\d+)}', prompt)
|
|
if ratio_match:
|
|
aspect_ratio = ratio_match.group(1)
|
|
if aspect_ratio not in VALID_IMAGE_RATIOS:
|
|
raise ValueError(
|
|
f"Invalid ratio: {aspect_ratio}. Must be one of: {', '.join(VALID_IMAGE_RATIOS)}"
|
|
)
|
|
prompt = prompt.replace(ratio_match.group(0), '').strip()
|
|
|
|
return prompt, n, aspect_ratio
|
|
|
|
|
|
def extract_image_urls_from_markdown(text: str) -> List[str]:
|
|
"""
|
|
从Markdown文本中提取图片URL
|
|
|
|
Args:
|
|
text: Markdown文本
|
|
|
|
Returns:
|
|
List[str]: 图片URL列表
|
|
"""
|
|
pattern = IMAGE_URL_PATTERN
|
|
matches = re.findall(pattern, text)
|
|
return [match[1] for match in matches]
|
|
|
|
|
|
def is_valid_api_key(key: str) -> bool:
|
|
"""
|
|
检查API密钥格式是否有效
|
|
|
|
Args:
|
|
key: API密钥
|
|
|
|
Returns:
|
|
bool: 如果密钥格式有效则返回True
|
|
"""
|
|
# 检查Gemini API密钥格式
|
|
if key.startswith('AIza'):
|
|
return len(key) >= 30
|
|
|
|
# 检查OpenAI API密钥格式
|
|
if key.startswith('sk-'):
|
|
return len(key) >= 30
|
|
|
|
return False
|
|
|
|
|