331 lines
11 KiB
Python
331 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
上传/更新《28套私域商业模式》全部28章到小程序
|
||
|
||
功能：
|
||
- Markdown → 语义化 HTML（保留 h2/h3/blockquote/ul/ol/li/table/img）
|
||
- 图片自动上传到 OSS，替换本地路径
|
||
- 预处理：去 # 标题行、去 ---、扁平化嵌套列表
|
||
- 后处理：去 <hr>、去 <li> 内 <p>、合并空行、修孤立 bullet
|
||
- 写入 DB 后需要通过 admin API 清 Redis 缓存（脚本不含此步，见下方）
|
||
|
||
用法：
|
||
python3 scripts/optimize_28_models.py --dry-run # 预览（不上传/不写库）
|
||
python3 scripts/optimize_28_models.py # 执行上传+写库
|
||
|
||
清缓存（上传完后手动执行）：
|
||
通过 admin API PUT /api/db/book 触发 cache.InvalidateChapterContent()
|
||
参考本对话上下文中的缓存失效脚本
|
||
|
||
小程序前端要求：
|
||
contentParser.js 需支持 heading/quote/listItem/table/image 段落类型
|
||
read.wxml/read.wxss 需有对应渲染组件与样式
|
||
"""
|
||
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:
|
||
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""
|
||
return MD_IMAGE_RE.sub(_replace, md_text)
|
||
|
||
|
||
def preprocess_markdown(md_text: str) -> str:
|
||
"""预处理 Markdown 源文本，解决嵌套列表产生的孤立 bullet 问题"""
|
||
lines = md_text.splitlines()
|
||
|
||
# 去掉第一行如果是 # 标题
|
||
if lines and lines[0].lstrip().startswith("# "):
|
||
lines = lines[1:]
|
||
|
||
result = []
|
||
for line in lines:
|
||
# 把 "---" 分隔线去掉（标题已经能分隔段落）
|
||
if re.match(r"^\s*---+\s*$", line):
|
||
continue
|
||
result.append(line)
|
||
|
||
# 合并嵌套列表：把 " - xxx" 子项合并为同级 "- xxx"
|
||
flat = []
|
||
for line in result:
|
||
# 检测 2 空格或 4 空格缩进的子列表项
|
||
m = re.match(r"^(\s{2,4})[-*]\s+(.*)", line)
|
||
if m:
|
||
flat.append(f"- {m.group(2)}")
|
||
else:
|
||
flat.append(line)
|
||
|
||
return "\n".join(flat).strip()
|
||
|
||
|
||
def md_to_html(md_text: str) -> str:
|
||
"""Markdown → 干净的 HTML"""
|
||
cleaned = preprocess_markdown(md_text)
|
||
html = markdown.markdown(cleaned, extensions=["tables", "fenced_code"])
|
||
html = clean_html(html)
|
||
return html.strip()
|
||
|
||
|
||
def clean_html(html: str) -> str:
|
||
"""清理 HTML：去 <hr>、去 <li> 内 <p>、合并空标签、修孤立 bullet"""
|
||
|
||
# 去掉所有 <hr /> 标签
|
||
html = re.sub(r"<hr\s*/?>", "", html, flags=re.IGNORECASE)
|
||
|
||
# 去掉 <li> 内部的 <p></p> 包裹（防止 contentParser 把 bullet 和文本拆开）
|
||
# <li>\n<p>content</p>\n</li> → <li>content</li>
|
||
html = re.sub(
|
||
r"<li>\s*<p>([\s\S]*?)</p>\s*</li>",
|
||
r"<li>\1</li>",
|
||
html,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
|
||
# 去掉空的 <li></li>
|
||
html = re.sub(r"<li>\s*</li>", "", html, flags=re.IGNORECASE)
|
||
|
||
# 去掉空的 <p></p>
|
||
html = re.sub(r"<p>\s*</p>", "", html, flags=re.IGNORECASE)
|
||
|
||
# 把 <p> 内的 "- xxx" 列表项拆出来为 <ul><li> 格式
|
||
def fix_inline_list(m):
|
||
before = m.group(1).strip()
|
||
body = m.group(2)
|
||
items = re.findall(r"-\s*(.+)", body)
|
||
if not items:
|
||
return m.group(0)
|
||
parts = [f"<p>{before}</p>"] if before else []
|
||
parts.append("<ul>")
|
||
for item in items:
|
||
parts.append(f"<li>{item.strip()}</li>")
|
||
parts.append("</ul>")
|
||
return "\n".join(parts)
|
||
|
||
html = re.sub(
|
||
r"<p>(.*?)\n((?:\s*-\s+.+\n?)+)</p>",
|
||
fix_inline_list,
|
||
html,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
|
||
# 压缩 block 标签之间的空白：</tag>\n\n<tag> → </tag>\n<tag>
|
||
html = re.sub(
|
||
r"(</(?:blockquote|ul|ol|table|h[1-6]|p|div)>)\s*\n\s*\n\s*(<(?:h[1-6]|blockquote|ul|ol|table|p|div)[^>]*>)",
|
||
r"\1\n\2",
|
||
html,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
|
||
# 合并连续 3+ 换行为 1 个
|
||
html = re.sub(r"\n{3,}", "\n", html)
|
||
|
||
return html
|
||
|
||
|
||
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]]:
|
||
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 verify_segments(html: str, chapter_id: str):
|
||
"""验证 HTML 渲染后不会出现孤立 bullet 和过多空行"""
|
||
import re as _re
|
||
text = html
|
||
text = _re.sub(r"<table[^>]*>[\s\S]*?</table>", "[TABLE]", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"</p>\s*<p[^>]*>", "\n\n", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"<p[^>]*>", "", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"</p>", "\n", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"<div[^>]*>", "", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"</div>", "\n", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"<br\s*/?>", "\n", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"</?h[1-6][^>]*>", "\n", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"</?blockquote[^>]*>", "\n", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"</?ul[^>]*>", "\n", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"</?ol[^>]*>", "\n", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"<li[^>]*>", "• ", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"</li>", "\n", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"<hr\s*/?>", "\n", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"<img[^>]*>", "[IMG]", text, flags=_re.IGNORECASE)
|
||
text = _re.sub(r"<[^>]+>", "", text)
|
||
|
||
# 模拟 contentParser 的 split(/\n+/) 行为
|
||
blocks = [b.strip() for b in re.split(r"\n+", text) if b.strip()]
|
||
issues = []
|
||
for i, block in enumerate(blocks):
|
||
if block == "•" or block == "·":
|
||
issues.append(f" 段{i}: 孤立 bullet '{block}'")
|
||
|
||
if issues:
|
||
print(f" ⚠ {chapter_id} 检测到 {len(issues)} 个格式问题:")
|
||
for issue in issues[:5]:
|
||
print(issue)
|
||
else:
|
||
print(f" ✓ {chapter_id} 格式检查通过")
|
||
|
||
|
||
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)
|
||
|
||
verify_segments(html_content, ip_id)
|
||
|
||
word_count = len(re.sub(r"<[^>]+>", "", html_content).replace("\n", ""))
|
||
|
||
if dry_run:
|
||
print(f" [dry] HTML 长度={len(html_content)}, 字数={word_count}")
|
||
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()
|