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

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

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