how/migrate_data_to_cache.py

#!/usr/bin/env python3
"""
数据目录迁移脚本

将 data/ 目录迁移到 cache/data/ 目录下，实现统一的缓存管理。

注意：
- 会保留 data/ 目录中的非缓存文件（如文档、配置等）
- 只移动缓存性质的数据（爬虫数据、分析结果等）
"""

import os
import shutil
from pathlib import Path


def migrate_data_to_cache():
    """将 data 目录迁移到 cache/data"""

    project_root = Path(__file__).parent
    old_data_dir = project_root / "data"
    new_data_dir = project_root / "cache" / "data"

    print("=" * 60)
    print("数据目录迁移脚本")
    print("=" * 60)
    print(f"源目录: {old_data_dir}")
    print(f"目标目录: {new_data_dir}")
    print()

    # 检查源目录是否存在
    if not old_data_dir.exists():
        print("✓ data/ 目录不存在，无需迁移")
        return

    # 确保目标目录的父目录存在
    new_data_dir.parent.mkdir(parents=True, exist_ok=True)

    # 如果目标目录已存在，询问是否覆盖
    if new_data_dir.exists():
        print(f"⚠️  目标目录已存在: {new_data_dir}")
        response = input("是否继续？这将合并两个目录的内容 (y/n): ")
        if response.lower() != 'y':
            print("取消迁移")
            return

    print("开始迁移...")
    print()

    # 统计信息
    total_files = 0
    total_dirs = 0
    skipped_items = []

    # 需要跳过的目录和文件（非缓存数据）
    skip_patterns = {
        '.git',
        '.DS_Store',
        '__pycache__',
        '*.md',  # 文档文件
        '*.txt', # 说明文件
        'README*',
    }

    # 遍历源目录
    for item in old_data_dir.iterdir():
        item_name = item.name

        # 检查是否应该跳过
        should_skip = False
        for pattern in skip_patterns:
            if pattern.startswith('*'):
                if item_name.endswith(pattern[1:]):
                    should_skip = True
                    break
            elif pattern.endswith('*'):
                if item_name.startswith(pattern[:-1]):
                    should_skip = True
                    break
            elif item_name == pattern:
                should_skip = True
                break

        if should_skip:
            skipped_items.append(item_name)
            print(f"⊘ 跳过: {item_name} (非缓存数据)")
            continue

        # 目标路径
        target = new_data_dir / item_name

        try:
            if item.is_dir():
                # 如果目标已存在，合并；否则直接移动
                if target.exists():
                    print(f"→ 合并目录: {item_name}/")
                    # 递归复制并删除源
                    shutil.copytree(item, target, dirs_exist_ok=True)
                    shutil.rmtree(item)
                else:
                    print(f"→ 移动目录: {item_name}/")
                    shutil.move(str(item), str(target))
                total_dirs += 1
            else:
                # 文件直接移动
                if target.exists():
                    print(f"→ 覆盖文件: {item_name}")
                else:
                    print(f"→ 移动文件: {item_name}")
                shutil.move(str(item), str(target))
                total_files += 1
        except Exception as e:
            print(f"✗ 移动失败: {item_name}")
            print(f"  错误: {e}")

    print()
    print("=" * 60)
    print("迁移完成！")
    print("=" * 60)
    print(f"移动的目录数: {total_dirs}")
    print(f"移动的文件数: {total_files}")
    if skipped_items:
        print(f"跳过的项目数: {len(skipped_items)}")
        print(f"跳过的项目: {', '.join(skipped_items)}")
    print()

    # 检查 data 目录是否为空
    remaining_items = list(old_data_dir.iterdir())
    if not remaining_items:
        print(f"✓ 源目录 {old_data_dir} 已空，可以删除")
        response = input("是否删除空的 data/ 目录？(y/n): ")
        if response.lower() == 'y':
            old_data_dir.rmdir()
            print(f"✓ 已删除 {old_data_dir}")
    else:
        print(f"⚠️  源目录 {old_data_dir} 中还有以下内容：")
        for item in remaining_items:
            print(f"  - {item.name}")
        print("这些可能是非缓存数据，已保留在原位置")

    print()
    print("=" * 60)
    print("下一步操作：")
    print("1. 验证迁移结果：检查 cache/data/ 目录中的数据是否完整")
    print("2. 如果一切正常，可以删除旧的 data/ 目录（如果为空）")
    print("3. 更新你的脚本或配置，使用新的路径")
    print("=" * 60)


if __name__ == "__main__":
    migrate_data_to_cache()