MuMuAINovel/backend/scripts/migrate_sqlite_to_postgres.py

#!/usr/bin/env python3
"""
SQLite to PostgreSQL 数据迁移脚本

使用方法:
    python backend/scripts/migrate_sqlite_to_postgres.py

前置条件:
    1. PostgreSQL数据库已创建
    2. .env文件中DATABASE_URL已配置为PostgreSQL
    3. SQLite数据文件存在于 backend/data/ 目录
"""
import asyncio
import sys
from pathlib import Path
from typing import List, Dict, Any
import logging
from datetime import datetime

# 添加项目根目录到Python路径
sys.path.insert(0, str(Path(__file__).parent.parent))

from sqlalchemy import create_engine, text, select
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from app.database import Base
from app.models import (
    Project, Outline, Character, Chapter, GenerationHistory,
    Settings, WritingStyle, ProjectDefaultStyle,
    RelationshipType, CharacterRelationship, Organization, OrganizationMember,
    StoryMemory, PlotAnalysis, AnalysisTask, BatchGenerationTask,
    MCPPlugin
)
from app.config import settings

# 创建日志目录
log_dir = Path(__file__).parent.parent / "logs"
log_dir.mkdir(exist_ok=True)

# 生成日志文件名（带时间戳）
log_filename = log_dir / f"migration_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"

# 设置日志 - 同时输出到控制台和文件
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),  # 控制台输出
        logging.FileHandler(log_filename, encoding='utf-8')  # 文件输出
    ]
)
logger = logging.getLogger(__name__)
logger.info(f"📝 日志文件: {log_filename}")


class SQLiteToPostgresMigrator:
    """SQLite到PostgreSQL的数据迁移器"""
    
    def __init__(self, sqlite_dir: Path, target_user_id: str):
        """
        初始化迁移器
        
        Args:
            sqlite_dir: SQLite数据库文件目录
            target_user_id: 目标用户ID（迁移后的数据归属）
        """
        self.sqlite_dir = sqlite_dir
        self.target_user_id = target_user_id
        self.sqlite_files = list(sqlite_dir.glob("ai_story_user_*.db"))
        
        # PostgreSQL连接
        if "postgresql" not in settings.database_url:
            raise ValueError("DATABASE_URL必须配置为PostgreSQL")
        
        self.pg_engine = create_async_engine(
            settings.database_url,
            echo=False,
            pool_pre_ping=True
        )
        
        self.pg_session_maker = async_sessionmaker(
            self.pg_engine,
            class_=AsyncSession,
            expire_on_commit=False
        )
    
    async def migrate_all(self):
        """迁移所有SQLite数据库"""
        if not self.sqlite_files:
            logger.warning(f"未找到SQLite数据库文件: {self.sqlite_dir}")
            return
        
        logger.info(f"找到 {len(self.sqlite_files)} 个SQLite数据库文件")
        
        # 创建PostgreSQL表结构
        await self._create_tables()
        
        # 初始化关系类型数据
        await self._init_relationship_types()
        
        # 逐个迁移
        for sqlite_file in self.sqlite_files:
            await self._migrate_single_db(sqlite_file)
        
        # 重置自增序列
        await self._reset_sequences()
        
        logger.info("✅ 所有数据迁移完成")
    
    async def _create_tables(self):
        """创建PostgreSQL表结构"""
        logger.info("创建PostgreSQL表结构...")
        async with self.pg_engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        logger.info("✅ 表结构创建完成")
    
    async def _init_relationship_types(self):
        """初始化关系类型数据"""
        logger.info("初始化关系类型数据...")
        
        # 预置关系类型数据
        relationship_types = [
            # 家族关系
            {"name": "父亲", "category": "family", "reverse_name": "子女", "intimacy_range": "high", "icon": "👨"},
            {"name": "母亲", "category": "family", "reverse_name": "子女", "intimacy_range": "high", "icon": "👩"},
            {"name": "兄弟", "category": "family", "reverse_name": "兄弟", "intimacy_range": "high", "icon": "👬"},
            {"name": "姐妹", "category": "family", "reverse_name": "姐妹", "intimacy_range": "high", "icon": "👭"},
            {"name": "子女", "category": "family", "reverse_name": "父母", "intimacy_range": "high", "icon": "👶"},
            {"name": "配偶", "category": "family", "reverse_name": "配偶", "intimacy_range": "high", "icon": "💑"},
            {"name": "恋人", "category": "family", "reverse_name": "恋人", "intimacy_range": "high", "icon": "💕"},
            
            # 社交关系
            {"name": "师父", "category": "social", "reverse_name": "徒弟", "intimacy_range": "high", "icon": "🎓"},
            {"name": "徒弟", "category": "social", "reverse_name": "师父", "intimacy_range": "high", "icon": "📚"},
            {"name": "朋友", "category": "social", "reverse_name": "朋友", "intimacy_range": "medium", "icon": "🤝"},
            {"name": "同学", "category": "social", "reverse_name": "同学", "intimacy_range": "medium", "icon": "🎒"},
            {"name": "邻居", "category": "social", "reverse_name": "邻居", "intimacy_range": "low", "icon": "🏘️"},
            {"name": "知己", "category": "social", "reverse_name": "知己", "intimacy_range": "high", "icon": "💙"},
            
            # 职业关系
            {"name": "上司", "category": "professional", "reverse_name": "下属", "intimacy_range": "low", "icon": "👔"},
            {"name": "下属", "category": "professional", "reverse_name": "上司", "intimacy_range": "low", "icon": "💼"},
            {"name": "同事", "category": "professional", "reverse_name": "同事", "intimacy_range": "medium", "icon": "🤵"},
            {"name": "合作伙伴", "category": "professional", "reverse_name": "合作伙伴", "intimacy_range": "medium", "icon": "🤜🤛"},
            
            # 敌对关系
            {"name": "敌人", "category": "hostile", "reverse_name": "敌人", "intimacy_range": "low", "icon": "⚔️"},
            {"name": "仇人", "category": "hostile", "reverse_name": "仇人", "intimacy_range": "low", "icon": "💢"},
            {"name": "竞争对手", "category": "hostile", "reverse_name": "竞争对手", "intimacy_range": "low", "icon": "🎯"},
            {"name": "宿敌", "category": "hostile", "reverse_name": "宿敌", "intimacy_range": "low", "icon": "⚡"},
        ]
        
        try:
            async with self.pg_session_maker() as session:
                # 检查是否已经有数据
                result = await session.execute(select(RelationshipType))
                existing = result.scalars().first()
                
                if existing:
                    logger.info("关系类型数据已存在，跳过初始化")
                    return
                
                # 插入预置数据
                logger.info("开始插入关系类型数据...")
                for rt_data in relationship_types:
                    relationship_type = RelationshipType(**rt_data)
                    session.add(relationship_type)
                
                await session.commit()
                logger.info(f"✅ 成功插入 {len(relationship_types)} 条关系类型数据")
                
        except Exception as e:
            logger.error(f"初始化关系类型数据失败: {str(e)}", exc_info=True)
            # 不抛出异常，继续迁移流程
            logger.warning("关系类型初始化失败，将跳过有外键依赖的记录")
    
    async def _migrate_single_db(self, sqlite_file: Path):
        """迁移单个SQLite数据库"""
        # 从文件名提取user_id
        filename = sqlite_file.stem  # ai_story_user_xxx
        if filename.startswith("ai_story_user_"):
            user_id = filename.replace("ai_story_user_", "")
        else:
            user_id = self.target_user_id
        
        logger.info(f"\n{'='*60}")
        logger.info(f"开始迁移: {sqlite_file.name} -> user_id: {user_id}")
        logger.info(f"{'='*60}")
        
        # 创建SQLite连接
        sqlite_url = f"sqlite+aiosqlite:///{sqlite_file.absolute()}"
        sqlite_engine = create_async_engine(sqlite_url, echo=False)
        sqlite_session_maker = async_sessionmaker(
            sqlite_engine,
            class_=AsyncSession,
            expire_on_commit=False
        )
        
        try:
            # 迁移各个表
            async with sqlite_session_maker() as sqlite_session:
                async with self.pg_session_maker() as pg_session:
                    # 按照依赖顺序迁移
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, Settings, "设置"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, Project, "项目"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, Character, "角色"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, Outline, "大纲"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, Chapter, "章节"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, CharacterRelationship, "角色关系"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, Organization, "组织"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, OrganizationMember, "组织成员"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, GenerationHistory, "生成历史"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, WritingStyle, "写作风格"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, ProjectDefaultStyle, "项目默认风格"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, StoryMemory, "记忆"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, PlotAnalysis, "剧情分析"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, AnalysisTask, "分析任务"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, BatchGenerationTask, "批量生成任务"
                    )
                    await self._migrate_table(
                        sqlite_session, pg_session, user_id, MCPPlugin, "MCP插件"
                    )
                    
                    await pg_session.commit()
            
            logger.info(f"✅ {sqlite_file.name} 迁移完成")
        
        except Exception as e:
            logger.error(f"❌ 迁移失败: {e}", exc_info=True)
        finally:
            await sqlite_engine.dispose()
    
    async def _migrate_table(
        self,
        sqlite_session: AsyncSession,
        pg_session: AsyncSession,
        user_id: str,
        model_class,
        table_name: str
    ):
        """迁移单个表的数据"""
        try:
            # 获取SQLite表中实际存在的列
            sqlite_table = model_class.__table__
            sqlite_conn = await sqlite_session.connection()
            
            # 查询SQLite表结构
            inspect_result = await sqlite_conn.execute(
                text(f"PRAGMA table_info({sqlite_table.name})")
            )
            sqlite_columns = {row[1] for row in inspect_result.fetchall()}  # row[1]是列名
            
            # 构建只包含SQLite中存在的列的查询
            available_columns = [
                c for c in model_class.__table__.columns
                if c.name in sqlite_columns
            ]
            
            if not available_columns:
                logger.warning(f"  ⚠️ {table_name}: 表结构不匹配，跳过")
                return
            
            # 从SQLite读取数据（只查询存在的列）
            result = await sqlite_session.execute(
                select(*available_columns)
            )
            records = result.all()
            
            if not records:
                logger.info(f"  - {table_name}: 无数据")
                return
            
            # 为每条记录创建字典并添加user_id
            migrated_count = 0
            skipped_count = 0
            
            for record in records:
                # 从查询结果构建字典
                record_dict = {}
                for i, col in enumerate(available_columns):
                    record_dict[col.name] = record[i]
                
                # 添加user_id（如果PostgreSQL模型有这个字段但SQLite没有）
                if hasattr(model_class, 'user_id') and 'user_id' not in record_dict:
                    record_dict['user_id'] = user_id
                
                # 验证字段长度（防止超长字段导致插入失败）
                if not self._validate_field_lengths(model_class, record_dict, table_name):
                    skipped_count += 1
                    record_id = record_dict.get('id', 'unknown')
                    logger.warning(f"    ⚠️ [{table_name}] 跳过超长字段记录 ID={record_id}")
                    continue
                
                # 验证外键引用（针对有外键的表）
                validation_result = await self._validate_foreign_keys(pg_session, model_class, record_dict)
                if not validation_result:
                    skipped_count += 1
                    record_id = record_dict.get('id', 'unknown')
                    logger.warning(f"    ⚠️ [{table_name}] 跳过无效外键记录 ID={record_id}")
                    # 输出记录详情以便调试
                    if model_class.__tablename__ == 'story_memories':
                        logger.warning(f"       记忆详情: project_id={record_dict.get('project_id')}, "
                                     f"chapter_id={record_dict.get('chapter_id')}, "
                                     f"type={record_dict.get('memory_type')}")
                    elif model_class.__tablename__ == 'character_relationships':
                        logger.warning(f"       关系详情: project_id={record_dict.get('project_id')}, "
                                     f"from={record_dict.get('character_from_id')}, "
                                     f"to={record_dict.get('character_to_id')}, "
                                     f"type_id={record_dict.get('relationship_type_id')}")
                    elif model_class.__tablename__ == 'organizations':
                        logger.warning(f"       组织详情: project_id={record_dict.get('project_id')}, "
                                     f"character_id={record_dict.get('character_id')}")
                    elif model_class.__tablename__ == 'organization_members':
                        logger.warning(f"       成员详情: org_id={record_dict.get('organization_id')}, "
                                     f"character_id={record_dict.get('character_id')}")
                    elif model_class.__tablename__ == 'writing_styles':
                        logger.warning(f"       写作风格详情: project_id={record_dict.get('project_id')}, "
                                        f"name={record_dict.get('name')}, "
                                        f"style_type={record_dict.get('style_type')}")
                    elif model_class.__tablename__ == 'characters':
                        logger.warning(f"       角色详情: project_id={record_dict.get('project_id')}, "
                                        f"name={record_dict.get('name')}, "
                                        f"is_organization={record_dict.get('is_organization')}")
                    elif model_class.__tablename__ == 'outlines':
                        logger.warning(f"       大纲详情: project_id={record_dict.get('project_id')}, "
                                        f"title={record_dict.get('title')}")
                    elif model_class.__tablename__ == 'chapters':
                        logger.warning(f"       章节详情: project_id={record_dict.get('project_id')}, "
                                        f"title={record_dict.get('title')}, "
                                        f"chapter_number={record_dict.get('chapter_number')}")
                    elif model_class.__tablename__ == 'generation_history':
                        logger.warning(f"       生成历史详情: project_id={record_dict.get('project_id')}, "
                                        f"chapter_id={record_dict.get('chapter_id')}, "
                                        f"model={record_dict.get('model')}")
                    elif model_class.__tablename__ == 'plot_analysis':
                        logger.warning(f"       剧情分析详情: project_id={record_dict.get('project_id')}, "
                                        f"chapter_id={record_dict.get('chapter_id')}, "
                                        f"plot_stage={record_dict.get('plot_stage')}")
                    elif model_class.__tablename__ == 'analysis_tasks':
                        logger.warning(f"       分析任务详情: chapter_id={record_dict.get('chapter_id')}, "
                                        f"project_id={record_dict.get('project_id')}, "
                                        f"status={record_dict.get('status')}")
                    elif model_class.__tablename__ == 'batch_generation_tasks':
                        logger.warning(f"       批量生成任务详情: project_id={record_dict.get('project_id')}, "
                                        f"status={record_dict.get('status')}, "
                                        f"completed={record_dict.get('completed_chapters')}/{record_dict.get('total_chapters')}")
                    elif model_class.__tablename__ == 'project_default_styles':
                        logger.warning(f"       项目默认风格详情: project_id={record_dict.get('project_id')}, "
                                        f"style_id={record_dict.get('style_id')}")
                    continue
                
                # 检查记录是否已存在（避免主键冲突）
                record_id = record_dict.get('id')
                if record_id and await self._record_exists(pg_session, model_class, record_id):
                    skipped_count += 1
                    logger.debug(f"    跳过已存在的记录: {record_id}")
                    continue
                
                # 创建新记录
                try:
                    new_record = model_class(**record_dict)
                    pg_session.add(new_record)
                    migrated_count += 1
                except Exception as e:
                    logger.warning(f"    ⚠️ 跳过无效记录: {str(e)[:100]}")
                    skipped_count += 1
                    continue
            
            await pg_session.flush()
            
            if skipped_count > 0:
                logger.info(f"  ✅ {table_name}: {migrated_count} 条记录（跳过 {skipped_count} 条无效记录）")
            else:
                logger.info(f"  ✅ {table_name}: {migrated_count} 条记录")
        
        except Exception as e:
            logger.error(f"  ❌ {table_name} 迁移失败: {e}")
            raise
    
    async def _record_exists(
        self,
        pg_session: AsyncSession,
        model_class,
        record_id: Any
    ) -> bool:
        """
        检查记录是否已存在
        
        Args:
            pg_session: PostgreSQL会话
            model_class: 模型类
            record_id: 记录ID
            
        Returns:
            bool: 记录是否存在
        """
        try:
            # 获取主键列
            pk_column = list(model_class.__table__.primary_key.columns)[0]
            result = await pg_session.execute(
                select(pk_column).where(pk_column == record_id)
            )
            return result.scalar_one_or_none() is not None
        except Exception:
            return False
    
    async def _validate_foreign_keys(
        self,
        pg_session: AsyncSession,
        model_class,
        record_dict: Dict[str, Any]
    ) -> bool:
        """
        验证记录的外键是否有效
        
        Args:
            pg_session: PostgreSQL会话
            model_class: 模型类
            record_dict: 记录字典
            
        Returns:
            bool: 外键是否全部有效
        """
        from app.models import Character, Project, Chapter
        
        # 使用no_autoflush防止过早flush
        with pg_session.no_autoflush:
            # 针对StoryMemory表验证外键
            if model_class.__tablename__ == 'story_memories':
                # 验证project_id
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [记忆] 无效的project_id: {project_id}")
                        return False
                
                # 验证chapter_id（可选）
                chapter_id = record_dict.get('chapter_id')
                if chapter_id:
                    result = await pg_session.execute(
                        select(Chapter.id).where(Chapter.id == chapter_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [记忆] 无效的chapter_id: {chapter_id}")
                        return False
            
            # 针对CharacterRelationship表验证外键
            elif model_class.__tablename__ == 'character_relationships':
                # 验证project_id
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ 无效的project_id: {project_id}")
                        return False
                
                # 验证character_from_id
                char_from_id = record_dict.get('character_from_id')
                if char_from_id:
                    result = await pg_session.execute(
                        select(Character.id).where(Character.id == char_from_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ 无效的character_from_id: {char_from_id}")
                        return False
                
                # 验证character_to_id
                char_to_id = record_dict.get('character_to_id')
                if char_to_id:
                    result = await pg_session.execute(
                        select(Character.id).where(Character.id == char_to_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ 无效的character_to_id: {char_to_id}")
                        return False
                
                # 验证relationship_type_id
                rel_type_id = record_dict.get('relationship_type_id')
                if rel_type_id:
                    result = await pg_session.execute(
                        select(RelationshipType.id).where(RelationshipType.id == rel_type_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ 无效的relationship_type_id: {rel_type_id}")
                        return False
        
            # 针对Organization表验证外键
            elif model_class.__tablename__ == 'organizations':
                # 验证character_id
                char_id = record_dict.get('character_id')
                if char_id:
                    result = await pg_session.execute(
                        select(Character.id).where(Character.id == char_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [组织] 无效的character_id: {char_id}")
                        return False
            
            # 针对OrganizationMember表验证外键
            elif model_class.__tablename__ == 'organization_members':
                from app.models import Organization
                
                # 验证organization_id
                org_id = record_dict.get('organization_id')
                if org_id:
                    result = await pg_session.execute(
                        select(Organization.id).where(Organization.id == org_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ 无效的organization_id: {org_id}")
                        return False
                
                # 验证character_id
                char_id = record_dict.get('character_id')
                if char_id:
                    result = await pg_session.execute(
                        select(Character.id).where(Character.id == char_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [组织成员] 无效的character_id: {char_id}")
                        return False
            
            # 针对Character表验证外键
            elif model_class.__tablename__ == 'characters':
                # 验证project_id
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [角色] 无效的project_id: {project_id}")
                        return False
            
            # 针对Outline表验证外键
            elif model_class.__tablename__ == 'outlines':
                # 验证project_id
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [大纲] 无效的project_id: {project_id}")
                        return False
            
            # 针对Chapter表验证外键
            elif model_class.__tablename__ == 'chapters':
                # 验证project_id
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [章节] 无效的project_id: {project_id}")
                        return False
            
            # 针对WritingStyle表验证外键
            elif model_class.__tablename__ == 'writing_styles':
                # 验证project_id（可选）
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [写作风格] 无效的project_id: {project_id}")
                        return False
            
            # 针对GenerationHistory表验证外键
            elif model_class.__tablename__ == 'generation_history':
                # 验证project_id
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [生成历史] 无效的project_id: {project_id}")
                        return False
                
                # 验证chapter_id（可选）
                chapter_id = record_dict.get('chapter_id')
                if chapter_id:
                    result = await pg_session.execute(
                        select(Chapter.id).where(Chapter.id == chapter_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [生成历史] 无效的chapter_id: {chapter_id}")
                        return False
            
            # 针对PlotAnalysis表验证外键
            elif model_class.__tablename__ == 'plot_analysis':
                # 验证project_id（必需）
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [剧情分析] 无效的project_id: {project_id}")
                        return False
                
                # 验证chapter_id（必需）
                chapter_id = record_dict.get('chapter_id')
                if chapter_id:
                    result = await pg_session.execute(
                        select(Chapter.id).where(Chapter.id == chapter_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [剧情分析] 无效的chapter_id: {chapter_id}")
                        return False
            
            # 针对AnalysisTask表验证外键
            elif model_class.__tablename__ == 'analysis_tasks':
                # 验证chapter_id（必需）
                chapter_id = record_dict.get('chapter_id')
                if chapter_id:
                    result = await pg_session.execute(
                        select(Chapter.id).where(Chapter.id == chapter_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [分析任务] 无效的chapter_id: {chapter_id}")
                        return False
                
                # 验证project_id
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [分析任务] 无效的project_id: {project_id}")
                        return False
            
            # 针对BatchGenerationTask表验证外键
            elif model_class.__tablename__ == 'batch_generation_tasks':
                # 验证project_id（必需）
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [批量生成任务] 无效的project_id: {project_id}")
                        return False
            
            # 针对ProjectDefaultStyle表验证外键
            elif model_class.__tablename__ == 'project_default_styles':
                from app.models import WritingStyle
                
                # 验证project_id（必需）
                project_id = record_dict.get('project_id')
                if project_id:
                    result = await pg_session.execute(
                        select(Project.id).where(Project.id == project_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [项目默认风格] 无效的project_id: {project_id}")
                        return False
                
                # 验证style_id（必需）
                style_id = record_dict.get('style_id')
                if style_id:
                    result = await pg_session.execute(
                        select(WritingStyle.id).where(WritingStyle.id == style_id)
                    )
                    if not result.scalar_one_or_none():
                        logger.warning(f"      ❌ [项目默认风格] 无效的style_id: {style_id}")
                        return False
        
            return True
    
    def _validate_field_lengths(
        self,
        model_class,
        record_dict: Dict[str, Any],
        table_name: str
    ) -> bool:
        """
        验证记录的字段长度是否符合模型定义
        
        Args:
            model_class: 模型类
            record_dict: 记录字典
            table_name: 表名（用于日志）
            
        Returns:
            bool: 字段长度是否全部有效
        """
        from sqlalchemy import String
        
        # 检查所有字符串类型字段
        for column in model_class.__table__.columns:
            # 只检查有长度限制的String类型字段
            if isinstance(column.type, String) and column.type.length:
                field_name = column.name
                field_value = record_dict.get(field_name)
                max_length = column.type.length
                
                # 如果字段有值且超过最大长度
                if field_value and isinstance(field_value, str) and len(field_value) > max_length:
                    logger.warning(
                        f"      ❌ [{table_name}] 字段 '{field_name}' 超长: "
                        f"{len(field_value)} > {max_length} (截断了 {len(field_value) - max_length} 字符)"
                    )
                    # 对于敏感字段如API密钥，记录部分内容
                    if field_name in ['api_key', 'api_base_url']:
                        preview = field_value[:50] + "..." + field_value[-20:] if len(field_value) > 70 else field_value
                        logger.warning(f"         值预览: {preview}")
                    return False
        
        return True
    
    async def _reset_sequences(self):
        """重置PostgreSQL的自增序列到正确的值"""
        logger.info("\n" + "="*60)
        logger.info("重置自增序列...")
        logger.info("="*60)
        
        # 需要重置序列的表（使用Integer自增主键的表）
        tables_with_sequences = [
            ('relationship_types', 'id'),
            ('writing_styles', 'id'),
            ('project_default_styles', 'id'),
        ]
        
        async with self.pg_session_maker() as session:
            for table_name, id_column in tables_with_sequences:
                try:
                    # 获取表中当前最大ID
                    result = await session.execute(
                        text(f"SELECT MAX({id_column}) FROM {table_name}")
                    )
                    max_id = result.scalar()
                    
                    if max_id is not None:
                        # 重置序列到 max_id + 1
                        sequence_name = f"{table_name}_{id_column}_seq"
                        await session.execute(
                            text(f"SELECT setval('{sequence_name}', :max_id, true)"),
                            {"max_id": max_id}
                        )
                        logger.info(f"  ✅ {table_name}: 序列重置到 {max_id}")
                    else:
                        logger.info(f"  - {table_name}: 表为空，跳过序列重置")
                        
                except Exception as e:
                    logger.warning(f"  ⚠️ {table_name}: 序列重置失败 - {str(e)}")
            
            await session.commit()
        
        logger.info("✅ 序列重置完成")
    
    async def cleanup(self):
        """清理资源"""
        await self.pg_engine.dispose()


async def main():
    """主函数"""
    banner = """
╔══════════════════════════════════════════════════════════════╗
║          SQLite to PostgreSQL 数据迁移工具                   ║
║                                                              ║
║  此工具将SQLite数据迁移到PostgreSQL                          ║
║  请确保:                                                     ║
║  1. PostgreSQL数据库已创建                                   ║
║  2. .env中DATABASE_URL已配置为PostgreSQL                     ║
║  3. SQLite数据文件存在                                       ║
╚══════════════════════════════════════════════════════════════╝
    """
    print(banner)
    logger.info(banner)
    
    # 配置
    sqlite_dir = Path(__file__).parent.parent / "data"
    target_user_id = "migrated_user"  # 默认用户ID
    
    config_info = f"""
配置信息:
  SQLite目录: {sqlite_dir}
  PostgreSQL: {settings.database_url}
  目标用户ID: {target_user_id}
  日志文件: {log_filename}
"""
    print(config_info)
    logger.info(config_info)
    
    # 确认
    response = input("是否继续迁移? (yes/no): ")
    if response.lower() not in ['yes', 'y']:
        print("已取消迁移")
        return
    
    # 执行迁移
    migrator = SQLiteToPostgresMigrator(sqlite_dir, target_user_id)
    
    try:
        await migrator.migrate_all()
        success_msg = """
🎉 数据迁移成功完成!

下一步:
  1. 测试应用功能
  2. 验证数据完整性
  3. 备份SQLite文件后可删除
  
详细日志已保存到: {}
        """.format(log_filename)
        print(success_msg)
        logger.info(success_msg)
    
    except Exception as e:
        error_msg = f"\n❌ 迁移失败: {e}\n详细日志已保存到: {log_filename}"
        print(error_msg)
        logger.error("迁移过程出错", exc_info=True)
    
    finally:
        await migrator.cleanup()
        logger.info(f"🔒 数据库连接已关闭，日志文件: {log_filename}")


if __name__ == "__main__":
    asyncio.run(main())