Delete 迁移_余额合并.py

迁移_余额合并.py

@@ -1,262 +0,0 @@
-#!/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()