Upload 10 files

core/account.py

@@ -13,6 +13,9 @@ from typing import Dict, List, Optional, TYPE_CHECKING
 
 from fastapi import HTTPException
 
+# 导入存储层（支持数据库）
+from core import storage
+
 if TYPE_CHECKING:
     from core.jwt import JWTManager
 
@@ -313,16 +316,41 @@ class MultiAccountManager:
 
 # ---------- 配置文件管理 ----------
 
-def save_accounts_to_file(accounts_data: list):
-    """保存账户配置到文件"""
+def _save_to_file(accounts_data: list):
+    """保存账户配置到本地文件"""
+    os.makedirs(os.path.dirname(ACCOUNTS_FILE) or ".", exist_ok=True)
     with open(ACCOUNTS_FILE, 'w', encoding='utf-8') as f:
         json.dump(accounts_data, f, ensure_ascii=False, indent=2)
     logger.info(f"[CONFIG] 配置已保存到 {ACCOUNTS_FILE}")
 
 
+def _load_from_file() -> list:
+    """从本地文件加载账户配置"""
+    if os.path.exists(ACCOUNTS_FILE):
+        try:
+            with open(ACCOUNTS_FILE, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        except Exception as e:
+            logger.warning(f"[CONFIG] 文件加载失败: {str(e)}")
+    return None
+
+
+def save_accounts_to_file(accounts_data: list):
+    """保存账户配置（优先数据库，降级到文件）"""
+    if storage.is_database_enabled():
+        try:
+            saved = storage.save_accounts_sync(accounts_data)
+            if saved:
+                return
+        except Exception as e:
+            logger.warning(f"[CONFIG] 数据库保存失败: {e}，降级到文件存储")
+
+    _save_to_file(accounts_data)
+
+
 def load_accounts_from_source() -> list:
-    """从环境变量或文件加载账户配置，优先使用环境变量"""
-    # 优先从环境变量加载
+    """从环境变量、数据库或文件加载账户配置"""
+    # 1. 优先从环境变量加载
     env_accounts = os.environ.get('ACCOUNTS_CONFIG')
     if env_accounts:
         try:
@@ -333,24 +361,33 @@ def load_accounts_from_source() -> list:
                 logger.warning(f"[CONFIG] 环境变量 ACCOUNTS_CONFIG 为空")
             return accounts_data
         except Exception as e:
-            logger.error(f"[CONFIG] 环境变量加载失败: {str(e)}，尝试从文件加载")
+            logger.error(f"[CONFIG] 环境变量加载失败: {str(e)}")
 
-    # 从文件加载
-    if os.path.exists(ACCOUNTS_FILE):
+    # 2. 尝试从数据库加载
+    if storage.is_database_enabled():
         try:
-            with open(ACCOUNTS_FILE, 'r', encoding='utf-8') as f:
-                accounts_data = json.load(f)
-            if accounts_data:
-                logger.info(f"[CONFIG] 从文件加载配置: {ACCOUNTS_FILE}，共 {len(accounts_data)} 个账户")
-            else:
-                logger.warning(f"[CONFIG] 账户配置为空，请在管理面板添加账户或编辑 {ACCOUNTS_FILE}")
-            return accounts_data
+            accounts_data = storage.load_accounts_sync()
+            if accounts_data is not None:
+                if accounts_data:
+                    logger.info(f"[CONFIG] 从数据库加载配置，共 {len(accounts_data)} 个账户")
+                else:
+                    logger.warning(f"[CONFIG] 数据库中账户配置为空")
+                return accounts_data
         except Exception as e:
-            logger.warning(f"[CONFIG] 文件加载失败: {str(e)}，创建空配置")
-
-    # 文件不存在，创建空配置
-    logger.warning(f"[CONFIG] 未找到 {ACCOUNTS_FILE}，已创建空配置文件")
-    logger.info(f"[CONFIG] 💡 请在管理面板添加账户，或直接编辑 {ACCOUNTS_FILE}，或使用批量上传功能，或设置环境变量 ACCOUNTS_CONFIG")
+            logger.warning(f"[CONFIG] 数据库加载失败: {e}，降级到文件存储")
+
+    # 3. 从文件加载
+    accounts_data = _load_from_file()
+    if accounts_data is not None:
+        if accounts_data:
+            logger.info(f"[CONFIG] 从文件加载配置: {ACCOUNTS_FILE}，共 {len(accounts_data)} 个账户")
+        else:
+            logger.warning(f"[CONFIG] 账户配置为空，请在管理面板添加账户或编辑 {ACCOUNTS_FILE}")
+        return accounts_data
+
+    # 4. 无配置，创建空配置
+    logger.warning(f"[CONFIG] 未找到配置，已创建空配置")
+    logger.info(f"[CONFIG] 💡 请在管理面板添加账户，或设置 DATABASE_URL 使用数据库存储")
     save_accounts_to_file([])
     return []
 
@@ -390,14 +427,14 @@ def load_multi_account_config(
             disabled=acc.get("disabled", False)  # 读取手动禁用状态，默认为 False
         )
 
-        # 检查账户是否已过期（已过期也加载到管理面板）
-        is_expired = config.is_expired()
-        if is_expired:
-            logger.warning(f"[CONFIG] 账户 {config.account_id} 已过期，仍加载用于展示")
-
-        manager.add_account(config, http_client, user_agent, account_failure_threshold, rate_limit_cooldown_seconds, global_stats)
-        if is_expired:
-            manager.accounts[config.account_id].is_available = False
+        # 检查账户是否已过期（已过期也加载到管理面板）
+        is_expired = config.is_expired()
+        if is_expired:
+            logger.warning(f"[CONFIG] 账户 {config.account_id} 已过期，仍加载用于展示")
+
+        manager.add_account(config, http_client, user_agent, account_failure_threshold, rate_limit_cooldown_seconds, global_stats)
+        if is_expired:
+            manager.accounts[config.account_id].is_available = False
 
     if not manager.accounts:
         logger.warning(f"[CONFIG] 没有有效的账户配置，服务将启动但无法处理请求，请在管理面板添加账户")

core/config.py

@@ -19,6 +19,8 @@ from typing import Optional, List
 from pydantic import BaseModel, Field, validator
 from dotenv import load_dotenv
 
+from core import storage
+
 # 加载 .env 文件
 load_dotenv()
 
@@ -152,6 +154,13 @@ class ConfigManager:
 
     def _load_yaml(self) -> dict:
         """加载 YAML 文件"""
+        if storage.is_database_enabled():
+            try:
+                data = storage.load_settings_sync()
+                if isinstance(data, dict):
+                    return data
+            except Exception as e:
+                print(f"[WARN] 加载数据库设置失败: {e}，使用本地配置")
         if self.yaml_path.exists():
             try:
                 with open(self.yaml_path, 'r', encoding='utf-8') as f:
@@ -166,6 +175,13 @@ class ConfigManager:
 
     def save_yaml(self, data: dict):
         """保存 YAML 配置"""
+        if storage.is_database_enabled():
+            try:
+                saved = storage.save_settings_sync(data)
+                if saved:
+                    return
+            except Exception as e:
+                print(f"[WARN] 保存数据库设置失败: {e}，降级到本地文件")
         self.yaml_path.parent.mkdir(exist_ok=True)
         with open(self.yaml_path, 'w', encoding='utf-8') as f:
             yaml.dump(data, f, allow_unicode=True, default_flow_style=False, sort_keys=False)

core/storage.py

@@ -0,0 +1,245 @@
+"""
+Storage abstraction supporting file and PostgreSQL backends.
+
+If DATABASE_URL is set, PostgreSQL is used.
+"""
+
+import asyncio
+import json
+import logging
+import os
+import threading
+from typing import Optional
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+logger = logging.getLogger(__name__)
+
+_db_pool = None
+_db_pool_lock = None
+_db_loop = None
+_db_thread = None
+_db_loop_lock = threading.Lock()
+
+
+def _get_database_url() -> str:
+    return os.environ.get("DATABASE_URL", "").strip()
+
+
+def is_database_enabled() -> bool:
+    """Return True when DATABASE_URL is configured."""
+    return bool(_get_database_url())
+
+
+def _ensure_db_loop() -> asyncio.AbstractEventLoop:
+    global _db_loop, _db_thread
+    if _db_loop and _db_thread and _db_thread.is_alive():
+        return _db_loop
+    with _db_loop_lock:
+        if _db_loop and _db_thread and _db_thread.is_alive():
+            return _db_loop
+        loop = asyncio.new_event_loop()
+
+        def _runner() -> None:
+            asyncio.set_event_loop(loop)
+            loop.run_forever()
+
+        thread = threading.Thread(target=_runner, name="storage-db-loop", daemon=True)
+        thread.start()
+        _db_loop = loop
+        _db_thread = thread
+        return _db_loop
+
+
+def _run_in_db_loop(coro):
+    loop = _ensure_db_loop()
+    future = asyncio.run_coroutine_threadsafe(coro, loop)
+    return future.result()
+
+
+async def _get_pool():
+    """Get (or create) the asyncpg connection pool."""
+    global _db_pool, _db_pool_lock
+    if _db_pool is not None:
+        return _db_pool
+    if _db_pool_lock is None:
+        _db_pool_lock = asyncio.Lock()
+    async with _db_pool_lock:
+        if _db_pool is not None:
+            return _db_pool
+        db_url = _get_database_url()
+        if not db_url:
+            raise ValueError("DATABASE_URL is not set")
+        try:
+            import asyncpg
+            _db_pool = await asyncpg.create_pool(
+                db_url,
+                min_size=1,
+                max_size=10,
+                command_timeout=30,
+            )
+            await _init_tables(_db_pool)
+            logger.info("[STORAGE] PostgreSQL pool initialized")
+        except ImportError:
+            logger.error("[STORAGE] asyncpg is required for database storage")
+            raise
+        except Exception as e:
+            logger.error(f"[STORAGE] Database connection failed: {e}")
+            raise
+    return _db_pool
+
+
+async def _init_tables(pool) -> None:
+    """Initialize database tables."""
+    async with pool.acquire() as conn:
+        await conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS kv_store (
+                key TEXT PRIMARY KEY,
+                value JSONB NOT NULL,
+                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+            """
+        )
+        logger.info("[STORAGE] Database tables initialized")
+
+
+async def db_get(key: str) -> Optional[dict]:
+    """Fetch a value from the database."""
+    pool = await _get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            "SELECT value FROM kv_store WHERE key = $1", key
+        )
+        if not row:
+            return None
+        value = row["value"]
+        if isinstance(value, str):
+            return json.loads(value)
+        return value
+
+
+async def db_set(key: str, value: dict) -> None:
+    """Persist a value to the database."""
+    pool = await _get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute(
+            """
+            INSERT INTO kv_store (key, value, updated_at)
+            VALUES ($1, $2, CURRENT_TIMESTAMP)
+            ON CONFLICT (key) DO UPDATE SET
+                value = EXCLUDED.value,
+                updated_at = CURRENT_TIMESTAMP
+            """,
+            key,
+            json.dumps(value, ensure_ascii=False),
+        )
+
+
+# ==================== Accounts storage ====================
+
+async def load_accounts() -> Optional[list]:
+    """
+    Load account configuration from database when enabled.
+    Return None to indicate file-based fallback.
+    """
+    if not is_database_enabled():
+        return None
+    try:
+        data = await db_get("accounts")
+        if data:
+            logger.info(f"[STORAGE] Loaded {len(data)} accounts from database")
+            return data
+        logger.info("[STORAGE] No accounts found in database")
+        return []
+    except Exception as e:
+        logger.error(f"[STORAGE] Database read failed: {e}")
+    return None
+
+
+async def save_accounts(accounts: list) -> bool:
+    """Save account configuration to database when enabled."""
+    if not is_database_enabled():
+        return False
+    try:
+        await db_set("accounts", accounts)
+        logger.info(f"[STORAGE] Saved {len(accounts)} accounts to database")
+        return True
+    except Exception as e:
+        logger.error(f"[STORAGE] Database write failed: {e}")
+    return False
+
+
+def load_accounts_sync() -> Optional[list]:
+    """Sync wrapper for load_accounts (safe in sync/async call sites)."""
+    return _run_in_db_loop(load_accounts())
+
+
+def save_accounts_sync(accounts: list) -> bool:
+    """Sync wrapper for save_accounts (safe in sync/async call sites)."""
+    return _run_in_db_loop(save_accounts(accounts))
+
+
+# ==================== Settings storage ====================
+
+async def load_settings() -> Optional[dict]:
+    if not is_database_enabled():
+        return None
+    try:
+        return await db_get("settings")
+    except Exception as e:
+        logger.error(f"[STORAGE] Settings read failed: {e}")
+    return None
+
+
+async def save_settings(settings: dict) -> bool:
+    if not is_database_enabled():
+        return False
+    try:
+        await db_set("settings", settings)
+        logger.info("[STORAGE] Settings saved to database")
+        return True
+    except Exception as e:
+        logger.error(f"[STORAGE] Settings write failed: {e}")
+    return False
+
+
+# ==================== Stats storage ====================
+
+async def load_stats() -> Optional[dict]:
+    if not is_database_enabled():
+        return None
+    try:
+        return await db_get("stats")
+    except Exception as e:
+        logger.error(f"[STORAGE] Stats read failed: {e}")
+    return None
+
+
+async def save_stats(stats: dict) -> bool:
+    if not is_database_enabled():
+        return False
+    try:
+        await db_set("stats", stats)
+        return True
+    except Exception as e:
+        logger.error(f"[STORAGE] Stats write failed: {e}")
+    return False
+
+
+def load_settings_sync() -> Optional[dict]:
+    return _run_in_db_loop(load_settings())
+
+
+def save_settings_sync(settings: dict) -> bool:
+    return _run_in_db_loop(save_settings(settings))
+
+
+def load_stats_sync() -> Optional[dict]:
+    return _run_in_db_loop(load_stats())
+
+
+def save_stats_sync(stats: dict) -> bool:
+    return _run_in_db_loop(save_stats(stats))