feat: support the redis
This commit is contained in:
1 parent
f612156e8e
commit
bb945854db
7 files changed
+352
-202
No files matched your search
@@ -698,7 +698,7 @@ async def get_storage_mode(_: bool = Depends(verify_admin_session)) -> Dict[str,
|
||||
"""
|
||||
获取当前存储模式
|
||||
|
||||
返回当前的存储模式(file/mysql)。
|
||||
返回当前的存储模式(file/mysql/redis)。
|
||||
"""
|
||||
try:
|
||||
logger.debug("[Admin] 获取存储模式")
|
||||
|
||||
+12
-2
@@ -32,21 +32,31 @@ class LoggerManager:
|
||||
if self.logger.handlers:
|
||||
return
|
||||
|
||||
# 创建格式器
|
||||
formatter = logging.Formatter(log_format)
|
||||
|
||||
# 控制台处理器
|
||||
console_handler = logging.StreamHandler(sys.stdout)
|
||||
console_handler.setLevel(log_level)
|
||||
console_handler.setFormatter(logging.Formatter(log_format))
|
||||
console_handler.setFormatter(formatter)
|
||||
|
||||
# 文件处理器
|
||||
file_handler = RotatingFileHandler(
|
||||
log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8"
|
||||
)
|
||||
file_handler.setLevel(log_level)
|
||||
file_handler.setFormatter(logging.Formatter(log_format))
|
||||
file_handler.setFormatter(formatter)
|
||||
|
||||
# 添加处理器到根日志器
|
||||
self.logger.addHandler(console_handler)
|
||||
self.logger.addHandler(file_handler)
|
||||
|
||||
# 配置第三方库日志级别,避免过多调试信息
|
||||
logging.getLogger("asyncio").setLevel(logging.WARNING)
|
||||
logging.getLogger("uvicorn").setLevel(logging.INFO)
|
||||
logging.getLogger("fastapi").setLevel(logging.INFO)
|
||||
logging.getLogger("aiomysql").setLevel(logging.WARNING)
|
||||
|
||||
LoggerManager._initialized = True
|
||||
|
||||
def debug(self, msg: str) -> None:
|
||||
|
||||
+320
-187
@@ -1,18 +1,20 @@
|
||||
"""存储抽象层 - 支持文件和MySQL两种存储模式"""
|
||||
"""存储抽象层 - 支持文件、MySQL和Redis存储"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import toml
|
||||
import asyncio
|
||||
import warnings
|
||||
import aiofiles
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, Optional, Literal
|
||||
from abc import ABC, abstractmethod
|
||||
from urllib.parse import urlparse, unquote
|
||||
|
||||
from app.core.logger import logger
|
||||
|
||||
|
||||
StorageMode = Literal["file", "mysql"]
|
||||
StorageMode = Literal["file", "mysql", "redis"]
|
||||
|
||||
|
||||
class BaseStorage(ABC):
|
||||
@@ -57,79 +59,86 @@ class FileStorage(BaseStorage):
|
||||
async def init_db(self) -> None:
|
||||
"""初始化文件存储"""
|
||||
self.data_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
# 初始化token文件
|
||||
if not self.token_file.exists():
|
||||
default_tokens = {"sso": {}, "ssoSuper": {}}
|
||||
async with aiofiles.open(self.token_file, "w", encoding="utf-8") as f:
|
||||
await f.write(json.dumps(default_tokens, indent=2, ensure_ascii=False))
|
||||
await self._write_file(self.token_file, json.dumps({"sso": {}, "ssoSuper": {}}, indent=2, ensure_ascii=False))
|
||||
logger.info("[Storage] 创建新的token文件")
|
||||
|
||||
# 初始化配置文件
|
||||
if not self.config_file.exists():
|
||||
default_config = {
|
||||
"global": {
|
||||
"api_keys": [],
|
||||
"admin_username": "admin",
|
||||
"admin_password": "admin"
|
||||
},
|
||||
"grok": {
|
||||
"proxy_url": "",
|
||||
"cf_clearance": "",
|
||||
"x_statsig_id": ""
|
||||
}
|
||||
"global": {"api_keys": [], "admin_username": "admin", "admin_password": "admin"},
|
||||
"grok": {"proxy_url": "", "cf_clearance": "", "x_statsig_id": ""}
|
||||
}
|
||||
async with aiofiles.open(self.config_file, "w", encoding="utf-8") as f:
|
||||
await f.write(toml.dumps(default_config))
|
||||
await self._write_file(self.config_file, toml.dumps(default_config))
|
||||
logger.info("[Storage] 创建新的配置文件")
|
||||
|
||||
async def load_tokens(self) -> Dict[str, Any]:
|
||||
"""从文件加载token数据"""
|
||||
async def _read_file(self, file_path: Path) -> str:
|
||||
"""读取文件内容"""
|
||||
async with aiofiles.open(file_path, "r", encoding="utf-8") as f:
|
||||
return await f.read()
|
||||
|
||||
async def _write_file(self, file_path: Path, content: str) -> None:
|
||||
"""写入文件内容"""
|
||||
async with aiofiles.open(file_path, "w", encoding="utf-8") as f:
|
||||
await f.write(content)
|
||||
|
||||
async def _load_json(self, file_path: Path, default: Dict[str, Any], lock: asyncio.Lock) -> Dict[str, Any]:
|
||||
"""加载JSON文件"""
|
||||
try:
|
||||
async with self._token_lock:
|
||||
if not self.token_file.exists():
|
||||
return {"sso": {}, "ssoSuper": {}}
|
||||
|
||||
async with aiofiles.open(self.token_file, "r", encoding="utf-8") as f:
|
||||
content = await f.read()
|
||||
return json.loads(content)
|
||||
async with lock:
|
||||
if not file_path.exists():
|
||||
return default
|
||||
return json.loads(await self._read_file(file_path))
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] 加载token失败: {e}")
|
||||
return {"sso": {}, "ssoSuper": {}}
|
||||
logger.error(f"[Storage] 加载{file_path.name}失败: {e}")
|
||||
return default
|
||||
|
||||
async def _save_json(self, file_path: Path, data: Dict[str, Any], lock: asyncio.Lock) -> None:
|
||||
"""保存JSON文件"""
|
||||
try:
|
||||
async with lock:
|
||||
await self._write_file(file_path, json.dumps(data, indent=2, ensure_ascii=False))
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] 保存{file_path.name}失败: {e}")
|
||||
raise
|
||||
|
||||
async def _load_toml(self, file_path: Path, default: Dict[str, Any], lock: asyncio.Lock) -> Dict[str, Any]:
|
||||
"""加载TOML文件"""
|
||||
try:
|
||||
async with lock:
|
||||
if not file_path.exists():
|
||||
return default
|
||||
return toml.loads(await self._read_file(file_path))
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] 加载{file_path.name}失败: {e}")
|
||||
return default
|
||||
|
||||
async def _save_toml(self, file_path: Path, data: Dict[str, Any], lock: asyncio.Lock) -> None:
|
||||
"""保存TOML文件"""
|
||||
try:
|
||||
async with lock:
|
||||
await self._write_file(file_path, toml.dumps(data))
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] 保存{file_path.name}失败: {e}")
|
||||
raise
|
||||
|
||||
async def load_tokens(self) -> Dict[str, Any]:
|
||||
"""加载token数据"""
|
||||
return await self._load_json(self.token_file, {"sso": {}, "ssoSuper": {}}, self._token_lock)
|
||||
|
||||
async def save_tokens(self, data: Dict[str, Any]) -> None:
|
||||
"""保存token数据到文件"""
|
||||
try:
|
||||
async with self._token_lock:
|
||||
async with aiofiles.open(self.token_file, "w", encoding="utf-8") as f:
|
||||
await f.write(json.dumps(data, indent=2, ensure_ascii=False))
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] 保存token失败: {e}")
|
||||
raise
|
||||
"""保存token数据"""
|
||||
await self._save_json(self.token_file, data, self._token_lock)
|
||||
|
||||
async def load_config(self) -> Dict[str, Any]:
|
||||
"""从文件加载配置数据"""
|
||||
try:
|
||||
async with self._config_lock:
|
||||
if not self.config_file.exists():
|
||||
return {"global": {}, "grok": {}}
|
||||
|
||||
async with aiofiles.open(self.config_file, "r", encoding="utf-8") as f:
|
||||
content = await f.read()
|
||||
return toml.loads(content)
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] 加载配置失败: {e}")
|
||||
return {"global": {}, "grok": {}}
|
||||
"""加载配置数据"""
|
||||
return await self._load_toml(self.config_file, {"global": {}, "grok": {}}, self._config_lock)
|
||||
|
||||
async def save_config(self, data: Dict[str, Any]) -> None:
|
||||
"""保存配置数据到文件"""
|
||||
try:
|
||||
async with self._config_lock:
|
||||
async with aiofiles.open(self.config_file, "w", encoding="utf-8") as f:
|
||||
await f.write(toml.dumps(data))
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] 保存配置失败: {e}")
|
||||
raise
|
||||
"""保存配置数据"""
|
||||
await self._save_toml(self.config_file, data, self._config_lock)
|
||||
|
||||
|
||||
class MysqlStorage(BaseStorage):
|
||||
@@ -138,172 +147,297 @@ class MysqlStorage(BaseStorage):
|
||||
def __init__(self, database_url: str, data_dir: Path):
|
||||
self.database_url = database_url
|
||||
self.data_dir = data_dir
|
||||
self.token_file = data_dir / "token.json"
|
||||
self.config_file = data_dir / "setting.toml"
|
||||
self._pool = None
|
||||
self._file_storage = FileStorage(data_dir)
|
||||
self._file = FileStorage(data_dir)
|
||||
|
||||
async def init_db(self) -> None:
|
||||
"""初始化MySQL数据库和连接池"""
|
||||
"""初始化MySQL"""
|
||||
try:
|
||||
import aiomysql
|
||||
from urllib.parse import urlparse, unquote
|
||||
parsed = self._parse_url(self.database_url)
|
||||
logger.info(f"[Storage] 解析数据库连接: {parsed['user']}@{parsed['host']}:{parsed['port']}/{parsed['db']}")
|
||||
|
||||
# 使用标准库解析URL
|
||||
parsed = urlparse(self.database_url)
|
||||
|
||||
# 解码用户名和密码(处理URL编码)
|
||||
user = unquote(parsed.username) if parsed.username else ""
|
||||
password = unquote(parsed.password) if parsed.password else ""
|
||||
host = parsed.hostname
|
||||
port = parsed.port if parsed.port else 3306
|
||||
database = parsed.path[1:] if parsed.path else "grok2api" # 去掉开头的 '/'
|
||||
|
||||
logger.info(f"[Storage] 解析数据库连接: {user}@{host}:{port}/{database}")
|
||||
# 创建数据库
|
||||
await self._create_db(parsed)
|
||||
|
||||
# 创建连接池
|
||||
self._pool = await aiomysql.create_pool(
|
||||
host=host,
|
||||
port=port,
|
||||
user=user,
|
||||
password=password,
|
||||
db=database,
|
||||
charset="utf8mb4",
|
||||
autocommit=True,
|
||||
maxsize=10
|
||||
host=parsed['host'], port=parsed['port'], user=parsed['user'],
|
||||
password=parsed['password'], db=parsed['db'], charset="utf8mb4",
|
||||
autocommit=True, maxsize=10
|
||||
)
|
||||
|
||||
logger.info(f"[Storage] MySQL连接池创建成功: {host}:{port}/{database}")
|
||||
|
||||
# 创建表
|
||||
async with self._pool.acquire() as conn:
|
||||
async with conn.cursor() as cursor:
|
||||
# 创建tokens表
|
||||
await cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS grok_tokens (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
data JSON NOT NULL,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||||
""")
|
||||
|
||||
# 创建settings表
|
||||
await cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS grok_settings (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
data JSON NOT NULL,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||||
""")
|
||||
|
||||
logger.info("[Storage] MySQL表创建/验证成功")
|
||||
|
||||
# 确保文件存储目录存在
|
||||
await self._file_storage.init_db()
|
||||
|
||||
# 从数据库同步数据到文件
|
||||
await self._sync_from_db()
|
||||
|
||||
await self._create_tables()
|
||||
|
||||
# 初始化文件存储和同步数据
|
||||
await self._file.init_db()
|
||||
await self._sync_data()
|
||||
|
||||
except ImportError:
|
||||
logger.error("[Storage] aiomysql未安装,请运行: pip install aiomysql")
|
||||
raise Exception("aiomysql未安装")
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] MySQL初始化失败: {e}")
|
||||
raise
|
||||
|
||||
async def _sync_from_db(self) -> None:
|
||||
"""从数据库同步数据到文件"""
|
||||
try:
|
||||
# 同步tokens
|
||||
tokens = await self._load_from_db("grok_tokens")
|
||||
if tokens:
|
||||
await self._file_storage.save_tokens(tokens)
|
||||
logger.info("[Storage] Token数据已从数据库同步到文件")
|
||||
|
||||
# 同步settings
|
||||
settings = await self._load_from_db("grok_settings")
|
||||
if settings:
|
||||
await self._file_storage.save_config(settings)
|
||||
logger.info("[Storage] 配置数据已从数据库同步到文件")
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"[Storage] 从数据库同步数据失败: {e}")
|
||||
def _parse_url(self, url: str) -> Dict[str, Any]:
|
||||
"""解析数据库URL"""
|
||||
parsed = urlparse(url)
|
||||
return {
|
||||
'user': unquote(parsed.username) if parsed.username else "",
|
||||
'password': unquote(parsed.password) if parsed.password else "",
|
||||
'host': parsed.hostname,
|
||||
'port': parsed.port or 3306,
|
||||
'db': parsed.path[1:] if parsed.path else "grok2api"
|
||||
}
|
||||
|
||||
async def _load_from_db(self, table: str) -> Optional[Dict[str, Any]]:
|
||||
"""从数据库表加载最新数据"""
|
||||
async def _create_db(self, parsed: Dict[str, Any]) -> None:
|
||||
"""创建数据库"""
|
||||
import aiomysql
|
||||
temp_pool = await aiomysql.create_pool(
|
||||
host=parsed['host'], port=parsed['port'], user=parsed['user'],
|
||||
password=parsed['password'], charset="utf8mb4", autocommit=True, maxsize=1
|
||||
)
|
||||
|
||||
try:
|
||||
async with temp_pool.acquire() as conn:
|
||||
async with conn.cursor() as cursor:
|
||||
with warnings.catch_warnings():
|
||||
warnings.filterwarnings('ignore', message='.*database exists')
|
||||
await cursor.execute(
|
||||
f"CREATE DATABASE IF NOT EXISTS `{parsed['db']}` "
|
||||
f"CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
|
||||
)
|
||||
logger.info(f"[Storage] MySQL数据库 '{parsed['db']}' 已就绪")
|
||||
finally:
|
||||
temp_pool.close()
|
||||
await temp_pool.wait_closed()
|
||||
|
||||
async def _create_tables(self) -> None:
|
||||
"""创建表"""
|
||||
tables = {
|
||||
"grok_tokens": """
|
||||
CREATE TABLE IF NOT EXISTS grok_tokens (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
data JSON NOT NULL,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||||
""",
|
||||
"grok_settings": """
|
||||
CREATE TABLE IF NOT EXISTS grok_settings (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
data JSON NOT NULL,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||||
"""
|
||||
}
|
||||
|
||||
async with self._pool.acquire() as conn:
|
||||
async with conn.cursor() as cursor:
|
||||
with warnings.catch_warnings():
|
||||
warnings.filterwarnings('ignore', message='.*already exists')
|
||||
for sql in tables.values():
|
||||
await cursor.execute(sql)
|
||||
logger.info("[Storage] MySQL表创建/验证成功")
|
||||
|
||||
async def _sync_data(self) -> None:
|
||||
"""同步数据"""
|
||||
try:
|
||||
for table, key in [("grok_tokens", "sso"), ("grok_settings", "global")]:
|
||||
data = await self._load_db(table)
|
||||
if data:
|
||||
if table == "grok_tokens":
|
||||
await self._file.save_tokens(data)
|
||||
else:
|
||||
await self._file.save_config(data)
|
||||
logger.info(f"[Storage] {table.split('_')[1]}数据已从数据库同步到文件")
|
||||
else:
|
||||
if table == "grok_tokens":
|
||||
file_data = await self._file.load_tokens()
|
||||
if file_data.get(key) or file_data.get("ssoSuper"):
|
||||
await self._save_db(table, file_data)
|
||||
logger.info("[Storage] Token数据已从文件初始化到数据库")
|
||||
else:
|
||||
file_data = await self._file.load_config()
|
||||
if file_data.get(key) or file_data.get("grok"):
|
||||
await self._save_db(table, file_data)
|
||||
logger.info("[Storage] 配置数据已从文件初始化到数据库")
|
||||
except Exception as e:
|
||||
logger.warning(f"[Storage] 数据同步失败: {e}")
|
||||
|
||||
async def _load_db(self, table: str) -> Optional[Dict[str, Any]]:
|
||||
"""从数据库加载数据"""
|
||||
try:
|
||||
async with self._pool.acquire() as conn:
|
||||
async with conn.cursor() as cursor:
|
||||
await cursor.execute(f"SELECT data FROM {table} ORDER BY id DESC LIMIT 1")
|
||||
result = await cursor.fetchone()
|
||||
if result:
|
||||
return json.loads(result[0])
|
||||
return None
|
||||
return json.loads(result[0]) if result else None
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] 从数据库加载{table}失败: {e}")
|
||||
return None
|
||||
|
||||
async def _save_to_db(self, table: str, data: Dict[str, Any]) -> None:
|
||||
"""保存数据到数据库表"""
|
||||
async def _save_db(self, table: str, data: Dict[str, Any]) -> None:
|
||||
"""保存数据到数据库"""
|
||||
try:
|
||||
async with self._pool.acquire() as conn:
|
||||
async with conn.cursor() as cursor:
|
||||
json_data = json.dumps(data, ensure_ascii=False)
|
||||
|
||||
# 检查是否已有数据
|
||||
await cursor.execute(f"SELECT id FROM {table} ORDER BY id DESC LIMIT 1")
|
||||
result = await cursor.fetchone()
|
||||
|
||||
|
||||
if result:
|
||||
# 更新现有数据
|
||||
await cursor.execute(
|
||||
f"UPDATE {table} SET data = %s WHERE id = %s",
|
||||
(json_data, result[0])
|
||||
)
|
||||
await cursor.execute(f"UPDATE {table} SET data = %s WHERE id = %s", (json_data, result[0]))
|
||||
else:
|
||||
# 插入新数据
|
||||
await cursor.execute(
|
||||
f"INSERT INTO {table} (data) VALUES (%s)",
|
||||
(json_data,)
|
||||
)
|
||||
await cursor.execute(f"INSERT INTO {table} (data) VALUES (%s)", (json_data,))
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] 保存数据到{table}失败: {e}")
|
||||
raise
|
||||
|
||||
async def load_tokens(self) -> Dict[str, Any]:
|
||||
"""从文件加载token数据(已从数据库同步)"""
|
||||
return await self._file_storage.load_tokens()
|
||||
"""加载token数据"""
|
||||
return await self._file.load_tokens()
|
||||
|
||||
async def save_tokens(self, data: Dict[str, Any]) -> None:
|
||||
"""保存token数据到文件和数据库"""
|
||||
# 先保存到文件
|
||||
await self._file_storage.save_tokens(data)
|
||||
# 再保存到数据库
|
||||
await self._save_to_db("grok_tokens", data)
|
||||
"""保存token数据"""
|
||||
await self._file.save_tokens(data)
|
||||
await self._save_db("grok_tokens", data)
|
||||
|
||||
async def load_config(self) -> Dict[str, Any]:
|
||||
"""从文件加载配置数据(已从数据库同步)"""
|
||||
return await self._file_storage.load_config()
|
||||
"""加载配置数据"""
|
||||
return await self._file.load_config()
|
||||
|
||||
async def save_config(self, data: Dict[str, Any]) -> None:
|
||||
"""保存配置数据到文件和数据库"""
|
||||
# 先保存到文件
|
||||
await self._file_storage.save_config(data)
|
||||
# 再保存到数据库
|
||||
await self._save_to_db("grok_settings", data)
|
||||
"""保存配置数据"""
|
||||
await self._file.save_config(data)
|
||||
await self._save_db("grok_settings", data)
|
||||
|
||||
async def close(self) -> None:
|
||||
"""关闭数据库连接池"""
|
||||
"""关闭连接"""
|
||||
if self._pool:
|
||||
self._pool.close()
|
||||
await self._pool.wait_closed()
|
||||
logger.info("[Storage] MySQL连接池已关闭")
|
||||
|
||||
|
||||
class RedisStorage(BaseStorage):
|
||||
"""Redis存储实现"""
|
||||
|
||||
def __init__(self, redis_url: str, data_dir: Path):
|
||||
self.redis_url = redis_url
|
||||
self.data_dir = data_dir
|
||||
self._redis = None
|
||||
self._file = FileStorage(data_dir)
|
||||
|
||||
async def init_db(self) -> None:
|
||||
"""初始化Redis"""
|
||||
try:
|
||||
import redis.asyncio as redis
|
||||
parsed = self._parse_url(self.redis_url)
|
||||
logger.info(f"[Storage] 解析Redis URL: host={parsed['host']}, port={parsed['port']}, db={parsed.get('db', 0)}, username={parsed.get('username')}, password={'***' if parsed.get('password') else None}")
|
||||
|
||||
# 创建Redis连接
|
||||
self._redis = redis.Redis(
|
||||
host=parsed['host'], port=parsed['port'], password=parsed.get('password'),
|
||||
username=parsed.get('username'), db=parsed.get('db', 0),
|
||||
encoding="utf-8", decode_responses=True
|
||||
)
|
||||
|
||||
# 测试连接
|
||||
await self._redis.ping()
|
||||
logger.info(f"[Storage] Redis连接成功: {parsed['host']}:{parsed['port']}/{parsed['db']}")
|
||||
|
||||
# 初始化文件存储和同步数据
|
||||
await self._file.init_db()
|
||||
await self._sync_data()
|
||||
|
||||
except ImportError:
|
||||
raise Exception("redis未安装")
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] Redis初始化失败: {e}")
|
||||
raise
|
||||
|
||||
def _parse_url(self, url: str) -> Dict[str, Any]:
|
||||
"""解析Redis URL"""
|
||||
if url.startswith('redis://'):
|
||||
url = url[8:]
|
||||
parsed = urlparse(f'//{url}')
|
||||
|
||||
result = {
|
||||
'host': parsed.hostname or 'localhost',
|
||||
'port': parsed.port or 6379,
|
||||
'db': int(parsed.path.lstrip('/')) if parsed.path and parsed.path != '/' else 0,
|
||||
'username': unquote(parsed.username) if parsed.username else None,
|
||||
'password': unquote(parsed.password) if parsed.password else None
|
||||
}
|
||||
|
||||
# Redis 6+ 默认用户名
|
||||
if result['password'] and not result['username']:
|
||||
result['username'] = 'default'
|
||||
|
||||
return result
|
||||
|
||||
async def _sync_data(self) -> None:
|
||||
"""同步数据"""
|
||||
try:
|
||||
for key, file_func, key_name in [
|
||||
("grok:tokens", self._file.load_tokens, "sso"),
|
||||
("grok:settings", self._file.load_config, "global")
|
||||
]:
|
||||
data = await self._redis.get(key)
|
||||
if data:
|
||||
parsed = json.loads(data)
|
||||
if key == "grok:tokens":
|
||||
await self._file.save_tokens(parsed)
|
||||
else:
|
||||
await self._file.save_config(parsed)
|
||||
logger.info(f"[Storage] {key.split(':')[1]}数据已从Redis同步到文件")
|
||||
else:
|
||||
file_data = await file_func()
|
||||
if file_data.get(key_name) or (key == "grok:tokens" and file_data.get("ssoSuper")):
|
||||
json_data = json.dumps(file_data, ensure_ascii=False)
|
||||
await self._redis.set(key, json_data)
|
||||
logger.info(f"[Storage] {key.split(':')[1]}数据已从文件初始化到Redis")
|
||||
except Exception as e:
|
||||
logger.warning(f"[Storage] 数据同步失败: {e}")
|
||||
|
||||
async def _save_redis(self, key: str, data: Dict[str, Any]) -> None:
|
||||
"""保存到Redis"""
|
||||
try:
|
||||
await self._redis.set(key, json.dumps(data, ensure_ascii=False))
|
||||
except Exception as e:
|
||||
logger.error(f"[Storage] 保存到Redis失败: {e}")
|
||||
raise
|
||||
|
||||
async def load_tokens(self) -> Dict[str, Any]:
|
||||
"""加载token数据"""
|
||||
return await self._file.load_tokens()
|
||||
|
||||
async def save_tokens(self, data: Dict[str, Any]) -> None:
|
||||
"""保存token数据"""
|
||||
await self._file.save_tokens(data)
|
||||
await self._save_redis("grok:tokens", data)
|
||||
|
||||
async def load_config(self) -> Dict[str, Any]:
|
||||
"""加载配置数据"""
|
||||
return await self._file.load_config()
|
||||
|
||||
async def save_config(self, data: Dict[str, Any]) -> None:
|
||||
"""保存配置数据"""
|
||||
await self._file.save_config(data)
|
||||
await self._save_redis("grok:settings", data)
|
||||
|
||||
async def close(self) -> None:
|
||||
"""关闭连接"""
|
||||
if self._redis:
|
||||
await self._redis.close()
|
||||
logger.info("[Storage] Redis连接已关闭")
|
||||
|
||||
|
||||
class StorageManager:
|
||||
"""存储管理器 - 单例模式"""
|
||||
"""存储管理器"""
|
||||
|
||||
_instance: Optional['StorageManager'] = None
|
||||
_storage: Optional[BaseStorage] = None
|
||||
@@ -315,43 +449,42 @@ class StorageManager:
|
||||
return cls._instance
|
||||
|
||||
async def init(self) -> None:
|
||||
"""初始化存储管理器"""
|
||||
"""初始化存储"""
|
||||
if self._initialized:
|
||||
return
|
||||
|
||||
# 读取环境变量
|
||||
storage_mode = os.getenv("STORAGE_MODE", "file").lower()
|
||||
database_url = os.getenv("DATABASE_URL", "")
|
||||
|
||||
mode = os.getenv("STORAGE_MODE", "file").lower()
|
||||
url = os.getenv("DATABASE_URL", "")
|
||||
data_dir = Path(__file__).parents[2] / "data"
|
||||
|
||||
if storage_mode == "mysql":
|
||||
if not database_url:
|
||||
logger.error("[Storage] MySQL模式需要DATABASE_URL环境变量")
|
||||
raise ValueError("MySQL模式需要DATABASE_URL环境变量")
|
||||
|
||||
logger.info(f"[Storage] 使用MySQL存储模式")
|
||||
self._storage = MysqlStorage(database_url, data_dir)
|
||||
else:
|
||||
logger.info(f"[Storage] 使用文件存储模式")
|
||||
self._storage = FileStorage(data_dir)
|
||||
|
||||
|
||||
storage_classes = {
|
||||
"mysql": MysqlStorage,
|
||||
"redis": RedisStorage,
|
||||
"file": FileStorage
|
||||
}
|
||||
|
||||
if mode in ("mysql", "redis") and not url:
|
||||
raise ValueError(f"{mode.upper()}模式需要DATABASE_URL环境变量")
|
||||
|
||||
storage_class = storage_classes.get(mode, FileStorage)
|
||||
self._storage = storage_class(url, data_dir) if mode != "file" else storage_class(data_dir)
|
||||
|
||||
await self._storage.init_db()
|
||||
self._initialized = True
|
||||
logger.info(f"[Storage] 使用{mode}存储模式")
|
||||
logger.info("[Storage] 存储管理器初始化完成")
|
||||
|
||||
def get_storage(self) -> BaseStorage:
|
||||
"""获取存储实例"""
|
||||
if not self._initialized or not self._storage:
|
||||
raise RuntimeError("StorageManager未初始化,请先调用init()")
|
||||
raise RuntimeError("StorageManager未初始化")
|
||||
return self._storage
|
||||
|
||||
async def close(self) -> None:
|
||||
"""关闭存储"""
|
||||
if self._storage and isinstance(self._storage, MysqlStorage):
|
||||
if self._storage and hasattr(self._storage, 'close'):
|
||||
await self._storage.close()
|
||||
|
||||
|
||||
# 全局存储管理器实例
|
||||
storage_manager = StorageManager()
|
||||
|
||||
@@ -523,7 +523,7 @@
|
||||
saveGlobalSettings=async()=>{const gc={admin_username:$('cfgAdminUser').value,log_level:$('cfgLogLevel').value,image_cache_max_size_mb:parseInt($('cfgImageCacheMaxSize').value)||500,video_cache_max_size_mb:parseInt($('cfgVideoCacheMaxSize').value)||1000,image_mode:$('cfgImageMode').value,base_url:$('cfgBaseUrl').value};if($('cfgAdminPass').value)gc.admin_password=$('cfgAdminPass').value;try{const r=await apiRequest('/api/settings');if(!r)return;const d=await r.json();if(!d.success)return showToast('加载配置失败','error');const s=await apiRequest('/api/settings',{method:'POST',body:JSON.stringify({global_config:gc,grok_config:d.data.grok})});if(!s)return;const sd=await s.json();sd.success?(showToast('全局配置保存成功','success'),$('cfgAdminPass').value=''):showToast('保存失败: '+(sd.error||'未知错误'),'error')}catch(e){showToast('保存失败: '+e.message,'error')}},
|
||||
saveGrokSettings=async()=>{const kc={api_key:$('cfgApiKey').value,proxy_url:$('cfgProxyUrl').value,cf_clearance:$('cfgCfClearance').value,x_statsig_id:$('cfgStatsigId').value,filtered_tags:$('cfgFilteredTags').value,temporary:$('cfgTemporary').value==='true',stream_chunk_timeout:parseInt($('cfgStreamChunkTimeout').value)||120,stream_first_response_timeout:parseInt($('cfgStreamFirstResponseTimeout').value)||30,stream_total_timeout:parseInt($('cfgStreamTotalTimeout').value)||600};try{const r=await apiRequest('/api/settings');if(!r)return;const d=await r.json();if(!d.success)return showToast('加载配置失败','error');const s=await apiRequest('/api/settings',{method:'POST',body:JSON.stringify({global_config:d.data.global,grok_config:kc})});if(!s)return;const sd=await s.json();sd.success?showToast('Grok配置保存成功','success'):showToast('保存失败: '+(sd.error||'未知错误'),'error')}catch(e){showToast('保存失败: '+e.message,'error')}};
|
||||
const updateHoverCardPosition=(cardElement)=>{const trigger=cardElement.querySelector('.hover-card-trigger'),content=cardElement.querySelector('.hover-card-content');if(!trigger||!content)return;console.log('更新悬浮卡片位置',cardElement,trigger,content);const rect=trigger.getBoundingClientRect(),windowHeight=window.innerHeight,spaceAbove=rect.top,spaceBelow=windowHeight-rect.bottom;content.classList.remove('top','bottom');const originalVisibility=getComputedStyle(content).visibility,originalOpacity=getComputedStyle(content).opacity;content.style.visibility='hidden';content.style.opacity='1';const contentHeight=content.offsetHeight;content.style.visibility=originalVisibility;content.style.opacity=originalOpacity;const position=spaceAbove>contentHeight+10?'top':spaceBelow>contentHeight+10?'bottom':'top';content.classList.add(position);console.log('悬浮卡片位置:',position,'高度:',contentHeight,'上方空间:',spaceAbove,'下方空间:',spaceBelow)},
|
||||
loadStorageMode=async()=>{try{const r=await apiRequest('/api/storage/mode');if(!r)return;const d=await r.json();if(d.success){const mode=d.data.mode;console.log('存储模式:',mode);$('storageModeText').textContent=mode;if(mode==='MYSQL'){$('storageMode').classList.add('bg-blue-50','text-blue-700','border-blue-200');$('storageModeTooltip').textContent='数据库连接模式 - 数据持久化存储,修改配置时可能稍慢但更安全'}else{$('storageMode').classList.add('bg-green-50','text-green-700','border-green-200');$('storageModeTooltip').textContent='文件存储模式 - 本地文件存储,读写速度快'};updateHoverCardPosition($('storageMode').closest('.hover-card'))}}catch(e){console.error('加载存储模式失败:',e);$('storageModeText').textContent='FILE';$('storageMode').classList.add('bg-green-50','text-green-700','border-green-200');$('storageModeTooltip').textContent='文件存储模式 - 本地文件存储,读写速度快';updateHoverCardPosition($('storageMode').closest('.hover-card'))}};
|
||||
loadStorageMode=async()=>{try{const r=await apiRequest('/api/storage/mode');if(!r)return;const d=await r.json();if(d.success){const mode=d.data.mode;console.log('存储模式:',mode);$('storageModeText').textContent=mode;if(mode==='MYSQL'){$('storageMode').classList.add('bg-blue-50','text-blue-700','border-blue-200');$('storageModeTooltip').textContent='数据库连接模式 - 数据持久化存储,修改配置时可能稍慢但更安全'}else if(mode==='REDIS'){$('storageMode').classList.add('bg-purple-50','text-purple-700','border-purple-200');$('storageModeTooltip').textContent='Redis缓存模式 - 高速内存存储,数据持久化且读写性能极佳'}else{$('storageMode').classList.add('bg-green-50','text-green-700','border-green-200');$('storageModeTooltip').textContent='文件存储模式 - 本地文件存储,读写速度快'};updateHoverCardPosition($('storageMode').closest('.hover-card'))}}catch(e){console.error('加载存储模式失败:',e);$('storageModeText').textContent='FILE';$('storageMode').classList.add('bg-green-50','text-green-700','border-green-200');$('storageModeTooltip').textContent='文件存储模式 - 本地文件存储,读写速度快';updateHoverCardPosition($('storageMode').closest('.hover-card'))}};
|
||||
window.addEventListener('DOMContentLoaded',()=>{checkAuth();loadStorageMode();refreshTokens();setInterval(()=>{loadStats();updateRemaining()},30000);window.addEventListener('resize',()=>{const hoverCard=$('storageMode').closest('.hover-card');hoverCard&&updateHoverCardPosition(hoverCard)});const hoverCard=$('storageMode').closest('.hover-card'),trigger=hoverCard?.querySelector('.hover-card-trigger'),content=hoverCard?.querySelector('.hover-card-content');if(trigger&&content){trigger.addEventListener('mouseenter',()=>{content.style.opacity='1';content.style.visibility='visible'});trigger.addEventListener('mouseleave',()=>{content.style.opacity='0';content.style.visibility='hidden'})}});
|
||||
</script>
|
||||
</body>
|
||||
|
||||
+2
-2
@@ -15,5 +15,5 @@ log_level = "DEBUG"
|
||||
image_mode = "url"
|
||||
admin_password = "admin"
|
||||
admin_username = "admin"
|
||||
image_cache_max_size_mb = 500
|
||||
video_cache_max_size_mb = 1000
|
||||
image_cache_max_size_mb = 512
|
||||
video_cache_max_size_mb = 1024
|
||||
+11
-5
@@ -7,10 +7,16 @@ services:
|
||||
- grok_data:/app/data
|
||||
- ./logs:/app/logs
|
||||
environment:
|
||||
# 数据存储模式( file/mysql 二选一 )
|
||||
- STORAGE_MODE=file
|
||||
# MySQL数据库 URL (仅在STORAGE_MODE=mysql时需要 示例:mysql://user:password@host:port/database)
|
||||
# - DATABASE_URL=mysql://
|
||||
# 存储模式: file, mysql 或 redis
|
||||
- STORAGE_MODE=file
|
||||
# 数据库连接 URL (仅在STORAGE_MODE=mysql或redis时需要)
|
||||
# MySQL格式: mysql://user:password@host:port/database
|
||||
# Redis格式: redis://host:port/db 或 redis://user:password@host:port/db
|
||||
# - DATABASE_URL=mysql://user:password@host:3306/grok2api
|
||||
# - DATABASE_URL=redis://redis:6379/0
|
||||
|
||||
|
||||
volumes:
|
||||
grok_data:
|
||||
grok_data:
|
||||
# mysql_data: # 取消注释以使用MySQL
|
||||
# redis_data: # 取消注释以使用Redis
|
||||
+5
-4
@@ -1,10 +1,11 @@
|
||||
toml==0.10.2
|
||||
fastapi==0.118.2
|
||||
fastapi==0.119.0
|
||||
uvicorn==0.37.0
|
||||
python-dotenv==1.1.1
|
||||
curl_cffi==0.13.0
|
||||
requests==2.32.5
|
||||
starlette==0.48.0
|
||||
pydantic==2.12.0
|
||||
aiofiles==24.1.0
|
||||
aiomysql==0.2.0
|
||||
pydantic==2.12.2
|
||||
aiofiles==25.1.0
|
||||
aiomysql==0.2.0
|
||||
redis==6.4.0
|
||||
Reference in new issue
Block a user