Video-Tree-TRM5/tools/migrate_subtitle_to_card.py

"""字幕数据迁移脚本：L3Node.subtitle → L3Card.subtitle + L2 聚合。

幂等操作——已迁移的树文件不会被重复修改。

用法:
    conda activate Video-Tree-TRM & python tools/migrate_subtitle_to_card.py [--dry-run]

迁移逻辑:
    1. 扫描 store/videos/*/tree.json
    2. 对每棵树:
       - L3: 将节点级 subtitle 移入 card["subtitle"]，删除节点级 key
       - L2: 聚合子 L3 的 card["subtitle"] 写入 L2 card["subtitle"]
       - 处理边界: None → ""、空字符串、已迁移(card 内已有 subtitle)
    3. 回写 JSON
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

from loguru import logger


def _migrate_tree_dict(tree_dict: dict) -> bool:
    """原地迁移单棵树的 dict 结构，返回是否发生了修改。

    参数:
        tree_dict: tree.json 加载后的字典。

    返回:
        True 表示发生了修改，需要回写。
    """
    modified = False

    for l1 in tree_dict.get("roots", []):
        for l2 in l1.get("children", []):
            l3_subtitles: list[str] = []

            for l3 in l2.get("children", []):
                card = l3.get("card", {})

                # Phase 1: 迁移 L3 节点级 subtitle → card.subtitle
                node_subtitle = l3.get("subtitle")  # 旧位置
                card_subtitle = card.get("subtitle", "")  # 新位置

                if node_subtitle is not None and not card_subtitle:
                    # 旧格式：节点级有值，card 内无值 → 迁移
                    card["subtitle"] = node_subtitle if node_subtitle else ""
                    modified = True
                elif "subtitle" not in card:
                    # 既无节点级也无 card 级 → 初始化为空
                    card["subtitle"] = ""
                    modified = True

                # Phase 2: 清理节点级 subtitle key
                if "subtitle" in l3:
                    del l3["subtitle"]
                    modified = True

                # 收集 L3 字幕用于 L2 聚合
                final_subtitle = card.get("subtitle", "")
                if final_subtitle:
                    l3_subtitles.append(final_subtitle)

            # Phase 3: L2 字幕聚合
            l2_card = l2.get("card", {})
            existing_l2_subtitle = l2_card.get("subtitle", "")
            aggregated = "\n".join(l3_subtitles)

            if "subtitle" not in l2_card or existing_l2_subtitle != aggregated:
                l2_card["subtitle"] = aggregated
                modified = True

    return modified


def migrate_all(
    videos_dir: Path,
    *,
    dry_run: bool = False,
) -> tuple[int, int, int]:
    """扫描并迁移所有树文件。

    参数:
        videos_dir: store/videos 目录路径。
        dry_run: 仅检查不写入。

    返回:
        (total, migrated, skipped) 计数元组。
    """
    tree_files = sorted(videos_dir.glob("*/tree.json"))
    total = len(tree_files)
    migrated = 0
    skipped = 0

    for tree_path in tree_files:
        video_id = tree_path.parent.name
        try:
            with open(tree_path, encoding="utf-8") as f:
                tree_dict = json.load(f)
        except (json.JSONDecodeError, OSError) as exc:
            logger.warning("跳过损坏文件: {} ({})", tree_path, exc)
            skipped += 1
            continue

        changed = _migrate_tree_dict(tree_dict)

        if changed:
            if dry_run:
                logger.info("[dry-run] 需要迁移: {}", video_id)
            else:
                with open(tree_path, "w", encoding="utf-8") as f:
                    json.dump(tree_dict, f, ensure_ascii=False, indent=2)
                logger.info("已迁移: {}", video_id)
            migrated += 1
        else:
            logger.debug("无需迁移: {}", video_id)

    return total, migrated, skipped


def main() -> None:
    """CLI 入口。"""
    parser = argparse.ArgumentParser(
        description="将 L3Node.subtitle 迁移到 L3Card.subtitle + L2 聚合",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="仅检查不写入",
    )
    parser.add_argument(
        "--videos-dir",
        type=Path,
        default=Path("store/videos"),
        help="视频树目录 (默认: store/videos)",
    )
    args = parser.parse_args()

    if not args.videos_dir.is_dir():
        logger.error("目录不存在: {}", args.videos_dir)
        sys.exit(1)

    total, migrated, skipped = migrate_all(
        args.videos_dir,
        dry_run=args.dry_run,
    )
    action = "需要迁移" if args.dry_run else "已迁移"
    logger.info(
        "迁移完成: 总计={}, {}={}, 跳过={}",
        total,
        action,
        migrated,
        skipped,
    )


if __name__ == "__main__":
    main()