Upload 5 files

app.py

@@ -31,7 +31,8 @@ from rate_limit import start_shared_work, stop_shared_work, close_shared_work
 from script_execution import start_admin_scripts, stop_admin_scripts, close_admin_scripts
 from comment_store import start_comment_reads, stop_comment_reads, close_comment_reads
 from periodic_tasks import start_periodic_task, request_periodic_stop, wait_periodic_tasks
-from owned_thread import run_owned_thread
+from owned_thread import run_owned_thread, OwnedThreadGroup, owned_session_endpoint
+from contextlib import contextmanager
 from resource_work import start_resource_work, stop_resource_work, close_resource_work
 from shutdown_lifecycle import (prepare_application_shutdown, run_application_shutdown,
                                 shutdown_cancel_requested, ShutdownFailures)
@@ -86,6 +87,25 @@ from models_sql import SystemConfig  # 🧹 P2清理：Ownership 在本文件无
 from fastapi.exceptions import RequestValidationError
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
+MAX_CONFIG_SQL_WORK = 8
+_config_sql_work = OwnedThreadGroup(MAX_CONFIG_SQL_WORK,
+    lambda: HTTPException(status_code=503, detail="service_unavailable"))
+_config_sql_lock = threading.Lock()
+start_config_sql = _config_sql_work.start
+stop_config_sql = _config_sql_work.stop
+close_config_sql = _config_sql_work.close
+
+
+@contextmanager
+def _config_sql_session():
+    with _config_sql_lock:
+        with SessionLocal() as session:
+            yield session
+
+
+_config_endpoint = owned_session_endpoint(_config_sql_work, _config_sql_session, 'db')
+
+
 app = FastAPI(
     title="ComfyUI Ranking API",
     description="ComfyUI 社区排名系统 API",
@@ -295,6 +315,7 @@ async def on_startup():
             start_identity_io()
             start_shared_work()
             start_open_sql()
+            start_config_sql()
             start_wallet_sql()
             start_resource_work()
             start_resource_reads()
@@ -365,7 +386,7 @@ async def _shutdown_resources():
         ('stop_wallet_sql', stop_wallet_sql),
         ('stop_resource_reads', stop_resource_reads), ('stop_comment_writes', stop_comment_writes),
         ('stop_posts', stop_posts_io), ('stop_prompts', stop_prompts_io), ('stop_tasks', stop_tasks_io), ('stop_proxy_access', stop_proxy_access),
-        ('stop_messages', stop_message_io), ('stop_resources', stop_resource_work), ('stop_open_sql', stop_open_sql), ('stop_shared', stop_shared_work),
+        ('stop_messages', stop_message_io), ('stop_resources', stop_resource_work), ('stop_open_sql', stop_open_sql), ('stop_config_sql', stop_config_sql), ('stop_shared', stop_shared_work),
         ('stop_identity', stop_identity_io), ('stop_auth', stop_auth_io),
         ('stop_verification', stop_verification_io), ('stop_scripts', stop_admin_scripts),
         ('stop_comments', stop_comment_reads),
@@ -382,7 +403,7 @@ async def _shutdown_resources():
         ('close_wallet_sql', close_wallet_sql, ()),
         ('close_resource_reads', close_resource_reads, ()), ('close_comment_writes', close_comment_writes, ()),
         ('close_posts', close_posts_io, ()), ('close_prompts', close_prompts_io, ()), ('close_tasks', close_tasks_io, ()), ('close_proxy_access', close_proxy_access, ()),
-        ('close_messages', close_message_io, ()), ('close_resources', close_resource_work, ()), ('close_open_sql', close_open_sql, ()), ('close_shared', close_shared_work, ()),
+        ('close_messages', close_message_io, ()), ('close_resources', close_resource_work, ()), ('close_open_sql', close_open_sql, ()), ('close_config_sql', close_config_sql, ()), ('close_shared', close_shared_work, ()),
         ('close_identity', close_identity_io, ()), ('close_auth', close_auth_io, ()),
         ('close_verification', close_verification_io, ()), ('close_profile', close_profile_io, ()),
         ('close_scripts', close_admin_scripts, ()), ('close_comments', close_comment_reads, ()),
@@ -488,7 +509,7 @@ def _probe_skill_repository(item, token):
 
 @app.post("/api/validate_resource")
 @shared_limit(30)
-def validate_resource(req_data: ValidateResourceRequest, request: Request, sql_db: Session = Depends(get_db), current_user: str = Depends(require_auth)):
+def validate_resource(req_data: ValidateResourceRequest, request: Request, current_user: str = Depends(require_auth)):
     return probe_resource(req_data.item_id, store=db, download=hf_hub_download,
                           open_url=open_github_probe, token_selector=select_github_token,
                           error_response=create_error_response, skill_probe=_probe_skill_repository)
@@ -609,7 +630,8 @@ def _save_system_config(config: dict, db: Session = None, updated_by: str = None
 
 
 @app.get("/api/admin/config/{config_key}")
-async def get_system_config(config_key: str, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
+@_config_endpoint
+def get_system_config(config_key: str, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     """ 🔒 获取系统配置（仅管理员） """
     # 🔒 权限验证
     if not _is_admin(current_user):
@@ -644,7 +666,8 @@ DEFAULT_VERSION_CONFIG = {
 
 
 @app.get("/api/public/project-version")
-async def get_public_project_version(db: Session = Depends(get_db)):
+@_config_endpoint
+def get_public_project_version(db: Session = Depends(get_db)):
     """获取项目版本信息（公开接口，无需认证）"""
     config = _load_system_config(db)
     version_data = config.get("project_version", DEFAULT_VERSION_CONFIG)
@@ -652,7 +675,8 @@ async def get_public_project_version(db: Session = Depends(get_db)):
 
 
 @app.get("/api/public/banner_config")
-async def get_public_banner_config(db: Session = Depends(get_db)):
+@_config_endpoint
+def get_public_banner_config(db: Session = Depends(get_db)):
     """公开接口：获取广告横幅配置（所有已登录/未登录用户均可访问）"""
     try:
         config = _load_system_config(db, strict=True)
@@ -680,7 +704,8 @@ async def get_public_banner_config(db: Session = Depends(get_db)):
 
 
 @app.put("/api/admin/config/{config_key}")
-async def set_system_config(config_key: str, value: dict, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
+@_config_endpoint
+def set_system_config(config_key: str, value: dict, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     """ 🔒 设置系统配置（仅管理员） """
     # 🔒 权限验证
     if not _is_admin(current_user):

router_admin_plugins.py

@@ -1,420 +1,428 @@
-# 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 hmac
-import hashlib
-import secrets
-import logging
-import datetime
-from typing import Optional, List
-
-from fastapi import APIRouter, Depends, HTTPException, Request
-from rate_limit import shared_limit
-from fastapi.responses import JSONResponse
-from sqlalchemy.orm import Session
-from pydantic import BaseModel, Field, field_validator  # 🧹 P2修复：pydantic v2 已废弃 v1 风格 validator
-
-from database_sql import get_db
-from models_sql import RegisteredPlugin
-from plugin_permissions import DEFAULT_PERMISSIONS, parse_permissions
-from 安全认证 import require_auth, is_admin, JWT_SECRET
-
-logger = logging.getLogger("ComfyUI-Ranking.AdminPlugins")
-
-router = APIRouter()
-
-# ==========================================
-# 🔧 工具函数
-# ==========================================
-
-# plugin_id 合法性正则：3-50字符，仅允许小写字母、数字、下划线
-_PLUGIN_ID_PATTERN = re.compile(r"^[a-z0-9_]{3,50}$")
-
-def _audit_actor_id(account: str) -> str:
-    """稳定关联管理员审计事件，但不把可识别账号写入运行日志。"""
-    return hmac.new(
-        JWT_SECRET.encode("utf-8"), str(account).encode("utf-8"), hashlib.sha256
-    ).hexdigest()[:16]
-
-
-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"插件注册表热更新失败: error_type={type(e).__name__}")
-
-
-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 = parse_permissions(p.permissions)
-        permissions_valid = True
-    except ValueError:
-        permissions = []
-        permissions_valid = False
-
-    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,
-        "permissions_valid": permissions_valid,
-        "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_factory=lambda: list(DEFAULT_PERMISSIONS))
-
-    @field_validator("plugin_id")
-    @classmethod
-    def _validate_plugin_id(cls, v):
-        if not _PLUGIN_ID_PATTERN.match(v):
-            raise ValueError("plugin_id 仅允许 [a-z0-9_]，长度 3-50")
-        return v
-
-    @field_validator("permissions")
-    @classmethod
-    def _validate_permissions(cls, v):
-        if v is None:
-            return list(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
-
-    @field_validator("permissions")
-    @classmethod
-    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.error(f"列出插件失败: error_type={type(e).__name__}")
-        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")
-@shared_limit(20)
-async def create_plugin(
-    request: Request,
-    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.error(f"创建插件失败: error_type={type(e).__name__}")
-        return JSONResponse(status_code=500, content={"success": False, "error": "创建插件失败"})
-
-    logger.info(f"ADMIN_PLUGIN_CREATE | actor={_audit_actor_id(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}")
-@shared_limit(30)
-async def update_plugin(
-    request: Request,
-    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.error(f"更新插件失败: error_type={type(e).__name__}")
-        return JSONResponse(status_code=500, content={"success": False, "error": "更新插件失败"})
-
-    logger.info(
-        f"ADMIN_PLUGIN_UPDATE | actor={_audit_actor_id(account)} | plugin_id={plugin_id} | "
-        f"fields={','.join(sorted(update_fields.keys()))}"
-    )
-    _notify_registry_reload()
-
-    return JSONResponse(content={
-        "success": True,
-        "message": "插件更新成功",
-        "plugin": _serialize_plugin(plugin),
-    })
-
-
-@router.delete("/api/admin/plugins/{plugin_id}")
-@shared_limit(10)
-async def delete_plugin(
-    request: Request,
-    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.error(f"删除插件失败: error_type={type(e).__name__}")
-        return JSONResponse(status_code=500, content={"success": False, "error": "删除插件失败"})
-
-    logger.info(f"ADMIN_PLUGIN_DELETE | actor={_audit_actor_id(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")
-@shared_limit(5)
-async def rotate_plugin_key(
-    request: Request,
-    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.error(f"轮换 plugin_key 失败: error_type={type(e).__name__}")
-        return JSONResponse(status_code=500, content={"success": False, "error": "轮换 plugin_key 失败"})
-
-    logger.info(f"ADMIN_PLUGIN_ROTATE_KEY | actor={_audit_actor_id(account)} | plugin_id={plugin_id}")
-    _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")
-@shared_limit(20)
-async def toggle_plugin(
-    request: Request,
-    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.error(f"切换插件状态失败: error_type={type(e).__name__}")
-        return JSONResponse(status_code=500, content={"success": False, "error": "切换插件状态失败"})
-
-    logger.info(
-        f"ADMIN_PLUGIN_TOGGLE | actor={_audit_actor_id(account)} | plugin_id={plugin_id} | "
-        f"enabled={bool(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_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 hmac
+import hashlib
+import secrets
+import logging
+import datetime
+from typing import Optional, List
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from rate_limit import shared_limit
+from fastapi.responses import JSONResponse
+from sqlalchemy.orm import Session
+from pydantic import BaseModel, Field, field_validator  # 🧹 P2修复：pydantic v2 已废弃 v1 风格 validator
+
+from database_sql import get_db
+from router_open_api import open_sql_endpoint
+from models_sql import RegisteredPlugin
+from plugin_permissions import DEFAULT_PERMISSIONS, parse_permissions
+from 安全认证 import require_auth, is_admin, JWT_SECRET
+
+logger = logging.getLogger("ComfyUI-Ranking.AdminPlugins")
+
+router = APIRouter()
+
+# ==========================================
+# 🔧 工具函数
+# ==========================================
+
+# plugin_id 合法性正则：3-50字符，仅允许小写字母、数字、下划线
+_PLUGIN_ID_PATTERN = re.compile(r"^[a-z0-9_]{3,50}$")
+
+def _audit_actor_id(account: str) -> str:
+    """稳定关联管理员审计事件，但不把可识别账号写入运行日志。"""
+    return hmac.new(
+        JWT_SECRET.encode("utf-8"), str(account).encode("utf-8"), hashlib.sha256
+    ).hexdigest()[:16]
+
+
+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"插件注册表热更新失败: error_type={type(e).__name__}")
+
+
+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 = parse_permissions(p.permissions)
+        permissions_valid = True
+    except ValueError:
+        permissions = []
+        permissions_valid = False
+
+    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,
+        "permissions_valid": permissions_valid,
+        "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_factory=lambda: list(DEFAULT_PERMISSIONS))
+
+    @field_validator("plugin_id")
+    @classmethod
+    def _validate_plugin_id(cls, v):
+        if not _PLUGIN_ID_PATTERN.match(v):
+            raise ValueError("plugin_id 仅允许 [a-z0-9_]，长度 3-50")
+        return v
+
+    @field_validator("permissions")
+    @classmethod
+    def _validate_permissions(cls, v):
+        if v is None:
+            return list(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
+
+    @field_validator("permissions")
+    @classmethod
+    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")
+@open_sql_endpoint
+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.error(f"列出插件失败: error_type={type(e).__name__}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "服务器内部错误"})
+
+
+@router.get("/api/admin/plugins/{plugin_id}")
+@open_sql_endpoint
+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")
+@shared_limit(20)
+@open_sql_endpoint
+def create_plugin(
+    request: Request,
+    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.error(f"创建插件失败: error_type={type(e).__name__}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "创建插件失败"})
+
+    logger.info(f"ADMIN_PLUGIN_CREATE | actor={_audit_actor_id(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}")
+@shared_limit(30)
+@open_sql_endpoint
+def update_plugin(
+    request: Request,
+    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.error(f"更新插件失败: error_type={type(e).__name__}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "更新插件失败"})
+
+    logger.info(
+        f"ADMIN_PLUGIN_UPDATE | actor={_audit_actor_id(account)} | plugin_id={plugin_id} | "
+        f"fields={','.join(sorted(update_fields.keys()))}"
+    )
+    _notify_registry_reload()
+
+    return JSONResponse(content={
+        "success": True,
+        "message": "插件更新成功",
+        "plugin": _serialize_plugin(plugin),
+    })
+
+
+@router.delete("/api/admin/plugins/{plugin_id}")
+@shared_limit(10)
+@open_sql_endpoint
+def delete_plugin(
+    request: Request,
+    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.error(f"删除插件失败: error_type={type(e).__name__}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "删除插件失败"})
+
+    logger.info(f"ADMIN_PLUGIN_DELETE | actor={_audit_actor_id(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")
+@shared_limit(5)
+@open_sql_endpoint
+def rotate_plugin_key(
+    request: Request,
+    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.error(f"轮换 plugin_key 失败: error_type={type(e).__name__}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "轮换 plugin_key 失败"})
+
+    logger.info(f"ADMIN_PLUGIN_ROTATE_KEY | actor={_audit_actor_id(account)} | plugin_id={plugin_id}")
+    _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")
+@shared_limit(20)
+@open_sql_endpoint
+def toggle_plugin(
+    request: Request,
+    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.error(f"切换插件状态失败: error_type={type(e).__name__}")
+        return JSONResponse(status_code=500, content={"success": False, "error": "切换插件状态失败"})
+
+    logger.info(
+        f"ADMIN_PLUGIN_TOGGLE | actor={_audit_actor_id(account)} | plugin_id={plugin_id} | "
+        f"enabled={bool(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

@@ -1,688 +1,702 @@
-# router_open_api.py
-# ==========================================
-# 🌐 通用开放 API 模块（供第三方插件调用）
-# ==========================================
-# 作用：对外提供受 Plugin Key 保护的通用接口，
-#       供 NodeCraft AI 等第三方插件复用 ComfyUI-Ranking 账号体系。
-# 关联文件：
-#   - 安全认证.py        (verify_token_with_fallback, require_auth)
-#   - database_sql.py    (SessionLocal)
-#   - models_sql.py      (Wallet, Transaction)
-#   - 数据库连接.py       (users.json 读写)
-# 安全机制：
-#   - X-Plugin-Key Header 鉴权
-#   - 基于 plugin_id 的 SQL 持久化限流（每分钟窗口，多 worker 共享）
-#   - 所有扣款/退款操作记录到 Transaction 流水（含哈希链）
-# ==========================================
-
-import os
-import json
-import time
-import uuid
-import datetime
-import hashlib
-import logging
-from typing import Optional
-
-from fastapi import APIRouter, Header, HTTPException, Query
-from fastapi.responses import JSONResponse
-from sqlalchemy.orm import Session
-from sqlalchemy import func, text
-from pydantic import BaseModel, Field
-
-from database_sql import SessionLocal
-from models_sql import Wallet, Transaction, RegisteredPlugin, OpenApiRateLimitBucket
-from open_transaction import lock_open_wallet, positive_integer, validate_live_plugin, quota_total
-from open_transaction import open_operation_identity, read_open_operation, save_open_operation
-from plugin_permissions import parse_permissions
-from tip_store import run_tip_effect
-from auth_io import run_identity_io
-from owned_thread import OwnedThreadGroup
-from 安全认证 import verify_token_with_fallback
-import 数据库连接 as json_db
-
-logger = logging.getLogger("ComfyUI-Ranking.OpenAPI")
-
-router = APIRouter(prefix="/api/open", tags=["open-api"])
-
-
-MAX_OPEN_SQL_WORK = 8
-_open_sql_work = OwnedThreadGroup(MAX_OPEN_SQL_WORK,
-    lambda: HTTPException(status_code=503, detail="插件限流服务暂不可用"))
-start_open_sql = _open_sql_work.start
-stop_open_sql = _open_sql_work.stop
-close_open_sql = _open_sql_work.close
-
-
-async def _run_open_sql(function, *args, **kwargs):
-    def execute():
-        with SessionLocal() as session:
-            return function(*args, db=session, **kwargs)
-    return await _open_sql_work.run(execute)
-
-
-# ═══════════════ 插件注册表（从数据库动态加载） ═══════════════
-# 反向索引：plugin_key -> {id, key, name, rate_limit, permissions, max_deduct_per_tx, max_deduct_per_day}
-_PLUGIN_KEY_INDEX: dict = {}
-_INSECURE_LEGACY_PLUGIN_KEYS = frozenset({"nca_2026_nodecraft_ai_plugin"})
-
-
-def load_plugins_from_db():
-    """启动时从数据库加载所有启用的插件到内存索引"""
-    global _PLUGIN_KEY_INDEX
-
-    db = None
-    try:
-        db = SessionLocal()
-        # 首次启动时仅在显式配置高熵 Key 时 seed 默认插件，不再使用源码内固定凭证。
-        if not db.query(RegisteredPlugin).first():
-            default_key = os.environ.get("NCA_PLUGIN_KEY", "").strip()
-            if len(default_key) >= 32 and default_key not in _INSECURE_LEGACY_PLUGIN_KEYS:
-                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")
-            else:
-                logger.warning("NCA_PLUGIN_KEY 缺失或不符合安全要求，跳过默认插件 seed")
-
-        plugins = db.query(RegisteredPlugin).filter(RegisteredPlugin.enabled == True).all()
-
-        # 构建内存索引
-        new_index = {}
-        for p in plugins:
-            # 固定旧 Key 已随源代码泄露，不能继续接受；管理员轮换成新 Key 后即可恢复。
-            if p.plugin_key in _INSECURE_LEGACY_PLUGIN_KEYS:
-                logger.critical(f"拒绝加载已泄露的默认 Plugin Key: plugin_id={p.plugin_id}")
-                continue
-            try:
-                perms = parse_permissions(p.permissions)
-            except ValueError:
-                logger.warning("OPEN_API_PLUGIN_PERMISSIONS_INVALID")
-                continue
-            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:
-        # 数据源不可信/不可用时 fail closed：清空内存索引，撤销旧 Key 的继续访问能力。
-        _PLUGIN_KEY_INDEX = {}
-        logger.error(f"❌ 加载插件注册表失败: error_type={type(e).__name__}; 插件 API 已 fail closed")
-    finally:
-        if db is not None:
-            try:
-                db.close()
-            except Exception as e:
-                _PLUGIN_KEY_INDEX = {}
-                logger.error(
-                    f"关闭插件注册表会话失败: error_type={type(e).__name__}; 插件 API 已 fail closed"
-                )
-
-
-def reload_plugin_registry():
-    """热更新插件注册表（管理操作后调用）"""
-    load_plugins_from_db()
-
-
-# ==========================================
-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("OPEN_API_INVALID_KEY")
-        raise HTTPException(status_code=403, detail="Plugin Key 无效或未注册")
-    return info
-
-
-# ==========================================
-# 🚦 限流：SQL 分钟计数器，多 worker 共享
-# ==========================================
-def _check_rate_limit(plugin_info: dict, permission: str, db: Session) -> None:
-    """
-    校验插件权限并执行限流。
-
-    - 权限不足 -> 403
-    - 触发限流 -> 429
-    """
-    plugin_id = plugin_info["id"]
-
-    now = int(time.time())
-    window_start = now - (now % 60)
-    try:
-        # 对配置列执行 no-op UPDATE：PostgreSQL 获取行锁，SQLite 也会取得写事务锁；
-        # 再 SELECT FOR UPDATE，保证首次建桶和窗口递增串行，而不是只依赖进程锁。
-        db.execute(
-            text("UPDATE registered_plugins SET rate_limit = rate_limit WHERE plugin_id = :plugin_id"),
-            {"plugin_id": plugin_id},
-        )
-        plugin_row = (db.query(RegisteredPlugin)
-            .filter(RegisteredPlugin.plugin_id == plugin_id)
-            .populate_existing()
-            .with_for_update()
-            .first())
-        validate_live_plugin(plugin_row, plugin_info.get("key"), permission)
-        rate_limit = positive_integer(plugin_row.rate_limit)
-        bucket = (db.query(OpenApiRateLimitBucket)
-            .filter(OpenApiRateLimitBucket.plugin_id == plugin_id)
-            .with_for_update()
-            .first())
-        if not bucket:
-            bucket = OpenApiRateLimitBucket(
-                plugin_id=plugin_id, window_start=window_start, request_count=0
-            )
-            db.add(bucket)
-            db.flush()
-        if bucket.window_start != window_start:
-            bucket.window_start = window_start
-            bucket.request_count = 0
-        if bucket.request_count >= rate_limit:
-            db.rollback()
-            raise HTTPException(
-                status_code=429,
-                detail=f"请求过于频繁，限流 {rate_limit}/分钟，请稍后再试",
-            )
-        bucket.request_count += 1
-        db.commit()
-    except HTTPException:
-        db.rollback()
-        raise
-    except Exception as exc:
-        db.rollback()
-        logger.error(f"OPEN_API_RATE_LIMIT_DB_FAIL | plugin={plugin_id} | error_type={type(exc).__name__}")
-        raise HTTPException(status_code=503, detail="插件限流服务暂不可用")
-
-
-def _require_plugin(plugin_key: Optional[str], permission: str, db: Session) -> dict:
-    """统一入口：校验 Plugin Key + 权限 + 限流"""
-    info = verify_plugin_key(plugin_key or "")
-    _check_rate_limit(info, permission, db)
-    return info
-
-
-async 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 = await run_identity_io(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, strict=True, description="扣款金额（正整数）")
-    reason: str = Field(..., description="业务标识/扣款原因")
-    reference_id: Optional[str] = Field(None, description="业务侧参考 ID")
-    request_id: Optional[str] = Field(None, strict=True, min_length=8, max_length=128,
-        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]*$", description="同次扣款重试使用相同请求标识")
-
-
-class RefundRequest(BaseModel):
-    token: str = Field(..., description="用户 JWT Token")
-    transaction_id: str = Field(..., description="原扣款交易 ID")
-    amount: int = Field(..., gt=0, strict=True, description="退款金额（正整数）")
-    request_id: Optional[str] = Field(None, strict=True, min_length=8, max_length=128,
-        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]*$", 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 = await _run_open_sql(_require_plugin, x_plugin_key, "verify")
-
-    is_valid, account, error_msg = await run_identity_io(verify_token_with_fallback, req.token)
-    if not is_valid or not account:
-        logger.info(f"OPEN_VERIFY_TOKEN_FAIL | plugin={plugin_info['id']} | category=invalid_token")
-        return {"valid": False, "error": error_msg or "Token 无效"}
-
-    users_db = await run_identity_io(json_db.load_data, "users.json", default_data={})
-    user = users_db.get(account, {}) or {}
-
-    logger.info(f"OPEN_VERIFY_TOKEN | plugin={plugin_info['id']}")
-    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"),
-):
-    """
-    查询用户钱包余额。
-    认证：Authorization: Bearer <user_token> + X-Plugin-Key
-    """
-    plugin_info = await _run_open_sql(_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 = await _resolve_user_from_token(parts[1])
-    return await _run_open_sql(_read_open_balance, account, plugin_info)
-
-
-# ==========================================
-# 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"),
-):
-    """
-    通用扣款接口（从 wallet.balance 扣减）。
-    流程：
-      1. 校验 Plugin Key + 权限 + 限流
-      2. 解析 Token -> account
-      3. 共用写锁锁定钱包与实时插件配置，检查额度/余额
-      4. 扣款 + 写入 Transaction（含哈希链）
-    """
-    plugin_info = await _run_open_sql(_require_plugin, x_plugin_key, "deduct")
-    account = await _resolve_user_from_token(req.token)
-    return await _run_open_sql(_deduct_open_balance, req, account, plugin_info, x_plugin_key)
-
-
-# ==========================================
-# 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"),
-):
-    """
-    退款接口：将之前 OPEN_DEDUCT 的金额（部分或全额）退还到用户钱包。
-
-    校验：
-      - 原交易必须存在且 tx_type == OPEN_DEDUCT
-      - 原交易归属当前 token 用户
-      - 原交易由当前插件发起（related_account == plugin_id）
-      - 累计退款金额 ≤ 原扣款金额
-    """
-    plugin_info = await _run_open_sql(_require_plugin, x_plugin_key, "refund")
-    account = await _resolve_user_from_token(req.token)
-    return await _run_open_sql(_refund_open_balance, req, account, plugin_info, x_plugin_key)
-
-
-# ==========================================
-# 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"),
-):
-    """
-    查询当前用户由当前插件产生的消费/退款记录。
-
-    认证：Authorization: Bearer <user_token> + X-Plugin-Key
-    安全：仅返回 source_plugin == 当前 plugin_id 的记录，防止跨插件读取。
-    """
-    plugin_info = await _run_open_sql(_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 = await _resolve_user_from_token(parts[1])
-
-    return await _run_open_sql(_read_open_transactions, account, plugin_info, page, page_size, start_time, end_time, tx_type)
-
-
-# ==========================================
-# 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 秒）"),
-):
-    """
-    本插件维度的消费汇总统计。
-
-    返回：总扣款 / 总退款 / 总笔数 / 净消费 + 按 period 分桶的明细。
-    安全：仅统计 source_plugin == 当前 plugin_id 的记录。
-    """
-    plugin_info = await _run_open_sql(_require_plugin, x_plugin_key, "verify")
-    return await _run_open_sql(_read_open_usage, plugin_info, account, period, start_time, end_time)
-
-
-def _read_open_balance(account, plugin_info, db):
-    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']}")
-    return result
-
-
-def _deduct_open_balance(req, account, plugin_info, x_plugin_key, db):
-    amount = req.amount
-    if isinstance(amount, bool) or not isinstance(amount, int) or amount <= 0:
-        raise HTTPException(status_code=422, detail="扣款金额必须为正整数")
-    operation = open_operation_identity(req.request_id, account, plugin_info['id'], 'OPEN_DEDUCT',
-        [amount, req.reason, req.reference_id])
-    try:
-        wallet, plugin_row = lock_open_wallet(db, account, plugin_info["id"], x_plugin_key, "deduct")
-        replay = read_open_operation(db, operation, account, plugin_info['id'], 'OPEN_DEDUCT', -amount, req.reference_id)
-        if replay is not None:
-            return replay
-        max_per_tx = positive_integer(plugin_row.max_deduct_per_tx)
-        max_daily = positive_integer(plugin_row.max_deduct_per_day)
-        if amount > max_per_tx:
-            db.rollback()
-            return JSONResponse(status_code=403, content={
-                "error": "AMOUNT_EXCEEDS_LIMIT", "message": f"单笔扣款不能超过{max_per_tx}积分", "max": max_per_tx})
-        now = datetime.datetime.utcnow()
-        day_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
-        used_today = db.query(func.coalesce(func.sum(func.abs(Transaction.amount)), 0)).filter(
-            Transaction.tx_type == "OPEN_DEDUCT", Transaction.source_plugin == plugin_info["id"],
-            Transaction.created_at >= day_start,
-            Transaction.created_at < day_start + datetime.timedelta(days=1)).scalar()
-        used_today = quota_total(used_today)
-        if used_today + amount > max_daily:
-            db.rollback()
-            return JSONResponse(status_code=403, content={
-                "error": "DAILY_LIMIT_EXCEEDED", "message": "今日扣款已达上限", "limit": max_daily, "used": used_today})
-        if wallet is None:
-            raise HTTPException(status_code=402, detail="余额不足")
-        if isinstance(wallet.balance, bool) or not isinstance(wallet.balance, int):
-            raise ValueError("invalid wallet balance")
-        if wallet.balance < amount:
-            raise HTTPException(status_code=402, detail="余额不足")
-        wallet.balance -= amount
-        tx_id = f"OPEN_DEDUCT_{operation[0] if operation is not None else uuid.uuid4().hex}"
-        last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(
-            Transaction.created_at.desc(), Transaction.tx_id.desc()).with_for_update().first()
-        prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
-        db.add(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=_calc_tx_hash(tx_id, account, "OPEN_DEDUCT", -amount, prev_hash),
-            description=f"[{plugin_row.name}] {req.reason}", source_plugin=plugin_info["id"]))
-        response = {"success": True, "remaining_balance": wallet.balance, "transaction_id": tx_id}
-        save_open_operation(db, operation, response)
-        db.commit()
-    except HTTPException:
-        db.rollback()
-        raise
-    except Exception as error:
-        db.rollback()
-        logger.error("OPEN_DEDUCT_FAIL | error_type=%s", type(error).__name__)
-        raise HTTPException(status_code=500, detail="扣款失败，请稍后重试")
-    run_tip_effect(lambda: logger.info("OPEN_DEDUCT_COMPLETED"))
-    return response
-
-
-def _refund_open_balance(req, account, plugin_info, x_plugin_key, db):
-    refund_amount = req.amount
-    if isinstance(refund_amount, bool) or not isinstance(refund_amount, int) or refund_amount <= 0:
-        raise HTTPException(status_code=422, detail="退款金额必须为正整数")
-    operation = open_operation_identity(req.request_id, account, plugin_info['id'], 'OPEN_REFUND',
-        [refund_amount, req.transaction_id])
-    try:
-        wallet, plugin_row = lock_open_wallet(db, account, plugin_info["id"], x_plugin_key, "refund")
-        replay = read_open_operation(db, operation, account, plugin_info['id'], 'OPEN_REFUND', refund_amount, req.transaction_id)
-        if replay is not None:
-            return replay
-        original_tx = db.query(Transaction).filter(Transaction.tx_id == req.transaction_id).populate_existing().with_for_update().first()
-        if original_tx is None:
-            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 != plugin_info["id"] or original_tx.source_plugin not in (None, plugin_info["id"]):
-            raise HTTPException(status_code=403, detail="不允许跨插件退款")
-        if isinstance(original_tx.amount, bool) or not isinstance(original_tx.amount, int) or original_tx.amount >= 0:
-            raise HTTPException(status_code=503, detail="原交易金额暂时无法核实")
-        original_amount = -original_tx.amount
-        # 校验全部关联退款，不能用 abs() 把损坏的负退款/跨账号记录视为合法。
-        refunded = db.query(Transaction).filter(
-            Transaction.tx_type == "OPEN_REFUND", Transaction.item_id == req.transaction_id).all()
-        if any(row.account != account or row.related_account != plugin_info["id"] or
-            row.source_plugin not in (None, plugin_info["id"]) or isinstance(row.amount, bool) or
-            not isinstance(row.amount, int) or row.amount <= 0 for row in refunded):
-            raise HTTPException(status_code=503, detail="退款流水暂时无法核实")
-        already_refunded = sum(row.amount for row in refunded)
-        if already_refunded + refund_amount > original_amount:
-            raise HTTPException(status_code=400, detail="退款金额超过可退余额")
-        if wallet is None:
-            raise HTTPException(status_code=503, detail="钱包记录暂时无法核实")
-        if isinstance(wallet.balance, bool) or not isinstance(wallet.balance, int):
-            raise ValueError("invalid wallet balance")
-        wallet.balance += refund_amount
-        tx_id = f"OPEN_REFUND_{operation[0] if operation is not None else uuid.uuid4().hex}"
-        last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(
-            Transaction.created_at.desc(), Transaction.tx_id.desc()).with_for_update().first()
-        prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
-        db.add(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=_calc_tx_hash(tx_id, account, "OPEN_REFUND", refund_amount, prev_hash),
-            description=f"[{plugin_row.name}] 退款 {req.transaction_id}", source_plugin=plugin_info["id"]))
-        response = {"success": True, "remaining_balance": wallet.balance, "transaction_id": tx_id}
-        save_open_operation(db, operation, response)
-        db.commit()
-    except HTTPException:
-        db.rollback()
-        raise
-    except Exception as error:
-        db.rollback()
-        logger.error("OPEN_REFUND_FAIL | error_type=%s", type(error).__name__)
-        raise HTTPException(status_code=500, detail="退款失败，请稍后重试")
-    run_tip_effect(lambda: logger.info("OPEN_REFUND_COMPLETED"))
-    return response
-
-
-def _read_open_transactions(account, plugin_info, page, page_size, start_time, end_time, tx_type, db):
-    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} | page={page}/{page_size} | total={total}")
-    return {
-        "success": True,
-        "transactions": items,
-        "total": total,
-        "page": page,
-        "page_size": page_size,
-    }
-
-
-def _read_open_usage(plugin_info, account, period, start_time, end_time, db):
-    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} | 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,
-    }
-
+# router_open_api.py
+# ==========================================
+# 🌐 通用开放 API 模块（供第三方插件调用）
+# ==========================================
+# 作用：对外提供受 Plugin Key 保护的通用接口，
+#       供 NodeCraft AI 等第三方插件复用 ComfyUI-Ranking 账号体系。
+# 关联文件：
+#   - 安全认证.py        (verify_token_with_fallback, require_auth)
+#   - database_sql.py    (SessionLocal)
+#   - models_sql.py      (Wallet, Transaction)
+#   - 数据库连接.py       (users.json 读写)
+# 安全机制：
+#   - X-Plugin-Key Header 鉴权
+#   - 基于 plugin_id 的 SQL 持久化限流（每分钟窗口，多 worker 共享）
+#   - 所有扣款/退款操作记录到 Transaction 流水（含哈希链）
+# ==========================================
+
+import os
+import threading
+import json
+import time
+import uuid
+import datetime
+import hashlib
+import logging
+from contextlib import contextmanager
+from typing import Optional
+
+from fastapi import APIRouter, Header, HTTPException, Query
+from fastapi.responses import JSONResponse
+from sqlalchemy.orm import Session
+from sqlalchemy import func, text
+from pydantic import BaseModel, Field
+
+from database_sql import SessionLocal
+from models_sql import Wallet, Transaction, RegisteredPlugin, OpenApiRateLimitBucket
+from open_transaction import lock_open_wallet, positive_integer, validate_live_plugin, quota_total
+from open_transaction import open_operation_identity, read_open_operation, save_open_operation
+from plugin_permissions import parse_permissions
+from tip_store import run_tip_effect
+from auth_io import run_identity_io
+from owned_thread import OwnedThreadGroup, owned_session_endpoint
+from 安全认证 import verify_token_with_fallback
+import 数据库连接 as json_db
+
+logger = logging.getLogger("ComfyUI-Ranking.OpenAPI")
+
+router = APIRouter(prefix="/api/open", tags=["open-api"])
+
+
+MAX_OPEN_SQL_WORK = 8
+_open_sql_work = OwnedThreadGroup(MAX_OPEN_SQL_WORK,
+    lambda: HTTPException(status_code=503, detail="插件限流服务暂不可用"))
+start_open_sql = _open_sql_work.start
+stop_open_sql = _open_sql_work.stop
+close_open_sql = _open_sql_work.close
+
+
+async def _run_open_sql(function, *args, **kwargs):
+    def execute():
+        with SessionLocal() as session:
+            return function(*args, db=session, **kwargs)
+    return await _open_sql_work.run(execute)
+
+
+# ═══════════════ 插件注册表（从数据库动态加载） ═══════════════
+# 反向索引：plugin_key -> {id, key, name, rate_limit, permissions, max_deduct_per_tx, max_deduct_per_day}
+_PLUGIN_KEY_INDEX: dict = {}
+_plugin_registry_lock = threading.RLock()
+_INSECURE_LEGACY_PLUGIN_KEYS = frozenset({"nca_2026_nodecraft_ai_plugin"})
+
+
+@contextmanager
+def _plugin_admin_session():
+    with _plugin_registry_lock:
+        with SessionLocal() as session:
+            yield session
+
+
+open_sql_endpoint = owned_session_endpoint(_open_sql_work, _plugin_admin_session, 'db')
+
+
+def load_plugins_from_db():
+    """启动时从数据库加载所有启用的插件到内存索引"""
+    with _plugin_registry_lock:
+        global _PLUGIN_KEY_INDEX
+
+        db = None
+        try:
+            db = SessionLocal()
+            # 首次启动时仅在显式配置高熵 Key 时 seed 默认插件，不再使用源码内固定凭证。
+            if not db.query(RegisteredPlugin).first():
+                default_key = os.environ.get("NCA_PLUGIN_KEY", "").strip()
+                if len(default_key) >= 32 and default_key not in _INSECURE_LEGACY_PLUGIN_KEYS:
+                    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")
+                else:
+                    logger.warning("NCA_PLUGIN_KEY 缺失或不符合安全要求，跳过默认插件 seed")
+
+            plugins = db.query(RegisteredPlugin).filter(RegisteredPlugin.enabled == True).all()
+
+            # 构建内存索引
+            new_index = {}
+            for p in plugins:
+                # 固定旧 Key 已随源代码泄露，不能继续接受；管理员轮换成新 Key 后即可恢复。
+                if p.plugin_key in _INSECURE_LEGACY_PLUGIN_KEYS:
+                    logger.critical(f"拒绝加载已泄露的默认 Plugin Key: plugin_id={p.plugin_id}")
+                    continue
+                try:
+                    perms = parse_permissions(p.permissions)
+                except ValueError:
+                    logger.warning("OPEN_API_PLUGIN_PERMISSIONS_INVALID")
+                    continue
+                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:
+            # 数据源不可信/不可用时 fail closed：清空内存索引，撤销旧 Key 的继续访问能力。
+            _PLUGIN_KEY_INDEX = {}
+            logger.error(f"❌ 加载插件注册表失败: error_type={type(e).__name__}; 插件 API 已 fail closed")
+        finally:
+            if db is not None:
+                try:
+                    db.close()
+                except Exception as e:
+                    _PLUGIN_KEY_INDEX = {}
+                    logger.error(
+                        f"关闭插件注册表会话失败: error_type={type(e).__name__}; 插件 API 已 fail closed"
+                    )
+
+
+def reload_plugin_registry():
+    """热更新插件注册表（管理操作后调用）"""
+    load_plugins_from_db()
+
+
+# ==========================================
+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("OPEN_API_INVALID_KEY")
+        raise HTTPException(status_code=403, detail="Plugin Key 无效或未注册")
+    return info
+
+
+# ==========================================
+# 🚦 限流：SQL 分钟计数器，多 worker 共享
+# ==========================================
+def _check_rate_limit(plugin_info: dict, permission: str, db: Session) -> None:
+    """
+    校验插件权限并执行限流。
+
+    - 权限不足 -> 403
+    - 触发限流 -> 429
+    """
+    plugin_id = plugin_info["id"]
+
+    now = int(time.time())
+    window_start = now - (now % 60)
+    try:
+        # 对配置列执行 no-op UPDATE：PostgreSQL 获取行锁，SQLite 也会取得写事务锁；
+        # 再 SELECT FOR UPDATE，保证首次建桶和窗口递增串行，而不是只依赖进程锁。
+        db.execute(
+            text("UPDATE registered_plugins SET rate_limit = rate_limit WHERE plugin_id = :plugin_id"),
+            {"plugin_id": plugin_id},
+        )
+        plugin_row = (db.query(RegisteredPlugin)
+            .filter(RegisteredPlugin.plugin_id == plugin_id)
+            .populate_existing()
+            .with_for_update()
+            .first())
+        validate_live_plugin(plugin_row, plugin_info.get("key"), permission)
+        rate_limit = positive_integer(plugin_row.rate_limit)
+        bucket = (db.query(OpenApiRateLimitBucket)
+            .filter(OpenApiRateLimitBucket.plugin_id == plugin_id)
+            .with_for_update()
+            .first())
+        if not bucket:
+            bucket = OpenApiRateLimitBucket(
+                plugin_id=plugin_id, window_start=window_start, request_count=0
+            )
+            db.add(bucket)
+            db.flush()
+        if bucket.window_start != window_start:
+            bucket.window_start = window_start
+            bucket.request_count = 0
+        if bucket.request_count >= rate_limit:
+            db.rollback()
+            raise HTTPException(
+                status_code=429,
+                detail=f"请求过于频繁，限流 {rate_limit}/分钟，请稍后再试",
+            )
+        bucket.request_count += 1
+        db.commit()
+    except HTTPException:
+        db.rollback()
+        raise
+    except Exception as exc:
+        db.rollback()
+        logger.error(f"OPEN_API_RATE_LIMIT_DB_FAIL | plugin={plugin_id} | error_type={type(exc).__name__}")
+        raise HTTPException(status_code=503, detail="插件限流服务暂不可用")
+
+
+def _require_plugin(plugin_key: Optional[str], permission: str, db: Session) -> dict:
+    """统一入口：校验 Plugin Key + 权限 + 限流"""
+    info = verify_plugin_key(plugin_key or "")
+    _check_rate_limit(info, permission, db)
+    return info
+
+
+async 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 = await run_identity_io(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, strict=True, description="扣款金额（正整数）")
+    reason: str = Field(..., description="业务标识/扣款原因")
+    reference_id: Optional[str] = Field(None, description="业务侧参考 ID")
+    request_id: Optional[str] = Field(None, strict=True, min_length=8, max_length=128,
+        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]*$", description="同次扣款重试使用相同请求标识")
+
+
+class RefundRequest(BaseModel):
+    token: str = Field(..., description="用户 JWT Token")
+    transaction_id: str = Field(..., description="原扣款交易 ID")
+    amount: int = Field(..., gt=0, strict=True, description="退款金额（正整数）")
+    request_id: Optional[str] = Field(None, strict=True, min_length=8, max_length=128,
+        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]*$", 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 = await _run_open_sql(_require_plugin, x_plugin_key, "verify")
+
+    is_valid, account, error_msg = await run_identity_io(verify_token_with_fallback, req.token)
+    if not is_valid or not account:
+        logger.info(f"OPEN_VERIFY_TOKEN_FAIL | plugin={plugin_info['id']} | category=invalid_token")
+        return {"valid": False, "error": error_msg or "Token 无效"}
+
+    users_db = await run_identity_io(json_db.load_data, "users.json", default_data={})
+    user = users_db.get(account, {}) or {}
+
+    logger.info(f"OPEN_VERIFY_TOKEN | plugin={plugin_info['id']}")
+    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"),
+):
+    """
+    查询用户钱包余额。
+    认证：Authorization: Bearer <user_token> + X-Plugin-Key
+    """
+    plugin_info = await _run_open_sql(_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 = await _resolve_user_from_token(parts[1])
+    return await _run_open_sql(_read_open_balance, account, plugin_info)
+
+
+# ==========================================
+# 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"),
+):
+    """
+    通用扣款接口（从 wallet.balance 扣减）。
+    流程：
+      1. 校验 Plugin Key + 权限 + 限流
+      2. 解析 Token -> account
+      3. 共用写锁锁定钱包与实时插件配置，检查额度/余额
+      4. 扣款 + 写入 Transaction（含哈希链）
+    """
+    plugin_info = await _run_open_sql(_require_plugin, x_plugin_key, "deduct")
+    account = await _resolve_user_from_token(req.token)
+    return await _run_open_sql(_deduct_open_balance, req, account, plugin_info, x_plugin_key)
+
+
+# ==========================================
+# 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"),
+):
+    """
+    退款接口：将之前 OPEN_DEDUCT 的金额（部分或全额）退还到用户钱包。
+
+    校验：
+      - 原交易必须存在且 tx_type == OPEN_DEDUCT
+      - 原交易归属当前 token 用户
+      - 原交易由当前插件发起（related_account == plugin_id）
+      - 累计退款金额 ≤ 原扣款金额
+    """
+    plugin_info = await _run_open_sql(_require_plugin, x_plugin_key, "refund")
+    account = await _resolve_user_from_token(req.token)
+    return await _run_open_sql(_refund_open_balance, req, account, plugin_info, x_plugin_key)
+
+
+# ==========================================
+# 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"),
+):
+    """
+    查询当前用户由当前插件产生的消费/退款记录。
+
+    认证：Authorization: Bearer <user_token> + X-Plugin-Key
+    安全：仅返回 source_plugin == 当前 plugin_id 的记录，防止跨插件读取。
+    """
+    plugin_info = await _run_open_sql(_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 = await _resolve_user_from_token(parts[1])
+
+    return await _run_open_sql(_read_open_transactions, account, plugin_info, page, page_size, start_time, end_time, tx_type)
+
+
+# ==========================================
+# 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 秒）"),
+):
+    """
+    本插件维度的消费汇总统计。
+
+    返回：总扣款 / 总退款 / 总笔数 / 净消费 + 按 period 分桶的明细。
+    安全：仅统计 source_plugin == 当前 plugin_id 的记录。
+    """
+    plugin_info = await _run_open_sql(_require_plugin, x_plugin_key, "verify")
+    return await _run_open_sql(_read_open_usage, plugin_info, account, period, start_time, end_time)
+
+
+def _read_open_balance(account, plugin_info, db):
+    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']}")
+    return result
+
+
+def _deduct_open_balance(req, account, plugin_info, x_plugin_key, db):
+    amount = req.amount
+    if isinstance(amount, bool) or not isinstance(amount, int) or amount <= 0:
+        raise HTTPException(status_code=422, detail="扣款金额必须为正整数")
+    operation = open_operation_identity(req.request_id, account, plugin_info['id'], 'OPEN_DEDUCT',
+        [amount, req.reason, req.reference_id])
+    try:
+        wallet, plugin_row = lock_open_wallet(db, account, plugin_info["id"], x_plugin_key, "deduct")
+        replay = read_open_operation(db, operation, account, plugin_info['id'], 'OPEN_DEDUCT', -amount, req.reference_id)
+        if replay is not None:
+            return replay
+        max_per_tx = positive_integer(plugin_row.max_deduct_per_tx)
+        max_daily = positive_integer(plugin_row.max_deduct_per_day)
+        if amount > max_per_tx:
+            db.rollback()
+            return JSONResponse(status_code=403, content={
+                "error": "AMOUNT_EXCEEDS_LIMIT", "message": f"单笔扣款不能超过{max_per_tx}积分", "max": max_per_tx})
+        now = datetime.datetime.utcnow()
+        day_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
+        used_today = db.query(func.coalesce(func.sum(func.abs(Transaction.amount)), 0)).filter(
+            Transaction.tx_type == "OPEN_DEDUCT", Transaction.source_plugin == plugin_info["id"],
+            Transaction.created_at >= day_start,
+            Transaction.created_at < day_start + datetime.timedelta(days=1)).scalar()
+        used_today = quota_total(used_today)
+        if used_today + amount > max_daily:
+            db.rollback()
+            return JSONResponse(status_code=403, content={
+                "error": "DAILY_LIMIT_EXCEEDED", "message": "今日扣款已达上限", "limit": max_daily, "used": used_today})
+        if wallet is None:
+            raise HTTPException(status_code=402, detail="余额不足")
+        if isinstance(wallet.balance, bool) or not isinstance(wallet.balance, int):
+            raise ValueError("invalid wallet balance")
+        if wallet.balance < amount:
+            raise HTTPException(status_code=402, detail="余额不足")
+        wallet.balance -= amount
+        tx_id = f"OPEN_DEDUCT_{operation[0] if operation is not None else uuid.uuid4().hex}"
+        last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(
+            Transaction.created_at.desc(), Transaction.tx_id.desc()).with_for_update().first()
+        prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
+        db.add(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=_calc_tx_hash(tx_id, account, "OPEN_DEDUCT", -amount, prev_hash),
+            description=f"[{plugin_row.name}] {req.reason}", source_plugin=plugin_info["id"]))
+        response = {"success": True, "remaining_balance": wallet.balance, "transaction_id": tx_id}
+        save_open_operation(db, operation, response)
+        db.commit()
+    except HTTPException:
+        db.rollback()
+        raise
+    except Exception as error:
+        db.rollback()
+        logger.error("OPEN_DEDUCT_FAIL | error_type=%s", type(error).__name__)
+        raise HTTPException(status_code=500, detail="扣款失败，请稍后重试")
+    run_tip_effect(lambda: logger.info("OPEN_DEDUCT_COMPLETED"))
+    return response
+
+
+def _refund_open_balance(req, account, plugin_info, x_plugin_key, db):
+    refund_amount = req.amount
+    if isinstance(refund_amount, bool) or not isinstance(refund_amount, int) or refund_amount <= 0:
+        raise HTTPException(status_code=422, detail="退款金额必须为正整数")
+    operation = open_operation_identity(req.request_id, account, plugin_info['id'], 'OPEN_REFUND',
+        [refund_amount, req.transaction_id])
+    try:
+        wallet, plugin_row = lock_open_wallet(db, account, plugin_info["id"], x_plugin_key, "refund")
+        replay = read_open_operation(db, operation, account, plugin_info['id'], 'OPEN_REFUND', refund_amount, req.transaction_id)
+        if replay is not None:
+            return replay
+        original_tx = db.query(Transaction).filter(Transaction.tx_id == req.transaction_id).populate_existing().with_for_update().first()
+        if original_tx is None:
+            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 != plugin_info["id"] or original_tx.source_plugin not in (None, plugin_info["id"]):
+            raise HTTPException(status_code=403, detail="不允许跨插件退款")
+        if isinstance(original_tx.amount, bool) or not isinstance(original_tx.amount, int) or original_tx.amount >= 0:
+            raise HTTPException(status_code=503, detail="原交易金额暂时无法核实")
+        original_amount = -original_tx.amount
+        # 校验全部关联退款，不能用 abs() 把损坏的负退款/跨账号记录视为合法。
+        refunded = db.query(Transaction).filter(
+            Transaction.tx_type == "OPEN_REFUND", Transaction.item_id == req.transaction_id).all()
+        if any(row.account != account or row.related_account != plugin_info["id"] or
+            row.source_plugin not in (None, plugin_info["id"]) or isinstance(row.amount, bool) or
+            not isinstance(row.amount, int) or row.amount <= 0 for row in refunded):
+            raise HTTPException(status_code=503, detail="退款流水暂时无法核实")
+        already_refunded = sum(row.amount for row in refunded)
+        if already_refunded + refund_amount > original_amount:
+            raise HTTPException(status_code=400, detail="退款金额超过可退余额")
+        if wallet is None:
+            raise HTTPException(status_code=503, detail="钱包记录暂时无法核实")
+        if isinstance(wallet.balance, bool) or not isinstance(wallet.balance, int):
+            raise ValueError("invalid wallet balance")
+        wallet.balance += refund_amount
+        tx_id = f"OPEN_REFUND_{operation[0] if operation is not None else uuid.uuid4().hex}"
+        last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(
+            Transaction.created_at.desc(), Transaction.tx_id.desc()).with_for_update().first()
+        prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
+        db.add(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=_calc_tx_hash(tx_id, account, "OPEN_REFUND", refund_amount, prev_hash),
+            description=f"[{plugin_row.name}] 退款 {req.transaction_id}", source_plugin=plugin_info["id"]))
+        response = {"success": True, "remaining_balance": wallet.balance, "transaction_id": tx_id}
+        save_open_operation(db, operation, response)
+        db.commit()
+    except HTTPException:
+        db.rollback()
+        raise
+    except Exception as error:
+        db.rollback()
+        logger.error("OPEN_REFUND_FAIL | error_type=%s", type(error).__name__)
+        raise HTTPException(status_code=500, detail="退款失败，请稍后重试")
+    run_tip_effect(lambda: logger.info("OPEN_REFUND_COMPLETED"))
+    return response
+
+
+def _read_open_transactions(account, plugin_info, page, page_size, start_time, end_time, tx_type, db):
+    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} | page={page}/{page_size} | total={total}")
+    return {
+        "success": True,
+        "transactions": items,
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+    }
+
+
+def _read_open_usage(plugin_info, account, period, start_time, end_time, db):
+    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} | 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,
+    }
+