优化

app.py

@@ -73,38 +73,80 @@ app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 # 🛡️ 稳定性优化：全局异常处理器
 # ==========================================
 # 作用：捕获所有未处理异常，防止单个请求崩溃整个服务
+# 🚀 P3优化：统一错误响应格式
+
+def create_error_response(status_code: int, detail: str, error_type: str = None, error_id: str = None, extra: dict = None):
+    """🚀 P3优化：统一错误响应构造器"""
+    response = {
+        "success": False,
+        "status": "error",
+        "code": status_code,
+        "detail": detail,
+    }
+    if error_type:
+        response["type"] = error_type
+    if error_id:
+        response["error_id"] = error_id
+    if extra:
+        response.update(extra)
+    return response
 
 @app.exception_handler(StarletteHTTPException)
 async def http_exception_handler(request, exc):
-    """HTTP 异常处理，返回友好错误信息"""
+    """🚀 P3优化：HTTP 异常处理，返回统一格式"""
     logger.warning(f"HTTP {exc.status_code} | {request.url.path} | {exc.detail}")
+    
+    # 根据状态码确定错误类型
+    error_type_map = {
+        400: "bad_request",
+        401: "unauthorized",
+        403: "forbidden",
+        404: "not_found",
+        409: "conflict",
+        429: "rate_limited",
+    }
+    error_type = error_type_map.get(exc.status_code, "http_error")
+    
     return JSONResponse(
         status_code=exc.status_code,
-        content={"status": "error", "detail": str(exc.detail)}
+        content=create_error_response(exc.status_code, str(exc.detail), error_type)
     )
 
 @app.exception_handler(RequestValidationError)
 async def validation_exception_handler(request, exc):
-    """请求参数校验失败处理"""
+    """🚀 P3优化：请求参数校验失败处理"""
     logger.warning(f"Validation Error | {request.url.path} | {exc.errors()}")
+    
+    # 提取简洁的错误信息
+    errors = []
+    for err in exc.errors():
+        field = ".".join(str(loc) for loc in err["loc"][1:])  # 跳过 'body'
+        errors.append({"field": field, "message": err["msg"], "type": err["type"]})
+    
     return JSONResponse(
         status_code=422,
-        content={"status": "error", "detail": "请求参数格式错误", "errors": exc.errors()}
+        content=create_error_response(
+            422, 
+            "请求参数格式错误", 
+            "validation_error",
+            extra={"errors": errors}
+        )
     )
 
 @app.exception_handler(Exception)
 async def global_exception_handler(request, exc):
-    """全局异常处理，捕获所有未预期异常"""
+    """🚀 P3优化：全局异常处理，捕获所有未预期异常"""
     error_id = f"ERR_{int(time.time())}_{id(exc) % 10000}"
     logger.error(f"Unhandled Exception [{error_id}] | {request.url.path} | {type(exc).__name__}: {exc}")
     logger.error(traceback.format_exc())
     return JSONResponse(
         status_code=500,
-        content={
-            "status": "error", 
-            "detail": "服务器内部错误，请稍后重试",
-            "error_id": error_id  # 方便排查
-        }
+        content=create_error_response(
+            500,
+            "服务器内部错误，请稍后重试",
+            "internal_error",
+            error_id=error_id
+        )
     )
 
 

db_utils.py

@@ -0,0 +1,428 @@
+# 云端Space代码/db_utils.py
+# ==========================================
+# 🔧 P2代码质量优化：数据库工具函数
+# ==========================================
+# 作用：封装 JSON 数据库常用操作，减少重复代码
+# 关联文件：
+#   - 数据库连接.py (基础读写)
+#   - router_tasks.py (任务操作)
+#   - router_posts.py (帖子操作)
+#   - router_items.py (商品操作)
+# ==========================================
+
+from typing import Any, Dict, List, Optional, Callable, Union
+import 数据库连接 as db
+
+
+# ==========================================
+# 📖 查询工具函数
+# ==========================================
+
+def get_by_id(file_name: str, item_id: str, id_field: str = "id") -> Optional[Dict]:
+    """
+    根据 ID 获取单个记录
+    
+    参数：
+        file_name: JSON 文件名（如 tasks.json）
+        item_id: 要查找的 ID
+        id_field: ID 字段名（默认 "id"）
+    
+    返回：
+        找到的记录，或 None
+    
+    示例：
+        task = get_by_id("tasks.json", "task_123")
+        user = get_by_id("users.json", "user@example.com", id_field="account")
+    """
+    data = db.load_data(file_name, default_data=[])
+    
+    if isinstance(data, dict):
+        return data.get(item_id)
+    
+    return next((item for item in data if item.get(id_field) == item_id), None)
+
+
+def get_by_field(file_name: str, field: str, value: Any) -> Optional[Dict]:
+    """
+    根据指定字段获取单个记录
+    
+    参数：
+        file_name: JSON 文件名
+        field: 字段名
+        value: 字段值
+    
+    返回：
+        找到的记录，或 None
+    """
+    data = db.load_data(file_name, default_data=[])
+    
+    if isinstance(data, dict):
+        for item in data.values():
+            if isinstance(item, dict) and item.get(field) == value:
+                return item
+        return None
+    
+    return next((item for item in data if item.get(field) == value), None)
+
+
+def filter_by(file_name: str, **conditions) -> List[Dict]:
+    """
+    根据条件筛选记录
+    
+    参数：
+        file_name: JSON 文件名
+        **conditions: 筛选条件（键值对）
+    
+    返回：
+        符合条件的记录列表
+    
+    示例：
+        open_tasks = filter_by("tasks.json", status="open")
+        user_posts = filter_by("posts.json", author="user123", deleted=False)
+    """
+    data = db.load_data(file_name, default_data=[])
+    
+    if isinstance(data, dict):
+        data = list(data.values())
+    
+    result = []
+    for item in data:
+        if all(item.get(key) == value for key, value in conditions.items()):
+            result.append(item)
+    
+    return result
+
+
+def count_by(file_name: str, **conditions) -> int:
+    """
+    统计符合条件的记录数量
+    
+    参数：
+        file_name: JSON 文件名
+        **conditions: 筛选条件
+    
+    返回：
+        符合条件的记录数量
+    """
+    return len(filter_by(file_name, **conditions))
+
+
+# ==========================================
+# ✏️ 更新工具函数
+# ==========================================
+
+def update_by_id(file_name: str, item_id: str, updates: Dict, id_field: str = "id") -> bool:
+    """
+    根据 ID 更新记录
+    
+    参数：
+        file_name: JSON 文件名
+        item_id: 要更新的 ID
+        updates: 要更新的字段（键值对）
+        id_field: ID 字段名
+    
+    返回：
+        True 更新成功 / False 记录不存在
+    
+    示例：
+        update_by_id("tasks.json", "task_123", {"status": "completed"})
+    """
+    data = db.load_data(file_name, default_data=[])
+    
+    if isinstance(data, dict):
+        if item_id in data:
+            data[item_id].update(updates)
+            db.save_data(file_name, data)
+            return True
+        return False
+    
+    for item in data:
+        if item.get(id_field) == item_id:
+            item.update(updates)
+            db.save_data(file_name, data)
+            return True
+    
+    return False
+
+
+def update_with_fn(file_name: str, item_id: str, update_fn: Callable[[Dict], None], id_field: str = "id") -> bool:
+    """
+    使用函数更新记录（支持复杂更新逻辑）
+    
+    参数：
+        file_name: JSON 文件名
+        item_id: 要更新的 ID
+        update_fn: 更新函数，接收记录 dict，直接修改
+        id_field: ID 字段名
+    
+    返回：
+        True 更新成功 / False 记录不存在
+    
+    示例：
+        def increment_views(item):
+            item["views"] = item.get("views", 0) + 1
+        
+        update_with_fn("items.json", "item_123", increment_views)
+    """
+    data = db.load_data(file_name, default_data=[])
+    
+    if isinstance(data, dict):
+        if item_id in data:
+            update_fn(data[item_id])
+            db.save_data(file_name, data)
+            return True
+        return False
+    
+    for item in data:
+        if item.get(id_field) == item_id:
+            update_fn(item)
+            db.save_data(file_name, data)
+            return True
+    
+    return False
+
+
+# ==========================================
+# ➕ 添加工具函数
+# ==========================================
+
+def insert(file_name: str, item: Dict, prepend: bool = True) -> bool:
+    """
+    插入新记录
+    
+    参数：
+        file_name: JSON 文件名
+        item: 要插入的记录
+        prepend: True 插入到开头 / False 插入到末尾
+    
+    返回：
+        True 插入成功
+    
+    示例：
+        insert("tasks.json", {"id": "task_123", "title": "新任务"})
+    """
+    data = db.load_data(file_name, default_data=[])
+    
+    if isinstance(data, dict):
+        item_id = item.get("id") or item.get("account")
+        if item_id:
+            data[item_id] = item
+            db.save_data(file_name, data)
+            return True
+        return False
+    
+    if prepend:
+        data.insert(0, item)
+    else:
+        data.append(item)
+    
+    db.save_data(file_name, data)
+    return True
+
+
+def insert_if_not_exists(file_name: str, item: Dict, id_field: str = "id") -> bool:
+    """
+    如果不存在则插入
+    
+    参数：
+        file_name: JSON 文件名
+        item: 要插入的记录
+        id_field: ID 字段名
+    
+    返回：
+        True 插入成功 / False 已存在
+    """
+    item_id = item.get(id_field)
+    if not item_id:
+        return False
+    
+    existing = get_by_id(file_name, item_id, id_field)
+    if existing:
+        return False
+    
+    return insert(file_name, item)
+
+
+# ==========================================
+# ❌ 删除工具函数
+# ==========================================
+
+def delete_by_id(file_name: str, item_id: str, id_field: str = "id") -> bool:
+    """
+    根据 ID 删除记录
+    
+    参数：
+        file_name: JSON 文件名
+        item_id: 要删除的 ID
+        id_field: ID 字段名
+    
+    返回：
+        True 删除成功 / False 记录不存在
+    """
+    data = db.load_data(file_name, default_data=[])
+    
+    if isinstance(data, dict):
+        if item_id in data:
+            del data[item_id]
+            db.save_data(file_name, data)
+            return True
+        return False
+    
+    original_len = len(data)
+    data = [item for item in data if item.get(id_field) != item_id]
+    
+    if len(data) < original_len:
+        db.save_data(file_name, data)
+        return True
+    
+    return False
+
+
+def soft_delete_by_id(file_name: str, item_id: str, id_field: str = "id") -> bool:
+    """
+    软删除（标记为已删除，不物理删除）
+    
+    参数：
+        file_name: JSON 文件名
+        item_id: 要删除的 ID
+        id_field: ID 字段名
+    
+    返回：
+        True 成功 / False 记录不存在
+    """
+    import time
+    return update_by_id(file_name, item_id, {
+        "deleted": True,
+        "deleted_at": int(time.time())
+    }, id_field)
+
+
+# ==========================================
+# 🔍 分页工具函数
+# ==========================================
+
+def paginate(
+    file_name: str, 
+    page: int = 1, 
+    limit: int = 20, 
+    sort_by: str = None,
+    sort_desc: bool = True,
+    **filters
+) -> Dict:
+    """
+    分页查询
+    
+    参数：
+        file_name: JSON 文件名
+        page: 页码（从 1 开始）
+        limit: 每页数量
+        sort_by: 排序字段
+        sort_desc: 是否降序
+        **filters: 筛选条件
+    
+    返回：
+        {
+            "data": [...],      # 当前页数据
+            "total": 100,       # 总数
+            "page": 1,          # 当前页
+            "limit": 20,        # 每页数量
+            "pages": 5          # 总页数
+        }
+    """
+    # 获取并筛选数据
+    if filters:
+        data = filter_by(file_name, **filters)
+    else:
+        data = db.load_data(file_name, default_data=[])
+        if isinstance(data, dict):
+            data = list(data.values())
+    
+    # 排序
+    if sort_by:
+        try:
+            data = sorted(data, key=lambda x: x.get(sort_by, 0), reverse=sort_desc)
+        except TypeError:
+            pass  # 排序失败时忽略
+    
+    # 计算分页
+    total = len(data)
+    pages = (total + limit - 1) // limit  # 向上取整
+    start = (page - 1) * limit
+    end = start + limit
+    
+    return {
+        "data": data[start:end],
+        "total": total,
+        "page": page,
+        "limit": limit,
+        "pages": pages
+    }
+
+
+# ==========================================
+# 🔄 批量操作工具函数
+# ==========================================
+
+def batch_update(file_name: str, item_ids: List[str], updates: Dict, id_field: str = "id") -> int:
+    """
+    批量更新记录
+    
+    参数：
+        file_name: JSON 文件名
+        item_ids: 要更新的 ID 列表
+        updates: 要更新的字段
+        id_field: ID 字段名
+    
+    返回：
+        更新的记录数量
+    """
+    data = db.load_data(file_name, default_data=[])
+    updated_count = 0
+    
+    if isinstance(data, dict):
+        for item_id in item_ids:
+            if item_id in data:
+                data[item_id].update(updates)
+                updated_count += 1
+    else:
+        id_set = set(item_ids)
+        for item in data:
+            if item.get(id_field) in id_set:
+                item.update(updates)
+                updated_count += 1
+    
+    if updated_count > 0:
+        db.save_data(file_name, data)
+    
+    return updated_count
+
+
+def batch_delete(file_name: str, item_ids: List[str], id_field: str = "id") -> int:
+    """
+    批量删除记录
+    
+    参数：
+        file_name: JSON 文件名
+        item_ids: 要删除的 ID 列表
+        id_field: ID 字段名
+    
+    返回：
+        删除的记录数量
+    """
+    data = db.load_data(file_name, default_data=[])
+    original_count = len(data) if isinstance(data, list) else len(data)
+    
+    if isinstance(data, dict):
+        for item_id in item_ids:
+            data.pop(item_id, None)
+    else:
+        id_set = set(item_ids)
+        data = [item for item in data if item.get(id_field) not in id_set]
+    
+    new_count = len(data) if isinstance(data, list) else len(data)
+    deleted_count = original_count - new_count
+    
+    if deleted_count > 0:
+        db.save_data(file_name, data)
+    
+    return deleted_count

router_tasks.py

@@ -90,12 +90,15 @@ def check_and_update_expired_tasks(tasks_db, db_session=None):
     - open 状态且超过截止日期：自动取消，退还冻结金额
     - in_progress 状态且超过截止日期：标记为过期（不自动取消，需双方处理）
     💳 P6支付增强：过期时自动退款
+    💳 P0修复：事务完整性保护，失败则回滚
     """
     today = datetime.date.today().isoformat()  # "2026-03-30"
     updated = False
     refund_tasks = []  # 需要退款的任务
+    expired_task_indices = []  # 记录过期任务的索引
     
-    for task in tasks_db:
+    # 第一阶段：扫描过期任务（不修改数据）
+    for idx, task in enumerate(tasks_db):
         deadline = task.get("deadline", "")
         status = task.get("status", "")
         
@@ -105,57 +108,81 @@ def check_and_update_expired_tasks(tasks_db, db_session=None):
         # 检查是否过期
         if deadline < today:
             if status == "open":
-                # 开放接单状态且过期：自动取消，退还冻结金额
-                task["status"] = "expired"
-                task["expired_at"] = int(time.time())
-                updated = True
-                
-                # 💳 记录需要退款的任务
+                # 开放接单状态且过期：记录需要处理
                 frozen_amount = task.get("frozen_amount", task.get("total_price", 0))
-                if frozen_amount > 0:
-                    refund_tasks.append({
-                        "task_id": task["id"],
-                        "publisher": task.get("publisher"),
-                        "amount": frozen_amount,
-                        "title": task.get("title", "")
-                    })
-                    task["refunded"] = True
-                    task["refund_amount"] = frozen_amount
+                expired_task_indices.append({
+                    "index": idx,
+                    "task_id": task["id"],
+                    "publisher": task.get("publisher"),
+                    "amount": frozen_amount,
+                    "title": task.get("title", "")
+                })
+                updated = True
                     
             elif status == "in_progress":
-                # 进行中且过期：标记过期但不取消
+                # 进行中且过期：直接标记（不涉及资金）
                 task["is_overdue"] = True
                 updated = True
     
-    # 💳 执行退款操作
-    if refund_tasks and db_session:
-        for refund in refund_tasks:
-            try:
-                wallet = db_session.query(Wallet).filter(Wallet.account == refund["publisher"]).with_for_update().first()
-                if wallet:
-                    wallet.frozen_balance = max(0, wallet.frozen_balance - refund["amount"])  # 减少冻结
-                    wallet.balance += refund["amount"]  # 返还余额
-                    
-                    # 记录退款交易
-                    create_task_transaction(
-                        db_session, refund["publisher"], "TASK_REFUND",
-                        refund["amount"], task_id=refund["task_id"]
-                    )
-                    
-                    # 发送退款通知
-                    add_notification(refund["publisher"], {
+    # 第二阶段：执行退款操作（事务保护）
+    if expired_task_indices and db_session:
+        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"
+                task["expired_at"] = int(time.time())
+                if item["amount"] > 0:
+                    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": refund["task_id"],
-                        "target_item_title": refund["title"],
-                        "content": f"💰 任务《{refund['title']}》已过期自动取消，{refund['amount']}积分已退还"
+                        "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)}")
                     
-                    logger.info(f"TASK_REFUND | publisher={refund['publisher']} | task={refund['task_id']} | amount={refund['amount']}")
-            except Exception as e:
-                logger.error(f"TASK_REFUND_ERROR | task={refund['task_id']} | error={str(e)}")
-        
-        db_session.commit()
+        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 的调用
     
     return updated
 
@@ -410,9 +437,10 @@ async def update_task(task_id: str, update_data: TaskUpdate, current_user: str =
     raise HTTPException(status_code=404, detail="任务不存在")
 
 @router.delete("/api/tasks/{task_id}")
-async def cancel_task(task_id: str, current_user: str = Depends(require_auth)):
+async def cancel_task(task_id: str, current_user: str = Depends(require_auth), db_session: Session = Depends(get_db)):
     """
     取消任务（仅发布者可操作，且仅在 open 状态时可取消）
+    💳 P0修复：取消时退还冻结资金
     """
     tasks_db = db.load_data("tasks.json", default_data=[])
     
@@ -424,9 +452,41 @@ async def cancel_task(task_id: str, current_user: str = Depends(require_auth)):
             if task.get("status") != "open":
                 raise HTTPException(status_code=400, detail="只能取消开放状态的任务")
             
-            task["status"] = "cancelled"
-            db.save_data("tasks.json", tasks_db)
-            return {"status": "success"}
+            # 💳 P0修复：退还冻结资金
+            frozen_amount = task.get("frozen_amount", task.get("total_price", 0))
+            refund_success = False
+            
+            if frozen_amount > 0:
+                try:
+                    wallet = db_session.query(Wallet).filter(Wallet.account == current_user).with_for_update().first()
+                    if wallet:
+                        wallet.frozen_balance = max(0, wallet.frozen_balance - frozen_amount)
+                        wallet.balance += frozen_amount
+                        
+                        # 记录退款交易
+                        create_task_transaction(
+                            db_session, current_user, "TASK_CANCEL_REFUND",
+                            frozen_amount, task_id=task_id
+                        )
+                        db_session.commit()
+                        refund_success = True
+                        logger.info(f"TASK_CANCEL_REFUND | user={current_user} | task={task_id} | amount={frozen_amount}")
+                except Exception as e:
+                    db_session.rollback()
+                    logger.error(f"TASK_CANCEL_REFUND_ERROR | task={task_id} | error={str(e)}")
+                    raise HTTPException(status_code=500, detail="退款失败，请稍后重试")
+            else:
+                refund_success = True  # 无需退款
+            
+            # 只有退款成功才更新状态
+            if refund_success:
+                task["status"] = "cancelled"
+                task["cancelled_at"] = int(time.time())
+                task["refunded"] = frozen_amount > 0
+                task["refund_amount"] = frozen_amount
+                db.save_data("tasks.json", tasks_db)
+                
+                return {"status": "success", "refunded_amount": frozen_amount}
     
     raise HTTPException(status_code=404, detail="任务不存在")
 
@@ -619,6 +679,7 @@ async def accept_task(task_id: str, is_accepted: bool, feedback: str = None, cur
     - is_accepted=True: 验收通过，支付尾款给接��者
     - is_accepted=False: 验收不通过，可以要求修改或发起申诉
     💳 P6支付增强：使用SQL钱包支付，记录交易流水，发送支付通知
+    💳 P0修复：事务完整性保护，失败时回滚
     """
     tasks_db = db.load_data("tasks.json", default_data=[])
     
@@ -638,50 +699,61 @@ async def accept_task(task_id: str, is_accepted: bool, feedback: str = None, cur
             
             if is_accepted:
                 # 💳 验收通过：支付尾款给接单者
-                publisher_wallet = db_session.query(Wallet).filter(Wallet.account == current_user).with_for_update().first()
-                if not publisher_wallet or publisher_wallet.frozen_balance < remaining:
-                    raise HTTPException(status_code=400, detail="冻结余额不足支付尾款")
-                
-                # 扣除发布者尾款（从冻结余额）
-                publisher_wallet.frozen_balance -= remaining
-                
-                # 给接单者全款（订金+尾款）
-                assignee_wallet = db_session.query(Wallet).filter(Wallet.account == assignee_account).with_for_update().first()
-                if not assignee_wallet:
-                    assignee_wallet = Wallet(account=assignee_account, balance=0)
-                    db_session.add(assignee_wallet)
-                    db_session.flush()
-                
-                assignee_wallet.balance += total_price  # 全款进入可用余额
-                
-                # 💳 记录交易流水
-                # 1. 发布者支付尾款
-                create_task_transaction(
-                    db_session, current_user, "TASK_PAYMENT",
-                    -remaining, related_account=assignee_account, task_id=task_id
-                )
-                # 2. 接单者收入
-                create_task_transaction(
-                    db_session, assignee_account, "TASK_INCOME",
-                    total_price, related_account=current_user, task_id=task_id
-                )
-                
-                task["status"] = "completed"
-                task["completed_at"] = int(time.time())
-                
-                db_session.commit()
-                
-                logger.info(f"TASK_COMPLETE | publisher={current_user} | assignee={assignee_account} | task={task_id} | total={total_price}")
-                message = f"验收通过，已支付 {total_price} 积分给接单者"
-                
-                # 🔔 支付通知：接单者收到款项
-                add_notification(assignee_account, {
-                    "type": "task_payment",
-                    "from_user": current_user,
-                    "target_item_id": task_id,
-                    "target_item_title": task.get("title", ""),
-                    "content": f"💰 任务《{task.get('title', '')}》验收通过，{total_price}积分已到账"
-                })
+                # 💳 P0修复：使用事务保护，失败则回滚
+                try:
+                    publisher_wallet = db_session.query(Wallet).filter(Wallet.account == current_user).with_for_update().first()
+                    if not publisher_wallet or publisher_wallet.frozen_balance < remaining:
+                        raise HTTPException(status_code=400, detail="冻结余额不足支付尾款")
+                    
+                    # 扣除发布者尾款（从冻结余额）
+                    publisher_wallet.frozen_balance -= remaining
+                    
+                    # 给接单者全款（订金+尾款）
+                    assignee_wallet = db_session.query(Wallet).filter(Wallet.account == assignee_account).with_for_update().first()
+                    if not assignee_wallet:
+                        assignee_wallet = Wallet(account=assignee_account, balance=0)
+                        db_session.add(assignee_wallet)
+                        db_session.flush()
+                    
+                    assignee_wallet.balance += total_price  # 全款进入可用余额
+                    
+                    # 💳 记录交易流水
+                    # 1. 发布者支付尾款
+                    create_task_transaction(
+                        db_session, current_user, "TASK_PAYMENT",
+                        -remaining, related_account=assignee_account, task_id=task_id
+                    )
+                    # 2. 接单者收入
+                    create_task_transaction(
+                        db_session, assignee_account, "TASK_INCOME",
+                        total_price, related_account=current_user, task_id=task_id
+                    )
+                    
+                    task["status"] = "completed"
+                    task["completed_at"] = int(time.time())
+                    
+                    # 💳 P0修复：先保存JSON，再 commit，确保原子性
+                    db.save_data("tasks.json", tasks_db)
+                    db_session.commit()
+                    
+                    logger.info(f"TASK_COMPLETE | publisher={current_user} | assignee={assignee_account} | task={task_id} | total={total_price}")
+                    message = f"验收通过，已支付 {total_price} 积分给接单者"
+                    
+                    # 🔔 支付通知：接单者收到款项
+                    add_notification(assignee_account, {
+                        "type": "task_payment",
+                        "from_user": current_user,
+                        "target_item_id": task_id,
+                        "target_item_title": task.get("title", ""),
+                        "content": f"💰 任务《{task.get('title', '')}》验收通过，{total_price}积分已到账"
+                    })
+                    
+                except HTTPException:
+                    raise  # 重新抛出 HTTP 异常
+                except Exception as e:
+                    db_session.rollback()
+                    logger.error(f"TASK_ACCEPT_ERROR | task={task_id} | error={str(e)}")
+                    raise HTTPException(status_code=500, detail="验收处理失败，请稍后重试")
             else:
                 # 验收不通过：回到进行中状态，允许修改后重新提交
                 task["status"] = "in_progress"
@@ -689,6 +761,8 @@ async def accept_task(task_id: str, is_accepted: bool, feedback: str = None, cur
                 task["submitted_at"] = None
                 message = "验收不通过，接单者可以修改后重新提交"
                 
+                db.save_data("tasks.json", tasks_db)
+                
                 # 🔔 通知接单者：验收未通过
                 add_notification(assignee_account, {
                     "type": "task_rejected",
@@ -698,8 +772,6 @@ async def accept_task(task_id: str, is_accepted: bool, feedback: str = None, cur
                     "content": f"任务《{task.get('title', '')}》验收未通过，请修改后重新提交"
                 })
             
-            db.save_data("tasks.json", tasks_db)
-            
             return {"status": "success", "message": message}
     
     raise HTTPException(status_code=404, detail="任务不存在")

安全认证.py

@@ -313,3 +313,182 @@ def verify_token_with_fallback(token: str) -> Tuple[bool, Optional[str], str]:
         return True, account, ""
     
     return False, None, error_msg
+
+
+# ==========================================
+# 🛡️ P2优化：统一权限检查装饰器
+# ==========================================
+# 作用：提供所有者、管理员等权限检查功能
+# 用法：减少路由中的重复权限检查代码
+
+import os
+from functools import wraps
+
+# 管理员账号列表
+ADMIN_ACCOUNTS = [a.strip() for a in os.getenv("ADMIN_ACCOUNTS", "admin").split(",")]
+
+
+def is_admin(account: str) -> bool:
+    """
+    检查用户是否为管理员
+    
+    参数：
+        account: 用户账号
+    
+    返回：
+        True 是管理员 / False 不是
+    """
+    return account in ADMIN_ACCOUNTS
+
+
+async def require_admin(authorization: str = Header(None, alias="Authorization")) -> str:
+    """
+    FastAPI 依赖：要求管理员权限
+    
+    使用方法：
+        @router.delete("/api/admin/users/{user_id}")
+        async def delete_user(user_id: str, admin: str = Depends(require_admin)):
+            ...
+    
+    异常：
+        401: 未登录
+        403: 非管理员
+    """
+    account = await require_auth(authorization)
+    
+    if not is_admin(account):
+        raise HTTPException(status_code=403, detail="需要管理员权限")
+    
+    return account
+
+
+def check_ownership(
+    item: dict, 
+    current_user: str, 
+    owner_field: str = "author",
+    allow_admin: bool = True
+) -> bool:
+    """
+    检查用户是否为资源所有者
+    
+    参数：
+        item: 资源对象
+        current_user: 当前用户账号
+        owner_field: 所有者字段名（默认 "author"）
+        allow_admin: 是否允许管理员操作（默认 True）
+    
+    返回：
+        True 有权限 / False 无权限
+    
+    示例：
+        if not check_ownership(task, current_user, "publisher"):
+            raise HTTPException(403, "无权操作")
+    """
+    # 检查是否为所有者
+    if item.get(owner_field) == current_user:
+        return True
+    
+    # 检查是否为管理员
+    if allow_admin and is_admin(current_user):
+        return True
+    
+    return False
+
+
+def require_ownership(
+    item: dict,
+    current_user: str,
+    owner_field: str = "author",
+    allow_admin: bool = True,
+    error_msg: str = "无权操作此资源"
+):
+    """
+    要求所有者权限（失败时抛出异常）
+    
+    参数：
+        item: 资源对象
+        current_user: 当前用户账号
+        owner_field: 所有者字段名
+        allow_admin: 是否允许管理员操作
+        error_msg: 错误提示
+    
+    异常：
+        403 Forbidden: 无权操作
+    
+    示例：
+        task = get_by_id("tasks.json", task_id)
+        require_ownership(task, current_user, "publisher")
+        # 继续执行更新操作...
+    """
+    if not check_ownership(item, current_user, owner_field, allow_admin):
+        raise HTTPException(status_code=403, detail=error_msg)
+
+
+def require_not_self(
+    target_user: str,
+    current_user: str,
+    error_msg: str = "不能对自己执行此操作"
+):
+    """
+    要求操作对象不是自己
+    
+    使用场景：
+        - 不能关注自己
+        - 不能给自己发私信
+        - 不能接自己发布的任务
+    
+    异常：
+        400 Bad Request: 不能对自己执行此操作
+    """
+    if target_user == current_user:
+        raise HTTPException(status_code=400, detail=error_msg)
+
+
+def require_item_exists(
+    item: Optional[dict],
+    item_type: str = "资源"
+):
+    """
+    要求资源存在
+    
+    参数：
+        item: 资源对象（可能为 None）
+        item_type: 资源类型名称（用于错误提示）
+    
+    异常：
+        404 Not Found: 资源不存在
+    
+    示例：
+        task = get_by_id("tasks.json", task_id)
+        require_item_exists(task, "任务")
+    """
+    if not item:
+        raise HTTPException(status_code=404, detail=f"{item_type}不存在")
+
+
+def require_status(
+    item: dict,
+    allowed_statuses: list,
+    status_field: str = "status",
+    error_msg: str = None
+):
+    """
+    要求资源处于指定状态
+    
+    参数：
+        item: 资源对象
+        allowed_statuses: 允许的状态列表
+        status_field: 状态字段名
+        error_msg: 自定义错误信息
+    
+    异常：
+        400 Bad Request: 状态不允许
+    
+    示例：
+        require_status(task, ["open"], error_msg="只能取消开放状态的任务")
+    """
+    current_status = item.get(status_field)
+    if current_status not in allowed_statuses:
+        if not error_msg:
+            error_msg = f"当前状态 ({current_status}) 不允许此操作"
+        raise HTTPException(status_code=400, detail=error_msg)

数据库连接.py

@@ -59,6 +59,97 @@ os.makedirs(LOCAL_DB_DIR, exist_ok=True)
 os.makedirs(BACKUP_DIR, exist_ok=True)
 
 
+# ==========================================
+# 🚀 P1性能优化：内存缓存层
+# ==========================================
+# 缓存配置（秒）
+CACHE_TTL = {
+    "items.json": 120,      # 列表数据 2 分钟
+    "tasks.json": 60,       # 任务数据 1 分钟
+    "users.json": 300,      # 用户数据 5 分钟
+    "posts.json": 120,      # 帖子数据 2 分钟
+    "comments.json": 60,    # 评论数据 1 分钟
+    "disputes.json": 60,    # 申诉数据 1 分钟
+    "_default": 60          # 默认 1 分钟
+}
+
+# 内存缓存存储
+_memory_cache = {}  # {file_name: {"data": ..., "time": ..., "mtime": ...}}
+_cache_lock = threading.Lock()
+
+
+def _get_cache_ttl(file_name: str) -> int:
+    """获取文件的缓存 TTL"""
+    return CACHE_TTL.get(file_name, CACHE_TTL["_default"])
+
+
+def _get_from_cache(file_name: str, local_path: str) -> Optional[Union[Dict, List]]:
+    """
+    从内存缓存获取数据
+    
+    返回：
+        缓存数据（如果有效）或 None
+    """
+    with _cache_lock:
+        if file_name not in _memory_cache:
+            return None
+        
+        cache_entry = _memory_cache[file_name]
+        now = time.time()
+        
+        # 检查 TTL 是否过期
+        if now - cache_entry["time"] > _get_cache_ttl(file_name):
+            del _memory_cache[file_name]
+            return None
+        
+        # 检查文件是否被修改（mtime 变化）
+        try:
+            current_mtime = os.path.getmtime(local_path)
+            if current_mtime != cache_entry.get("mtime"):
+                del _memory_cache[file_name]
+                return None
+        except OSError:
+            return None
+        
+        logger.debug(f"✨ 内存缓存命中: {file_name}")
+        # 返回深拷贝，防止外部修改影响缓存
+        import copy
+        return copy.deepcopy(cache_entry["data"])
+
+
+def _set_to_cache(file_name: str, data: Union[Dict, List], local_path: str):
+    """将数据存入内存缓存"""
+    with _cache_lock:
+        try:
+            mtime = os.path.getmtime(local_path) if os.path.exists(local_path) else 0
+        except OSError:
+            mtime = 0
+        
+        import copy
+        _memory_cache[file_name] = {
+            "data": copy.deepcopy(data),
+            "time": time.time(),
+            "mtime": mtime
+        }
+        logger.debug(f"💾 内存缓存更新: {file_name}")
+
+
+def invalidate_cache(file_name: str = None):
+    """
+    使缓存失效
+    
+    参数：
+        file_name: 指定文件名，为 None 时清空所有缓存
+    """
+    with _cache_lock:
+        if file_name:
+            _memory_cache.pop(file_name, None)
+            logger.debug(f"🗑️ 缓存失效: {file_name}")
+        else:
+            _memory_cache.clear()
+            logger.debug("🗑️ 所有缓存已清空")
+
+
 # ==========================================
 # 🔐 并发控制：线程锁 + 文件锁
 # ==========================================
@@ -126,17 +217,19 @@ else:
 # 📖 数据读取函数
 # ==========================================
 # 特点：
+#   - 🚀 P1优化：内存缓存优先
 #   - 线程安全（threading.Lock）
 #   - 文件不存在时从 HF 下载
 #   - JSON 解析失败时返回默认值
 
-def load_data(file_name: str, default_data: Optional[Union[Dict, List]] = None) -> Union[Dict, List]:
+def load_data(file_name: str, default_data: Optional[Union[Dict, List]] = None, skip_cache: bool = False) -> Union[Dict, List]:
     """
     从 JSON 文件加载数据
     
     参数：
         file_name: 文件名（如 users.json）
         default_data: 数据不存在时的默认值
+        skip_cache: 是否跳过内存缓存（默认 False）
     
     返回：
         解析后的 JSON 数据（dict 或 list）
@@ -146,6 +239,13 @@ def load_data(file_name: str, default_data: Optional[Union[Dict, List]] = None)
         default_data = {} if file_name == "users.json" else []
     
     local_path = os.path.join(LOCAL_DB_DIR, file_name)
+    
+    # 🚀 P1优化：先检查内存缓存
+    if not skip_cache:
+        cached_data = _get_from_cache(file_name, local_path)
+        if cached_data is not None:
+            return cached_data
+    
     file_lock = _get_file_lock(file_name)
     
     with file_lock:
@@ -167,6 +267,8 @@ def load_data(file_name: str, default_data: Optional[Union[Dict, List]] = None)
                     with open(local_path, "w", encoding="utf-8") as f:
                         json.dump(data, f, ensure_ascii=False, indent=2)
                     print(f"✅ 从 HF Dataset 下载 {file_name} 成功")
+                    # 🚀 P1优化：存入内存缓存
+                    _set_to_cache(file_name, data, local_path)
                     return data
                 except Exception as e:
                     print(f"⚠️ 从 HF 下载 {file_name} 失败: {e}")
@@ -179,9 +281,13 @@ def load_data(file_name: str, default_data: Optional[Union[Dict, List]] = None)
                 # 获取共享锁（读锁）
                 _lock_file(f, exclusive=False)
                 try:
-                    return json.load(f)
+                    data = json.load(f)
                 finally:
                     _unlock_file(f)
+            
+            # 🚀 P1优化：存入内存缓存
+            _set_to_cache(file_name, data, local_path)
+            return data
         except json.JSONDecodeError as e:
             print(f"🚨 JSON 解析错误 {file_name}: {e}")
             # 尝试从备份恢复
@@ -218,7 +324,8 @@ def _recover_from_backup(file_name: str, default_data: Union[Dict, List]) -> Uni
 #   1. 先写临时文件，验证成功后原子重命名
 #   2. 写入前备份上一版本
 #   3. 写入后验证数据完整性
-#   4. 异步同步到 HuggingFace
+#   4. 🚀 P1优化：保存后更新内存缓存
+#   5. 异步同步到 HuggingFace
 
 def save_data(file_name: str, data: Union[Dict, List]) -> bool:
     """
@@ -235,6 +342,7 @@ def save_data(file_name: str, data: Union[Dict, List]) -> bool:
         - 原子写入：先写临时文件再重命名
         - 自动备份：保留上一版本
         - 完整性校验：验证 JSON 可解析
+        - 🚀 P1优化：保存后更新内存缓存
         - 异步同步：后台上传到 HF Dataset
     """
     local_path = os.path.join(LOCAL_DB_DIR, file_name)
@@ -285,6 +393,9 @@ def save_data(file_name: str, data: Union[Dict, List]) -> bool:
             print(f"🚨 保存 {file_name} 失败: {e}")
             raise
     
+    # ========== 🚀 P1优化：更新内存缓存 ==========
+    _set_to_cache(file_name, data, local_path)
+    
     # ========== 第五步：异步同步到云端 ==========
     # 🔧 P3优化：使用线程池替代直接创建线程
     if HF_TOKEN: