Upload 8 files

app.py

@@ -70,7 +70,7 @@ from router_users_social import router as users_social_router
 from router_items import (router as items_router, start_resource_reads, stop_resource_reads, close_resource_reads)       # 📦 内容管理(工具/Skill/应用/推荐)
 from router_comments import (router as comments_router, start_comment_writes, stop_comment_writes, close_comment_writes) # 💬 评论系统
 from router_messages import (router as messages_router, start_message_io, stop_message_io, close_message_io) # ✉️ 私信系统
-from router_wallet import router as wallet_router     # 💰 钱包/提现
+from router_wallet import (router as wallet_router, start_wallet_sql, stop_wallet_sql, close_wallet_sql)     # 💰 钱包/提现
 from router_proxy import router as proxy_router       # 🔗 代理下载
 from router_posts import (router as posts_router, start_posts_io, stop_posts_io, close_posts_io)       # 💬 讨论区
 from router_prompts import (router as prompts_router, start_prompts_io, stop_prompts_io, close_prompts_io)   # ✨ 提示词市场
@@ -294,6 +294,7 @@ async def on_startup():
             start_identity_io()
             start_shared_work()
             start_open_sql()
+            start_wallet_sql()
             start_resource_work()
             start_resource_reads()
             start_comment_writes()
@@ -358,6 +359,7 @@ async def _shutdown_resources():
     failures.call('log_start', logger.info, "🛑 ComfyUI-Ranking API 正在关闭...")
     failures.call('stop_warmup', stop_startup_warmup, app.state)
     for stage, stop in (
+        ('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_messages', stop_message_io), ('stop_resources', stop_resource_work), ('stop_open_sql', stop_open_sql), ('stop_shared', stop_shared_work),
@@ -374,6 +376,7 @@ async def _shutdown_resources():
         failures.call('stop_' + name, request_periodic_stop, app.state, (name,))
     for stage, close, args in (
         ('wait_warmup', wait_startup_warmup, (app.state,)),
+        ('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_messages', close_message_io, ()), ('close_resources', close_resource_work, ()), ('close_open_sql', close_open_sql, ()), ('close_shared', close_shared_work, ()),

owned_thread.py

@@ -1,7 +1,7 @@
 """等待所属同步工作真实终态；内部取消也不能遗留仍读写资源的线程。"""
 import asyncio
 from functools import wraps
-from inspect import iscoroutinefunction, isasyncgenfunction, isgeneratorfunction
+from inspect import Parameter, signature, iscoroutinefunction, isasyncgenfunction, isgeneratorfunction
 import threading
 
 
@@ -142,3 +142,27 @@ def owned_sync_endpoint(group):
             return await group.run(endpoint, *args, **kwargs)
         return invoke
     return decorate
+
+
+def owned_session_endpoint(group, session_factory, session_parameter):
+    def decorate(endpoint):
+        if iscoroutinefunction(endpoint) or isasyncgenfunction(endpoint) or isgeneratorfunction(endpoint):
+            raise TypeError('owned session endpoint must be a synchronous function')
+        original = signature(endpoint)
+        if session_parameter not in original.parameters:
+            raise ValueError('missing owned session parameter')
+        if any(parameter.kind not in (Parameter.POSITIONAL_OR_KEYWORD, Parameter.KEYWORD_ONLY)
+               for parameter in original.parameters.values()):
+            raise TypeError('owned session endpoint requires named parameters')
+        public = original.replace(parameters=[parameter for name, parameter in original.parameters.items()
+                                               if name != session_parameter])
+        @wraps(endpoint)
+        async def invoke(*args, **kwargs):
+            arguments = public.bind(*args, **kwargs).arguments
+            def execute():
+                with session_factory() as session:
+                    return endpoint(**arguments, **{session_parameter: session})
+            return await group.run(execute)
+        invoke.__signature__ = public
+        return invoke
+    return decorate

resource_work.py

@@ -1,6 +1,7 @@
 """资源写入与可选预缓存的准入和关闭归属。"""
 import asyncio
 import logging
+from functools import wraps
 
 from fastapi import HTTPException
 from owned_thread import OwnedThreadGroup, owned_sync_endpoint
@@ -34,6 +35,49 @@ class ResourceWork:
     async def run(self, function, *args, **kwargs):
         return await self.writes.run(function, *args, **kwargs)
 
+    def endpoint(self, function):
+        @wraps(function)
+        async def invoke(*args, **kwargs):
+            return await self.operation(function, *args, **kwargs)
+        return invoke
+
+    async def operation(self, function, *args, **kwargs):
+        if not self._accepting or len(self._tasks) >= self.background_capacity:
+            raise HTTPException(status_code=503, detail='资源处理繁忙，请稍后重试')
+
+        async def execute():
+            try:
+                return True, await function(*args, **kwargs)
+            except BaseException as error:
+                return False, error
+
+        coroutine = execute()
+        try:
+            task = asyncio.create_task(coroutine)
+        except BaseException:
+            coroutine.close()
+            raise
+        self._tasks.add(task)
+        cancelled = None
+        try:
+            while not task.done():
+                try:
+                    await asyncio.shield(task)
+                except asyncio.CancelledError as error:
+                    if cancelled is None:
+                        cancelled = error
+                        task.cancel()
+            if task.cancelled():
+                raise cancelled or asyncio.CancelledError()
+            succeeded, result = task.result()
+            if cancelled is not None:
+                raise cancelled
+            if not succeeded:
+                raise result
+            return result
+        finally:
+            self._tasks.discard(task)
+
     def background(self, factory):
         if not self._accepting or len(self._tasks) >= self.background_capacity:
             return False
@@ -83,6 +127,7 @@ class ResourceWork:
 _resource_work = ResourceWork()
 run_resource_write = _resource_work.run
 resource_write_endpoint = owned_sync_endpoint(_resource_work)
+resource_operation_endpoint = _resource_work.endpoint
 start_resource_background = _resource_work.background
 start_resource_work = _resource_work.start
 stop_resource_work = _resource_work.stop

router_items.py

@@ -1,5 +1,5 @@
-from resource_work import run_resource_write, start_resource_background, resource_write_endpoint
-from owned_thread import OwnedThreadGroup, owned_sync_endpoint
+from resource_work import run_resource_write, start_resource_background, resource_write_endpoint, resource_operation_endpoint
+from owned_thread import OwnedThreadGroup, owned_sync_endpoint, run_owned_thread
 # router_items.py
 from fastapi import APIRouter, HTTPException, Depends, Request
 from rate_limit import shared_limit
@@ -11,7 +11,6 @@ import re
 import urllib.request
 import urllib.error
 import json
-import asyncio
 import 数据库连接 as db
 from models import ItemCreate, ItemUpdate, RatingRequest
 from 安全认证 import require_auth, check_ownership, is_admin
@@ -496,13 +495,14 @@ async def update_item(request: Request, item_id: str, update_data: ItemUpdate, c
 # ==========================================
 @router.post("/api/check_updates")
 @shared_limit(30)
+@resource_operation_endpoint
 async def check_github_updates(request: Request, current_user: str = Depends(require_auth)):
     """
     遍历所有 GitHub 类型的工具，获取最新 commit hash，存入 versions.json
     """
     if not is_admin(current_user):
         raise HTTPException(status_code=403, detail="无权限")
-    items_db = await asyncio.to_thread(db.load_data, "items.json", default_data=[], strict=True, skip_cache=True)
+    items_db = await run_owned_thread(db.load_data, "items.json", default_data=[], strict=True, skip_cache=True)
     updated_count = 0
     for item in items_db:
         if not item.get('link', '').startswith('https://github.com/'):
@@ -788,9 +788,10 @@ def toggle_item_favorite(request: Request, item_id: str, current_user: str = Dep
 # ==========================================
 @router.post("/api/items/{item_id}/refresh_cache")
 @shared_limit(3)
+@resource_operation_endpoint
 async def refresh_item_cache(request: Request, item_id: str, current_user: str = Depends(require_auth)):
     """手动触发指定资源的 ZIP 预缓存刷新，仅限作者调用"""
-    items_db = db.load_data("items.json", default_data=[])
+    items_db = await run_owned_thread(db.load_data, "items.json", default_data=[])
     
     # 查找目标 item
     target_item = None