Upload 8 files

router_posts.py

@@ -13,13 +13,13 @@ import 数据库连接 as db
 from comment_store import COMMENT_MAX_LENGTH, append_comment, increment_comment_count, attach_comment_counts, read_comment_view
 from 安全认证 import require_auth
 from db_utils import read_public_source, record_view, sort_cache
-from database_sql import get_db
+from database_sql import get_db, SessionLocal
 from models_sql import Wallet, Transaction
 from notifications import add_notification
 from tip_transaction import lock_tip_wallets
 from tip_store import run_tip_effect
 import time
-from owned_thread import OwnedThreadGroup, owned_sync_endpoint
+from owned_thread import OwnedThreadGroup, owned_sync_endpoint, owned_session_endpoint
 from post_store import publish_post, edit_post, delete_post as remove_post, PostStoreError
 import uuid
 import hashlib
@@ -36,6 +36,7 @@ start_posts_io = _posts_io.start
 stop_posts_io = _posts_io.stop
 close_posts_io = _posts_io.close
 _posts_endpoint = owned_sync_endpoint(_posts_io)
+_posts_session_endpoint = owned_session_endpoint(_posts_io, lambda: SessionLocal(), 'db_session')
 
 router = APIRouter()
 
@@ -426,7 +427,8 @@ def calculate_tx_hash(tx_id, account, tx_type, amount, prev_hash):
 
 @router.post("/api/posts/{post_id}/tip")
 @shared_limit(10)
-async def tip_post(request: Request, post_id: str, amount: int, is_anon: bool = False, current_user: str = Depends(require_auth), db_session: Session = Depends(get_db)):
+@_posts_session_endpoint
+def tip_post(request: Request, post_id: str, amount: int, is_anon: bool = False, current_user: str = Depends(require_auth), db_session: Session = Depends(get_db)):
     """
     打赏帖子（原子操作，并发安全）- 使用 SQL Wallet 系统
     """

router_prompts.py

@@ -21,14 +21,14 @@ import 数据库连接 as db
 from comment_store import COMMENT_MAX_LENGTH, append_comment, increment_comment_count, attach_comment_counts, read_comment_view
 from 安全认证 import require_auth, optional_auth, is_admin
 from db_utils import read_public_source, record_view, sort_cache
-from database_sql import get_db
+from database_sql import get_db, SessionLocal
 from models_sql import Wallet, Transaction, Ownership
 from notifications import add_notification
 from tip_transaction import lock_tip_wallets
 from purchase_transaction import effective_resource_price, lock_resource_purchase_wallets
 from tip_store import run_tip_effect
 import time
-from owned_thread import OwnedThreadGroup, owned_sync_endpoint
+from owned_thread import OwnedThreadGroup, owned_sync_endpoint, owned_session_endpoint
 from prompt_store import (PromptStoreError, publish_prompt, edit_prompt,
     delete_prompt as remove_prompt, project_price, load_categories, modify_category, read_prompts)
 import uuid
@@ -46,6 +46,7 @@ start_prompts_io = _prompts_io.start
 stop_prompts_io = _prompts_io.stop
 close_prompts_io = _prompts_io.close
 _prompts_endpoint = owned_sync_endpoint(_prompts_io)
+_prompts_session_endpoint = owned_session_endpoint(_prompts_io, lambda: SessionLocal(), 'db_session')
 
 router = APIRouter()
 
@@ -340,7 +341,8 @@ def get_my_prompts(current_user: str = Depends(require_auth)):
 
 
 @router.get("/api/prompts/{prompt_id}")
-async def get_prompt_detail(prompt_id: str, current_user: str = Depends(optional_auth),
+@_prompts_session_endpoint
+def get_prompt_detail(prompt_id: str, current_user: str = Depends(optional_auth),
                             db_session: Session = Depends(get_db)):
     """
     获取提示词详情
@@ -657,7 +659,8 @@ def toggle_favorite(request: Request, prompt_id: str, current_user: str = Depend
 
 @router.post("/api/prompts/{prompt_id}/tip")
 @shared_limit(10)
-async def tip_prompt(request: Request, prompt_id: str, amount: int, is_anon: bool = False,
+@_prompts_session_endpoint
+def tip_prompt(request: Request, prompt_id: str, amount: int, is_anon: bool = False,
                      current_user: str = Depends(require_auth), db_session: Session = Depends(get_db)):
     """
     打赏提示词（原子操作，并发安全）- 使用 SQL Wallet 系统
@@ -823,7 +826,8 @@ async def tip_prompt(request: Request, prompt_id: str, amount: int, is_anon: boo
 
 @router.post("/api/prompts/{prompt_id}/purchase")
 @shared_limit(10)  # 🔒 P0安全优化：购买每分钟最多10次
-async def purchase_prompt(request: Request, prompt_id: str,
+@_prompts_session_endpoint
+def purchase_prompt(request: Request, prompt_id: str,
                           purchase: Optional[PromptPurchaseRequest] = None,
                           current_user: str = Depends(require_auth), db_session: Session = Depends(get_db)):
     """资金与所有权原子提交；通知失败不改变已提交的购买结果。"""

router_wallet.py

@@ -16,7 +16,6 @@ from fastapi.responses import Response
 from sqlalchemy.orm import Session
 from pydantic import BaseModel, Field
 import time
-import asyncio
 import uuid
 import hashlib
 import os
@@ -137,7 +136,8 @@ def calculate_tx_hash(tx_id, account, tx_type, amount, prev_hash):
 
 @router.post("/api/wallet/create_recharge_order")
 @shared_limit(10)
-async def create_recharge_order(request: Request, req: RechargeRequest, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
+@_wallet_endpoint
+def create_recharge_order(request: Request, req: RechargeRequest, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     if not alipay:
         raise HTTPException(status_code=500, detail="支付网关配置错误，请联系管理员")
     if type(req.amount) is not int or not 1 <= req.amount <= 10000:
@@ -146,7 +146,7 @@ async def create_recharge_order(request: Request, req: RechargeRequest, current_
     try:
         # 先持久化；失败时不能生成可付款订单。
         persist_order(db, order_id, current_user, req.amount)
-        result = await asyncio.to_thread(alipay.api_alipay_trade_precreate,
+        result = alipay.api_alipay_trade_precreate(
             out_trade_no=order_id, total_amount=str(req.amount),
             subject=f"ComfyUI Community Points - {current_user}")
         if not isinstance(result, dict) or result.get("code") != "10000" or not isinstance(result.get("qr_code"), str) or not result["qr_code"]:
@@ -157,11 +157,9 @@ async def create_recharge_order(request: Request, req: RechargeRequest, current_
         raise HTTPException(status_code=503, detail="创建支付订单失败，请稍后查询确认")
 
 
-@router.post("/api/wallet/alipay_notify")
-async def alipay_notify(request: Request, db: Session = Depends(get_db)):
+@_wallet_endpoint
+def _process_alipay_notification(pairs, db: Session = Depends(get_db)):
     try:
-        form = await request.form()
-        pairs = list(form.multi_items())
         if len({key for key, _ in pairs}) != len(pairs):
             raise ValueError("duplicate payment fields")
         data = dict(pairs)
@@ -169,7 +167,7 @@ async def alipay_notify(request: Request, db: Session = Depends(get_db)):
         sign_type = data.pop("sign_type", None)
         if not alipay or not isinstance(signature, str) or not signature or sign_type != "RSA2":
             raise ValueError("invalid signature fields")
-        if not await asyncio.to_thread(alipay.verify, data, signature):
+        if not alipay.verify(data, signature):
             raise ValueError("invalid signature")
         app_id = os.environ.get("ALIPAY_APPID", "").strip()
         seller_id = os.environ.get("ALIPAY_SELLER_ID", "").strip()
@@ -191,8 +189,20 @@ async def alipay_notify(request: Request, db: Session = Depends(get_db)):
         return Response(content="fail", media_type="text/plain")
 
 
+@router.post("/api/wallet/alipay_notify")
+async def alipay_notify(request: Request):
+    try:
+        form = await request.form()
+        pairs = list(form.multi_items())
+    except Exception as error:
+        run_tip_effect(lambda: logger.error(f"RECHARGE_ERROR | error_type={type(error).__name__}"))
+        return Response(content="fail", media_type="text/plain")
+    return await _process_alipay_notification(pairs)
+
+
 @router.get("/api/wallet/check_order/{order_id}")
-async def check_order(order_id: str, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
+@_wallet_endpoint
+def check_order(order_id: str, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     try:
         info = read_order(db, order_id)
         if info is None or info["account"] != current_user:
@@ -201,7 +211,7 @@ async def check_order(order_id: str, current_user: str = Depends(require_auth),
             return {"status": "SUCCESS"}
         if not alipay:
             return {"status": "PENDING"}
-        result = await asyncio.to_thread(alipay.api_alipay_trade_query, out_trade_no=order_id)
+        result = alipay.api_alipay_trade_query(out_trade_no=order_id)
         if not isinstance(result, dict) or result.get("code") != "10000":
             return {"status": "PENDING"}
         if result.get("out_trade_no") != order_id: