soul-yongping/scripts/migrate_2026_sections.py

#!/usr/bin/env python3
"""
将第102场及以后的派对场次迁移到「2026每日派对干货」目录

用法:
  python3 scripts/migrate_2026_sections.py           # 仅预览，不执行
  python3 scripts/migrate_2026_sections.py --execute # 执行迁移

迁移规则:
- 从章节中筛选 section_title 包含「第102场」「第103场」... 的条目
- 按场次号排序，依次赋 id 10.01, 10.02, 10.03, ...
- 更新 part_id=part-2026-daily, part_title=2026每日派对干货
- 更新 chapter_id=chapter-2026-daily, chapter_title=2026每日派对干货

依赖: pip install pymysql
"""

import argparse
import re
import sys
from pathlib import Path

# 项目根目录
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

try:
    import pymysql
except ImportError:
    print("需要安装 pymysql: pip3 install pymysql")
    sys.exit(1)

DB_CONFIG = {
    "host": "56b4c23f6853c.gz.cdb.myqcloud.com",
    "port": 14413,
    "user": "cdb_outerroot",
    "password": "Zhiqun1984",
    "database": "soul_miniprogram",
    "charset": "utf8mb4",
}

PART_2026 = "part-2026-daily"
CHAPTER_2026 = "chapter-2026-daily"
TITLE_2026 = "2026每日派对干货"


def extract_session_num(section_title: str) -> int | None:
    """从 section_title 解析场次号，如 第102场 -> 102"""
    m = re.search(r"第(\d+)场", section_title)
    return int(m.group(1)) if m else None


def get_connection():
    return pymysql.connect(**DB_CONFIG)


def get_max_10_section(cur) -> int:
    """获取当前 10.xx 最大序号"""
    cur.execute(
        "SELECT id FROM chapters WHERE id REGEXP '^10\\.[0-9]+$' ORDER BY CAST(SUBSTRING_INDEX(id, '.', -1) AS UNSIGNED) DESC LIMIT 1"
    )
    row = cur.fetchone()
    if row:
        return int(row[0].split(".")[-1])
    return 0


def find_sections_to_migrate(cur) -> list[tuple]:
    """查找需要迁移的章节：第102场及以后，且不在 part-2026-daily 的"""
    cur.execute("""
        SELECT id, section_title, part_id, chapter_id, sort_order
        FROM chapters
        WHERE section_title REGEXP '第[0-9]+场'
        ORDER BY sort_order, id
    """)
    rows = cur.fetchall()
    to_migrate = []
    for row in rows:
        sid, title, part_id, ch_id, order = row
        num = extract_session_num(title)
        if num is not None and num >= 102 and part_id != PART_2026:
            to_migrate.append((sid, title, part_id, ch_id, order, num))
    to_migrate.sort(key=lambda x: (x[5], x[4]))  # 按场次号、sort_order
    return to_migrate


def run(dry_run: bool):
    conn = get_connection()
    cur = conn.cursor()
    rows = find_sections_to_migrate(cur)
    max_10 = get_max_10_section(cur)
    conn.close()

    if not rows:
        print("未找到需要迁移的章节（第102场及以后、且不在 2026每日派对干货 中）")
        return

    print(f"当前 10.xx 最大序号: {max_10}")
    print(f"找到 {len(rows)} 节待迁移到「2026每日派对干货」：\n")
    plan = []
    for i, (old_id, title, part_id, ch_id, order, num) in enumerate(rows, 1):
        new_id = f"10.{max_10 + i:02d}"
        plan.append((old_id, new_id, title, part_id))
        print(f"  {old_id} -> {new_id}  {title}")

    if dry_run:
        print("\n[预览模式] 未执行写入，使用 --execute 执行迁移")
        return

    print("\n执行迁移...")
    conn = get_connection()
    cur = conn.cursor()

    try:
        # 先全部改为临时 id（避免与已有 10.xx 冲突）
        for i, (old_id, new_id, title, part_id) in enumerate(plan, 1):
            tmp_id = f"tmp-migrate-{old_id.replace('.', '-')}"
            cur.execute(
                "UPDATE chapters SET id = %s WHERE id = %s",
                (tmp_id, old_id),
            )
        conn.commit()

        # 再改为最终 id 并更新 part/chapter
        for i, (old_id, new_id, title, part_id) in enumerate(plan, 1):
            tmp_id = f"tmp-migrate-{old_id.replace('.', '-')}"
            cur.execute("""
                UPDATE chapters SET
                    id = %s, part_id = %s, part_title = %s,
                    chapter_id = %s, chapter_title = %s
                WHERE id = %s
            """, (new_id, PART_2026, TITLE_2026, CHAPTER_2026, TITLE_2026, tmp_id))
        conn.commit()
        print(f"已迁移 {len(plan)} 节到 part-2026-daily，id 为 10.01 ~ 10.{len(plan):02d}")
    except Exception as e:
        conn.rollback()
        print(f"迁移失败: {e}")
        raise
    finally:
        conn.close()


def main():
    parser = argparse.ArgumentParser(description="将第102场及以后的场次迁移到 2026每日派对干货")
    parser.add_argument("--execute", action="store_true", help="执行迁移（默认仅预览）")
    args = parser.parse_args()
    run(dry_run=not args.execute)


if __name__ == "__main__":
    main()