Delete 迁移_补全历史SALE记录.py

迁移_补全历史SALE记录.py

@@ -1,351 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-迁移脚本：为历史购买记录补全卖家 SALE 交易记录
-====================================================
-背景：
-    之前购买接口只为买家生成 PURCHASE 记录，未为卖家生成 SALE 记录。
-    现已修复，新购买会同步生成双方记录。
-    本脚本为历史缺失的 SALE 记录进行补全。
-
-迁移逻辑：
-    1. 查询所有 tx_type="PURCHASE" 的 Transaction 记录
-    2. 检查是否已存在对应 SALE 记录（幂等保障）
-       - 匹配条件：相同 item_id + 卖家account(related_account) + tx_type="SALE" + 金额相同
-    3. 不存在则创建 SALE 记录，created_at 与 PURCHASE 保持一致
-
-用法：
-    python 迁移_补全历史SALE记录.py           # 预览模式（dry-run，默认）
-    python 迁移_补全历史SALE记录.py --execute # 实际执行迁移
-
-作者：数据迁移脚本
-创建日期：2026-05-20
-"""
-
-import os
-import sys
-import time
-import uuid
-import argparse
-from dataclasses import dataclass, field
-from typing import List, Dict, Optional, Tuple
-
-# 添加云端Space代码目录到路径
-script_dir = os.path.dirname(os.path.abspath(__file__))
-parent_dir = os.path.dirname(script_dir)
-if parent_dir not in sys.path:
-    sys.path.insert(0, parent_dir)
-
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker, Session
-from sqlalchemy.exc import SQLAlchemyError
-
-from database_sql import SQLALCHEMY_DATABASE_URL
-from models_sql import Transaction
-
-
-# ==========================================
-# 📊 统计数据结构
-# ==========================================
-
-@dataclass
-class MigrationStats:
-    """迁移统计信息"""
-    total_purchase: int = 0       # PURCHASE 记录总数
-    already_exists: int = 0       # 已有对应 SALE 记录，跳过
-    skipped_no_seller: int = 0    # 无卖家账号（related_account 为空），跳过
-    created: int = 0              # 新创建 SALE 记录数
-    failed: int = 0               # 单条处理失败数
-    failed_tx_ids: List[str] = field(default_factory=list)  # 失败的 PURCHASE tx_id
-
-    def __str__(self) -> str:
-        return f"""
-========== 迁移统计报告 ==========
-PURCHASE 记录总数:        {self.total_purchase}
-已有对应 SALE（跳过）:    {self.already_exists}
-无卖家账号（跳过）:       {self.skipped_no_seller}
-新创建 SALE 记录:         {self.created}
-处理失败:                 {self.failed}
-{f"  失败的 tx_id: {self.failed_tx_ids}" if self.failed_tx_ids else ""}
-==================================
-"""
-
-
-# ==========================================
-# 🔗 数据库连接
-# ==========================================
-
-def get_database_session() -> Session:
-    """创建数据库会话"""
-    engine = create_engine(SQLALCHEMY_DATABASE_URL)
-    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-    return SessionLocal()
-
-
-# ==========================================
-# 🔍 核心检测逻辑
-# ==========================================
-
-def sale_record_exists(session: Session, purchase: Transaction) -> bool:
-    """
-    检查是否已存在对应的 SALE 记录（幂等校验）
-
-    匹配条件：
-        - tx_type = "SALE"
-        - account = purchase.related_account（卖家）
-        - item_id = purchase.item_id
-        - amount = abs(purchase.amount)（PURCHASE 为负数，SALE 为正数）
-    """
-    seller_account = purchase.related_account
-    if not seller_account:
-        return False
-
-    sale_amount = abs(purchase.amount)
-
-    existing = session.query(Transaction).filter(
-        Transaction.tx_type == "SALE",
-        Transaction.account == seller_account,
-        Transaction.item_id == purchase.item_id,
-        Transaction.amount == sale_amount,
-        Transaction.related_account == purchase.account,
-    ).first()
-
-    return existing is not None
-
-
-def build_sale_record(purchase: Transaction) -> Optional[Dict]:
-    """
-    根据 PURCHASE 记录构造 SALE 记录的字段
-
-    返回 None 表示无法构造（如无卖家账号）
-    """
-    seller_account = purchase.related_account
-    if not seller_account:
-        return None
-
-    item_title = purchase.item_title or purchase.item_id or ""
-    sale_amount = abs(purchase.amount)
-
-    # tx_id 格式：SALE_MIGRATE_{原PURCHASE的tx_id尾段}_{随机6位}
-    # 取 PURCHASE tx_id 的最后一段（通常为 BUY_{ts}_{hex6}，取 hex6 部分）
-    purchase_suffix = purchase.tx_id.rsplit("_", 1)[-1] if "_" in purchase.tx_id else purchase.tx_id[-6:]
-    sale_tx_id = f"SALE_MIGRATE_{purchase_suffix}_{uuid.uuid4().hex[:6]}"
-
-    return {
-        "tx_id": sale_tx_id,
-        "account": seller_account,
-        "tx_type": "SALE",
-        "amount": sale_amount,
-        "related_account": purchase.account,
-        "item_id": purchase.item_id,
-        "item_title": purchase.item_title,
-        "item_type": purchase.item_type,
-        "related_user_name": purchase.related_user_name,
-        "description": f"销售收入(历史补全): {item_title}",
-        "created_at": purchase.created_at,
-        # 迁移记录无需防篡改哈希链，留空
-        "prev_hash": "MIGRATE_NO_HASH",
-        "tx_hash": "MIGRATE_NO_HASH",
-    }
-
-
-# ==========================================
-# 👁️ 预览模式
-# ==========================================
-
-def preview_migration(session: Session) -> Tuple[List[Dict], MigrationStats]:
-    """
-    预览模式：扫描所有 PURCHASE 记录，返回待创建列表和统计
-
-    不写入数据库。
-    """
-    print("\n========== 迁移预览模式 ==========")
-    print("⚠️  当前为预览模式，不会实际修改数据")
-    print("   添加 --execute 参数以实际执行迁移\n")
-
-    stats = MigrationStats()
-    to_create: List[Dict] = []
-
-    purchases = session.query(Transaction).filter(
-        Transaction.tx_type == "PURCHASE"
-    ).order_by(Transaction.created_at.asc()).all()
-
-    stats.total_purchase = len(purchases)
-    print(f"共发现 {stats.total_purchase} 条 PURCHASE 记录，开始扫描...\n")
-
-    for purchase in purchases:
-        # 无卖家
-        if not purchase.related_account:
-            stats.skipped_no_seller += 1
-            continue
-
-        # 已有 SALE
-        if sale_record_exists(session, purchase):
-            stats.already_exists += 1
-            continue
-
-        # 构造 SALE 记录
-        sale_data = build_sale_record(purchase)
-        if sale_data:
-            to_create.append(sale_data)
-            stats.created += 1
-
-    # 打印预览详情（最多前 20 条）
-    print(f"发现 {len(to_create)} 条需要补全的 SALE 记录：\n")
-    display_limit = 20
-    for i, sale in enumerate(to_create[:display_limit], 1):
-        print(f"  {i:3d}. 卖家: {sale['account']}")
-        print(f"       买家: {sale['related_account']}")
-        print(f"       资源: {sale['item_id']} | {sale['item_title'] or '(无标题)'}")
-        print(f"       金额: +{sale['amount']} 分")
-        print(f"       时间: {sale['created_at']}")
-        print(f"       新 tx_id: {sale['tx_id']}\n")
-
-    if len(to_create) > display_limit:
-        print(f"  ... 还有 {len(to_create) - display_limit} 条未显示\n")
-
-    # 预览模式下 created 字段表示"待创建"数量
-    print(stats)
-    return to_create, stats
-
-
-# ==========================================
-# ⚡ 执行模式
-# ==========================================
-
-BATCH_SIZE = 100  # 每批提交条数
-
-
-def execute_migration(session: Session):
-    """
-    执行实际迁移：
-      - 逐条处理，幂等检测
-      - 每 BATCH_SIZE 条 commit 一次
-      - 单条失败不影响其他记录
-    """
-    print("\n========== 开始执行迁移 ==========\n")
-
-    stats = MigrationStats()
-
-    purchases = session.query(Transaction).filter(
-        Transaction.tx_type == "PURCHASE"
-    ).order_by(Transaction.created_at.asc()).all()
-
-    stats.total_purchase = len(purchases)
-    print(f"共发现 {stats.total_purchase} 条 PURCHASE 记录，开始处理...\n")
-
-    batch_pending = 0  # 当前批次待提交数
-
-    for idx, purchase in enumerate(purchases, 1):
-        try:
-            # ---- 无卖家，跳过 ----
-            if not purchase.related_account:
-                stats.skipped_no_seller += 1
-                continue
-
-            # ---- 已有 SALE，跳过 ----
-            if sale_record_exists(session, purchase):
-                stats.already_exists += 1
-                continue
-
-            # ---- 构造并插入 SALE 记录 ----
-            sale_data = build_sale_record(purchase)
-            if not sale_data:
-                stats.skipped_no_seller += 1
-                continue
-
-            sale_tx = Transaction(**sale_data)
-            session.add(sale_tx)
-            stats.created += 1
-            batch_pending += 1
-
-            # ---- 每 BATCH_SIZE 条提交一次 ----
-            if batch_pending >= BATCH_SIZE:
-                session.commit()
-                print(f"  [进度] 已处理 {idx}/{stats.total_purchase}，"
-                      f"本批次提交 {batch_pending} 条，"
-                      f"累计创建 {stats.created} 条，"
-                      f"跳过 {stats.already_exists + stats.skipped_no_seller} 条")
-                batch_pending = 0
-
-        except SQLAlchemyError as e:
-            # 单条失败：回滚当前批次，记录错误，继续下一条
-            session.rollback()
-            batch_pending = 0
-            stats.failed += 1
-            stats.failed_tx_ids.append(purchase.tx_id)
-            print(f"  ⚠️  处理 PURCHASE {purchase.tx_id} 失败: {type(e).__name__}: {e}")
-
-        except Exception as e:
-            session.rollback()
-            batch_pending = 0
-            stats.failed += 1
-            stats.failed_tx_ids.append(purchase.tx_id)
-            print(f"  ⚠️  处理 PURCHASE {purchase.tx_id} 异常: {type(e).__name__}: {e}")
-
-    # ---- 提交剩余记录 ----
-    if batch_pending > 0:
-        try:
-            session.commit()
-            print(f"  [进度] 最后一批提交 {batch_pending} 条")
-        except SQLAlchemyError as e:
-            session.rollback()
-            stats.failed += batch_pending
-            stats.created -= batch_pending
-            print(f"  🚨 最后一批提交失败: {e}")
-
-    print("\n✅ 迁移执行完毕！")
-    print(stats)
-
-
-# ==========================================
-# 🚀 入口
-# ==========================================
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="为历史购买记录补全卖家 SALE 交易记录",
-        formatter_class=argparse.RawDescriptionHelpFormatter,
-        epilog="""
-示例：
-  python 迁移_补全历史SALE记录.py           # 预览模式（默认）
-  python 迁移_补全历史SALE记录.py --execute # 实际执行
-        """
-    )
-    parser.add_argument(
-        "--execute",
-        action="store_true",
-        help="实际执行迁移（默认仅预览）"
-    )
-    args = parser.parse_args()
-
-    session = None
-    try:
-        print("正在连接数据库...")
-        session = get_database_session()
-        print("✅ 数据库连接成功\n")
-
-        if args.execute:
-            execute_migration(session)
-        else:
-            preview_migration(session)
-
-    except KeyboardInterrupt:
-        print("\n⚠️  用户中断，已退出")
-        if session:
-            session.rollback()
-    except Exception as e:
-        print(f"\n🚨 程序执行出错：")
-        print(f"   错误类型: {type(e).__name__}")
-        print(f"   错误信息: {str(e)}")
-        if session:
-            session.rollback()
-        sys.exit(1)
-    finally:
-        if session:
-            session.close()
-            print("\n数据库连接已关闭")
-
-
-if __name__ == "__main__":
-    main()