支持第三方扣费

app.py

@@ -54,6 +54,8 @@ from router_wallet import router as wallet_router     # 💰 钱包/提现
 from router_proxy import router as proxy_router       # 🔗 代理下载
 from router_posts import router as posts_router       # 💬 讨论区
 from router_tasks import router as tasks_router       # 📝 任务榜
+from router_open_api import router as open_api_router  # 🌐 通用开放API（第三方插件）
+from router_admin_plugins import router as admin_plugins_router  # 🔒 管理后台 - 第三方插件管理API
 
 from database_sql import init_sql_db, get_db, SessionLocal
 from models_sql import Ownership, SystemConfig
@@ -212,6 +214,14 @@ async def on_startup():
     except Exception as e:
         logger.error(f"❌ SQL 数据库初始化失败: {e}")
         # 不抛出异常，允许降级运行
+
+    # ========== 加载第三方插件注册表（依赖 SQL 数据库） ==========
+    try:
+        from router_open_api import load_plugins_from_db
+        load_plugins_from_db()
+        logger.info("✅ 第三方插件注册表已加载")
+    except Exception as e:
+        logger.warning(f"⚠️ 第三方插件注册表加载失败: {e}")
     
     # ========== 预热检查：JSON 数据库 ==========
     try:
@@ -275,6 +285,8 @@ app.include_router(wallet_router)         # 💰 钱包/提现
 app.include_router(proxy_router)          # 🔗 代理下载
 app.include_router(posts_router)          # 💬 讨论区
 app.include_router(tasks_router)          # 📝 任务榜
+app.include_router(open_api_router)       # 🌐 通用开放API（供第三方插件调用）
+app.include_router(admin_plugins_router)  # 🔒 管理后台 - 第三方插件管理API
 
 # ==========================================
 # 🟢 私有图床代理中心 (Image Proxy)
@@ -634,4 +646,13 @@ async def set_system_config(config_key: str, value: dict, current_user: str = De
     if not _is_admin(current_user):
         raise HTTPException(status_code=403, detail="无权修改系统配置，仅管理员可操作")
     _save_system_config_to_db(db, config_key, value, updated_by=current_user)
-    return {"status": "success", "message": "配置已更新"}
+    return {"status": "success", "message": "配置已更新"}
+
+
+# ==========================================
+# 🖥️ 管理界面静态文件挂载
+# ⚠️ 必须放在所有 API 路由注册之后，避免拦截 API
+# 访问 /admin/ 会自动返回 管理界面/index.html
+# ==========================================
+from fastapi.staticfiles import StaticFiles
+app.mount("/admin", StaticFiles(directory="管理界面", html=True), name="admin")

database_sql.py

@@ -1,260 +1,266 @@
-# database_sql.py
-# ==========================================
-# 🛡️ 稳定性优化：SQL 数据库连接模块
-# ==========================================
-import os
-import time
-import logging
-from functools import wraps
-from sqlalchemy import create_engine, text
-from sqlalchemy.orm import sessionmaker
-from sqlalchemy.pool import QueuePool, NullPool
-from sqlalchemy.exc import OperationalError, InterfaceError, DBAPIError
-from models_sql import Base, SystemConfig
-
-logger = logging.getLogger("ComfyUI-Ranking.Database")
-
-# 核心：优先读取环境变量中的 PostgreSQL 数据库连接
-SQLALCHEMY_DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:////tmp/comfy_financial.db")
-
-# 🚀 P2优化：连接池配置支持环境变量
-POOL_SIZE = int(os.environ.get("DB_POOL_SIZE", "5"))
-POOL_OVERFLOW = int(os.environ.get("DB_POOL_OVERFLOW", "10"))
-
-# 🚀 P1性能优化：根据数据库类型配置连接池
-if "sqlite" in SQLALCHEMY_DATABASE_URL:
-    # SQLite：使用空连接池，单线程模式
-    connect_args = {"check_same_thread": False}
-    engine = create_engine(
-        SQLALCHEMY_DATABASE_URL, 
-        connect_args=connect_args,
-        poolclass=NullPool  # SQLite 不需要连接池
-    )
-    logger.info(f"数据库引擎初始化完成 (SQLite)")
-else:
-    # PostgreSQL/MySQL：配置连接池参数
-    connect_args = {}
-    if "postgresql" in SQLALCHEMY_DATABASE_URL or "postgres" in SQLALCHEMY_DATABASE_URL:
-        sslmode = os.environ.get("DB_SSLMODE", "require")
-        connect_args["sslmode"] = sslmode
-        connect_args["connect_timeout"] = int(os.environ.get("DB_CONNECT_TIMEOUT", "10"))
-        logger.info(f"PostgreSQL SSL 配置: sslmode={sslmode}, connect_timeout={connect_args['connect_timeout']}")
-    engine = create_engine(
-        SQLALCHEMY_DATABASE_URL,
-        poolclass=QueuePool,
-        pool_size=POOL_SIZE,           # 核心连接数（支持环境变量配置）
-        max_overflow=POOL_OVERFLOW,    # 超出 pool_size 后可创建的最大连接数（支持环境变量配置）
-        pool_timeout=30,               # 获取连接超时(秒)
-        pool_recycle=1800,             # 连接回收时间(30分钟)
-        pool_pre_ping=True,            # 使用前检测连接有效性
-        connect_args=connect_args
-    )
-    logger.info(f"数据库引擎初始化完成 (PostgreSQL/MySQL) - 连接池配置: pool_size={POOL_SIZE}, max_overflow={POOL_OVERFLOW}")
-
-SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-
-
-# ==========================================
-# 🛡️ 稳定性优化：数据库操作重试装饰器
-# ==========================================
-def db_retry(max_retries: int = 3, delay: float = 0.5):
-    """
-    数据库操作重试装饰器
-    
-    用法：
-        @db_retry(max_retries=3)
-        def my_db_function(db):
-            ...
-    """
-    def decorator(func):
-        @wraps(func)
-        def wrapper(*args, **kwargs):
-            last_error = None
-            for attempt in range(max_retries):
-                try:
-                    return func(*args, **kwargs)
-                except (OperationalError, InterfaceError, DBAPIError) as e:
-                    last_error = e
-                    if attempt < max_retries - 1:
-                        wait_time = delay * (2 ** attempt)  # 指数退避
-                        logger.warning(f"DB 操作失败 (第{attempt+1}次):{e}, {wait_time}秒后重试")
-                        time.sleep(wait_time)
-                    else:
-                        logger.error(f"DB 操作失败 (重试{max_retries}次后放弃): {e}")
-            raise last_error
-        return wrapper
-    return decorator
-
-
-def init_sql_db():
-    """初始化数据库，包含重试机制"""
-    for attempt in range(3):
-        try:
-            Base.metadata.create_all(bind=engine)
-            
-            # 🔄 P7后悔模式：自动迁移新增字段
-            _auto_migrate_p7_fields()
-            
-            logger.info("数据库初始化成功")
-            return
-        except Exception as e:
-            if attempt < 2:
-                logger.warning(f"数据库初始化失败 (第{attempt+1}次): {e}")
-                time.sleep(2 ** attempt)
-            else:
-                logger.error(f"数据库初始化失败 (已重试3次): {e}")
-                raise
-
-
-def _auto_migrate_p7_fields():
-    """
-    🔄 P7后悔模式：自动迁移新增字段
-    检查并添加 ownerships 表和 transactions 表的新字段
-    所有 ALTER TABLE ADD COLUMN 均使用 IF NOT EXISTS 或 inspector 预检，避免重复执行报错
-    """
-    from sqlalchemy import inspect
-    
-    try:
-        inspector = inspect(engine)
-        
-        # ========== 1. 迁移 ownerships 表 ==========
-        if 'ownerships' in inspector.get_table_names():
-            columns = [col['name'] for col in inspector.get_columns('ownerships')]
-            
-            with engine.connect() as conn:
-                if 'price_paid' not in columns:
-                    try:
-                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
-                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN price_paid INTEGER DEFAULT 0"))
-                        else:
-                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN IF NOT EXISTS price_paid INTEGER DEFAULT 0"))
-                        logger.info("迁移完成: 添加 ownerships.price_paid 字段")
-                    except Exception as e:
-                        logger.debug(f"跳过 ownerships.price_paid (可能已存在): {e}")
-                
-                if 'is_refunded' not in columns:
-                    try:
-                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
-                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN is_refunded BOOLEAN DEFAULT 0"))
-                        else:
-                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN IF NOT EXISTS is_refunded BOOLEAN DEFAULT FALSE"))
-                        logger.info("迁移完成: 添加 ownerships.is_refunded 字段")
-                    except Exception as e:
-                        logger.debug(f"跳过 ownerships.is_refunded (可能已存在): {e}")
-                
-                if 'refunded_at' not in columns:
-                    try:
-                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
-                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN refunded_at TIMESTAMP"))
-                        else:
-                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN IF NOT EXISTS refunded_at TIMESTAMP"))
-                        logger.info("迁移完成: 添加 ownerships.refunded_at 字段")
-                    except Exception as e:
-                        logger.debug(f"跳过 ownerships.refunded_at (可能已存在): {e}")
-                
-                conn.commit()
-        
-        # ========== 2. 迁移 wallets 表（新增 task_balance 字段） ==========
-        if 'wallets' in inspector.get_table_names():
-            columns = [col['name'] for col in inspector.get_columns('wallets')]
-            
-            with engine.connect() as conn:
-                if 'task_balance' not in columns:
-                    try:
-                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
-                            conn.execute(text("ALTER TABLE wallets ADD COLUMN task_balance INTEGER DEFAULT 0"))
-                        else:
-                            conn.execute(text("ALTER TABLE wallets ADD COLUMN IF NOT EXISTS task_balance INTEGER DEFAULT 0"))
-                        logger.info("[DB Migration] 添加列 wallets.task_balance")
-                    except Exception as e:
-                        logger.debug(f"跳过 wallets.task_balance (可能已存在): {e}")
-                
-                conn.commit()
-        
-        # ========== 3. 迁移 transactions 表（提现相关新字段） ==========
-        if 'transactions' in inspector.get_table_names():
-            columns = [col['name'] for col in inspector.get_columns('transactions')]
-            
-            with engine.connect() as conn:
-                # 定义 transactions 表的新列
-                new_columns = {
-                    'alipay_account': 'VARCHAR',
-                    'real_name': 'VARCHAR',
-                    'withdraw_status': 'VARCHAR',
-                    'payment_order_id': 'VARCHAR',
-                    'net_amount': 'INTEGER',
-                    'description': 'VARCHAR',
-                    'item_title': 'VARCHAR',
-                    'item_type': 'VARCHAR',
-                    'related_user_name': 'VARCHAR',
-                }
-                
-                for col_name, col_type in new_columns.items():
-                    if col_name not in columns:
-                        try:
-                            # VARCHAR 类型添加 DEFAULT NULL 避免 PostgreSQL NOT NULL 冲突
-                            if col_type == 'VARCHAR':
-                                if 'sqlite' in SQLALCHEMY_DATABASE_URL:
-                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {col_name} {col_type} DEFAULT NULL"))
-                                else:
-                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN IF NOT EXISTS {col_name} {col_type} DEFAULT NULL"))
-                            else:
-                                if 'sqlite' in SQLALCHEMY_DATABASE_URL:
-                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {col_name} {col_type}"))
-                                else:
-                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN IF NOT EXISTS {col_name} {col_type}"))
-                            logger.info(f"[DB Migration] 添加列 transactions.{col_name}")
-                        except Exception as e:
-                            logger.debug(f"跳过 transactions.{col_name} (可能已存在): {e}")
-                
-                conn.commit()
-                
-                # ========== 4. 回填旧提现记录数据 ==========
-                # 将旧 WITHDRAW 记录的 withdraw_status 从 NULL 更新为 "completed"
-                result = conn.execute(text(
-                    "UPDATE transactions SET withdraw_status = 'completed' "
-                    "WHERE tx_type = 'WITHDRAW' AND withdraw_status IS NULL"
-                ))
-                if result.rowcount > 0:
-                    logger.info(f"[DB Migration] 回填旧提现记录: 更新 {result.rowcount} 条记录的 withdraw_status 为 'completed'")
-                
-                # 将旧记录的 net_amount 从 NULL 更新为 ABS(amount)
-                result = conn.execute(text(
-                    "UPDATE transactions SET net_amount = ABS(amount) "
-                    "WHERE tx_type = 'WITHDRAW' AND net_amount IS NULL"
-                ))
-                if result.rowcount > 0:
-                    logger.info(f"[DB Migration] 回填旧提现记录: 更新 {result.rowcount} 条记录的 net_amount 为 ABS(amount)")
-                
-                conn.commit()
-                
-    except Exception as e:
-        logger.warning(f"字段迁移跳过 (可能已存在): {e}")
-
-
-def get_db():
-    """获取数据库会话，带连接有效性检测"""
-    db = SessionLocal()
-    try:
-        yield db
-    except (OperationalError, InterfaceError) as e:
-        logger.error(f"数据库连接错误: {e}")
-        db.rollback()
-        raise
-    finally:
-        db.close()
-
-
-def check_db_connection() -> bool:
-    """
-    检查数据库连接是否正常
-    用于健康检查接口
-    """
-    try:
-        db = SessionLocal()
-        db.execute(text("SELECT 1"))
-        db.close()
-        return True
-    except Exception as e:
-        logger.error(f"数据库连接检查失败: {e}")
+# database_sql.py
+# ==========================================
+# 🛡️ 稳定性优化：SQL 数据库连接模块
+# ==========================================
+import os
+import time
+import logging
+from functools import wraps
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import QueuePool, NullPool
+from sqlalchemy.exc import OperationalError, InterfaceError, DBAPIError
+from models_sql import Base, SystemConfig, RegisteredPlugin
+
+logger = logging.getLogger("ComfyUI-Ranking.Database")
+
+# 核心：优先读取环境变量中的 PostgreSQL 数据库连接
+SQLALCHEMY_DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:////tmp/comfy_financial.db")
+
+# 🚀 P2优化：连接池配置支持环境变量
+POOL_SIZE = int(os.environ.get("DB_POOL_SIZE", "5"))
+POOL_OVERFLOW = int(os.environ.get("DB_POOL_OVERFLOW", "10"))
+
+# 🚀 P1性能优化：根据数据库类型配置连接池
+if "sqlite" in SQLALCHEMY_DATABASE_URL:
+    # SQLite：使用空连接池，单线程模式
+    connect_args = {"check_same_thread": False}
+    engine = create_engine(
+        SQLALCHEMY_DATABASE_URL, 
+        connect_args=connect_args,
+        poolclass=NullPool  # SQLite 不需要连接池
+    )
+    logger.info(f"数据库引擎初始化完成 (SQLite)")
+else:
+    # PostgreSQL/MySQL：配置连接池参数
+    connect_args = {}
+    if "postgresql" in SQLALCHEMY_DATABASE_URL or "postgres" in SQLALCHEMY_DATABASE_URL:
+        sslmode = os.environ.get("DB_SSLMODE", "require")
+        connect_args["sslmode"] = sslmode
+        connect_args["connect_timeout"] = int(os.environ.get("DB_CONNECT_TIMEOUT", "10"))
+        logger.info(f"PostgreSQL SSL 配置: sslmode={sslmode}, connect_timeout={connect_args['connect_timeout']}")
+    engine = create_engine(
+        SQLALCHEMY_DATABASE_URL,
+        poolclass=QueuePool,
+        pool_size=POOL_SIZE,           # 核心连接数（支持环境变量配置）
+        max_overflow=POOL_OVERFLOW,    # 超出 pool_size 后可创建的最大连接数（支持环境变量配置）
+        pool_timeout=30,               # 获取连接超时(秒)
+        pool_recycle=1800,             # 连接回收时间(30分钟)
+        pool_pre_ping=True,            # 使用前检测连接有效性
+        connect_args=connect_args
+    )
+    logger.info(f"数据库引擎初始化完成 (PostgreSQL/MySQL) - 连接池配置: pool_size={POOL_SIZE}, max_overflow={POOL_OVERFLOW}")
+
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+# ==========================================
+# 🛡️ 稳定性优化：数据库操作重试装饰器
+# ==========================================
+def db_retry(max_retries: int = 3, delay: float = 0.5):
+    """
+    数据库操作重试装饰器
+    
+    用法：
+        @db_retry(max_retries=3)
+        def my_db_function(db):
+            ...
+    """
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            last_error = None
+            for attempt in range(max_retries):
+                try:
+                    return func(*args, **kwargs)
+                except (OperationalError, InterfaceError, DBAPIError) as e:
+                    last_error = e
+                    if attempt < max_retries - 1:
+                        wait_time = delay * (2 ** attempt)  # 指数退避
+                        logger.warning(f"DB 操作失败 (第{attempt+1}次):{e}, {wait_time}秒后重试")
+                        time.sleep(wait_time)
+                    else:
+                        logger.error(f"DB 操作失败 (重试{max_retries}次后放弃): {e}")
+            raise last_error
+        return wrapper
+    return decorator
+
+
+def init_sql_db():
+    """初始化数据库，包含重试机制"""
+    for attempt in range(3):
+        try:
+            Base.metadata.create_all(bind=engine)
+            
+            # 🔄 P7后悔模式：自动迁移新增字段
+            _auto_migrate_p7_fields()
+            
+            logger.info("数据库初始化成功")
+            return
+        except Exception as e:
+            if attempt < 2:
+                logger.warning(f"数据库初始化失败 (第{attempt+1}次): {e}")
+                time.sleep(2 ** attempt)
+            else:
+                logger.error(f"数据库初始化失败 (已重试3次): {e}")
+                raise
+
+
+def _auto_migrate_p7_fields():
+    """
+    🔄 P7后悔模式：自动迁移新增字段
+    检查并添加 ownerships 表和 transactions 表的新字段
+    所有 ALTER TABLE ADD COLUMN 均使用 IF NOT EXISTS 或 inspector 预检，避免重复执行报错
+    """
+    from sqlalchemy import inspect
+    
+    try:
+        inspector = inspect(engine)
+        
+        # ========== 1. 迁移 ownerships 表 ==========
+        if 'ownerships' in inspector.get_table_names():
+            columns = [col['name'] for col in inspector.get_columns('ownerships')]
+            
+            with engine.connect() as conn:
+                if 'price_paid' not in columns:
+                    try:
+                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN price_paid INTEGER DEFAULT 0"))
+                        else:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN IF NOT EXISTS price_paid INTEGER DEFAULT 0"))
+                        logger.info("迁移完成: 添加 ownerships.price_paid 字段")
+                    except Exception as e:
+                        logger.debug(f"跳过 ownerships.price_paid (可能已存在): {e}")
+                
+                if 'is_refunded' not in columns:
+                    try:
+                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN is_refunded BOOLEAN DEFAULT 0"))
+                        else:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN IF NOT EXISTS is_refunded BOOLEAN DEFAULT FALSE"))
+                        logger.info("迁移完成: 添加 ownerships.is_refunded 字段")
+                    except Exception as e:
+                        logger.debug(f"跳过 ownerships.is_refunded (可能已存在): {e}")
+                
+                if 'refunded_at' not in columns:
+                    try:
+                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN refunded_at TIMESTAMP"))
+                        else:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN IF NOT EXISTS refunded_at TIMESTAMP"))
+                        logger.info("迁移完成: 添加 ownerships.refunded_at 字段")
+                    except Exception as e:
+                        logger.debug(f"跳过 ownerships.refunded_at (可能已存在): {e}")
+                
+                conn.commit()
+        
+        # ========== 2. 迁移 wallets 表（新增 task_balance 字段） ==========
+        if 'wallets' in inspector.get_table_names():
+            columns = [col['name'] for col in inspector.get_columns('wallets')]
+            
+            with engine.connect() as conn:
+                if 'task_balance' not in columns:
+                    try:
+                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                            conn.execute(text("ALTER TABLE wallets ADD COLUMN task_balance INTEGER DEFAULT 0"))
+                        else:
+                            conn.execute(text("ALTER TABLE wallets ADD COLUMN IF NOT EXISTS task_balance INTEGER DEFAULT 0"))
+                        logger.info("[DB Migration] 添加列 wallets.task_balance")
+                    except Exception as e:
+                        logger.debug(f"跳过 wallets.task_balance (可能已存在): {e}")
+                
+                conn.commit()
+        
+        # ========== 3. 迁移 transactions 表（提现相关新字段） ==========
+        if 'transactions' in inspector.get_table_names():
+            columns = [col['name'] for col in inspector.get_columns('transactions')]
+            
+            with engine.connect() as conn:
+                # 定义 transactions 表的新列
+                new_columns = {
+                    'alipay_account': 'VARCHAR',
+                    'real_name': 'VARCHAR',
+                    'withdraw_status': 'VARCHAR',
+                    'payment_order_id': 'VARCHAR',
+                    'net_amount': 'INTEGER',
+                    'description': 'VARCHAR',
+                    'item_title': 'VARCHAR',
+                    'item_type': 'VARCHAR',
+                    'related_user_name': 'VARCHAR',
+                    'source_plugin': 'VARCHAR(50)',  # 🌐 开放API：来源插件标识
+                }
+                
+                for col_name, col_type in new_columns.items():
+                    if col_name not in columns:
+                        try:
+                            # VARCHAR 类型添加 DEFAULT NULL 避免 PostgreSQL NOT NULL 冲突
+                            if col_type == 'VARCHAR':
+                                if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {col_name} {col_type} DEFAULT NULL"))
+                                else:
+                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN IF NOT EXISTS {col_name} {col_type} DEFAULT NULL"))
+                            else:
+                                if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {col_name} {col_type}"))
+                                else:
+                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN IF NOT EXISTS {col_name} {col_type}"))
+                            logger.info(f"[DB Migration] 添加列 transactions.{col_name}")
+                        except Exception as e:
+                            logger.debug(f"跳过 transactions.{col_name} (可能已存在): {e}")
+                
+                conn.commit()
+                
+                # ========== 4. 回填旧提现记录数据 ==========
+                # 将旧 WITHDRAW 记录的 withdraw_status 从 NULL 更新为 "completed"
+                result = conn.execute(text(
+                    "UPDATE transactions SET withdraw_status = 'completed' "
+                    "WHERE tx_type = 'WITHDRAW' AND withdraw_status IS NULL"
+                ))
+                if result.rowcount > 0:
+                    logger.info(f"[DB Migration] 回填旧提现记录: 更新 {result.rowcount} 条记录的 withdraw_status 为 'completed'")
+                
+                # 将旧记录的 net_amount 从 NULL 更新为 ABS(amount)
+                result = conn.execute(text(
+                    "UPDATE transactions SET net_amount = ABS(amount) "
+                    "WHERE tx_type = 'WITHDRAW' AND net_amount IS NULL"
+                ))
+                if result.rowcount > 0:
+                    logger.info(f"[DB Migration] 回填旧提现记录: 更新 {result.rowcount} 条记录的 net_amount 为 ABS(amount)")
+                
+                conn.commit()
+                
+        # ========== 5. 确保 registered_plugins 表存在 ==========
+        if not inspect(engine).has_table("registered_plugins"):
+            RegisteredPlugin.__table__.create(engine)
+            logger.info("✅ 已创建 registered_plugins 表")
+
+    except Exception as e:
+        logger.warning(f"字段迁移跳过 (可能已存在): {e}")
+
+
+def get_db():
+    """获取数据库会话，带连接有效性检测"""
+    db = SessionLocal()
+    try:
+        yield db
+    except (OperationalError, InterfaceError) as e:
+        logger.error(f"数据库连接错误: {e}")
+        db.rollback()
+        raise
+    finally:
+        db.close()
+
+
+def check_db_connection() -> bool:
+    """
+    检查数据库连接是否正常
+    用于健康检查接口
+    """
+    try:
+        db = SessionLocal()
+        db.execute(text("SELECT 1"))
+        db.close()
+        return True
+    except Exception as e:
+        logger.error(f"数据库连接检查失败: {e}")
         return False

models_sql.py

@@ -1,98 +1,121 @@
-# models_sql.py
-# ==========================================
-# 🗄️ SQL 数据库模型定义
-# ==========================================
-# 🚀 P1性能优化：添加复合索引加速查询
-# ==========================================
-
-from sqlalchemy import Column, Integer, String, DateTime, Boolean, Index, func
-from sqlalchemy.orm import declarative_base
-import datetime
-
-Base = declarative_base()
-
-class Wallet(Base):
-    """用户钱包表"""
-    __tablename__ = "wallets"
-    
-    account = Column(String, primary_key=True, index=True)
-    balance = Column(Integer, default=0, nullable=False)         # 用于消费的余额 (充值获得)
-    earn_balance = Column(Integer, default=0, nullable=False)    # 创作者销售收益余额 (别人购买获得)
-    tip_balance = Column(Integer, default=0, nullable=False)     # 【新增】创作者打赏收益余额 (粉丝赞助获得)
-    task_balance = Column(Integer, default=0, nullable=False)    # 【新增】任务相关余额
-    frozen_balance = Column(Integer, default=0, nullable=False)  # 提现审核冻结中的余额
-    
-    # 乐观锁版本号，防止并发扣款被击穿
-    version = Column(Integer, default=1)
-
-class Ownership(Base):
-    """资源所有权表"""
-    __tablename__ = "ownerships"
-    
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    account = Column(String, index=True)
-    item_id = Column(String, index=True)
-    purchased_at = Column(DateTime, default=datetime.datetime.utcnow)
-    # 🔄 P7后悔模式增强
-    price_paid = Column(Integer, default=0)       # 购买时支付的价格
-    is_refunded = Column(Boolean, default=False)  # 是否已退款
-    refunded_at = Column(DateTime, nullable=True) # 退款时间
-    
-    # 🚀 P1性能优化：复合索引
-    __table_args__ = (
-        Index('ix_ownerships_account_item', 'account', 'item_id'),
-        Index('ix_ownerships_account_refunded', 'account', 'is_refunded'),
-    )
-
-class Refund(Base):
-    """🔄 P7后悔模式：退款记录表"""
-    __tablename__ = "refunds"
-    
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    account = Column(String, index=True)          # 退款用户
-    item_id = Column(String, index=True)          # 退款商品
-    amount = Column(Integer)                       # 退款金额
-    refunded_at = Column(DateTime, default=datetime.datetime.utcnow)
-    ban_until = Column(DateTime)                   # 禁止购买截止时间（30天后）
-
-class Transaction(Base):
-    """交易流水表"""
-    __tablename__ = "transactions"
-    
-    tx_id = Column(String, primary_key=True)
-    account = Column(String, index=True)
-    tx_type = Column(String)                     # 枚举: RECHARGE, PURCHASE, SALE, TIP_OUT, TIP_IN, WITHDRAW, WITHDRAW_FEE, REFUND, TASK_FREEZE, TASK_DEPOSIT, TASK_PAYMENT, TASK_INCOME, TASK_REFUND, TASK_CANCEL_REFUND
-    amount = Column(Integer)
-    related_account = Column(String, nullable=True)
-    item_id = Column(String, nullable=True)
-    created_at = Column(DateTime, default=datetime.datetime.utcnow, index=True)  # 🚀 P1: 添加时间索引
-    prev_hash = Column(String)                   # 上一笔订单的哈希
-    tx_hash = Column(String)                     # 本笔订单的哈希 (防篡改)
-    
-    # 💰 提现相关字段 (仅 WITHDRAW_APPLY 类型交易使用)
-    alipay_account = Column(String, nullable=True)      # 收款支付宝账号
-    real_name = Column(String, nullable=True)            # 收款人姓名
-    withdraw_status = Column(String, nullable=True)      # pending / completed
-    payment_order_id = Column(String, nullable=True)     # 管理员填写的打款订单号
-    net_amount = Column(Integer, nullable=True)          # 实际到账金额（用于解冻时准确扣减）
-    
-    # 交易详情增强字段
-    description = Column(String, nullable=True)          # 交易自描述文本
-    item_title = Column(String, nullable=True)           # 资源/任务标题
-    item_type = Column(String, nullable=True)            # 资源类型 (tool/app/recommend/post/task)
-    related_user_name = Column(String, nullable=True)    # 对方用户名
-    
-    # 🚀 P1性能优化：复合索引
-    __table_args__ = (
-        Index('ix_transactions_account_type', 'account', 'tx_type'),
-        Index('ix_transactions_account_created', 'account', 'created_at'),
-    )
-
-class SystemConfig(Base):
-    """系统配置表 - 持久化存储，替代 /tmp/system_config.json"""
-    __tablename__ = "system_configs"
-    
-    config_key = Column(String, primary_key=True)
-    config_value = Column(String, nullable=False, default="{}")
-    updated_at = Column(DateTime, default=datetime.datetime.utcnow)
-    updated_by = Column(String, nullable=True)
+# models_sql.py
+# ==========================================
+# 🗄️ SQL 数据库模型定义
+# ==========================================
+# 🚀 P1性能优化：添加复合索引加速查询
+# ==========================================
+
+from sqlalchemy import Column, Integer, String, DateTime, Boolean, Index, func
+from sqlalchemy.orm import declarative_base
+import datetime
+
+Base = declarative_base()
+
+class Wallet(Base):
+    """用户钱包表"""
+    __tablename__ = "wallets"
+    
+    account = Column(String, primary_key=True, index=True)
+    balance = Column(Integer, default=0, nullable=False)         # 用于消费的余额 (充值获得)
+    earn_balance = Column(Integer, default=0, nullable=False)    # 创作者销售收益余额 (别人购买获得)
+    tip_balance = Column(Integer, default=0, nullable=False)     # 【新增】创作者打赏收益余额 (粉丝赞助获得)
+    task_balance = Column(Integer, default=0, nullable=False)    # 【新增】任务相关余额
+    frozen_balance = Column(Integer, default=0, nullable=False)  # 提现审核冻结中的余额
+    
+    # 乐观锁版本号，防止并发扣款被击穿
+    version = Column(Integer, default=1)
+
+class Ownership(Base):
+    """资源所有权表"""
+    __tablename__ = "ownerships"
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    account = Column(String, index=True)
+    item_id = Column(String, index=True)
+    purchased_at = Column(DateTime, default=datetime.datetime.utcnow)
+    # 🔄 P7后悔模式增强
+    price_paid = Column(Integer, default=0)       # 购买时支付的价格
+    is_refunded = Column(Boolean, default=False)  # 是否已退款
+    refunded_at = Column(DateTime, nullable=True) # 退款时间
+    
+    # 🚀 P1性能优化：复合索引
+    __table_args__ = (
+        Index('ix_ownerships_account_item', 'account', 'item_id'),
+        Index('ix_ownerships_account_refunded', 'account', 'is_refunded'),
+    )
+
+class Refund(Base):
+    """🔄 P7后悔模式：退款记录表"""
+    __tablename__ = "refunds"
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    account = Column(String, index=True)          # 退款用户
+    item_id = Column(String, index=True)          # 退款商品
+    amount = Column(Integer)                       # 退款金额
+    refunded_at = Column(DateTime, default=datetime.datetime.utcnow)
+    ban_until = Column(DateTime)                   # 禁止购买截止时间（30天后）
+
+class Transaction(Base):
+    """交易流水表"""
+    __tablename__ = "transactions"
+    
+    tx_id = Column(String, primary_key=True)
+    account = Column(String, index=True)
+    tx_type = Column(String)                     # 枚举: RECHARGE, PURCHASE, SALE, TIP_OUT, TIP_IN, WITHDRAW, WITHDRAW_FEE, REFUND, TASK_FREEZE, TASK_DEPOSIT, TASK_PAYMENT, TASK_INCOME, TASK_REFUND, TASK_CANCEL_REFUND
+    amount = Column(Integer)
+    related_account = Column(String, nullable=True)
+    item_id = Column(String, nullable=True)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow, index=True)  # 🚀 P1: 添加时间索引
+    prev_hash = Column(String)                   # 上一笔订单的哈希
+    tx_hash = Column(String)                     # 本笔订单的哈希 (防篡改)
+    
+    # 💰 提现相关字段 (仅 WITHDRAW_APPLY 类型交易使用)
+    alipay_account = Column(String, nullable=True)      # 收款支付宝账号
+    real_name = Column(String, nullable=True)            # 收款人姓名
+    withdraw_status = Column(String, nullable=True)      # pending / completed
+    payment_order_id = Column(String, nullable=True)     # 管理员填写的打款订单号
+    net_amount = Column(Integer, nullable=True)          # 实际到账金额（用于解冻时准确扣减）
+    
+    # 交易详情增强字段
+    description = Column(String, nullable=True)          # 交易自描述文本
+    item_title = Column(String, nullable=True)           # 资源/任务标题
+    item_type = Column(String, nullable=True)            # 资源类型 (tool/app/recommend/post/task)
+    related_user_name = Column(String, nullable=True)    # 对方用户名
+    
+    # 🌐 开放API：来源插件标识（如 "nodecraft_ai"），用于按工具结构化追踪消费
+    source_plugin = Column(String(50), nullable=True, index=True)
+    
+    # 🚀 P1性能优化：复合索引
+    __table_args__ = (
+        Index('ix_transactions_account_type', 'account', 'tx_type'),
+        Index('ix_transactions_account_created', 'account', 'created_at'),
+        Index('ix_transactions_source_plugin_created', 'source_plugin', 'created_at'),
+    )
+
+class SystemConfig(Base):
+    """系统配置表 - 持久化存储，替代 /tmp/system_config.json"""
+    __tablename__ = "system_configs"
+    
+    config_key = Column(String, primary_key=True)
+    config_value = Column(String, nullable=False, default="{}")
+    updated_at = Column(DateTime, default=datetime.datetime.utcnow)
+    updated_by = Column(String, nullable=True)
+
+
+class RegisteredPlugin(Base):
+    """🌐 开放API：第三方插件注册信息表"""
+    __tablename__ = "registered_plugins"
+
+    plugin_id          = Column(String(50), primary_key=True)   # 英文标识如 "nodecraft_ai"
+    plugin_key         = Column(String(100), unique=True, nullable=False, index=True)
+    name               = Column(String(100), nullable=False)     # 显示名
+    description        = Column(String(500), nullable=True)
+    enabled            = Column(Boolean, default=True, nullable=False)
+    rate_limit         = Column(Integer, default=100)            # 每分钟最大请求数
+    max_deduct_per_tx  = Column(Integer, default=100)            # 单笔扣款上限
+    max_deduct_per_day = Column(Integer, default=1000)           # 日扣款上限
+    permissions        = Column(String(200), default='["verify","balance","deduct","refund"]')  # JSON数组
+    created_at         = Column(DateTime, default=func.now())
+    created_by         = Column(String(50))                      # 创建者管理员账号
+    updated_at         = Column(DateTime, onupdate=func.now())
+    last_rotated_at    = Column(DateTime, nullable=True)         # 上次轮换key时间

router_admin_plugins.py

@@ -0,0 +1,391 @@
+# router_admin_plugins.py
+# ==========================================
+# 🌐 管理员插件管理路由模块
+# ==========================================
+# 作用：提供第三方插件的注册、查询、更新、删除、Key轮换等管理API
+# 关联文件：
+#   - models_sql.py (RegisteredPlugin 模型)
+#   - database_sql.py (SQL数据库连接 get_db)
+#   - 安全认证.py (require_auth / is_admin)
+#   - router_open_api.py (reload_plugin_registry 热更新)
+# 权限：所有端点均需管理员权限（is_admin() == True）
+# ==========================================
+
+import re
+import json
+import secrets
+import logging
+import datetime
+from typing import Optional, List
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import JSONResponse
+from sqlalchemy.orm import Session
+from pydantic import BaseModel, Field, validator
+
+from database_sql import get_db
+from models_sql import RegisteredPlugin
+from 安全认证 import require_auth, is_admin
+
+logger = logging.getLogger("ComfyUI-Ranking.AdminPlugins")
+
+router = APIRouter()
+
+# ==========================================
+# 🔧 工具函数
+# ==========================================
+
+# plugin_id 合法性正则：3-50字符，仅允许小写字母、数字、下划线
+_PLUGIN_ID_PATTERN = re.compile(r"^[a-z0-9_]{3,50}$")
+
+# 默认权限列表
+_DEFAULT_PERMISSIONS = ["verify", "balance", "deduct", "refund"]
+
+
+def generate_plugin_key() -> str:
+    """生成 plugin_key，格式: rk_plug_ + 32位hex随机串"""
+    return f"rk_plug_{secrets.token_hex(16)}"
+
+
+def mask_key(key: str) -> str:
+    """脱敏 plugin_key：保留前12位（rk_plug_ + 4位），其余以 *** 替代"""
+    if not key:
+        return "***"
+    return key[:12] + "***" if len(key) > 12 else "***"
+
+
+def _require_admin(account: str):
+    """工具函数：校验管理员权限，否则抛 403"""
+    if not is_admin(account):
+        raise HTTPException(status_code=403, detail="需要管理员权限")
+
+
+def _notify_registry_reload():
+    """通知 open_api 模块重新加载插件注册表（热更新）"""
+    try:
+        from router_open_api import reload_plugin_registry
+        reload_plugin_registry()
+        logger.info("插件注册表热更新成功")
+    except (ImportError, Exception) as e:
+        logger.warning(f"插件注册表热更新失败: {e}")
+
+
+def _serialize_plugin(p: RegisteredPlugin, include_key: bool = False, raw_key: Optional[str] = None) -> dict:
+    """
+    序列化插件对象为字典
+
+    参数：
+        p: RegisteredPlugin ORM 对象
+        include_key: 是否返回完整 plugin_key（仅在创建/轮换时为 True）
+        raw_key: 完整明文 key（仅在创建/轮换时传入；其他场景使用脱敏）
+    """
+    try:
+        permissions = json.loads(p.permissions) if p.permissions else _DEFAULT_PERMISSIONS
+    except Exception:
+        permissions = _DEFAULT_PERMISSIONS
+
+    data = {
+        "plugin_id": p.plugin_id,
+        "name": p.name,
+        "description": p.description,
+        "enabled": bool(p.enabled),
+        "rate_limit": p.rate_limit,
+        "max_deduct_per_tx": p.max_deduct_per_tx,
+        "max_deduct_per_day": p.max_deduct_per_day,
+        "permissions": permissions,
+        "created_at": p.created_at.isoformat() if p.created_at else None,
+        "created_by": p.created_by,
+        "updated_at": p.updated_at.isoformat() if p.updated_at else None,
+        "last_rotated_at": p.last_rotated_at.isoformat() if p.last_rotated_at else None,
+    }
+
+    if include_key and raw_key:
+        data["plugin_key"] = raw_key
+    else:
+        data["plugin_key_masked"] = mask_key(p.plugin_key)
+
+    return data
+
+
+# ==========================================
+# 📥 请求模型
+# ==========================================
+
+class PluginCreateRequest(BaseModel):
+    plugin_id: str = Field(..., min_length=3, max_length=50, description="插件唯一标识，仅允许小写字母、数字、下划线")
+    name: str = Field(..., min_length=1, max_length=100, description="插件显示名")
+    description: Optional[str] = Field(None, max_length=500)
+    rate_limit: Optional[int] = Field(100, ge=1, le=10000)
+    max_deduct_per_tx: Optional[int] = Field(100, ge=1, le=10000)
+    max_deduct_per_day: Optional[int] = Field(1000, ge=1, le=100000)
+    permissions: Optional[List[str]] = Field(default=None)
+
+    @validator("plugin_id")
+    def _validate_plugin_id(cls, v):
+        if not _PLUGIN_ID_PATTERN.match(v):
+            raise ValueError("plugin_id 仅允许 [a-z0-9_]，长度 3-50")
+        return v
+
+    @validator("permissions", always=True)
+    def _validate_permissions(cls, v):
+        if v is None:
+            return _DEFAULT_PERMISSIONS
+        if not isinstance(v, list) or any(not isinstance(x, str) for x in v):
+            raise ValueError("permissions 必���为字符串数组")
+        return v
+
+
+class PluginUpdateRequest(BaseModel):
+    name: Optional[str] = Field(None, min_length=1, max_length=100)
+    description: Optional[str] = Field(None, max_length=500)
+    enabled: Optional[bool] = None
+    rate_limit: Optional[int] = Field(None, ge=1, le=10000)
+    max_deduct_per_tx: Optional[int] = Field(None, ge=1, le=10000)
+    max_deduct_per_day: Optional[int] = Field(None, ge=1, le=100000)
+    permissions: Optional[List[str]] = None
+
+    @validator("permissions")
+    def _validate_permissions(cls, v):
+        if v is None:
+            return v
+        if not isinstance(v, list) or any(not isinstance(x, str) for x in v):
+            raise ValueError("permissions 必须为字符串数组")
+        return v
+
+
+# ==========================================
+# 🌐 路由实现
+# ==========================================
+
+@router.get("/api/admin/plugins")
+async def list_plugins(
+    account: str = Depends(require_auth),
+    db: Session = Depends(get_db),
+):
+    """列出所有已注册插件（plugin_key 脱敏）"""
+    _require_admin(account)
+    try:
+        plugins = db.query(RegisteredPlugin).order_by(RegisteredPlugin.created_at.desc()).all()
+        return JSONResponse(content={
+            "success": True,
+            "total": len(plugins),
+            "plugins": [_serialize_plugin(p) for p in plugins],
+        })
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception(f"列出插件失败: {e}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "服务器内部错误"})
+
+
+@router.get("/api/admin/plugins/{plugin_id}")
+async def get_plugin(
+    plugin_id: str,
+    account: str = Depends(require_auth),
+    db: Session = Depends(get_db),
+):
+    """获取插件详情（plugin_key 脱敏）"""
+    _require_admin(account)
+    plugin = db.query(RegisteredPlugin).filter(RegisteredPlugin.plugin_id == plugin_id).first()
+    if not plugin:
+        return JSONResponse(status_code=404, content={"success": False, "error": "插件不存在"})
+    return JSONResponse(content={"success": True, "plugin": _serialize_plugin(plugin)})
+
+
+@router.post("/api/admin/plugins")
+async def create_plugin(
+    payload: PluginCreateRequest,
+    account: str = Depends(require_auth),
+    db: Session = Depends(get_db),
+):
+    """创建新插件（自动生成 plugin_key，仅本次返回明文 key）"""
+    _require_admin(account)
+
+    # 检查 plugin_id 是否已存在
+    exists = db.query(RegisteredPlugin).filter(RegisteredPlugin.plugin_id == payload.plugin_id).first()
+    if exists:
+        return JSONResponse(status_code=409, content={"success": False, "error": f"plugin_id '{payload.plugin_id}' 已存在"})
+
+    plugin_key = generate_plugin_key()
+    now = datetime.datetime.utcnow()
+
+    plugin = RegisteredPlugin(
+        plugin_id=payload.plugin_id,
+        plugin_key=plugin_key,
+        name=payload.name,
+        description=payload.description,
+        enabled=True,
+        rate_limit=payload.rate_limit,
+        max_deduct_per_tx=payload.max_deduct_per_tx,
+        max_deduct_per_day=payload.max_deduct_per_day,
+        permissions=json.dumps(payload.permissions, ensure_ascii=False),
+        created_at=now,
+        created_by=account,
+        updated_at=now,
+    )
+
+    try:
+        db.add(plugin)
+        db.commit()
+        db.refresh(plugin)
+    except Exception as e:
+        db.rollback()
+        logger.exception(f"创建插件失败: {e}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "创建插件失败"})
+
+    logger.info(f"管理员 {account} 创建插件 plugin_id={payload.plugin_id}")
+    _notify_registry_reload()
+
+    return JSONResponse(status_code=201, content={
+        "success": True,
+        "plugin_id": plugin.plugin_id,
+        "plugin_key": plugin_key,
+        "name": plugin.name,
+        "message": "插件创建成功，请妥善保存 plugin_key（仅显示一次）",
+        "plugin": _serialize_plugin(plugin, include_key=True, raw_key=plugin_key),
+    })
+
+
+@router.put("/api/admin/plugins/{plugin_id}")
+async def update_plugin(
+    plugin_id: str,
+    payload: PluginUpdateRequest,
+    account: str = Depends(require_auth),
+    db: Session = Depends(get_db),
+):
+    """更新插件配置（不可修改 plugin_id 与 plugin_key）"""
+    _require_admin(account)
+
+    plugin = db.query(RegisteredPlugin).filter(RegisteredPlugin.plugin_id == plugin_id).first()
+    if not plugin:
+        return JSONResponse(status_code=404, content={"success": False, "error": "插件不存在"})
+
+    update_fields = payload.dict(exclude_unset=True)
+    if not update_fields:
+        return JSONResponse(status_code=400, content={"success": False, "error": "未提供任何可更新字段"})
+
+    try:
+        if "permissions" in update_fields and update_fields["permissions"] is not None:
+            plugin.permissions = json.dumps(update_fields.pop("permissions"), ensure_ascii=False)
+
+        for key, value in update_fields.items():
+            if value is None:
+                continue
+            setattr(plugin, key, value)
+
+        plugin.updated_at = datetime.datetime.utcnow()
+        db.commit()
+        db.refresh(plugin)
+    except Exception as e:
+        db.rollback()
+        logger.exception(f"更新插件失败: {e}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "更新插件失败"})
+
+    logger.info(f"管理员 {account} 更新插件 plugin_id={plugin_id}, fields={list(update_fields.keys())}")
+    _notify_registry_reload()
+
+    return JSONResponse(content={
+        "success": True,
+        "message": "插件更新成功",
+        "plugin": _serialize_plugin(plugin),
+    })
+
+
+@router.delete("/api/admin/plugins/{plugin_id}")
+async def delete_plugin(
+    plugin_id: str,
+    account: str = Depends(require_auth),
+    db: Session = Depends(get_db),
+):
+    """删除插件"""
+    _require_admin(account)
+
+    plugin = db.query(RegisteredPlugin).filter(RegisteredPlugin.plugin_id == plugin_id).first()
+    if not plugin:
+        return JSONResponse(status_code=404, content={"success": False, "error": "插件不存在"})
+
+    try:
+        db.delete(plugin)
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        logger.exception(f"删除插件失败: {e}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "删除插件失败"})
+
+    logger.info(f"管理员 {account} 删除插件 plugin_id={plugin_id}")
+    _notify_registry_reload()
+
+    return JSONResponse(content={"success": True, "message": "插件已删除", "plugin_id": plugin_id})
+
+
+@router.post("/api/admin/plugins/{plugin_id}/rotate-key")
+async def rotate_plugin_key(
+    plugin_id: str,
+    account: str = Depends(require_auth),
+    db: Session = Depends(get_db),
+):
+    """轮换 plugin_key，返回新 key（仅本次返回明文）"""
+    _require_admin(account)
+
+    plugin = db.query(RegisteredPlugin).filter(RegisteredPlugin.plugin_id == plugin_id).first()
+    if not plugin:
+        return JSONResponse(status_code=404, content={"success": False, "error": "插件不存在"})
+
+    new_key = generate_plugin_key()
+    now = datetime.datetime.utcnow()
+
+    try:
+        plugin.plugin_key = new_key
+        plugin.last_rotated_at = now
+        plugin.updated_at = now
+        db.commit()
+        db.refresh(plugin)
+    except Exception as e:
+        db.rollback()
+        logger.exception(f"轮换 plugin_key 失败: {e}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "轮换 plugin_key 失败"})
+
+    logger.info(f"管理员 {account} 轮换插件 plugin_id={plugin_id} 的 plugin_key")
+    _notify_registry_reload()
+
+    return JSONResponse(content={
+        "success": True,
+        "plugin_id": plugin.plugin_id,
+        "plugin_key": new_key,
+        "message": "plugin_key 已轮换，请妥善保存（仅显示一次）",
+        "last_rotated_at": now.isoformat(),
+    })
+
+
+@router.post("/api/admin/plugins/{plugin_id}/toggle")
+async def toggle_plugin(
+    plugin_id: str,
+    account: str = Depends(require_auth),
+    db: Session = Depends(get_db),
+):
+    """启用/禁用切换"""
+    _require_admin(account)
+
+    plugin = db.query(RegisteredPlugin).filter(RegisteredPlugin.plugin_id == plugin_id).first()
+    if not plugin:
+        return JSONResponse(status_code=404, content={"success": False, "error": "插件不存在"})
+
+    try:
+        plugin.enabled = not bool(plugin.enabled)
+        plugin.updated_at = datetime.datetime.utcnow()
+        db.commit()
+        db.refresh(plugin)
+    except Exception as e:
+        db.rollback()
+        logger.exception(f"切换插件状态失败: {e}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "切换插件状态失败"})
+
+    logger.info(f"管理员 {account} 切换插件 plugin_id={plugin_id} 状态为 enabled={plugin.enabled}")
+    _notify_registry_reload()
+
+    return JSONResponse(content={
+        "success": True,
+        "plugin_id": plugin.plugin_id,
+        "enabled": bool(plugin.enabled),
+        "message": "插件已启用" if plugin.enabled else "插件已禁用",
+    })

router_open_api.py

@@ -0,0 +1,740 @@
+# router_open_api.py
+# ==========================================
+# 🌐 通用开放 API 模块（供第三方插件调用）
+# ==========================================
+# 作用：对外提供受 Plugin Key 保护的通用接口，
+#       供 NodeCraft AI 等第三方插件复用 ComfyUI-Ranking 账号体系。
+# 关联文件：
+#   - 安全认证.py        (verify_token_with_fallback, require_auth)
+#   - database_sql.py    (get_db, SessionLocal)
+#   - models_sql.py      (Wallet, Transaction)
+#   - 数据库连接.py       (users.json 读写)
+# 安全机制：
+#   - X-Plugin-Key Header 鉴权
+#   - 基于 plugin_key 的内存限流（每分钟重置）
+#   - 所有扣款/退款操作记录到 Transaction 流水（含哈希链）
+# ==========================================
+
+import os
+import json
+import time
+import uuid
+import datetime
+import hashlib
+import logging
+import threading
+from datetime import date
+from typing import Optional
+
+from fastapi import APIRouter, Depends, Header, HTTPException, Query
+from fastapi.responses import JSONResponse
+from sqlalchemy.orm import Session
+from pydantic import BaseModel, Field
+
+from database_sql import get_db, SessionLocal
+from models_sql import Wallet, Transaction, RegisteredPlugin
+from 安全认证 import verify_token_with_fallback
+import 数据库连接 as json_db
+
+logger = logging.getLogger("ComfyUI-Ranking.OpenAPI")
+
+router = APIRouter(prefix="/api/open", tags=["open-api"])
+
+
+# ═══════════════ 插件注册表（从数据库动态加载） ═══════════════
+# 反向索引：plugin_key -> {id, key, name, rate_limit, permissions, max_deduct_per_tx, max_deduct_per_day}
+_PLUGIN_KEY_INDEX: dict = {}
+
+
+def load_plugins_from_db():
+    """启动时从数据库加载所有启用的插件到内存索引"""
+    global _PLUGIN_KEY_INDEX
+
+    db = SessionLocal()
+    try:
+        # 首次启动时 seed 默认插件
+        if not db.query(RegisteredPlugin).first():
+            default_key = os.environ.get("NCA_PLUGIN_KEY", "nca_2026_nodecraft_ai_plugin")
+            default_plugin = RegisteredPlugin(
+                plugin_id="nodecraft_ai",
+                plugin_key=default_key,
+                name="NodeCraft AI",
+                enabled=True,
+                rate_limit=100,
+                max_deduct_per_tx=100,
+                max_deduct_per_day=1000,
+                permissions='["verify","balance","deduct","refund"]',
+                created_by="system",
+            )
+            db.add(default_plugin)
+            db.commit()
+            logger.info("✅ 已 seed 默认插件: NodeCraft AI")
+
+        plugins = db.query(RegisteredPlugin).filter(RegisteredPlugin.enabled == True).all()
+
+        # 构建内存索引
+        new_index = {}
+        for p in plugins:
+            try:
+                perms = json.loads(p.permissions) if p.permissions else ["verify", "balance", "deduct", "refund"]
+            except Exception:
+                perms = ["verify", "balance", "deduct", "refund"]
+            new_index[p.plugin_key] = {
+                "id": p.plugin_id,
+                "key": p.plugin_key,
+                "name": p.name,
+                "rate_limit": p.rate_limit or 100,
+                "permissions": perms,
+                "max_deduct_per_tx": p.max_deduct_per_tx or 100,
+                "max_deduct_per_day": p.max_deduct_per_day or 1000,
+            }
+
+        _PLUGIN_KEY_INDEX = new_index
+        logger.info(f"✅ 已加载 {len(new_index)} 个插件到注册表")
+    except Exception as e:
+        logger.error(f"❌ 加载插件注册表失败: {e}")
+        # 如果加载失败且索引为空，使用环境变量 fallback
+        if not _PLUGIN_KEY_INDEX:
+            fallback_key = os.environ.get("NCA_PLUGIN_KEY", "nca_2026_nodecraft_ai_plugin")
+            _PLUGIN_KEY_INDEX = {
+                fallback_key: {
+                    "id": "nodecraft_ai",
+                    "key": fallback_key,
+                    "name": "NodeCraft AI",
+                    "rate_limit": 100,
+                    "permissions": ["verify", "balance", "deduct", "refund"],
+                    "max_deduct_per_tx": 100,
+                    "max_deduct_per_day": 1000,
+                }
+            }
+            logger.warning("⚠️ 使用 fallback 插件配置")
+    finally:
+        db.close()
+
+
+def reload_plugin_registry():
+    """热更新插件注册表（管理操作后调用）"""
+    load_plugins_from_db()
+
+
+# ==========================================
+# 💰 日扣款总额限制：内存计数器（每自然日重置）
+# ==========================================
+_daily_deduct_lock = threading.Lock()
+_daily_deduct_counters: dict = {}  # {plugin_id: {"date": "2026-06-17", "total": 500}}
+
+
+def _peek_daily_used(plugin_id: str) -> int:
+    """读取插件当日已使用的扣款总额（不修改计数）。跨日自动归零。"""
+    today = str(date.today())
+    with _daily_deduct_lock:
+        counter = _daily_deduct_counters.get(plugin_id)
+        if not counter or counter.get("date") != today:
+            return 0
+        return int(counter.get("total", 0))
+
+
+def _commit_daily_used(plugin_id: str, amount: int) -> int:
+    """扣款成功后将本次金额累加到当日计数，返回累加后的总额。"""
+    today = str(date.today())
+    with _daily_deduct_lock:
+        counter = _daily_deduct_counters.get(plugin_id)
+        if not counter or counter.get("date") != today:
+            counter = {"date": today, "total": 0}
+        counter["total"] = int(counter.get("total", 0)) + int(amount)
+        _daily_deduct_counters[plugin_id] = counter
+        return counter["total"]
+
+
+def verify_plugin_key(plugin_key: str) -> dict:
+    """
+    校验 Plugin Key 并返回插件配置。
+
+    参数:
+        plugin_key: 请求头 X-Plugin-Key 的值
+    返回:
+        匹配到的插件配置字典（包含 id 字段）
+    异常:
+        401: 未提供 Plugin Key
+        403: Plugin Key 无效或未注册
+    """
+    if not plugin_key:
+        raise HTTPException(status_code=401, detail="缺少 X-Plugin-Key Header")
+    info = _PLUGIN_KEY_INDEX.get(plugin_key)
+    if not info:
+        logger.warning(f"OPEN_API_INVALID_KEY | key_prefix={plugin_key[:8]}")
+        raise HTTPException(status_code=403, detail="Plugin Key 无效或未注册")
+    return info
+
+
+# ==========================================
+# 🚦 限流：基于 plugin_id 的内存计数器（每分钟自动重置）
+# ==========================================
+_rate_lock = threading.Lock()
+_rate_buckets: dict = {}  # {plugin_id: {"window_start": int, "count": int}}
+
+
+def _check_rate_limit(plugin_info: dict, permission: str) -> None:
+    """
+    校验插件权限并执行限流。
+
+    - 权限不足 -> 403
+    - 触发限流 -> 429
+    """
+    plugin_id = plugin_info["id"]
+    rate_limit = int(plugin_info.get("rate_limit", 60))
+    perms = plugin_info.get("permissions", [])
+    if permission not in perms:
+        raise HTTPException(
+            status_code=403,
+            detail=f"插件 {plugin_id} 无权限调用 {permission} 接口",
+        )
+
+    now = int(time.time())
+    window_start = now - (now % 60)
+
+    with _rate_lock:
+        bucket = _rate_buckets.get(plugin_id)
+        if not bucket or bucket["window_start"] != window_start:
+            bucket = {"window_start": window_start, "count": 0}
+            _rate_buckets[plugin_id] = bucket
+        if bucket["count"] >= rate_limit:
+            raise HTTPException(
+                status_code=429,
+                detail=f"请求过于频繁，限流 {rate_limit}/分钟，请稍后再试",
+            )
+        bucket["count"] += 1
+
+
+def _require_plugin(plugin_key: Optional[str], permission: str) -> dict:
+    """统一入口：校验 Plugin Key + 权限 + 限流"""
+    info = verify_plugin_key(plugin_key or "")
+    _check_rate_limit(info, permission)
+    return info
+
+
+def _resolve_user_from_token(token: str) -> str:
+    """从 Token 中解析出账号；失败时抛 401"""
+    if not token:
+        raise HTTPException(status_code=401, detail="缺少用户 Token")
+    is_valid, account, error_msg = verify_token_with_fallback(token)
+    if not is_valid or not account:
+        raise HTTPException(status_code=401, detail=error_msg or "Token 无效")
+    return account
+
+
+def _calc_tx_hash(tx_id: str, account: str, tx_type: str, amount: int, prev_hash: str) -> str:
+    """计算交易哈希（与 router_wallet.calculate_tx_hash 算法一致）"""
+    data = f"{tx_id}{account}{tx_type}{amount}{prev_hash}"
+    return hashlib.sha256(data.encode()).hexdigest()
+
+
+# ==========================================
+# 📦 请求模型
+# ==========================================
+class VerifyTokenRequest(BaseModel):
+    token: str = Field(..., description="待验证的用户 JWT Token")
+
+
+class DeductRequest(BaseModel):
+    token: str = Field(..., description="用户 JWT Token")
+    amount: int = Field(..., gt=0, description="扣款金额（正整数）")
+    reason: str = Field(..., description="业务标识/扣款原因")
+    reference_id: Optional[str] = Field(None, description="业务侧参考 ID")
+
+
+class RefundRequest(BaseModel):
+    token: str = Field(..., description="用户 JWT Token")
+    transaction_id: str = Field(..., description="原扣款交易 ID")
+    amount: int = Field(..., gt=0, description="退款金额（正整数）")
+
+
+# ==========================================
+# 1️⃣ Token 验证：POST /api/open/verify-token
+# ==========================================
+@router.post("/verify-token")
+async def open_verify_token(
+    req: VerifyTokenRequest,
+    x_plugin_key: Optional[str] = Header(None, alias="X-Plugin-Key"),
+):
+    """
+    供第三方插件验证用户 Token 有效性。
+    成功返回账号与基础资料（昵称、头像）。
+    """
+    plugin_info = _require_plugin(x_plugin_key, "verify")
+
+    is_valid, account, error_msg = verify_token_with_fallback(req.token)
+    if not is_valid or not account:
+        logger.info(
+            f"OPEN_VERIFY_TOKEN_FAIL | plugin={plugin_info['id']} | reason={error_msg}"
+        )
+        return {"valid": False, "error": error_msg or "Token 无效"}
+
+    users_db = json_db.load_data("users.json", default_data={})
+    user = users_db.get(account, {}) or {}
+
+    logger.info(f"OPEN_VERIFY_TOKEN | plugin={plugin_info['id']} | account={account}")
+    return {
+        "valid": True,
+        "account": account,
+        "name": user.get("name", account),
+        "avatar": user.get("avatarDataUrl", "") or user.get("avatar", ""),
+    }
+
+
+# ==========================================
+# 2️⃣ 余额查询：GET /api/open/balance
+# ==========================================
+@router.get("/balance")
+async def open_get_balance(
+    authorization: Optional[str] = Header(None, alias="Authorization"),
+    x_plugin_key: Optional[str] = Header(None, alias="X-Plugin-Key"),
+    db: Session = Depends(get_db),
+):
+    """
+    查询用户钱包余额。
+    认证：Authorization: Bearer <user_token> + X-Plugin-Key
+    """
+    plugin_info = _require_plugin(x_plugin_key, "balance")
+
+    if not authorization:
+        raise HTTPException(status_code=401, detail="缺少 Authorization Header")
+    parts = authorization.split(" ", 1)
+    if len(parts) != 2 or parts[0].lower() != "bearer":
+        raise HTTPException(status_code=401, detail="Authorization 格式错误，请使用 Bearer Token")
+
+    account = _resolve_user_from_token(parts[1])
+    wallet = db.query(Wallet).filter(Wallet.account == account).first()
+
+    if not wallet:
+        result = {
+            "account": account,
+            "balance": 0,
+            "earn_balance": 0,
+            "tip_balance": 0,
+            "task_balance": 0,
+            "frozen_balance": 0,
+        }
+    else:
+        result = {
+            "account": account,
+            "balance": int(wallet.balance or 0),
+            "earn_balance": int(wallet.earn_balance or 0),
+            "tip_balance": int(wallet.tip_balance or 0),
+            "task_balance": int(wallet.task_balance or 0),
+            "frozen_balance": int(wallet.frozen_balance or 0),
+        }
+
+    logger.info(
+        f"OPEN_BALANCE | plugin={plugin_info['id']} | account={account} | balance={result['balance']}"
+    )
+    return result
+
+
+# ==========================================
+# 3️⃣ 通用扣款：POST /api/open/deduct
+# ==========================================
+@router.post("/deduct")
+async def open_deduct(
+    req: DeductRequest,
+    x_plugin_key: Optional[str] = Header(None, alias="X-Plugin-Key"),
+    db: Session = Depends(get_db),
+):
+    """
+    通用扣款接口（从 wallet.balance 扣减）。
+    流程：
+      1. 校验 Plugin Key + 权限 + 限流
+      2. 解析 Token -> account
+      3. 悲观锁锁定钱包，检查余额
+      4. 扣款 + 写入 Transaction（含哈希链）
+    """
+    plugin_info = _require_plugin(x_plugin_key, "deduct")
+    account = _resolve_user_from_token(req.token)
+    amount = int(req.amount)
+
+    # 🛡️ 单笔扣款上限校验（在余额检查之前快速拒绝）
+    max_per_tx = int(plugin_info.get("max_deduct_per_tx", 100))
+    if amount > max_per_tx:
+        logger.warning(
+            f"OPEN_DEDUCT_TX_LIMIT | plugin={plugin_info['id']} | account={account} | "
+            f"amount={amount} | max={max_per_tx}"
+        )
+        return JSONResponse(status_code=403, content={
+            "error": "AMOUNT_EXCEEDS_LIMIT",
+            "message": f"单笔扣款不能超过{max_per_tx}积分",
+            "max": max_per_tx,
+        })
+
+    # 🛡️ 日扣款总额校验（仅检查不累加，扣款成功后再累加）
+    max_daily = int(plugin_info.get("max_deduct_per_day", 1000))
+    used_today = _peek_daily_used(plugin_info["id"])
+    if used_today + amount > max_daily:
+        logger.warning(
+            f"OPEN_DEDUCT_DAILY_LIMIT | plugin={plugin_info['id']} | account={account} | "
+            f"amount={amount} | used={used_today} | limit={max_daily}"
+        )
+        return JSONResponse(status_code=403, content={
+            "error": "DAILY_LIMIT_EXCEEDED",
+            "message": "今日扣款已达上限",
+            "limit": max_daily,
+            "used": used_today,
+        })
+
+    try:
+        wallet = (
+            db.query(Wallet)
+            .filter(Wallet.account == account)
+            .with_for_update()
+            .first()
+        )
+        if not wallet or int(wallet.balance or 0) < amount:
+            raise HTTPException(status_code=402, detail="余额不足")
+
+        wallet.balance = int(wallet.balance or 0) - amount
+
+        tx_id = f"OPEN_DEDUCT_{int(time.time())}_{uuid.uuid4().hex[:8]}"
+        last_tx = (
+            db.query(Transaction)
+            .filter(Transaction.account == account)
+            .order_by(Transaction.created_at.desc())
+            .first()
+        )
+        prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
+        tx_hash = _calc_tx_hash(tx_id, account, "OPEN_DEDUCT", -amount, prev_hash)
+
+        new_tx = Transaction(
+            tx_id=tx_id,
+            account=account,
+            tx_type="OPEN_DEDUCT",
+            amount=-amount,
+            related_account=plugin_info["id"],
+            item_id=req.reference_id,
+            prev_hash=prev_hash,
+            tx_hash=tx_hash,
+            description=f"[{plugin_info['name']}] {req.reason}",
+            source_plugin=plugin_info["id"],  # 🌐 来源插件标识，用于按工具结构化追踪
+        )
+        db.add(new_tx)
+        db.commit()
+        db.refresh(wallet)
+    except HTTPException:
+        db.rollback()
+        raise
+    except Exception as e:
+        db.rollback()
+        logger.error(
+            f"OPEN_DEDUCT_FAIL | plugin={plugin_info['id']} | account={account} | err={e}"
+        )
+        raise HTTPException(status_code=500, detail=f"扣款失败: {e}")
+
+    # ✅ 扣款成功后再累加日计数器，避免余额不足等失败场景占用额度
+    daily_total = _commit_daily_used(plugin_info["id"], amount)
+
+    logger.info(
+        f"OPEN_DEDUCT | plugin={plugin_info['id']} | account={account} | "
+        f"amount={amount} | tx={tx_id} | reason={req.reason} | ref={req.reference_id} | "
+        f"daily_used={daily_total}/{max_daily}"
+    )
+    return {
+        "success": True,
+        "remaining_balance": int(wallet.balance or 0),
+        "transaction_id": tx_id,
+    }
+
+
+# ==========================================
+# 4️⃣ 退款：POST /api/open/refund
+# ==========================================
+@router.post("/refund")
+async def open_refund(
+    req: RefundRequest,
+    x_plugin_key: Optional[str] = Header(None, alias="X-Plugin-Key"),
+    db: Session = Depends(get_db),
+):
+    """
+    退款接口：将之前 OPEN_DEDUCT 的金额（部分或全额）退还到用户钱包。
+
+    校验：
+      - 原交易必须存在且 tx_type == OPEN_DEDUCT
+      - 原交易归属当前 token 用户
+      - 原交易由当前插件发起（related_account == plugin_id）
+      - 累计退款金额 ≤ 原扣款金额
+    """
+    plugin_info = _require_plugin(x_plugin_key, "refund")
+    account = _resolve_user_from_token(req.token)
+    refund_amount = int(req.amount)
+
+    original_tx = (
+        db.query(Transaction).filter(Transaction.tx_id == req.transaction_id).first()
+    )
+    if not original_tx:
+        raise HTTPException(status_code=404, detail="原交易不存在")
+    if original_tx.account != account:
+        raise HTTPException(status_code=403, detail="无权对该交易执行退款")
+    if original_tx.tx_type != "OPEN_DEDUCT":
+        raise HTTPException(status_code=400, detail="该交易类型不支持开放退款")
+    if (original_tx.related_account or "") != plugin_info["id"]:
+        raise HTTPException(status_code=403, detail="不允许跨插件退款")
+
+    original_amount = abs(int(original_tx.amount or 0))
+    if refund_amount > original_amount:
+        raise HTTPException(status_code=400, detail="退款金额超过原扣款金额")
+
+    # 累计已退金额校验（防止重复退款超额）
+    refunded_records = (
+        db.query(Transaction)
+        .filter(
+            Transaction.tx_type == "OPEN_REFUND",
+            Transaction.related_account == plugin_info["id"],
+            Transaction.item_id == req.transaction_id,
+        )
+        .all()
+    )
+    already_refunded = sum(abs(int(t.amount or 0)) for t in refunded_records)
+    if already_refunded + refund_amount > original_amount:
+        raise HTTPException(
+            status_code=400,
+            detail=f"退款金额超过可退余额（已退 {already_refunded}/{original_amount}）",
+        )
+
+    try:
+        wallet = (
+            db.query(Wallet)
+            .filter(Wallet.account == account)
+            .with_for_update()
+            .first()
+        )
+        if not wallet:
+            wallet = Wallet(
+                account=account,
+                balance=0,
+                earn_balance=0,
+                tip_balance=0,
+                frozen_balance=0,
+            )
+            db.add(wallet)
+            db.flush()
+        wallet.balance = int(wallet.balance or 0) + refund_amount
+
+        tx_id = f"OPEN_REFUND_{int(time.time())}_{uuid.uuid4().hex[:8]}"
+        last_tx = (
+            db.query(Transaction)
+            .filter(Transaction.account == account)
+            .order_by(Transaction.created_at.desc())
+            .first()
+        )
+        prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
+        tx_hash = _calc_tx_hash(tx_id, account, "OPEN_REFUND", refund_amount, prev_hash)
+
+        new_tx = Transaction(
+            tx_id=tx_id,
+            account=account,
+            tx_type="OPEN_REFUND",
+            amount=refund_amount,
+            related_account=plugin_info["id"],
+            item_id=req.transaction_id,  # 关联原扣款交易
+            prev_hash=prev_hash,
+            tx_hash=tx_hash,
+            description=f"[{plugin_info['name']}] 退款 {req.transaction_id}",
+            source_plugin=plugin_info["id"],  # 🌐 来源插件标识
+        )
+        db.add(new_tx)
+        db.commit()
+        db.refresh(wallet)
+    except HTTPException:
+        db.rollback()
+        raise
+    except Exception as e:
+        db.rollback()
+        logger.error(
+            f"OPEN_REFUND_FAIL | plugin={plugin_info['id']} | account={account} | err={e}"
+        )
+        raise HTTPException(status_code=500, detail=f"退款失败: {e}")
+
+    logger.info(
+        f"OPEN_REFUND | plugin={plugin_info['id']} | account={account} | "
+        f"amount={refund_amount} | tx={tx_id} | original={req.transaction_id}"
+    )
+    return {
+        "success": True,
+        "remaining_balance": int(wallet.balance or 0),
+        "transaction_id": tx_id,
+    }
+
+
+# ==========================================
+# 5️⃣ 消费记录查询：GET /api/open/transactions
+# ==========================================
+@router.get("/transactions")
+async def open_list_transactions(
+    authorization: Optional[str] = Header(None, alias="Authorization"),
+    x_plugin_key: Optional[str] = Header(None, alias="X-Plugin-Key"),
+    page: int = Query(1, ge=1, description="页码（从 1 开始）"),
+    page_size: int = Query(20, ge=1, le=100, description="每页条数（1-100）"),
+    start_time: Optional[int] = Query(None, description="起始时间（unix 秒）"),
+    end_time: Optional[int] = Query(None, description="结束时间（unix 秒）"),
+    tx_type: Optional[str] = Query(None, description="交易类型，如 OPEN_DEDUCT / OPEN_REFUND"),
+    db: Session = Depends(get_db),
+):
+    """
+    查询当前用户由当前插件产生的消费/退款记录。
+
+    认证：Authorization: Bearer <user_token> + X-Plugin-Key
+    安全：仅返回 source_plugin == 当前 plugin_id 的记录，防止跨插件读取。
+    """
+    plugin_info = _require_plugin(x_plugin_key, "verify")
+
+    if not authorization:
+        raise HTTPException(status_code=401, detail="缺少 Authorization Header")
+    parts = authorization.split(" ", 1)
+    if len(parts) != 2 or parts[0].lower() != "bearer":
+        raise HTTPException(status_code=401, detail="Authorization 格式错误，请使用 Bearer Token")
+    account = _resolve_user_from_token(parts[1])
+
+    plugin_id = plugin_info["id"]
+
+    # 构造查询：按 source_plugin 过滤；兼容历史 NULL 数据走 related_account 兜底
+    q = db.query(Transaction).filter(
+        Transaction.account == account,
+        ((Transaction.source_plugin == plugin_id)
+         | ((Transaction.source_plugin.is_(None)) & (Transaction.related_account == plugin_id))),
+    )
+
+    if tx_type:
+        q = q.filter(Transaction.tx_type == tx_type)
+    if start_time is not None:
+        q = q.filter(Transaction.created_at >= datetime.datetime.utcfromtimestamp(int(start_time)))
+    if end_time is not None:
+        q = q.filter(Transaction.created_at <= datetime.datetime.utcfromtimestamp(int(end_time)))
+
+    total = q.count()
+    rows = (
+        q.order_by(Transaction.created_at.desc())
+        .offset((page - 1) * page_size)
+        .limit(page_size)
+        .all()
+    )
+
+    items = []
+    for t in rows:
+        created_ts = int(t.created_at.timestamp()) if t.created_at else 0
+        # reason 从 description 中解析（去掉 [插件名] 前缀），兼容旧记录
+        reason = (t.description or "").replace(f"[{plugin_info['name']}] ", "").strip() or None
+        items.append({
+            "id": t.tx_id,
+            "type": t.tx_type,
+            "amount": int(t.amount or 0),
+            "reason": reason,
+            "reference_id": t.item_id,
+            "created_at": created_ts,
+            "description": t.description or "",
+        })
+
+    logger.info(
+        f"OPEN_LIST_TX | plugin={plugin_id} | account={account} | "
+        f"page={page}/{page_size} | total={total}"
+    )
+    return {
+        "success": True,
+        "transactions": items,
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+    }
+
+
+# ==========================================
+# 6️⃣ 消费汇总：GET /api/open/usage-summary
+# ==========================================
+@router.get("/usage-summary")
+async def open_usage_summary(
+    x_plugin_key: Optional[str] = Header(None, alias="X-Plugin-Key"),
+    account: Optional[str] = Query(None, description="指定用户账号，不传则统计所有用户"),
+    period: str = Query("daily", description="聚合粒度：daily / monthly / total"),
+    start_time: Optional[int] = Query(None, description="起始时间（unix 秒）"),
+    end_time: Optional[int] = Query(None, description="结束时间（unix 秒）"),
+    db: Session = Depends(get_db),
+):
+    """
+    本插件维度的消费汇总统计。
+
+    返回：总扣款 / 总退款 / 总笔数 / 净消费 + 按 period 分桶的明细。
+    安全：仅统计 source_plugin == 当前 plugin_id 的记录。
+    """
+    plugin_info = _require_plugin(x_plugin_key, "verify")
+    plugin_id = plugin_info["id"]
+
+    period = (period or "daily").lower()
+    if period not in ("daily", "monthly", "total"):
+        raise HTTPException(status_code=400, detail="period 仅支持 daily / monthly / total")
+
+    base = db.query(Transaction).filter(
+        ((Transaction.source_plugin == plugin_id)
+         | ((Transaction.source_plugin.is_(None)) & (Transaction.related_account == plugin_id))),
+        Transaction.tx_type.in_(["OPEN_DEDUCT", "OPEN_REFUND"]),
+    )
+    if account:
+        base = base.filter(Transaction.account == account)
+    if start_time is not None:
+        base = base.filter(Transaction.created_at >= datetime.datetime.utcfromtimestamp(int(start_time)))
+    if end_time is not None:
+        base = base.filter(Transaction.created_at <= datetime.datetime.utcfromtimestamp(int(end_time)))
+
+    rows = base.all()
+
+    total_deduct = 0
+    total_refund = 0
+    total_count = len(rows)
+    bucket: dict = {}  # key -> {"deduct":int,"refund":int,"count":int}
+
+    for t in rows:
+        amt = int(t.amount or 0)
+        if t.tx_type == "OPEN_DEDUCT":
+            total_deduct += abs(amt)
+        elif t.tx_type == "OPEN_REFUND":
+            total_refund += abs(amt)
+
+        if period == "total":
+            continue
+        if not t.created_at:
+            continue
+        if period == "daily":
+            key = t.created_at.strftime("%Y-%m-%d")
+        else:  # monthly
+            key = t.created_at.strftime("%Y-%m")
+
+        b = bucket.setdefault(key, {"deduct": 0, "refund": 0, "count": 0})
+        if t.tx_type == "OPEN_DEDUCT":
+            b["deduct"] += abs(amt)
+        elif t.tx_type == "OPEN_REFUND":
+            b["refund"] += abs(amt)
+        b["count"] += 1
+
+    breakdown = []
+    if period != "total":
+        for key in sorted(bucket.keys(), reverse=True):
+            v = bucket[key]
+            label = "date" if period == "daily" else "month"
+            breakdown.append({
+                label: key,
+                "deduct": v["deduct"],
+                "refund": v["refund"],
+                "count": v["count"],
+            })
+
+    logger.info(
+        f"OPEN_USAGE_SUMMARY | plugin={plugin_id} | account={account or '*'} | "
+        f"period={period} | tx_count={total_count}"
+    )
+    return {
+        "success": True,
+        "plugin_id": plugin_id,
+        "plugin_name": plugin_info.get("name", plugin_id),
+        "summary": {
+            "total_deduct": total_deduct,
+            "total_refund": total_refund,
+            "total_transactions": total_count,
+            "net_consumption": total_deduct - total_refund,
+        },
+        "breakdown": breakdown,
+    }