194 lines
6.5 KiB
Python
194 lines
6.5 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
批量上传《卡若的IP财富旅程》PART1-5 到小程序。
|
||
|
||
part_id : part-new-17755513182
|
||
part_title: 卡若的IP财富旅程
|
||
各 PART 对应独立 chapter:
|
||
PART1 -> chapter-ip-1 | PART1 | 我少年时期的互联网启蒙
|
||
PART2 -> chapter-ip-2 | PART2 | 从0-1的过程
|
||
PART3 -> chapter-ip-3 | PART3 | 我的创业旅程
|
||
PART4 -> chapter-ip-4 | PART4 | 私域流量到AI时代的商业进化
|
||
PART5 -> chapter-ip-5 | PART5 | 未来的一些思考
|
||
|
||
ID 格式: ip<part_num>.<section_num> 例: ip1.1, ip2.10, ip3.5
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import importlib.util
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
BOOK_DIR = Path("/Users/karuo/Documents/个人/2、我写的书/《卡若的IP财富旅程》")
|
||
|
||
PART_ID = "part-new-17755513182"
|
||
PART_TITLE = "卡若的IP财富旅程"
|
||
|
||
CHAPTERS = {
|
||
1: ("chapter-ip-1", "PART1 | 我少年时期的互联网启蒙"),
|
||
2: ("chapter-ip-2", "PART2 | 从0-1的过程"),
|
||
3: ("chapter-ip-3", "PART3 | 我的创业旅程"),
|
||
4: ("chapter-ip-4", "PART4 | 私域流量到AI时代的商业进化"),
|
||
5: ("chapter-ip-5", "PART5 | 未来的一些思考"),
|
||
}
|
||
|
||
# 匹配 ✅1.5-标题 或 ✅1.5- 标题 形式的文件名，提取 section_num 和 title
|
||
FILENAME_RE = re.compile(
|
||
r"^[✅⬜]?\s*(\d+)\.(\d+)[- ]+(.+?)(?:\.md)?$",
|
||
re.UNICODE,
|
||
)
|
||
|
||
|
||
def load_db():
|
||
mig = ROOT / "scripts" / "migrate_2026_sections.py"
|
||
spec = importlib.util.spec_from_file_location("_mig_db", mig)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
assert spec.loader is not None
|
||
spec.loader.exec_module(mod)
|
||
return mod.DB_CONFIG
|
||
|
||
|
||
def strip_md_title_line(text: str) -> str:
|
||
lines = text.splitlines()
|
||
if lines and lines[0].lstrip().startswith("#"):
|
||
return "\n".join(lines[1:]).lstrip("\n")
|
||
return text
|
||
|
||
|
||
def for_miniprogram_body(text: str) -> str:
|
||
out_lines: list[str] = []
|
||
for line in text.splitlines():
|
||
if line.strip() == "---":
|
||
out_lines.append("")
|
||
else:
|
||
out_lines.append(line)
|
||
body = "\n".join(out_lines)
|
||
body = body.replace("**", "")
|
||
body = re.sub(r"\n{3,}", "\n\n", body)
|
||
return body.strip() + "\n"
|
||
|
||
|
||
def collect_files(part_num: int) -> list[tuple[str, str, Path]]:
|
||
"""返回 [(ip_id, title, path)] 已去重且按节号排序"""
|
||
part_dir = BOOK_DIR / f"PART{part_num}"
|
||
if not part_dir.exists():
|
||
print(f" ⚠ 目录不存在: {part_dir}", file=sys.stderr)
|
||
return []
|
||
|
||
seen_sections: dict[str, tuple[str, str, Path]] = {}
|
||
for f in part_dir.iterdir():
|
||
name = f.name
|
||
# 去掉 .md 后缀再匹配（处理双后缀 .md.md）
|
||
clean_name = re.sub(r"\.md(\.md)*$", "", name, flags=re.IGNORECASE)
|
||
m = FILENAME_RE.match(clean_name)
|
||
if not m:
|
||
print(f" ⚠ 跳过无法解析的文件: {name}")
|
||
continue
|
||
p_num, s_num, title_raw = m.group(1), m.group(2), m.group(3).strip()
|
||
if int(p_num) != part_num:
|
||
print(f" ⚠ PART号不匹配，跳过: {name}")
|
||
continue
|
||
# 把文件名里的冒号/括号等清理出来作为标题
|
||
title = title_raw.strip("：: \t")
|
||
# 还原完整标题（包含冒号）
|
||
title = title_raw
|
||
ip_id = f"ip{part_num}.{s_num}"
|
||
section_key = f"{part_num}.{s_num}"
|
||
|
||
# 优先选有 .md 后缀的文件（去重）
|
||
if section_key in seen_sections:
|
||
existing_path = seen_sections[section_key][2]
|
||
# 若新文件有 .md 后缀，替换
|
||
if str(f).lower().endswith(".md") and not str(existing_path).lower().endswith(".md"):
|
||
seen_sections[section_key] = (ip_id, title, f)
|
||
else:
|
||
seen_sections[section_key] = (ip_id, title, f)
|
||
|
||
# 按节号排序
|
||
result = list(seen_sections.values())
|
||
result.sort(key=lambda x: float(re.sub(r"ip\d+\.", "", x[0])))
|
||
return result
|
||
|
||
|
||
def upsert_chapter(cur, chapter_id: str, chapter_title: str, ip_id: str, title: str, body: str, price: float):
|
||
cur.execute("SELECT id FROM chapters WHERE id = %s", (ip_id,))
|
||
if cur.fetchone():
|
||
cur.execute(
|
||
"UPDATE chapters SET section_title=%s, content=%s, part_id=%s, part_title=%s, "
|
||
"chapter_id=%s, chapter_title=%s, price=%s WHERE id=%s",
|
||
(title, body, PART_ID, PART_TITLE, chapter_id, chapter_title, price, ip_id),
|
||
)
|
||
return "updated"
|
||
else:
|
||
cur.execute(
|
||
"INSERT INTO chapters (id, section_title, content, part_id, part_title, chapter_id, chapter_title, price) "
|
||
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s)",
|
||
(ip_id, title, body, PART_ID, PART_TITLE, chapter_id, chapter_title, price),
|
||
)
|
||
return "created"
|
||
|
||
|
||
def main():
|
||
try:
|
||
import pymysql
|
||
except ImportError:
|
||
print("需要: pip install pymysql", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
dry_run = "--dry-run" in sys.argv
|
||
price = 1.0
|
||
|
||
cfg = load_db()
|
||
conn = pymysql.connect(**cfg)
|
||
cur = conn.cursor()
|
||
|
||
total_ok = 0
|
||
total_skip = 0
|
||
total_err = 0
|
||
|
||
for part_num, (chapter_id, chapter_title) in CHAPTERS.items():
|
||
print(f"\n{'='*60}")
|
||
print(f"PART{part_num} → {chapter_id} | {chapter_title}")
|
||
print(f"{'='*60}")
|
||
files = collect_files(part_num)
|
||
if not files:
|
||
print(" 无文件，跳过")
|
||
continue
|
||
|
||
for ip_id, title, path in files:
|
||
try:
|
||
raw = path.read_text(encoding="utf-8")
|
||
except Exception as e:
|
||
print(f" ✗ 读取失败 {path.name}: {e}")
|
||
total_err += 1
|
||
continue
|
||
|
||
body = for_miniprogram_body(strip_md_title_line(raw))
|
||
if dry_run:
|
||
print(f" [dry] {ip_id} {title[:30]} ({path.name})")
|
||
total_ok += 1
|
||
continue
|
||
|
||
try:
|
||
action = upsert_chapter(cur, chapter_id, chapter_title, ip_id, title, body, price)
|
||
conn.commit()
|
||
print(f" ✓ [{action}] {ip_id} {title[:40]}")
|
||
total_ok += 1
|
||
except Exception as e:
|
||
conn.rollback()
|
||
print(f" ✗ 上传失败 {ip_id} ({path.name}): {e}")
|
||
total_err += 1
|
||
|
||
conn.close()
|
||
print(f"\n{'='*60}")
|
||
print(f"完成：成功 {total_ok} | 跳过 {total_skip} | 失败 {total_err}")
|
||
if dry_run:
|
||
print("（dry-run 模式，未实际写库）")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|