Mycontent/scripts/reupload_28_models_html.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
重新上传《28套私域商业模式》：Markdown → HTML（含表格/图片/粗体/标题）
1. 遍历 3 篇目录，读取 .md 文件
2. 把 Markdown 转为 HTML（保留表格、图片、引用、列表、粗体等）
3. 上传本地图片到 OSS，替换为线上 URL
4. 更新 chapters 表的 content 字段

用法:
  python3 scripts/reupload_28_models_html.py --dry-run   # 预览
  python3 scripts/reupload_28_models_html.py              # 执行
"""
from __future__ import annotations

import importlib.util
import os
import re
import sys
from pathlib import Path

import markdown
import requests

ROOT = Path(__file__).resolve().parent.parent
BOOK_DIR = Path("/Users/karuo/Documents/个人/2、我写的书/《28套私域商业模式》")
API_BASE = os.environ.get("SOUL_API_BASE", "https://soulapi.quwanzhi.com")

PARTS = [
    ("01_起盘篇｜10套私域基础模式", "part-28-basic", "chapter-28-basic"),
    ("02_增长篇｜10套私域进阶模式", "part-28-growth", "chapter-28-growth"),
    ("03_精英篇｜8套私域高阶模式", "part-28-elite", "chapter-28-elite"),
]

MODEL_NUM_RE = re.compile(r"模式(\d+)")
IMAGE_EXT = frozenset({".png", ".jpg", ".jpeg", ".gif", ".webp"})
MD_IMAGE_RE = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)")


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 guess_mime(path: Path) -> str:
    return {
        ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
        ".gif": "image/gif", ".webp": "image/webp",
    }.get(path.suffix.lower(), "application/octet-stream")


def upload_image(local: Path) -> str:
    url = f"{API_BASE.rstrip('/')}/api/upload"
    mime = guess_mime(local)
    with local.open("rb") as f:
        r = requests.post(url, files={"file": (local.name, f, mime)},
                          data={"folder": "book-images"}, timeout=120)
    r.raise_for_status()
    j = r.json()
    if not j.get("success"):
        raise RuntimeError(j.get("error") or j.get("message") or str(j))
    out = j.get("url") or (j.get("data") or {}).get("url")
    if not out:
        raise RuntimeError("响应无 url: " + str(j)[:300])
    return str(out)


def process_images(md_text: str, md_path: Path, upload_cache: dict, dry_run: bool) -> str:
    """替换 md 中的本地图片引用为 OSS URL"""
    def _replace(m):
        alt, ref = m.group(1), m.group(2).strip()
        if ref.startswith(("http://", "https://")):
            return m.group(0)
        local = (md_path.parent / ref).resolve()
        if not local.is_file() or local.suffix.lower() not in IMAGE_EXT:
            print(f"      ⚠ 图片不存在: {ref}")
            return m.group(0)
        key = str(local)
        if key not in upload_cache:
            if dry_run:
                upload_cache[key] = f"https://dry-run/{local.name}"
                print(f"      [dry] 图片: {local.name}")
            else:
                print(f"      上传图片: {local.name} …", end=" ", flush=True)
                try:
                    upload_cache[key] = upload_image(local)
                    print("✓", upload_cache[key])
                except Exception as e:
                    print(f"✗ {e}")
                    upload_cache[key] = ref
        return f"![{alt}]({upload_cache[key]})"
    return MD_IMAGE_RE.sub(_replace, md_text)


def md_to_html(md_text: str) -> str:
    """Markdown → HTML，保留表格/粗体/标题/列表/引用/分隔线/图片"""
    lines = md_text.splitlines()
    if lines and lines[0].lstrip().startswith("# "):
        lines = lines[1:]
    cleaned = "\n".join(lines).strip()
    html = markdown.markdown(cleaned, extensions=["tables", "fenced_code"])
    html = re.sub(r"\n{3,}", "\n\n", html)
    return html.strip()


def extract_model_num(filename: str) -> int:
    m = MODEL_NUM_RE.search(filename)
    return int(m.group(1)) if m else 999


def collect_md_files(part_dir: Path) -> list[tuple[int, str, Path]]:
    """返回 [(model_num, section_title, path)]"""
    result = []
    for f in part_dir.iterdir():
        if not f.name.endswith(".md") or f.name.startswith("."):
            continue
        num = extract_model_num(f.name)
        title = f.stem
        result.append((num, title, f))
    result.sort(key=lambda x: x[0])
    return result


def main():
    try:
        import pymysql
    except ImportError:
        sys.exit("需要: pip install pymysql")

    dry_run = "--dry-run" in sys.argv

    cfg = load_db()
    conn = pymysql.connect(**cfg, autocommit=False)
    cur = conn.cursor()

    upload_cache: dict[str, str] = {}
    total_ok = 0
    total_err = 0

    for dir_name, part_id, chapter_id in PARTS:
        part_dir = BOOK_DIR / dir_name
        if not part_dir.exists():
            print(f"⚠ 目录不存在: {part_dir}")
            continue

        print(f"\n{'='*60}")
        print(f"{dir_name}")
        print(f"{'='*60}")

        files = collect_md_files(part_dir)
        for model_num, title, path in files:
            ip_id = f"m28.{model_num:02d}"
            print(f"\n  📄 {ip_id} | {title}")

            try:
                raw = path.read_text(encoding="utf-8")
            except Exception as e:
                print(f"    ✗ 读取失败: {e}")
                total_err += 1
                continue

            processed = process_images(raw, path, upload_cache, dry_run)
            html_content = md_to_html(processed)

            word_count = len(re.sub(r"<[^>]+>", "", html_content).replace("\n", ""))

            if dry_run:
                print(f"    [dry] HTML 长度={len(html_content)}, 字数={word_count}")
                preview = html_content[:200].replace("\n", " ")
                print(f"    [dry] 预览: {preview}...")
                total_ok += 1
                continue

            try:
                cur.execute(
                    "UPDATE chapters SET content=%s, word_count=%s, updated_at=NOW(3) WHERE id=%s",
                    (html_content, word_count, ip_id),
                )
                if cur.rowcount == 0:
                    print(f"    ⚠ 未找到 id={ip_id}，跳过")
                else:
                    conn.commit()
                    print(f"    ✓ 更新成功 (HTML {len(html_content)} 字符, {word_count} 字)")
                    total_ok += 1
            except Exception as e:
                conn.rollback()
                print(f"    ✗ 更新失败: {e}")
                total_err += 1

    conn.close()
    print(f"\n{'='*60}")
    print(f"完成：成功 {total_ok} | 失败 {total_err}")
    if dry_run:
        print("（dry-run 模式，未实际写库/上传图片）")
    else:
        print(f"图片上传缓存: {len(upload_cache)} 张")


if __name__ == "__main__":
    main()