Upload 26 files

router_posts.py

@@ -6,12 +6,19 @@
 # ==========================================
 
 from fastapi import APIRouter, HTTPException, Depends
+from sqlalchemy.orm import Session
 from models import PostCreate, PostUpdate
 import 数据库连接 as db
 from 安全认证 import require_auth
 from db_utils import record_view, sort_cache
+from database_sql import get_db
+from models_sql import Wallet, Transaction
+from notifications import add_notification
 import time
 import uuid
+import hashlib
+import datetime
+import logging
 
 router = APIRouter()
 
@@ -313,15 +320,23 @@ async def toggle_favorite(post_id: str, current_user: str = Depends(require_auth
 # 🎁 打赏接口
 # ==========================================
 
+# 📝 审计日志
+logger = logging.getLogger("ComfyUI-Ranking.Posts")
+
+def calculate_tx_hash(tx_id, account, tx_type, amount, prev_hash):
+    data = f"{tx_id}{account}{tx_type}{amount}{prev_hash}"
+    return hashlib.sha256(data.encode()).hexdigest()
+
 @router.post("/api/posts/{post_id}/tip")
-async def tip_post(post_id: str, amount: int, is_anon: bool = False, current_user: str = Depends(require_auth)):
+async def tip_post(post_id: str, amount: int, is_anon: bool = False, current_user: str = Depends(require_auth), db_session: Session = Depends(get_db)):
     """
-    打赏帖子（原子操作，并发安全）
+    打赏帖子（原子操作，并发安全）- 使用 SQL Wallet 系统
     """
     if amount <= 0:
         raise HTTPException(status_code=400, detail="打赏金额必须大于0")
     
     result_container = [None]
+    author_account = [None]  # 用于在原子操作外获取作者账号
     
     def updater(data):
         # 在锁内查找帖子
@@ -340,22 +355,7 @@ async def tip_post(post_id: str, amount: int, is_anon: bool = False, current_use
             result_container[0] = {"error": "self_tip"}
             return
         
-        # 在锁内操作用户余额
-        users_db = db.load_data("users.json", default_data={})
-        tipper = users_db.get(current_user)
-        if not tipper or tipper.get("balance", 0) < amount:
-            result_container[0] = {"error": "insufficient_balance"}
-            return
-        
-        author_account = target_post.get("author")
-        author = users_db.get(author_account)
-        if not author:
-            result_container[0] = {"error": "author_not_found"}
-            return
-        
-        # 扣款加款
-        tipper["balance"] -= amount
-        author["balance"] += amount
+        author_account[0] = target_post.get("author")
         
         # 更新打赏榜单
         tip_board = target_post.get("tip_board", [])
@@ -367,25 +367,141 @@ async def tip_post(post_id: str, amount: int, is_anon: bool = False, current_use
         tip_board.sort(key=lambda x: x["amount"], reverse=True)
         target_post["tip_board"] = tip_board
         
-        # 保存用户数据
-        db.save_data("users.json", users_db)
-        
         result_container[0] = {"status": "success", "message": f"成功打赏 {amount} 积分"}
     
     db.atomic_update("posts.json", updater, default_data=[])
     
-    # 🗂️ 清除排序缓存（打赏可能影响排序）
-    sort_cache.invalidate("posts:")
-    
     result = result_container[0]
     if result is None or result.get("error") == "not_found":
         raise HTTPException(status_code=404, detail="帖子不存在")
     if result.get("error") == "self_tip":
         raise HTTPException(status_code=400, detail="不能打赏自己的帖子")
-    if result.get("error") == "insufficient_balance":
-        raise HTTPException(status_code=400, detail="余额不足")
-    if result.get("error") == "author_not_found":
-        raise HTTPException(status_code=404, detail="作者账户不存在")
+    
+    # 💳 使用 SQL Wallet 系统处理余额转账
+    try:
+        author = author_account[0]
+        if not author:
+            raise HTTPException(status_code=404, detail="作者账户不存在")
+        
+        # 🔒 P1幂等性防护：检查最近5秒内是否存在相同交易
+        recent_cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=5)
+        duplicate_tx = db_session.query(Transaction).filter(
+            Transaction.account == current_user,
+            Transaction.tx_type == "TIP_OUT",
+            Transaction.amount == -amount,
+            Transaction.related_account == author,
+            Transaction.created_at >= recent_cutoff
+        ).first()
+        if duplicate_tx:
+            return {"status": "success", "message": "打赏已处理（重复请求）"}
+        
+        # 🔒 并发安全：使用悲观锁获取双方钱包
+        tipper_wallet = db_session.query(Wallet).filter(Wallet.account == current_user).with_for_update().first()
+        author_wallet = db_session.query(Wallet).filter(Wallet.account == author).with_for_update().first()
+        
+        if not tipper_wallet or tipper_wallet.balance < amount:
+            raise HTTPException(status_code=400, detail="余额不足")
+        if not author_wallet:
+            # 如果作者钱包不存在，创建一个
+            author_wallet = Wallet(account=author, balance=0, earn_balance=0, tip_balance=0, frozen_balance=0)
+            db_session.add(author_wallet)
+        
+        # 执行转账
+        tipper_wallet.balance -= amount
+        author_wallet.balance += amount         # 实际收入进统一余额
+        author_wallet.tip_balance += amount     # 累计打赏收益统计（只增不减）
+        
+        # 创建交易记录
+        tx_id_tipper = f"TIP_OUT_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+        tx_id_author = f"TIP_IN_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+        
+        # 获取最后交易记录的哈希
+        last_tx_tipper = db_session.query(Transaction).filter(Transaction.account == current_user).order_by(Transaction.created_at.desc()).first()
+        last_tx_author = db_session.query(Transaction).filter(Transaction.account == author).order_by(Transaction.created_at.desc()).first()
+        prev_hash_tipper = last_tx_tipper.tx_hash if last_tx_tipper else "GENESIS_HASH"
+        prev_hash_author = last_tx_author.tx_hash if last_tx_author else "GENESIS_HASH"
+        
+        # 获取用户信息和帖子标题
+        users_db = db.load_data("users.json", default_data={})
+        posts_db = db.load_data("posts.json", default_data=[])
+        author_info = users_db.get(author, {})
+        tipper_info = users_db.get(current_user, {})
+        author_name = author_info.get("name", author)
+        tipper_name = tipper_info.get("name", current_user)
+        
+        post_title = None
+        for post in posts_db:
+            if post["id"] == post_id:
+                post_title = post.get("title")
+                break
+        
+        # 打赏方交易记录 (TIP_OUT)
+        tx_tipper = Transaction(
+            tx_id=tx_id_tipper,
+            account=current_user,
+            tx_type="TIP_OUT",
+            amount=-amount,
+            related_account=author,
+            item_id=post_id,
+            prev_hash=prev_hash_tipper,
+            tx_hash=calculate_tx_hash(tx_id_tipper, current_user, "TIP_OUT", -amount, prev_hash_tipper),
+            description=f"打赏给 {author_name}" + (f" 的帖子《{post_title}》" if post_title else ""),
+            item_title=post_title,
+            item_type="post",
+            related_user_name=author_name
+        )
+        
+        # 接收方交易记录 (TIP_IN)
+        tx_author = Transaction(
+            tx_id=tx_id_author,
+            account=author,
+            tx_type="TIP_IN",
+            amount=amount,
+            related_account=current_user,
+            item_id=post_id,
+            prev_hash=prev_hash_author,
+            tx_hash=calculate_tx_hash(tx_id_author, author, "TIP_IN", amount, prev_hash_author),
+            description=f"收到 {tipper_name} 的帖子打赏" + (f" ({post_title})" if post_title else ""),
+            item_title=post_title,
+            item_type="post",
+            related_user_name=tipper_name if not is_anon else "匿名用户"
+        )
+        
+        db_session.add(tx_tipper)
+        db_session.add(tx_author)
+        db_session.commit()
+        
+        # 📝 审计日志
+        logger.info(f"POST_TIP | from={current_user} | to={author} | amount={amount} | post={post_id} | anon={is_anon}")
+        
+        # 🔔 打赏通知（考虑匿名）
+        if not is_anon:
+            add_notification(author, {
+                "type": "tip",
+                "from_user": current_user,
+                "target_item_id": post_id,
+                "target_item_title": post_title or "",
+                "content": f"您收到来自 {tipper_name} 的 {amount} 积分帖子打赏"
+            })
+        else:
+            add_notification(author, {
+                "type": "tip",
+                "from_user": "anonymous",
+                "target_item_id": post_id,
+                "target_item_title": "",
+                "content": f"您收到了一份 {amount} 积分的匿名帖子打赏"
+            })
+        
+        # 🗂️ 清除排序缓存（打赏可能影响排序）
+        sort_cache.invalidate("posts:")
+        
+    except HTTPException:
+        db_session.rollback()
+        raise
+    except Exception as e:
+        db_session.rollback()
+        logger.error(f"POST_TIP_ERROR | from={current_user} | post={post_id} | amount={amount} | error={str(e)}")
+        raise HTTPException(status_code=500, detail="打赏处理失败，请稍后重试")
     
     return result
 

router_tasks.py

@@ -806,6 +806,7 @@ async def accept_task(task_id: str, is_accepted: bool, feedback: str = None, cur
                         db_session.flush()
                     
                     assignee_wallet.balance += total_price  # 全款进入可用余额
+                    assignee_wallet.earn_balance += total_price     # 累计任务收益统计（只增不减）
                     
                     # 获取双方用户名
                     users_db = db.load_data("users.json", default_data={})
@@ -1073,12 +1074,13 @@ async def get_admin_disputes(status: str = None, current_user: str = Depends(req
     return {"status": "success", "data": result}
 
 @router.post("/api/admin/disputes/{dispute_id}/resolve")
-async def resolve_dispute(dispute_id: str, resolution: str, ratio: int = None, note: str = None, current_user: str = Depends(require_auth)):
+async def resolve_dispute(dispute_id: str, resolution: str, ratio: int = None, note: str = None, current_user: str = Depends(require_auth), db_session: Session = Depends(get_db)):
     """
     管理员裁决申诉
     - resolution: favor_initiator(支持申诉方) / favor_respondent(支持被申诉方) / split(按比例分成)
     - ratio: 分成比例(0-100)，表示申诉方获得的比例。仅split时有效
     - note: 裁决说明
+    💳 P0改造：使用 SQL Wallet 系统处理资金，替代 JSON balance 操作
     """
     if not is_admin(current_user):
         raise HTTPException(status_code=403, detail="需要管理员权限")
@@ -1106,75 +1108,164 @@ async def resolve_dispute(dispute_id: str, resolution: str, ratio: int = None, n
             total_price = task.get("total_price", 0)
             deposit = task.get("deposit_amount", 0)
             
-            publisher = users_db.get(dispute.get("publisher"))
-            assignee = users_db.get(dispute.get("assignee"))
+            publisher_account = dispute.get("publisher")
+            assignee_account = dispute.get("assignee")
             
             is_publisher_initiator = dispute.get("initiator_role") == "publisher"
-            initiator = publisher if is_publisher_initiator else assignee
-            respondent = assignee if is_publisher_initiator else publisher
-            
-            # 计算资金分配
-            if resolution == "favor_initiator":
-                # 支持申诉方
-                if is_publisher_initiator:
-                    # 发布者胜诉：退还订金
-                    if publisher:
-                        publisher["balance"] = publisher.get("balance", 0) + deposit
-                    initiator_amount = deposit
-                    respondent_amount = 0
-                else:
-                    # 接单者胜诉：获得全款
-                    if publisher:
+            
+            # 💳 P0改造：使用 SQL Wallet 处理资金
+            try:
+                # 获取双方钱包（带悲观锁）
+                publisher_wallet = db_session.query(Wallet).filter(Wallet.account == publisher_account).with_for_update().first()
+                assignee_wallet = db_session.query(Wallet).filter(Wallet.account == assignee_account).with_for_update().first()
+                
+                # 如果钱包不存在，创建新钱包
+                if not publisher_wallet:
+                    publisher_wallet = Wallet(account=publisher_account, balance=0)
+                    db_session.add(publisher_wallet)
+                    db_session.flush()
+                if not assignee_wallet:
+                    assignee_wallet = Wallet(account=assignee_account, balance=0)
+                    db_session.add(assignee_wallet)
+                    db_session.flush()
+                
+                # 获取双方用户名
+                publisher_info = users_db.get(publisher_account, {})
+                assignee_info = users_db.get(assignee_account, {})
+                publisher_name = publisher_info.get("name", publisher_account)
+                assignee_name = assignee_info.get("name", assignee_account)
+                
+                # 计算资金分配
+                if resolution == "favor_initiator":
+                    # 支持申诉方
+                    if is_publisher_initiator:
+                        # 发布者胜诉：退还订金（从冻结余额释放到可用余额）
+                        publisher_wallet.frozen_balance = max(0, publisher_wallet.frozen_balance - deposit)
+                        publisher_wallet.balance += deposit
+                        
+                        # 记录交易
+                        create_task_transaction(
+                            db_session, publisher_account, "TASK_REFUND",
+                            deposit, task_id=task.get("id"),
+                            task_title=task.get("title"),
+                            related_user_name=assignee_name
+                        )
+                        initiator_amount = deposit
+                        respondent_amount = 0
+                    else:
+                        # 接单者胜诉：获得全款
                         remaining = total_price - deposit
-                        publisher["balance"] = publisher.get("balance", 0) - remaining
-                    if assignee:
-                        assignee["balance"] = assignee.get("balance", 0) + total_price
-                    initiator_amount = total_price
-                    respondent_amount = 0
-            elif resolution == "favor_respondent":
-                # 支持被申诉方
-                if is_publisher_initiator:
-                    # 接单者胜诉：获得全款
-                    if publisher:
+                        # 发布者支付尾款（从冻结余额扣除）
+                        publisher_wallet.frozen_balance = max(0, publisher_wallet.frozen_balance - remaining)
+                        # 接单者获得全款
+                        assignee_wallet.balance += total_price
+                        assignee_wallet.earn_balance += total_price
+                        
+                        # 记录交易
+                        create_task_transaction(
+                            db_session, publisher_account, "TASK_PAYMENT",
+                            -remaining, related_account=assignee_account, task_id=task.get("id"),
+                            task_title=task.get("title"),
+                            related_user_name=assignee_name
+                        )
+                        create_task_transaction(
+                            db_session, assignee_account, "TASK_INCOME",
+                            total_price, related_account=publisher_account, task_id=task.get("id"),
+                            task_title=task.get("title"),
+                            related_user_name=publisher_name
+                        )
+                        initiator_amount = total_price
+                        respondent_amount = 0
+                        
+                elif resolution == "favor_respondent":
+                    # 支持被申诉方
+                    if is_publisher_initiator:
+                        # 接单者胜诉：获得全款
                         remaining = total_price - deposit
-                        publisher["balance"] = publisher.get("balance", 0) - remaining
-                    if assignee:
-                        assignee["balance"] = assignee.get("balance", 0) + total_price
-                    initiator_amount = 0
-                    respondent_amount = total_price
+                        # 发布者支付尾款（从冻结余额扣除）
+                        publisher_wallet.frozen_balance = max(0, publisher_wallet.frozen_balance - remaining)
+                        # 接单者获得全款
+                        assignee_wallet.balance += total_price
+                        assignee_wallet.earn_balance += total_price
+                        
+                        # 记录交易
+                        create_task_transaction(
+                            db_session, publisher_account, "TASK_PAYMENT",
+                            -remaining, related_account=assignee_account, task_id=task.get("id"),
+                            task_title=task.get("title"),
+                            related_user_name=assignee_name
+                        )
+                        create_task_transaction(
+                            db_session, assignee_account, "TASK_INCOME",
+                            total_price, related_account=publisher_account, task_id=task.get("id"),
+                            task_title=task.get("title"),
+                            related_user_name=publisher_name
+                        )
+                        initiator_amount = 0
+                        respondent_amount = total_price
+                    else:
+                        # 发布者胜诉：退还订金（从冻结余额释放到可用余额）
+                        publisher_wallet.frozen_balance = max(0, publisher_wallet.frozen_balance - deposit)
+                        publisher_wallet.balance += deposit
+                        
+                        # 记录交易
+                        create_task_transaction(
+                            db_session, publisher_account, "TASK_REFUND",
+                            deposit, task_id=task.get("id"),
+                            task_title=task.get("title"),
+                            related_user_name=assignee_name
+                        )
+                        initiator_amount = 0
+                        respondent_amount = deposit
+                        
                 else:
-                    # 发布者胜诉：退还订金
-                    if publisher:
-                        publisher["balance"] = publisher.get("balance", 0) + deposit
-                    initiator_amount = 0
-                    respondent_amount = deposit
-            else:
-                # 按比例分成
-                initiator_share = int(total_price * ratio / 100)
-                respondent_share = total_price - initiator_share
-                
-                # 处理资金：发布者支付尾款，然后按比例分配
-                if publisher:
+                    # 按比例分成
+                    initiator_share = int(total_price * ratio / 100)
+                    respondent_share = total_price - initiator_share
                     remaining = total_price - deposit
-                    publisher["balance"] = publisher.get("balance", 0) - remaining
-                
-                # 发布者获得退款部分，接单者获得报酬部分
-                if is_publisher_initiator:
-                    # 申诉方是发布者
-                    pub_refund = initiator_share
-                    assignee_earn = respondent_share
-                else:
-                    # 申诉方是接单者
-                    assignee_earn = initiator_share
-                    pub_refund = respondent_share
+                    
+                    # 发布者支付尾款（从冻结余额扣除）
+                    publisher_wallet.frozen_balance = max(0, publisher_wallet.frozen_balance - remaining)
+                    
+                    # 发布者获得退款部分，接单者获得报酬部分
+                    if is_publisher_initiator:
+                        # 申诉方是发布者
+                        pub_refund = initiator_share
+                        assignee_earn = respondent_share
+                    else:
+                        # 申诉方是接单者
+                        assignee_earn = initiator_share
+                        pub_refund = respondent_share
+                    
+                    # 分配资金
+                    publisher_wallet.balance += pub_refund
+                    assignee_wallet.balance += assignee_earn
+                    assignee_wallet.earn_balance += assignee_earn
+                    
+                    # 记录交易
+                    create_task_transaction(
+                        db_session, publisher_account, "TASK_PAYMENT",
+                        -remaining, related_account=assignee_account, task_id=task.get("id"),
+                        task_title=task.get("title"),
+                        related_user_name=assignee_name
+                    )
+                    create_task_transaction(
+                        db_session, assignee_account, "TASK_INCOME",
+                        assignee_earn, related_account=publisher_account, task_id=task.get("id"),
+                        task_title=task.get("title"),
+                        related_user_name=publisher_name
+                    )
+                    
+                    initiator_amount = initiator_share
+                    respondent_amount = respondent_share
                 
-                if publisher:
-                    publisher["balance"] = publisher.get("balance", 0) + pub_refund
-                if assignee:
-                    assignee["balance"] = assignee.get("balance", 0) + assignee_earn
+                # 提交 SQL 事务
+                db_session.commit()
                 
-                initiator_amount = initiator_share
-                respondent_amount = respondent_share
+            except Exception as e:
+                db_session.rollback()
+                logger.error(f"DISPUTE_RESOLVE_ERROR | dispute={dispute_id} | error={str(e)}")
+                raise HTTPException(status_code=500, detail="仲裁资金处理失败，请稍后重试")
             
             # 更新申诉状态
             dispute["resolution"] = resolution
@@ -1191,7 +1282,6 @@ async def resolve_dispute(dispute_id: str, resolution: str, ratio: int = None, n
             
             db.save_data("disputes.json", disputes_db)
             db.save_data("tasks.json", tasks_db)
-            db.save_data("users.json", users_db)
             # 🗂️ 清除排序缓存（任务状态变为已完成）
             sort_cache.invalidate("tasks:")
             
@@ -1221,6 +1311,8 @@ async def resolve_dispute(dispute_id: str, resolution: str, ratio: int = None, n
                 "content": f"任务申诉已裁决：{resolution_text}，您获得{respondent_amount}积分"
             })
             
+            logger.info(f"DISPUTE_RESOLVED | dispute={dispute_id} | resolution={resolution} | initiator_amount={initiator_amount} | respondent_amount={respondent_amount}")
+            
             return {"status": "success", "message": f"裁决完成：{resolution_text}"}
     
     raise HTTPException(status_code=404, detail="申诉不存在")
@@ -1348,14 +1440,15 @@ async def toggle_favorite(task_id: str, current_user: str = Depends(require_auth
 # ==========================================
 
 @router.post("/api/tasks/{task_id}/tip")
-async def tip_task(task_id: str, amount: int, is_anon: bool = False, current_user: str = Depends(require_auth)):
+async def tip_task(task_id: str, amount: int, is_anon: bool = False, current_user: str = Depends(require_auth), db_session: Session = Depends(get_db)):
     """
-    打赏任务（原子操作，并发安全）
+    打赏任务（原子操作，并发安全）- 使用 SQL Wallet 系统
     """
     if amount <= 0:
         raise HTTPException(status_code=400, detail="打赏金额必须大于0")
     
     result_container = [None]
+    publisher_account = [None]  # 用于在原子操作外获取发布者账号
     
     def updater(data):
         # 在锁内查找任务
@@ -1374,22 +1467,7 @@ async def tip_task(task_id: str, amount: int, is_anon: bool = False, current_use
             result_container[0] = {"error": "self_tip"}
             return
         
-        # 在锁内操作用户余额
-        users_db = db.load_data("users.json", default_data={})
-        tipper = users_db.get(current_user)
-        if not tipper or tipper.get("balance", 0) < amount:
-            result_container[0] = {"error": "insufficient_balance"}
-            return
-        
-        publisher = target_task.get("publisher")
-        author = users_db.get(publisher)
-        if not author:
-            result_container[0] = {"error": "author_not_found"}
-            return
-        
-        # 扣款加款
-        tipper["balance"] -= amount
-        author["balance"] += amount
+        publisher_account[0] = target_task.get("publisher")
         
         # 更新打赏榜单
         tip_board = target_task.get("tip_board", [])
@@ -1401,25 +1479,141 @@ async def tip_task(task_id: str, amount: int, is_anon: bool = False, current_use
         tip_board.sort(key=lambda x: x["amount"], reverse=True)
         target_task["tip_board"] = tip_board
         
-        # 保存用户数据
-        db.save_data("users.json", users_db)
-        
         result_container[0] = {"status": "success", "message": f"成功打赏 {amount} 积分"}
     
     db.atomic_update("tasks.json", updater, default_data=[])
     
-    # 🗂️ 清除排序缓存（打赏可能影响排序）
-    sort_cache.invalidate("tasks:")
-    
     result = result_container[0]
     if result is None or result.get("error") == "not_found":
         raise HTTPException(status_code=404, detail="任务不存在")
     if result.get("error") == "self_tip":
         raise HTTPException(status_code=400, detail="不能打赏自己的任务")
-    if result.get("error") == "insufficient_balance":
-        raise HTTPException(status_code=400, detail="余额不足")
-    if result.get("error") == "author_not_found":
-        raise HTTPException(status_code=404, detail="作者账户不存在")
+    
+    # 💳 使用 SQL Wallet 系统处理余额转账
+    try:
+        publisher = publisher_account[0]
+        if not publisher:
+            raise HTTPException(status_code=404, detail="作者账户不存在")
+        
+        # 🔒 P1幂等性防护：检查最近5秒内是否存在相同交易
+        recent_cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=5)
+        duplicate_tx = db_session.query(Transaction).filter(
+            Transaction.account == current_user,
+            Transaction.tx_type == "TIP_OUT",
+            Transaction.amount == -amount,
+            Transaction.related_account == publisher,
+            Transaction.created_at >= recent_cutoff
+        ).first()
+        if duplicate_tx:
+            return {"status": "success", "message": "打赏已处理（重复请求）"}
+        
+        # 🔒 并发安全：使用悲观锁获取双方钱包
+        tipper_wallet = db_session.query(Wallet).filter(Wallet.account == current_user).with_for_update().first()
+        author_wallet = db_session.query(Wallet).filter(Wallet.account == publisher).with_for_update().first()
+        
+        if not tipper_wallet or tipper_wallet.balance < amount:
+            raise HTTPException(status_code=400, detail="余额不足")
+        if not author_wallet:
+            # 如果作者钱包不存在，创建一个
+            author_wallet = Wallet(account=publisher, balance=0, earn_balance=0, tip_balance=0, frozen_balance=0)
+            db_session.add(author_wallet)
+        
+        # 执行转账
+        tipper_wallet.balance -= amount
+        author_wallet.balance += amount         # 实际收入进统一余额
+        author_wallet.tip_balance += amount     # 累计打赏收益统计（只增不减）
+        
+        # 创建交易记录
+        tx_id_tipper = f"TIP_OUT_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+        tx_id_author = f"TIP_IN_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+        
+        # 获取最后交易记录的哈希
+        last_tx_tipper = db_session.query(Transaction).filter(Transaction.account == current_user).order_by(Transaction.created_at.desc()).first()
+        last_tx_author = db_session.query(Transaction).filter(Transaction.account == publisher).order_by(Transaction.created_at.desc()).first()
+        prev_hash_tipper = last_tx_tipper.tx_hash if last_tx_tipper else "GENESIS_HASH"
+        prev_hash_author = last_tx_author.tx_hash if last_tx_author else "GENESIS_HASH"
+        
+        # 获取用户信息和任务标题
+        users_db = db.load_data("users.json", default_data={})
+        tasks_db = db.load_data("tasks.json", default_data=[])
+        author_info = users_db.get(publisher, {})
+        tipper_info = users_db.get(current_user, {})
+        author_name = author_info.get("name", publisher)
+        tipper_name = tipper_info.get("name", current_user)
+        
+        task_title = None
+        for task in tasks_db:
+            if task["id"] == task_id:
+                task_title = task.get("title")
+                break
+        
+        # 打赏方交易记录 (TIP_OUT)
+        tx_tipper = Transaction(
+            tx_id=tx_id_tipper,
+            account=current_user,
+            tx_type="TIP_OUT",
+            amount=-amount,
+            related_account=publisher,
+            item_id=task_id,
+            prev_hash=prev_hash_tipper,
+            tx_hash=calculate_tx_hash(tx_id_tipper, current_user, "TIP_OUT", -amount, prev_hash_tipper),
+            description=f"打赏给 {author_name}" + (f" 的任务《{task_title}》" if task_title else ""),
+            item_title=task_title,
+            item_type="task",
+            related_user_name=author_name
+        )
+        
+        # 接收方交易记录 (TIP_IN)
+        tx_author = Transaction(
+            tx_id=tx_id_author,
+            account=publisher,
+            tx_type="TIP_IN",
+            amount=amount,
+            related_account=current_user,
+            item_id=task_id,
+            prev_hash=prev_hash_author,
+            tx_hash=calculate_tx_hash(tx_id_author, publisher, "TIP_IN", amount, prev_hash_author),
+            description=f"收到 {tipper_name} 的任务打赏" + (f" ({task_title})" if task_title else ""),
+            item_title=task_title,
+            item_type="task",
+            related_user_name=tipper_name if not is_anon else "匿名用户"
+        )
+        
+        db_session.add(tx_tipper)
+        db_session.add(tx_author)
+        db_session.commit()
+        
+        # 📝 审计日志
+        logger.info(f"TASK_TIP | from={current_user} | to={publisher} | amount={amount} | task={task_id} | anon={is_anon}")
+        
+        # 🔔 打赏通知（考虑匿名）
+        if not is_anon:
+            add_notification(publisher, {
+                "type": "tip",
+                "from_user": current_user,
+                "target_item_id": task_id,
+                "target_item_title": task_title or "",
+                "content": f"您收到来自 {tipper_name} 的 {amount} 积分任务打赏"
+            })
+        else:
+            add_notification(publisher, {
+                "type": "tip",
+                "from_user": "anonymous",
+                "target_item_id": task_id,
+                "target_item_title": "",
+                "content": f"您收到了一份 {amount} 积分的匿名任务打赏"
+            })
+        
+        # 🗂️ 清除排序缓存（打赏可能影响排序）
+        sort_cache.invalidate("tasks:")
+        
+    except HTTPException:
+        db_session.rollback()
+        raise
+    except Exception as e:
+        db_session.rollback()
+        logger.error(f"TASK_TIP_ERROR | from={current_user} | task={task_id} | amount={amount} | error={str(e)}")
+        raise HTTPException(status_code=500, detail="打赏处理失败，请稍后重试")
     
     return result
 

router_wallet.py

@@ -292,7 +292,20 @@ async def purchase_item(request: Request, req: PurchaseRequest, db: Session = De
     seller_account = item.get("author")
     
     if price <= 0 or req.account == seller_account:
-        # ☁️ 免费资源或作者本人，也返回网盘密码
+        # ☁️ 免费资源或作者本人，确保 Ownership 记录存在
+        existing_ownership = db.query(Ownership).filter(
+            Ownership.account == req.account,
+            Ownership.item_id == req.item_id,
+            Ownership.is_refunded == False
+        ).first()
+        if not existing_ownership:
+            new_ownership = Ownership(
+                account=req.account,
+                item_id=req.item_id,
+                price_paid=0
+            )
+            db.add(new_ownership)
+            db.commit()
         return {
             "status": "success", 
             "already_owned": True,
@@ -337,7 +350,8 @@ async def purchase_item(request: Request, req: PurchaseRequest, db: Session = De
             db.add(seller_wallet)
             
         buyer_wallet.balance = (buyer_wallet.balance or 0) - price
-        seller_wallet.earn_balance = (seller_wallet.earn_balance or 0) + price
+        seller_wallet.balance = (seller_wallet.balance or 0) + price          # 实际收入进统一余额
+        seller_wallet.earn_balance = (seller_wallet.earn_balance or 0) + price     # 累计销售收益统计（只增不减）
         
         # 🔄 P7后悔模式：记录购买价格
         new_ownership = Ownership(account=req.account, item_id=req.item_id, price_paid=price)
@@ -401,6 +415,18 @@ async def tip_user(request: Request, req: TipRequest, db: Session = Depends(get_
     if req.sender_account == req.target_account:
         raise HTTPException(status_code=400, detail="不能打赏给自己")
     
+    # 🔒 P1幂等性防护：检查最近5秒内是否存在相同交易
+    recent_cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=5)
+    duplicate_tx = db.query(Transaction).filter(
+        Transaction.account == req.sender_account,
+        Transaction.tx_type == "TIP_OUT",
+        Transaction.amount == -req.amount,
+        Transaction.related_account == req.target_account,
+        Transaction.created_at >= recent_cutoff
+    ).first()
+    if duplicate_tx:
+        return {"status": "success", "message": "打赏已处理（重复请求）"}
+    
     # 🔒 P1并发安全：使用悲观锁+异常处理+事务回滚防止双花问题
     try:
         sender_wallet = db.query(Wallet).filter(Wallet.account == req.sender_account).with_for_update().first()
@@ -413,7 +439,8 @@ async def tip_user(request: Request, req: TipRequest, db: Session = Depends(get_
             db.add(target_wallet)
             
         sender_wallet.balance -= req.amount
-        target_wallet.tip_balance += req.amount
+        target_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]}"
@@ -538,6 +565,21 @@ async def tip_user(request: Request, req: TipRequest, db: Session = Depends(get_
 @router.post("/api/wallet/withdraw")
 @limiter.limit("3/minute")  # 🔒 P0安全优化：提现每分钟最多3次
 async def withdraw(request: Request, req: WithdrawRequest, db: Session = Depends(get_db)):
+    # 🔒 P1幂等性防护：检查最近10秒内是否存在相同提现请求
+    recent_cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=10)
+    duplicate_tx = db.query(Transaction).filter(
+        Transaction.account == req.account,
+        Transaction.tx_type == "WITHDRAW",
+        Transaction.amount == -req.amount,
+        Transaction.created_at >= recent_cutoff
+    ).first()
+    if duplicate_tx:
+        return {"status": "success", "message": "提现申请已提交（重复请求）"}
+    
+    # 🔒 最小金额检查
+    if req.amount < 1:
+        raise HTTPException(status_code=400, detail="最小提现金额为1积分")
+    
     # 🔒 验证码缓存键格式：{contact}_{action_type}，与 send_code 接口一致
     # 先通过账号查询用户邮箱，再用邮箱构建缓存键
     users_db = json_db.load_data("users.json", default_data={})
@@ -566,8 +608,8 @@ async def withdraw(request: Request, req: WithdrawRequest, db: Session = Depends
         ).scalar() or 0
         total_withdrawn = abs(withdrawals_sum)
         
-        # 手续费规则：100元 = 10000积分 免手续费额度，超出部分收取 10%
-        free_quota = max(0, 10000 - total_withdrawn)  # 剩余免责额度
+        # 手续费规则：100积分免手续费额度，超出部分收取 10%
+        free_quota = max(0, 100 - total_withdrawn)  # 剩余免责额度
         fee_amount = 0
         if req.amount > free_quota:
             fee_amount = int((req.amount - free_quota) * 0.10)  # 只对超出部分收 10%
@@ -575,17 +617,11 @@ async def withdraw(request: Request, req: WithdrawRequest, db: Session = Depends
         actual_withdraw = req.amount  # 从账户扣除的金额
         net_amount = req.amount - fee_amount  # 用户实际到账金额
             
-        total_withdrawable = wallet.earn_balance + wallet.tip_balance
-        if actual_withdraw > total_withdrawable:
+        # 可提现额度改为 wallet.balance（统一余额）
+        if actual_withdraw > wallet.balance:
             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.balance -= actual_withdraw  # 直接从统一余额扣除
         wallet.frozen_balance += net_amount  # 冻结的是到账金额，非手续费部分
         
         tx_id = f"WD_{int(time.time())}_{uuid.uuid4().hex[:6]}"
@@ -880,16 +916,11 @@ async def refund_purchase(request: Request, account: str, item_id: str, db: Sess
         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)
+            seller_wallet.balance -= refund_amount          # 卖家扣回
+            seller_wallet.earn_balance = max(0, seller_wallet.earn_balance - refund_amount)     # 累计统计回退（防止负数）
         
         if buyer_wallet:
-            buyer_wallet.balance += refund_amount
+            buyer_wallet.balance += refund_amount           # 买家退款
         else:
             buyer_wallet = Wallet(account=account, balance=refund_amount)
             db.add(buyer_wallet)