Upload 迁移_余额合并.py

迁移_余额合并.py

@@ -0,0 +1,262 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+迁移脚本：合并用户钱包余额
+====================================
+功能：将现有用户的 earn_balance + tip_balance 合并到 balance 中
+
+执行逻辑：
+    wallet.balance += wallet.earn_balance + wallet.tip_balance
+    # earn_balance 和 tip_balance 保留原值不变（作为历史累计统计）
+
+用法：
+    python 迁移_余额合并.py           # 预览模式（dry-run，默认）
+    python 迁移_余额合并.py --execute # 实际执行迁移
+
+作者：数据迁移脚本
+创建日期：2026-04-10
+"""
+
+import os
+import sys
+import argparse
+from typing import Dict, List, Tuple
+from dataclasses import dataclass
+
+# 添加当前目录到路径，确保可以导入项目模块
+script_dir = os.path.dirname(os.path.abspath(__file__))
+if script_dir not in sys.path:
+    sys.path.insert(0, script_dir)
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker, Session
+from sqlalchemy.exc import SQLAlchemyError
+
+# 导入项目配置
+from database_sql import Base, SQLALCHEMY_DATABASE_URL
+from models_sql import Wallet
+
+
+@dataclass
+class MigrationStats:
+    """迁移统计信息"""
+    total_users: int = 0           # 总用户数
+    affected_users: int = 0        # 受影响用户数（需要迁移的）
+    skipped_users: int = 0         # 跳过的用户数（earn 和 tip 都为0）
+    
+    # 金额统计（单位：分）
+    total_balance_before: int = 0  # 迁移前总 balance
+    total_balance_after: int = 0   # 迁移后总 balance
+    total_earn_merged: int = 0     # 合并的 earn_balance 总额
+    total_tip_merged: int = 0      # 合并的 tip_balance 总额
+    
+    def __str__(self) -> str:
+        return f"""
+========== 迁移统计报告 ==========
+用户统计：
+  - 总用户数: {self.total_users}
+  - 受影响用户数: {self.affected_users}
+  - 跳过用户数 (earn=0 且 tip=0): {self.skipped_users}
+
+金额统计（单位：分）：
+  - 迁移前总 balance: {self.total_balance_before}
+  - 迁移后总 balance: {self.total_balance_after}
+  - 合并的 earn_balance: {self.total_earn_merged}
+  - 合并的 tip_balance: {self.total_tip_merged}
+  - 总增加金额: {self.total_earn_merged + self.total_tip_merged}
+
+==================================
+"""
+
+
+def get_database_session() -> Session:
+    """创建数据库会话"""
+    engine = create_engine(SQLALCHEMY_DATABASE_URL)
+    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    return SessionLocal()
+
+
+def fetch_wallet_data(session: Session) -> List[Dict]:
+    """获取所有钱包数据"""
+    wallets = session.query(Wallet).all()
+    return [
+        {
+            "account": w.account,
+            "balance": w.balance or 0,
+            "earn_balance": w.earn_balance or 0,
+            "tip_balance": w.tip_balance or 0,
+        }
+        for w in wallets
+    ]
+
+
+def calculate_migration(wallets: List[Dict]) -> Tuple[List[Dict], MigrationStats]:
+    """
+    计算迁移结果
+    
+    返回：
+        - 需要更新的钱包列表
+        - 统计信息
+    """
+    stats = MigrationStats()
+    updates = []
+    
+    stats.total_users = len(wallets)
+    
+    for wallet in wallets:
+        account = wallet["account"]
+        balance = wallet["balance"]
+        earn = wallet["earn_balance"]
+        tip = wallet["tip_balance"]
+        
+        stats.total_balance_before += balance
+        
+        # 跳过 earn 和 tip 都为 0 的用户
+        if earn == 0 and tip == 0:
+            stats.skipped_users += 1
+            stats.total_balance_after += balance
+            continue
+        
+        # 计算新 balance
+        new_balance = balance + earn + tip
+        
+        stats.affected_users += 1
+        stats.total_earn_merged += earn
+        stats.total_tip_merged += tip
+        stats.total_balance_after += new_balance
+        
+        updates.append({
+            "account": account,
+            "old_balance": balance,
+            "new_balance": new_balance,
+            "earn": earn,
+            "tip": tip,
+        })
+    
+    return updates, stats
+
+
+def preview_migration(updates: List[Dict], stats: MigrationStats):
+    """打印迁移预览信息"""
+    print("\n========== 迁移预览模式 ==========")
+    print("⚠️  当前为预览模式，不会实际修改数据")
+    print("   添加 --execute 参数以实际执行迁移\n")
+    
+    print(f"发现 {stats.affected_users} 个用户需要迁移：\n")
+    
+    # 打印前10条详细信息
+    display_limit = 10
+    for i, update in enumerate(updates[:display_limit], 1):
+        print(f"  {i}. 用户: {update['account']}")
+        print(f"     当前 balance: {update['old_balance']} 分")
+        print(f"     earn_balance: {update['earn']} 分")
+        print(f"     tip_balance: {update['tip']} 分")
+        print(f"     新 balance: {update['new_balance']} 分 (+{update['earn'] + update['tip']})\n")
+    
+    if len(updates) > display_limit:
+        print(f"  ... 还有 {len(updates) - display_limit} 个用户未显示\n")
+    
+    print(stats)
+
+
+def execute_migration(session: Session, updates: List[Dict], stats: MigrationStats):
+    """
+    执行实际迁移
+    
+    在事务中批量执行，确保原子性
+    """
+    print("\n========== 开始执行迁移 ==========")
+    print(f"将更新 {len(updates)} 个用户的钱包数据\n")
+    
+    try:
+        # 开始事务
+        for update in updates:
+            account = update["account"]
+            new_balance = update["new_balance"]
+            
+            # 使用乐观锁更新，防止并发问题
+            result = session.query(Wallet).filter(
+                Wallet.account == account
+            ).update({
+                Wallet.balance: new_balance
+            }, synchronize_session=False)
+            
+            if result == 0:
+                print(f"⚠️  用户 {account} 更新失败：未找到记录")
+        
+        # 提交事务
+        session.commit()
+        
+        print("✅ 迁移执行成功！")
+        print(stats)
+        
+    except SQLAlchemyError as e:
+        session.rollback()
+        print(f"\n🚨 迁移失败，已回滚所有更改：")
+        print(f"   错误类型: {type(e).__name__}")
+        print(f"   错误信息: {str(e)}")
+        raise
+
+
+def main():
+    """主函数"""
+    parser = argparse.ArgumentParser(
+        description="合并用户钱包余额迁移脚本",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+示例：
+  python 迁移_余额合并.py           # 预览模式
+  python 迁移_余额合并.py --execute # 实际执行
+        """
+    )
+    parser.add_argument(
+        "--execute",
+        action="store_true",
+        help="实际执行迁移（默认仅预览）"
+    )
+    
+    args = parser.parse_args()
+    
+    session = None
+    try:
+        # 获取数据库会话
+        print("正在连接数据库...")
+        session = get_database_session()
+        print("✅ 数据库连接成功\n")
+        
+        # 获取钱包数据
+        print("正在获取钱包数据...")
+        wallets = fetch_wallet_data(session)
+        print(f"✅ 获取到 {len(wallets)} 个用户钱包数据\n")
+        
+        if len(wallets) == 0:
+            print("⚠️  没有用户钱包数据，无需迁移")
+            return
+        
+        # 计算迁移
+        updates, stats = calculate_migration(wallets)
+        
+        if stats.affected_users == 0:
+            print("✅ 没有需要迁移的用户（所有用户的 earn_balance 和 tip_balance 都为 0）")
+            print(stats)
+            return
+        
+        # 预览或执行
+        if args.execute:
+            execute_migration(session, updates, stats)
+        else:
+            preview_migration(updates, stats)
+            
+    except Exception as e:
+        print(f"\n🚨 程序执行出错：")
+        print(f"   错误类型: {type(e).__name__}")
+        print(f"   错误信息: {str(e)}")
+        sys.exit(1)
+    finally:
+        if session:
+            session.close()
+            print("\n数据库连接已关闭")
+
+
+if __name__ == "__main__":
+    main()