152 lines
4.7 KiB
Python
152 lines
4.7 KiB
Python
#!/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()
|