nanocode/new-media-platform/website-new/backend/migrate_images_auto.py

#!/usr/bin/env python3
"""
图片存储方案迁移脚本 - 自动执行版
将图片从旧的层级目录结构迁移到新的扁平结构

旧结构: /{module}/{year}/{month}/{filename}
        /{module}/{year}/{month}/thumbnails/thumb_{filename}.jpg
新结构: /{filename}
        /thumbnails/{filename_stem}.jpg

使用方法: python migrate_images_auto.py [--dry-run]
"""

import os
import sys
import shutil
import argparse
from pathlib import Path
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from app.config import get_settings

settings = get_settings()

# 数据库路径
DB_PATH = "./cms.db"
UPLOAD_DIR = Path(settings.UPLOAD_DIR)


def migrate_images(dry_run=False):
    """迁移图片路径"""
    engine = create_engine(f"sqlite:///{DB_PATH}")
    SessionLocal = sessionmaker(bind=engine)
    db = SessionLocal()
    
    migrated = []
    skipped = []
    errors = []
    
    try:
        # 获取所有图片记录
        result = db.execute(text("SELECT id, file_path, thumb_path, filename FROM images"))
        images = result.fetchall()
        
        print(f"找到 {len(images)} 条图片记录")
        
        for img in images:
            try:
                img_id = img[0]
                old_file_path = img[1]
                old_thumb_path = img[2]
                filename = img[3]
                
                # 如果已经是新格式（只有一级路径），跳过
                if old_file_path and old_file_path.startswith("/"):
                    path_parts = old_file_path.lstrip("/").split("/")
                    if len(path_parts) == 1 and "/thumbnails/" not in old_file_path:
                        skipped.append({"id": img_id, "path": old_file_path})
                        continue
                
                # 新的路径格式
                new_file_path = f"/{filename}"
                new_thumb_path = None
                if old_thumb_path:
                    new_thumb_path = f"/thumbnails/{Path(filename).stem}.jpg"
                
                if dry_run:
                    print(f"[DRY RUN] ID={img_id}: {old_file_path} -> {new_file_path}")
                    if old_thumb_path:
                        print(f"[DRY RUN] 缩略图: {old_thumb_path} -> {new_thumb_path}")
                    continue
                
                # 物理文件路径
                old_file_full = UPLOAD_DIR / old_file_path.lstrip("/") if old_file_path else None
                new_file_full = UPLOAD_DIR / filename
                
                # 移动原图文件
                if old_file_full and old_file_full.exists() and old_file_full != new_file_full:
                    try:
                        shutil.move(str(old_file_full), str(new_file_full))
                        print(f"移动文件: {old_file_full} -> {new_file_full}")
                    except Exception as e:
                        print(f"移动文件失败 {old_file_full}: {e}")
                        errors.append({"id": img_id, "error": f"移动原图失败: {e}"})
                        continue
                
                # 处理缩略图
                if old_thumb_path:
                    old_thumb_full = UPLOAD_DIR / old_thumb_path.lstrip("/")
                    new_thumb_full = UPLOAD_DIR / "thumbnails" / f"{Path(filename).stem}.jpg"
                    
                    if old_thumb_full.exists() and old_thumb_full != new_thumb_full:
                        try:
                            shutil.move(str(old_thumb_full), str(new_thumb_full))
                            print(f"移动缩略图: {old_thumb_full} -> {new_thumb_full}")
                        except Exception as e:
                            print(f"移动缩略图失败 {old_thumb_full}: {e}")
                
                # 更新数据库
                db.execute(
                    text("UPDATE images SET file_path = :new_path, thumb_path = :new_thumb WHERE id = :id"),
                    {"new_path": new_file_path, "new_thumb": new_thumb_path, "id": img_id}
                )
                
                migrated.append({"id": img_id, "new_path": new_file_path})
                print(f"迁移成功 ID={img_id}: {old_file_path} -> {new_file_path}")
                
            except Exception as e:
                error_msg = f"迁移失败 ID={img_id}: {e}"
                print(error_msg)
                errors.append({"id": img_id, "error": str(e)})
        
        if not dry_run:
            db.commit()
        
        print("\n" + "="*60)
        print(f"迁移结果:")
        print(f"成功: {len(migrated)}")
        print(f"跳过 (已是新格式): {len(skipped)}")
        print(f"失败: {len(errors)}")
        print("="*60)
        
        if errors:
            print("\n错误详情:")
            for err in errors:
                print(f"  ID={err['id']}: {err['error']}")
        
        # 清理空目录（仅在非 dry-run 时）
        if not dry_run:
            cleanup_empty_dirs(UPLOAD_DIR)
        
    except Exception as e:
        print(f"迁移过程出错: {e}")
        if not dry_run:
            db.rollback()
    finally:
        db.close()


def cleanup_empty_dirs(root_dir: Path):
    """清理空目录"""
    removed_count = 0
    for dirpath, dirnames, filenames in os.walk(str(root_dir), topdown=False):
        dir_path = Path(dirpath)
        # 跳过根目录和 thumbnails 目录
        if dir_path == root_dir or dir_path.name == "thumbnails":
            continue
        
        # 检查是否为空
        try:
            if dir_path.exists() and not any(dir_path.iterdir()):
                dir_path.rmdir()
                print(f"删除空目录: {dir_path}")
                removed_count += 1
        except Exception as e:
            pass
    
    if removed_count > 0:
        print(f"清理完成，共删除 {removed_count} 个空目录")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="图片存储方案迁移工具")
    parser.add_argument("--dry-run", action="store_true", help="试运行，不实际执行迁移")
    args = parser.parse_args()
    
    print("图片存储方案迁移工具")
    print("="*60)
    print(f"数据库: {DB_PATH}")
    print(f"上传目录: {UPLOAD_DIR}")
    print(f"模式: {'试运行' if args.dry_run else '实际执行'}")
    print("="*60)
    
    migrate_images(dry_run=args.dry_run)