buffer_platform/app/analysis_migration.py

"""
期货智析数据库迁移 - SQLite → MySQL
"""
import logging

from sqlalchemy.orm import sessionmaker

from app.analysis_db import analysis_engine, AnalysisBase
from app.analysis_models import (
    FuturesAnalysis, WatchedSymbol, AIModelConfig, AnalysisSettings,
    AIAnalysisCache, ReviewDate, SymbolRanking, TradingPlan,
    SymbolScoreV2, TradingPlanV2, SectorHeat, ReviewPlanV2,
    TradeRecord, TradeImportBatch,
)
from app.mysql_database import mysql_engine

logger = logging.getLogger(__name__)

# 所有 analysis 模型类，按依赖顺序排列
ANALYSIS_MODELS = [
    FuturesAnalysis, WatchedSymbol, AIModelConfig, AnalysisSettings,
    AIAnalysisCache, ReviewDate, SymbolRanking, TradingPlan,
    SymbolScoreV2, TradingPlanV2, SectorHeat, ReviewPlanV2,
    TradeRecord, TradeImportBatch,
]


def migrate_analysis_sqlite_to_mysql():
    """
    将 SQLite futures_analysis.db 中的数据迁移到 MySQL。

    Returns:
        bool: 迁移是否成功（跳过也视为成功）
    """
    if mysql_engine is None:
        logger.warning("MySQL 未初始化，跳过 analysis 数据迁移")
        return True

    try:
        mysql_count = _count_mysql_analysis_records()
        if mysql_count > 0:
            logger.info(f"MySQL analysis 表已有数据 ({mysql_count} 条)，跳过迁移")
            return True

        logger.info("开始将 SQLite analysis 数据迁移到 MySQL...")
        data = _read_sqlite_analysis_data()
        if not data:
            logger.info("SQLite analysis 数据库为空，无需迁移")
            return True

        if _write_mysql_analysis_data(data):
            total = sum(len(records) for records in data.values())
            logger.info(f"Analysis 数据迁移完成，共迁移 {total} 条记录")
            return True
        else:
            logger.error("Analysis 数据迁移失败")
            return False

    except Exception as e:
        logger.error(f"Analysis 数据迁移异常: {e}")
        return False


def _count_mysql_analysis_records():
    """统计 MySQL 中 analysis 表的总记录数。"""
    SessionLocal = sessionmaker(bind=mysql_engine)
    total = 0
    with SessionLocal() as db:
        for model in ANALYSIS_MODELS:
            try:
                total += db.query(model).count()
            except Exception as e:
                logger.warning(f"统计 MySQL {model.__tablename__} 记录数失败: {e}")
    return total


def _read_sqlite_analysis_data():
    """从 SQLite 读取所有 analysis 数据。"""
    SessionLocal = sessionmaker(bind=analysis_engine)
    data = {}

    with SessionLocal() as db:
        for model in ANALYSIS_MODELS:
            try:
                records = db.query(model).all()
                if records:
                    data[model.__tablename__] = records
                    logger.info(f"从 SQLite 读取 {model.__tablename__}: {len(records)} 条")
            except Exception as e:
                logger.warning(f"读取 SQLite {model.__tablename__} 失败: {e}")

    return data


def _write_mysql_analysis_data(data):
    """将 analysis 数据写入 MySQL。"""
    SessionLocal = sessionmaker(bind=mysql_engine)

    with SessionLocal() as db:
        try:
            for model in ANALYSIS_MODELS:
                records = data.get(model.__tablename__, [])
                if not records:
                    continue

                for record in records:
                    # 创建新对象，避免绑定到 SQLite session
                    db.expunge(record)
                    db.add(record)

            db.commit()
            return True
        except Exception as e:
            db.rollback()
            logger.error(f"写入 MySQL analysis 数据失败: {e}")
            return False