fnvtk/Mycontent
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: 以本地工作区为准全量快照同步 GitHub

Browse Source 
包含小程序、管理端、soul-api、脚本与静态资源等当前本地全部已跟踪与新增文件（排除 .DS_Store 与 .obsidian）。

Made-with: Cursor
This commit is contained in:
卡若
2026-04-13 11:49:38 +08:00
parent 83375cc388
commit b5f5180654
142 changed files with 10684 additions and 2670 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
scripts/batch_upload_ip_book.py Normal file
View File

@@ -0,0 +1,193 @@
#!/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()

269
scripts/batch_upload_ip_book_preserve_images.py Normal file
View File

@@ -0,0 +1,269 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
《卡若的IP财富旅程》PART1-5 重新上传：正文以本地 MD 为准，已入库的图片块按顺序保留（不重新上传）。
- 从 chapters.content 提取现有 <p><img .../></p>（或裸 <img>）块，顺序与 MD 中独立图片行一一对应。
- 无图片的章节：与 batch_upload_ip_book.py 相同（纯文本清洗）。
- 同步 sort_order（与 fix_ip_book_sort_and_images 一致）。
"""
from __future__ import annotations
import html
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 | 未来的一些思考"),
}
FILENAME_RE = re.compile(
r"^[✅⬜]?\s*(\d+)\.(\d+)[- ]+(.+?)(?:\.md)?$",
re.UNICODE,
)
LINE_IMAGE_ONLY = re.compile(r"^\s*!\[([^\]]*)\]\(([^)]+)\)\s*$")
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_plain(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 extract_img_blocks_from_db(old_content: str) -> list[str]:
"""从已入库正文中按文档顺序提取图片块（供复用 URL）。"""
if not old_content or "<img" not in old_content.lower():
return []
s = old_content.strip()
blocks: list[str] = []
for m in re.finditer(r"<p[^>]*>\s*(<img\b[^>]+/?>)\s*</p>", s, re.IGNORECASE | re.DOTALL):
blocks.append(m.group(0).strip())
if blocks:
return blocks
for m in re.finditer(r"<img\b[^>]+/?>", s, re.IGNORECASE):
blocks.append(f"<p>{m.group(0).strip()}</p>")
return blocks
def body_html_preserve_images(md_path: Path, old_content: str) -> tuple[str, list[str]]:
"""
按 fix_ip_book 的 HTML 段落规则生成正文，图片行使用旧库中的块。
返回 (body, warnings)
"""
warnings: list[str] = []
old_imgs = extract_img_blocks_from_db(old_content)
raw = md_path.read_text(encoding="utf-8")
lines = strip_md_title_line(raw).splitlines()
img_idx = 0
md_img_lines = sum(1 for ln in lines if LINE_IMAGE_ONLY.match(ln))
chunks: list[str] = []
for line in lines:
if line.strip() == "---":
chunks.append("")
continue
m = LINE_IMAGE_ONLY.match(line)
if m:
if img_idx < len(old_imgs):
chunks.append(old_imgs[img_idx])
img_idx += 1
else:
warnings.append(f"MD 图片行多于库中已存图片 ({md_path.name})，已跳过该行图片")
chunks.append("<p>（图片：库中无对应已上传块，请检查）</p>")
continue
stripped = line.replace("**", "").strip()
if stripped:
chunks.append(f"<p>{html.escape(stripped)}</p>")
else:
chunks.append("")
html_parts: list[str] = []
for c in chunks:
if c == "":
if html_parts and html_parts[-1] != "":
html_parts.append("")
else:
html_parts.append(c)
body = "\n".join(html_parts).strip() + "\n"
rest = len(old_imgs) - img_idx
if rest > 0:
warnings.append(f"{md_path.name}: 库中多 {rest} 张图片未在 MD 中引用，已从新正文移除（按 MD 为准）")
if md_img_lines and not old_imgs:
warnings.append(f"{md_path.name}: MD 含图片行但库中正文无 <img>，请先跑 fix_ip_book_sort_and_images 上传图片，或使用纯文本上传")
return body, warnings
def collect_files(part_num: int) -> list[tuple[str, str, 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
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
ip_id = f"ip{part_num}.{s_num}"
section_key = f"{part_num}.{s_num}"
title = title_raw
if section_key in seen_sections:
existing_path = seen_sections[section_key][2]
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 md_has_image_only_lines(md_path: Path) -> bool:
raw = md_path.read_text(encoding="utf-8")
text = strip_md_title_line(raw)
return any(LINE_IMAGE_ONLY.match(ln) for ln in text.splitlines())
def upsert_chapter(cur, chapter_id: str, chapter_title: str, ip_id: str, title: str, body: str, price: float, sort_order: int):
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, sort_order=%s WHERE id=%s",
(title, body, PART_ID, PART_TITLE, chapter_id, chapter_title, price, sort_order, ip_id),
)
return "updated"
cur.execute(
"INSERT INTO chapters (id, section_title, content, part_id, part_title, chapter_id, chapter_title, price, sort_order) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
(ip_id, title, body, PART_ID, PART_TITLE, chapter_id, chapter_title, price, sort_order),
)
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_err = 0
all_warnings: list[str] = []
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 sort_order, (ip_id, title, path) in enumerate(files, start=1):
try:
has_imgs = md_has_image_only_lines(path)
if has_imgs:
cur.execute("SELECT content FROM chapters WHERE id = %s", (ip_id,))
row = cur.fetchone()
old_content = row[0] if row and row[0] else ""
body, warns = body_html_preserve_images(path, old_content)
for w in warns:
all_warnings.append(f"{ip_id}: {w}")
else:
raw = path.read_text(encoding="utf-8")
body = for_miniprogram_body_plain(strip_md_title_line(raw))
except Exception as e:
print(f" ✗ 处理失败 {ip_id} ({path.name}): {e}")
total_err += 1
continue
if dry_run:
mode = "html+保留图" if has_imgs else "plain"
print(f" [dry] {ip_id} sort={sort_order} [{mode}] {title[:36]} ({path.name})")
total_ok += 1
continue
try:
action = upsert_chapter(
cur, chapter_id, chapter_title, ip_id, title, body, price, sort_order
)
conn.commit()
mode = "html" if has_imgs else "plain"
print(f" ✓ [{action}] {ip_id} sort={sort_order} [{mode}] {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_err}")
if all_warnings:
print("\n⚠ 提示：")
for w in all_warnings[:50]:
print(f" - {w}")
if len(all_warnings) > 50:
print(f" ... 另有 {len(all_warnings) - 50} 条")
if dry_run:
print("（dry-run 模式，未实际写库）")
if __name__ == "__main__":
main()

244
scripts/fix_ip_book_sort_and_images.py Normal file
View File

@@ -0,0 +1,244 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
修复《卡若的IP财富旅程》两个问题：
1. 排序错误：设置 sort_order（ip1.1=1, ip1.2=2, ...）使章节顺序正确
2. 图片缺失：上传 ../images/ 里的图片到 OSS，更新 content 为含 <img> 的 HTML
用法:
python3 scripts/fix_ip_book_sort_and_images.py
python3 scripts/fix_ip_book_sort_and_images.py --dry-run # 仅预览，不写库
"""
from __future__ import annotations
import html
import importlib.util
import os
import re
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
BOOK_DIR = Path("/Users/karuo/Documents/个人/2、我写的书/《卡若的IP财富旅程》")
API_BASE = os.environ.get("SOUL_API_BASE", "https://soulapi.quwanzhi.com")
PART_ID = "part-new-17755513182"
CHAPTERS = {
1: "chapter-ip-1",
2: "chapter-ip-2",
3: "chapter-ip-3",
4: "chapter-ip-4",
5: "chapter-ip-5",
}
FILENAME_RE = re.compile(
r"^[✅⬜]?\s*(\d+)\.(\d+)[- ]+(.+?)(?:\.md)?$",
re.UNICODE,
)
IMAGE_EXT = frozenset({".png", ".jpg", ".jpeg", ".gif", ".webp"})
LINE_IMAGE_ONLY = re.compile(r"^\s*!\[([^\]]*)\]\(([^)]+)\)\s*$")
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 resolve_image(md_path: Path, ref: str) -> Path | None:
if ref.startswith(("http://", "https://")):
return None
p = (md_path.parent / ref).expanduser().resolve()
if p.is_file() and p.suffix.lower() in IMAGE_EXT:
return p
return None
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:
try:
import requests
except ImportError:
sys.exit("需要: pip install requests")
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 has_image_refs(md_path: Path) -> bool:
text = md_path.read_text(encoding="utf-8")
return bool(re.search(r"^\s*!\[", text, re.MULTILINE))
def md_to_html(md_path: Path, upload_cache: dict[str, str], dry_run: bool) -> str:
raw = md_path.read_text(encoding="utf-8")
lines = raw.splitlines()
if lines and lines[0].lstrip().startswith("#"):
lines = lines[1:]
chunks: list[str] = []
for line in lines:
if line.strip() == "---":
chunks.append("")
continue
m = LINE_IMAGE_ONLY.match(line)
if m:
alt, ref = m.group(1), m.group(2).strip()
if ref.startswith(("http://", "https://")):
chunks.append(f'<p><img src="{html.escape(ref)}" alt="{html.escape(alt)}"/></p>')
continue
loc = resolve_image(md_path, ref)
if not loc:
chunks.append(f"<p>（图片路径无效：{html.escape(ref)}）</p>")
continue
key = str(loc)
if key not in upload_cache:
if dry_run:
upload_cache[key] = f"[dry]{loc.name}"
print(f" [dry] 上传图片: {loc.name}")
else:
print(f" 上传图片: {loc.name} …", end=" ", flush=True)
try:
upload_cache[key] = upload_image(loc)
print("✓", upload_cache[key])
except Exception as e:
print(f"✗ {e}")
upload_cache[key] = f"[上传失败]{loc.name}"
src = upload_cache[key]
chunks.append(f'<p><img src="{html.escape(src)}" alt="{html.escape(alt)}"/></p>')
continue
# 普通文字行
stripped = line.replace("**", "").strip()
if stripped:
chunks.append(f"<p>{html.escape(stripped)}</p>")
else:
chunks.append("")
# 压缩多余空行
html_parts: list[str] = []
for c in chunks:
if c == "":
if html_parts and html_parts[-1] != "":
html_parts.append("")
else:
html_parts.append(c)
return "\n".join(html_parts).strip() + "\n"
def plain_body(md_path: Path) -> str:
"""无图片章节：纯文本清洗（与原 batch_upload 一致）"""
raw = md_path.read_text(encoding="utf-8")
lines = raw.splitlines()
if lines and lines[0].lstrip().startswith("#"):
lines = lines[1:]
out_lines = []
for line in lines:
if line.strip() == "---":
out_lines.append("")
else:
out_lines.append(line)
body = "\n".join(out_lines).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]]:
part_dir = BOOK_DIR / f"PART{part_num}"
seen: dict[str, tuple[str, str, Path]] = {}
for f in part_dir.iterdir():
clean = re.sub(r"\.md(\.md)*$", "", f.name, flags=re.IGNORECASE)
m = FILENAME_RE.match(clean)
if not m or int(m.group(1)) != part_num:
continue
p_num, s_num, title_raw = m.group(1), m.group(2), m.group(3).strip()
ip_id = f"ip{p_num}.{s_num}"
key = f"{p_num}.{s_num}"
if key in seen:
if str(f).lower().endswith(".md") and not str(seen[key][2]).lower().endswith(".md"):
seen[key] = (ip_id, title_raw, f)
else:
seen[key] = (ip_id, title_raw, f)
result = list(seen.values())
result.sort(key=lambda x: float(re.sub(r"ip\d+\.", "", 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)
cur = conn.cursor()
upload_cache: dict[str, str] = {}
total_sort_fixed = 0
total_img_updated = 0
total_plain_updated = 0
for part_num, chapter_id in CHAPTERS.items():
print(f"\n{'='*60}")
print(f"PART{part_num} ({chapter_id})")
print(f"{'='*60}")
files = collect_files(part_num)
for sort_order, (ip_id, title, path) in enumerate(files, start=1):
has_imgs = has_image_refs(path)
img_tag = "🖼" if has_imgs else " "
print(f"\n {img_tag} {ip_id} sort={sort_order} {title[:35]} ({path.name})")
if has_imgs:
body = md_to_html(path, upload_cache, dry_run)
if not dry_run:
cur.execute(
"UPDATE chapters SET sort_order=%s, content=%s WHERE id=%s",
(sort_order, body, ip_id),
)
conn.commit()
total_img_updated += 1
total_sort_fixed += 1
else:
body = plain_body(path)
if not dry_run:
cur.execute(
"UPDATE chapters SET sort_order=%s, content=%s WHERE id=%s",
(sort_order, body, ip_id),
)
conn.commit()
total_plain_updated += 1
total_sort_fixed += 1
conn.close()
print(f"\n{'='*60}")
print(f"排序修复: {total_sort_fixed} 条")
print(f"图片章节更新: {total_img_updated} 条")
print(f"纯文本章节更新: {total_plain_updated} 条")
if dry_run:
print("（dry-run 模式，未实际写库）")
if __name__ == "__main__":
main()

330
scripts/optimize_28_models.py Normal file
View File

@@ -0,0 +1,330 @@
#!/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"![{alt}]({upload_cache[key]})"
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()

204
scripts/reupload_28_models_html.py Normal file
View File

@@ -0,0 +1,204 @@
#!/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()

279
scripts/reupload_gps_html.py Normal file
View File

@@ -0,0 +1,279 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
重新上传《游戏明星GPS导航》：Markdown → 高质量 HTML（表格/列表/粗体/标题/引用/图片）。
与 reupload_28_models_html.py 同样使用 python-markdown 做完整转换。
用法:
python3 scripts/reupload_gps_html.py --dry-run # 预览
python3 scripts/reupload_gps_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、我写的书/《游戏明星GPS导航》")
API_BASE = os.environ.get("SOUL_API_BASE", "https://soulapi.quwanzhi.com")
IMAGE_EXT = frozenset({".png", ".jpg", ".jpeg", ".gif", ".webp"})
MD_IMAGE_RE = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)")
BOOK_STRUCTURE = [
("part-gps-0", "序、什么是游戏明星", "chapter-gps-0", "什么是游戏明星", [
(1, "敬告", "序_什么是游戏明星/敬告.md"),
(2, "誓言", "序_什么是游戏明星/誓言.md"),
(3, "一、什么是游戏明星", "序_什么是游戏明星/一、什么是游戏明星.md"),
(4, "二、为什么要做游戏明星", "序_什么是游戏明星/二、为什么要做游戏明星.md"),
(5, "三、你好，未来的游戏明星", "序_什么是游戏明星/三、方向不对，努力全费...三、你好，未来的游戏明星.md"),
(6, "四、赚回十倍学费的一个小礼物", "序_什么是游戏明星/四、赚回十倍学费的一个小礼物.md"),
(7, "五、卡若与玩值电竞的故事", "序_什么是游戏明星/五、卡若与玩值电竞的故事.md"),
]),
("part-gps-1", "第一章、打开粉丝宝库的钥匙", "chapter-gps-1", "打开粉丝宝库的钥匙", [
(1, "粉丝是如何判断你有没有吸引力的", "第1章_打开粉丝宝库的钥匙/粉丝是如何判断你有没有吸引力的.md"),
(2, "一、我是如何发现这些粉丝的秘籍的", "第1章_打开粉丝宝库的钥匙/一、我是如何发现这些粉丝的秘籍的（打开粉丝宝库的钥匙）.md"),
(3, "二、目标粉丝聚集地（你的粉丝在哪里？）", "第1章_打开粉丝宝库的钥匙/二、目标粉丝聚集地（你的粉丝在哪里？）.md"),
(4, "三、价值输出载体（找出你与粉丝交流的桥梁）", "第1章_打开粉丝宝库的钥匙/三、价值输出载体（找出你与粉丝交流的桥梁）.md"),
(5, "四、不为人知的具体吸粉秘籍（总论）", "第1章_打开粉丝宝库的钥匙/四，不为人知的具体吸粉秘籍...（手把手带你吸粉）.md"),
(6, "4.1、斗鱼直播间粉丝自动流入", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.1，斗鱼直播间粉丝自动流入【吸粉秘籍】.md"),
(7, "4.2、原创游戏视频大脑缺口弥补术", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.2，原创游戏视频大脑缺口弥补术【吸粉秘籍】.md"),
(8, "4.3、批量渗透玩家微信群", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.3，批量渗透玩家微信群【吸粉秘籍】.md"),
(9, "4.4、朋友圈粉丝自我说服裂变", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.4，朋友圈粉丝自我说服裂变【吸粉秘籍】.md"),
(10, "4.5、淘宝老客户快速转移进入你的领地", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.5、淘宝老客户快速转移进入你的领地【吸粉秘籍】.md"),
(11, "4.6、人性换粉（威力巨大）", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.6，人性换粉（不管你在任何时间，任何场合永远有效，威力巨大）.md"),
(12, "QQ空间红包引流术", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/QQ空间红包引流术.md"),
(13, "微博问答降维吸粉", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/10，微博问答降维人吸粉.md"),
(14, "一个提醒", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/一个提醒.md"),
(15, "五、在这章你学到的东西，和将要得到的东西", "第1章_打开粉丝宝库的钥匙/五、在这章你学到的东西，和将要得到的东西.md"),
]),
("part-gps-2", "第二章、粉丝互动养熟秘籍", "chapter-gps-2", "粉丝互动养熟秘籍", [
(1, "粉丝对你的信任程度，与你的收入成正比", "第2章_粉丝互动养熟秘籍/粉丝对你的信任程度，以你的收入成正比。.md"),
(2, '让粉丝疯狂追捧的\u201c聊天技巧\u201d在哪里？', '第2章_粉丝互动养熟秘籍/让粉丝疯狂追棒的\u201c聊天技巧\u201d在哪里？.md'),
(3, "一、粉丝信任升级逻辑", "第2章_粉丝互动养熟秘籍/一、粉丝信任升级逻辑.md"),
(4, "二、如何在互动中，快速制造粉丝对你的好印象", "第2章_粉丝互动养熟秘籍/二、如何在互动中，快速制造粉丝对你的好印象.md"),
(5, "三、神奇的信任升级导图", "第2章_粉丝互动养熟秘籍/三、神奇的信任升级导图.md"),
(6, "四、3种立即见效的日常互动模型", "第2章_粉丝互动养熟秘籍/四、3种立既见效的日常互动模型.md"),
(7, "五、3个神秘的互动秘诀", "第2章_粉丝互动养熟秘籍/五、3个神秘的互动秘诀.md"),
(8, "六、4个步骤过滤超级粉丝", "第2章_粉丝互动养熟秘籍/六、4个步骤过滤超级粉丝.md"),
(9, "七、日常规划", "第2章_粉丝互动养熟秘籍/七、日常规划.md"),
]),
("part-gps-3", "第三章、粉丝转化系统", "chapter-gps-3", "粉丝转化系统", [
(1, "一、转化的四个前提", "第3章_粉丝转化系统/一、转化的四个前提.md"),
(2, "二、破解转化障碍", "第3章_粉丝转化系统/二、破解转化障碍.md"),
(3, "三、行之有效的转化秘籍", "第3章_粉丝转化系统/三、行之有效的转化秘籍.md"),
(4, "3.1、钟摆楼：一分钟获得变现的机会", "第3章_粉丝转化系统/3.1 钟摆楼，如何在一分钟之内获得变现的机会.md"),
(5, "3.2、社群价值裂变", "第3章_粉丝转化系统/3.2 激起兴趣的一些具体方法，微信群价值裂变.md"),
(6, "3.3、直播间转化实战", "第3章_粉丝转化系统/3.3 直播间转化实战.md"),
(7, "如何真正地掌握这套转化系统", "第3章_粉丝转化系统/一些想说的话，如何真正地掌握这套转化系统.md"),
]),
("part-gps-4", "第四章、超级粉丝", "chapter-gps-4", "超级粉丝", [
(1, "一、粉丝未来价值的重要性", "第4章_超级粉丝/一、粉丝未来价值的重要性.md"),
(2, "二、多维度升级为超级粉丝的谋略", "第4章_超级粉丝/二、多维度升级为超级粉丝的谋略.md"),
(3, "三、立体放大未来价值", "第4章_超级粉丝/三、立体放大未来价值.md"),
(4, "四、从超级粉丝到合伙人", "第4章_超级粉丝/四、从超级粉丝到合伙人.md"),
]),
("part-gps-5", "第五章、提升影响力的几个细节", "chapter-gps-5", "提升影响力的几个细节", [
(1, "一、当你觉得Hold不住自己野心时的处理方法", "第5章_提升影响力的几个细节/一、当你觉得Hold不住自己野心时的处理方法.md"),
(2, "二、保持你的个性，创造自己的小领地", "第5章_提升影响力的几个细节/二、保持你的个性，创造自己的小领地.md"),
(3, "三、如何让粉丝支持你", "第5章_提升影响力的几个细节/三、如何让粉丝和支持你.md"),
(4, "四、提升影响力前必须注意的细节", "第5章_提升影响力的几个细节/四、提升影响力前必须注意的细节.md"),
(5, "五、潜意识认同", "第5章_提升影响力的几个细节/五、潜意识认同.md"),
(6, "细节一、头像（视觉锤）", "第5章_提升影响力的几个细节/提升影响力的具体细节/一、头像.md"),
(7, "细节二、关联性昵称", "第5章_提升影响力的几个细节/提升影响力的具体细节/二、关联性昵称.md"),
(8, "细节三、个性签名", "第5章_提升影响力的几个细节/提升影响力的具体细节/三、个性签名.md"),
(9, "细节四、定位", "第5章_提升影响力的几个细节/提升影响力的具体细节/四、定位.md"),
(10, "案例：咨询卡若", "第5章_提升影响力的几个细节/案例/咨询卡若 （目前为了拒绝一些没什么问题的用户，你花半小时解答，他花30秒看完）.md"),
]),
("part-gps-6", "第六章、建造属于自己的游戏明星领地", "chapter-gps-6", "建造属于自己的游戏明星领地", [
(1, "一、什么是老铁", "第6章_建造属于自己的游戏明星领地/一、什么是老铁.md"),
(2, "二、如何设置领地规则", "第6章_建造属于自己的游戏明星领地/二、如何设置领地规则.md"),
(3, "三、玩值电竞——你的游戏明星大本营", "第6章_建造属于自己的游戏明星领地/三、玩值电竞——你的游戏明星大本营.md"),
(4, "四、领地的商业模式设计", "第6章_建造属于自己的游戏明星领地/四、领地的商业模式设计.md"),
(5, "如何坚持下去", "第6章_建造属于自己的游戏明星领地/如何坚持下去.md"),
]),
("part-gps-appendix-a", "附录A、玩值电竞入门指南", "chapter-gps-appendix-a", "玩值电竞入门指南", [
(1, "玩值电竞入门指南", "附录_玩值电竞入门指南/玩值电竞入门指南.md"),
]),
("part-gps-appendix-b", "附录B、干货工具箱", "chapter-gps-appendix-b", "干货工具箱", [
(1, "吸粉话术模板库", "附录_干货工具箱/吸粉话术模板库.md"),
(2, "游戏明星日常运营SOP", "附录_干货工具箱/游戏明星日常运营SOP.md"),
(3, "游戏明星自检清单", "附录_干货工具箱/游戏明星自检清单.md"),
]),
]
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（保持 markdown 格式）"""
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("✓")
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 clean_excessive_blank_lines(md_text: str) -> str:
"""压缩连续 3+ 空行为 2 空行，确保 markdown 解析正常"""
return re.sub(r"\n{3,}", "\n\n", 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()
cleaned = clean_excessive_blank_lines(cleaned)
html_out = markdown.markdown(
cleaned,
extensions=["tables", "fenced_code", "nl2br"],
)
html_out = re.sub(r"\n{3,}", "\n\n", html_out)
return html_out.strip()
def main():
try:
import pymysql
except ImportError:
sys.exit("需要: pip install pymysql markdown requests")
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 part_id, part_title, chapter_id, chapter_title, sections in BOOK_STRUCTURE:
print(f"\n{'='*60}")
print(f"{part_title}")
print(f"{'='*60}")
for seq, section_title, rel_path in sections:
part_num = part_id.replace("part-gps-", "")
section_id = f"gps-{part_num}.{seq}"
md_path = BOOK_DIR / rel_path
if not md_path.exists():
print(f" ✗ 文件不存在: {rel_path}")
total_err += 1
continue
print(f"\n 📄 {section_id} | {section_title}")
try:
raw = md_path.read_text(encoding="utf-8")
except Exception as e:
print(f" ✗ 读取失败: {e}")
total_err += 1
continue
processed = process_images(raw, md_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}")
has_table = "<table" in html_content
has_list = "<ul" in html_content or "<ol" in html_content
has_strong = "<strong" in html_content
has_img = "<img" in html_content
has_h2 = "<h2" in html_content or "<h3" in html_content
has_blockquote = "<blockquote" in html_content
tags = []
if has_table: tags.append("表格")
if has_list: tags.append("列表")
if has_strong: tags.append("粗体")
if has_img: tags.append("图片")
if has_h2: tags.append("标题")
if has_blockquote: tags.append("引用")
print(f" [dry] 富文本元素: {', '.join(tags) if tags else '纯文字'}")
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, section_id),
)
if cur.rowcount == 0:
print(f" ⚠ 未找到 id={section_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()

421
scripts/upload_gps_book.py Normal file
View File

@@ -0,0 +1,421 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
批量上传《游戏明星GPS导航》到小程序数据库。
结构：
序（part-gps-0）→ 7 节
第1章（part-gps-1）→ 15 节（含「四、吸粉秘籍」子目录 4.1-4.6 等）
第2章（part-gps-2）→ 9 节
第3章（part-gps-3）→ 7 节
第4章（part-gps-4）→ 4 节
第5章（part-gps-5）→ 10 节（含「具体细节」子目录 + 案例）
第6章（part-gps-6）→ 5 节
附录A（part-gps-appendix-a）→ 1 节
附录B（part-gps-appendix-b）→ 3 节
用法:
python3 scripts/upload_gps_book.py
python3 scripts/upload_gps_book.py --dry-run
"""
from __future__ import annotations
import html
import importlib.util
import os
import re
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
BOOK_DIR = Path("/Users/karuo/Documents/个人/2、我写的书/《游戏明星GPS导航》")
API_BASE = os.environ.get("SOUL_API_BASE", "https://soulapi.quwanzhi.com")
BOOK_KEY = "gps"
BOOK_TITLE = "游戏明星GPS导航"
BOOK_SUBTITLE = "教你在游戏领域成为明星"
BOOK_EMOJI = "🎮"
IMAGE_EXT = frozenset({".png", ".jpg", ".jpeg", ".gif", ".webp"})
LINE_IMAGE_ONLY = re.compile(r"^\s*!\[([^\]]*)\]\(([^)]+)\)\s*$")
# ---- 按目录.md 严格定义的章节排序 ----
# 每个 part: (part_id, part_title, chapter_id, chapter_title, sections)
# sections: [(section_id_suffix, section_title, relative_file_path)]
# section ID = "gps-{part_num}.{seq}" e.g. gps-0.1, gps-1.5
BOOK_STRUCTURE = [
(
"part-gps-0", "序、什么是游戏明星",
"chapter-gps-0", "什么是游戏明星",
[
(1, "敬告", "序_什么是游戏明星/敬告.md"),
(2, "誓言", "序_什么是游戏明星/誓言.md"),
(3, "一、什么是游戏明星", "序_什么是游戏明星/一、什么是游戏明星.md"),
(4, "二、为什么要做游戏明星", "序_什么是游戏明星/二、为什么要做游戏明星.md"),
(5, "三、你好，未来的游戏明星", "序_什么是游戏明星/三、方向不对，努力全费...三、你好，未来的游戏明星.md"),
(6, "四、赚回十倍学费的一个小礼物", "序_什么是游戏明星/四、赚回十倍学费的一个小礼物.md"),
(7, "五、卡若与玩值电竞的故事", "序_什么是游戏明星/五、卡若与玩值电竞的故事.md"),
],
),
(
"part-gps-1", "第一章、打开粉丝宝库的钥匙",
"chapter-gps-1", "打开粉丝宝库的钥匙",
[
(1, "粉丝是如何判断你有没有吸引力的", "第1章_打开粉丝宝库的钥匙/粉丝是如何判断你有没有吸引力的.md"),
(2, "一、我是如何发现这些粉丝的秘籍的", "第1章_打开粉丝宝库的钥匙/一、我是如何发现这些粉丝的秘籍的（打开粉丝宝库的钥匙）.md"),
(3, "二、目标粉丝聚集地（你的粉丝在哪里？）", "第1章_打开粉丝宝库的钥匙/二、目标粉丝聚集地（你的粉丝在哪里？）.md"),
(4, "三、价值输出载体（找出你与粉丝交流的桥梁）", "第1章_打开粉丝宝库的钥匙/三、价值输出载体（找出你与粉丝交流的桥梁）.md"),
(5, "四、不为人知的具体吸粉秘籍（总论）", "第1章_打开粉丝宝库的钥匙/四，不为人知的具体吸粉秘籍...（手把手带你吸粉）.md"),
(6, "4.1、斗鱼直播间粉丝自动流入", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.1，斗鱼直播间粉丝自动流入【吸粉秘籍】.md"),
(7, "4.2、原创游戏视频大脑缺口弥补术", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.2，原创游戏视频大脑缺口弥补术【吸粉秘籍】.md"),
(8, "4.3、批量渗透玩家微信群", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.3，批量渗透玩家微信群【吸粉秘籍】.md"),
(9, "4.4、朋友圈粉丝自我说服裂变", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.4，朋友圈粉丝自我说服裂变【吸粉秘籍】.md"),
(10, "4.5、淘宝老客户快速转移进入你的领地", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.5、淘宝老客户快速转移进入你的领地【吸粉秘籍】.md"),
(11, "4.6、人性换粉（威力巨大）", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/4.6，人性换粉（不管你在任何时间，任何场合永远有效，威力巨大）.md"),
(12, "QQ空间红包引流术", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/QQ空间红包引流术.md"),
(13, "微博问答降维吸粉", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/10，微博问答降维人吸粉.md"),
(14, "一个提醒", "第1章_打开粉丝宝库的钥匙/四，吸粉秘籍/一个提醒.md"),
(15, "五、在这章你学到的东西，和将要得到的东西", "第1章_打开粉丝宝库的钥匙/五、在这章你学到的东西，和将要得到的东西.md"),
],
),
(
"part-gps-2", "第二章、粉丝互动养熟秘籍",
"chapter-gps-2", "粉丝互动养熟秘籍",
[
(1, "粉丝对你的信任程度，与你的收入成正比", "第2章_粉丝互动养熟秘籍/粉丝对你的信任程度，以你的收入成正比。.md"),
(2, '让粉丝疯狂追捧的\u201c聊天技巧\u201d在哪里？', '第2章_粉丝互动养熟秘籍/让粉丝疯狂追棒的\u201c聊天技巧\u201d在哪里？.md'),
(3, "一、粉丝信任升级逻辑", "第2章_粉丝互动养熟秘籍/一、粉丝信任升级逻辑.md"),
(4, "二、如何在互动中，快速制造粉丝对你的好印象", "第2章_粉丝互动养熟秘籍/二、如何在互动中，快速制造粉丝对你的好印象.md"),
(5, "三、神奇的信任升级导图", "第2章_粉丝互动养熟秘籍/三、神奇的信任升级导图.md"),
(6, "四、3种立即见效的日常互动模型", "第2章_粉丝互动养熟秘籍/四、3种立既见效的日常互动模型.md"),
(7, "五、3个神秘的互动秘诀", "第2章_粉丝互动养熟秘籍/五、3个神秘的互动秘诀.md"),
(8, "六、4个步骤过滤超级粉丝", "第2章_粉丝互动养熟秘籍/六、4个步骤过滤超级粉丝.md"),
(9, "七、日常规划", "第2章_粉丝互动养熟秘籍/七、日常规划.md"),
],
),
(
"part-gps-3", "第三章、粉丝转化系统",
"chapter-gps-3", "粉丝转化系统",
[
(1, "一、转化的四个前提", "第3章_粉丝转化系统/一、转化的四个前提.md"),
(2, "二、破解转化障碍", "第3章_粉丝转化系统/二、破解转化障碍.md"),
(3, "三、行之有效的转化秘籍", "第3章_粉丝转化系统/三、行之有效的转化秘籍.md"),
(4, "3.1、钟摆楼：一分钟获得变现的机会", "第3章_粉丝转化系统/3.1 钟摆楼，如何在一分钟之内获得变现的机会.md"),
(5, "3.2、社群价值裂变", "第3章_粉丝转化系统/3.2 激起兴趣的一些具体方法，微信群价值裂变.md"),
(6, "3.3、直播间转化实战", "第3章_粉丝转化系统/3.3 直播间转化实战.md"),
(7, "如何真正地掌握这套转化系统", "第3章_粉丝转化系统/一些想说的话，如何真正地掌握这套转化系统.md"),
],
),
(
"part-gps-4", "第四章、超级粉丝",
"chapter-gps-4", "超级粉丝",
[
(1, "一、粉丝未来价值的重要性", "第4章_超级粉丝/一、粉丝未来价值的重要性.md"),
(2, "二、多维度升级为超级粉丝的谋略", "第4章_超级粉丝/二、多维度升级为超级粉丝的谋略.md"),
(3, "三、立体放大未来价值", "第4章_超级粉丝/三、立体放大未来价值.md"),
(4, "四、从超级粉丝到合伙人", "第4章_超级粉丝/四、从超级粉丝到合伙人.md"),
],
),
(
"part-gps-5", "第五章、提升影响力的几个细节",
"chapter-gps-5", "提升影响力的几个细节",
[
(1, "一、当你觉得Hold不住自己野心时的处理方法", "第5章_提升影响力的几个细节/一、当你觉得Hold不住自己野心时的处理方法.md"),
(2, "二、保持你的个性，创造自己的小领地", "第5章_提升影响力的几个细节/二、保持你的个性，创造自己的小领地.md"),
(3, "三、如何让粉丝支持你", "第5章_提升影响力的几个细节/三、如何让粉丝和支持你.md"),
(4, "四、提升影响力前必须注意的细节", "第5章_提升影响力的几个细节/四、提升影响力前必须注意的细节.md"),
(5, "五、潜意识认同", "第5章_提升影响力的几个细节/五、潜意识认同.md"),
(6, "细节一、头像（视觉锤）", "第5章_提升影响力的几个细节/提升影响力的具体细节/一、头像.md"),
(7, "细节二、关联性昵称", "第5章_提升影响力的几个细节/提升影响力的具体细节/二、关联性昵称.md"),
(8, "细节三、个性签名", "第5章_提升影响力的几个细节/提升影响力的具体细节/三、个性签名.md"),
(9, "细节四、定位", "第5章_提升影响力的几个细节/提升影响力的具体细节/四、定位.md"),
(10, "案例：咨询卡若", "第5章_提升影响力的几个细节/案例/咨询卡若 （目前为了拒绝一些没什么问题的用户，你花半小时解答，他花30秒看完）.md"),
],
),
(
"part-gps-6", "第六章、建造属于自己的游戏明星领地",
"chapter-gps-6", "建造属于自己的游戏明星领地",
[
(1, "一、什么是老铁", "第6章_建造属于自己的游戏明星领地/一、什么是老铁.md"),
(2, "二、如何设置领地规则", "第6章_建造属于自己的游戏明星领地/二、如何设置领地规则.md"),
(3, "三、玩值电竞——你的游戏明星大本营", "第6章_建造属于自己的游戏明星领地/三、玩值电竞——你的游戏明星大本营.md"),
(4, "四、领地的商业模式设计", "第6章_建造属于自己的游戏明星领地/四、领地的商业模式设计.md"),
(5, "如何坚持下去", "第6章_建造属于自己的游戏明星领地/如何坚持下去.md"),
],
),
(
"part-gps-appendix-a", "附录A、玩值电竞入门指南",
"chapter-gps-appendix-a", "玩值电竞入门指南",
[
(1, "玩值电竞入门指南", "附录_玩值电竞入门指南/玩值电竞入门指南.md"),
],
),
(
"part-gps-appendix-b", "附录B、干货工具箱",
"chapter-gps-appendix-b", "干货工具箱",
[
(1, "吸粉话术模板库", "附录_干货工具箱/吸粉话术模板库.md"),
(2, "游戏明星日常运营SOP", "附录_干货工具箱/游戏明星日常运营SOP.md"),
(3, "游戏明星自检清单", "附录_干货工具箱/游戏明星自检清单.md"),
],
),
]
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 resolve_image(md_path: Path, ref: str) -> Path | None:
if ref.startswith(("http://", "https://")):
return None
p = (md_path.parent / ref).expanduser().resolve()
if p.is_file() and p.suffix.lower() in IMAGE_EXT:
return p
return None
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:
import requests
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 md_to_html(md_path: Path, upload_cache: dict[str, str], dry_run: bool) -> str:
raw = md_path.read_text(encoding="utf-8")
lines = raw.splitlines()
if lines and lines[0].lstrip().startswith("#"):
lines = lines[1:]
chunks: list[str] = []
for line in lines:
if line.strip() == "---":
chunks.append("")
continue
m = LINE_IMAGE_ONLY.match(line)
if m:
alt, ref = m.group(1), m.group(2).strip()
if ref.startswith(("http://", "https://")):
chunks.append(f'<p><img src="{html.escape(ref)}" alt="{html.escape(alt)}"/></p>')
continue
loc = resolve_image(md_path, ref)
if not loc:
chunks.append(f"<p>（图片路径无效：{html.escape(ref)}）</p>")
continue
key = str(loc)
if key not in upload_cache:
if dry_run:
upload_cache[key] = f"[dry]{loc.name}"
print(f" [dry] 上传图片: {loc.name}")
else:
print(f" 上传图片: {loc.name} …", end=" ", flush=True)
try:
upload_cache[key] = upload_image(loc)
print("✓", upload_cache[key])
except Exception as e:
print(f"✗ {e}")
upload_cache[key] = f"[上传失败]{loc.name}"
src = upload_cache[key]
chunks.append(f'<p><img src="{html.escape(src)}" alt="{html.escape(alt)}"/></p>')
continue
stripped = line.strip()
if stripped.startswith("##"):
heading = stripped.lstrip("#").strip()
chunks.append(f"<h3>{html.escape(heading)}</h3>")
elif stripped:
chunks.append(f"<p>{html.escape(stripped)}</p>")
else:
chunks.append("")
html_parts: list[str] = []
for c in chunks:
if c == "":
if html_parts and html_parts[-1] != "":
html_parts.append("")
else:
html_parts.append(c)
return "\n".join(html_parts).strip() + "\n"
def has_image_refs(md_path: Path) -> bool:
text = md_path.read_text(encoding="utf-8")
return bool(re.search(r"^\s*!\[", text, re.MULTILINE))
def plain_body(md_path: Path) -> str:
raw = md_path.read_text(encoding="utf-8")
lines = raw.splitlines()
if lines and lines[0].lstrip().startswith("#"):
lines = lines[1:]
out_lines = []
for line in lines:
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 ensure_book(cur, conn) -> int:
cur.execute("SELECT id FROM books WHERE book_key = %s", (BOOK_KEY,))
row = cur.fetchone()
if row:
print(f"书已存在: book_id={row[0]}")
return row[0]
cur.execute("SELECT MAX(sort_order) FROM books")
max_sort = cur.fetchone()[0] or 0
cur.execute(
"INSERT INTO books (book_key, title, subtitle, icon_emoji, sort_order, status) "
"VALUES (%s, %s, %s, %s, %s, %s)",
(BOOK_KEY, BOOK_TITLE, BOOK_SUBTITLE, BOOK_EMOJI, max_sort + 1, "published"),
)
conn.commit()
cur.execute("SELECT id FROM books WHERE book_key = %s", (BOOK_KEY,))
book_id = cur.fetchone()[0]
print(f"已创建书: book_id={book_id}, key={BOOK_KEY}, title={BOOK_TITLE}")
return book_id
def upsert_section(cur, conn, book_id: int, section_id: str, part_id: str, part_title: str,
chapter_id: str, chapter_title: str, section_title: str,
content: str, sort_order: int, price: float):
cur.execute("SELECT mid FROM chapters WHERE id = %s", (section_id,))
existing = cur.fetchone()
if existing:
cur.execute(
"UPDATE chapters SET book_id=%s, section_title=%s, content=%s, "
"part_id=%s, part_title=%s, chapter_id=%s, chapter_title=%s, "
"price=%s, sort_order=%s, word_count=%s, status=%s, "
"edition_standard=%s, edition_premium=%s WHERE id=%s",
(book_id, section_title, content,
part_id, part_title, chapter_id, chapter_title,
price, sort_order, len(content), "published",
True, False, section_id),
)
conn.commit()
return "updated"
else:
cur.execute(
"INSERT INTO chapters (id, book_id, section_title, content, "
"part_id, part_title, chapter_id, chapter_title, "
"price, is_free, sort_order, word_count, status, "
"edition_standard, edition_premium) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(section_id, book_id, section_title, content,
part_id, part_title, chapter_id, chapter_title,
price, False, sort_order, len(content), "published",
True, False),
)
conn.commit()
return "created"
def main():
try:
import pymysql
except ImportError:
sys.exit("需要: pip install pymysql")
dry_run = "--dry-run" in sys.argv
price = 1.0
cfg = load_db()
conn = pymysql.connect(**cfg)
cur = conn.cursor()
if not dry_run:
book_id = ensure_book(cur, conn)
else:
book_id = 0
print("[dry-run] 跳过创建书")
upload_cache: dict[str, str] = {}
global_sort = 0
total_ok = 0
total_err = 0
for part_id, part_title, chapter_id, chapter_title, sections in BOOK_STRUCTURE:
print(f"\n{'='*60}")
print(f"{part_title}")
print(f" part_id={part_id} chapter_id={chapter_id}")
print(f"{'='*60}")
for seq, section_title, rel_path in sections:
global_sort += 1
part_num = part_id.replace("part-gps-", "")
section_id = f"gps-{part_num}.{seq}"
md_path = BOOK_DIR / rel_path
if not md_path.exists():
print(f" ✗ 文件不存在: {rel_path}")
total_err += 1
continue
has_imgs = has_image_refs(md_path)
img_tag = "🖼" if has_imgs else " "
try:
if has_imgs:
content = md_to_html(md_path, upload_cache, dry_run)
else:
content = plain_body(md_path)
except Exception as e:
print(f" ✗ 读取失败 {rel_path}: {e}")
total_err += 1
continue
if dry_run:
print(f" {img_tag} [{section_id}] sort={global_sort} {section_title[:40]}")
total_ok += 1
continue
try:
action = upsert_section(
cur, conn, book_id, section_id,
part_id, part_title, chapter_id, chapter_title,
section_title, content, global_sort, price,
)
print(f" ✓ [{action}] [{section_id}] sort={global_sort} {section_title[:40]}")
total_ok += 1
except Exception as e:
conn.rollback()
print(f" ✗ 失败 {section_id}: {e}")
total_err += 1
conn.close()
print(f"\n{'='*60}")
print(f"完成：成功 {total_ok} | 失败 {total_err}")
if dry_run:
print("（dry-run 模式，未实际写库）")
if __name__ == "__main__":
main()

367
scripts/upload_solo_company_book.py Normal file
View File

@@ -0,0 +1,367 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
上传《一人公司——你能掌控的自由人生》到小程序。
book_key : solo-company
共 10 章 + 1 术语手册，分 4 个篇。
用法:
python3 scripts/upload_solo_company_book.py --dry-run # 预览
python3 scripts/upload_solo_company_book.py # 执行上传
"""
from __future__ import annotations
import re
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
BOOK_DIR = Path("/Users/karuo/Documents/个人/2、我写的书/《一人公司》")
BOOK_KEY = "solo-company"
BOOK_TITLE = "一人公司"
BOOK_SUBTITLE = "你能掌控的自由人生"
BOOK_EMOJI = "🏢"
BOOK_SORT_ORDER = 10
# 篇章结构定义
PARTS = {
"part-solo-1": {
"title": "启程·方向",
"chapters": {
"chapter-solo-1": {
"title": "找准方向，别再瞎忙",
"sections": [1, 2, 3],
}
},
},
"part-solo-2": {
"title": "实战·方法",
"chapters": {
"chapter-solo-2": {
"title": "用对方法，高效变现",
"sections": [4, 5, 6],
}
},
},
"part-solo-3": {
"title": "进阶·系统",
"chapters": {
"chapter-solo-3": {
"title": "搭建系统，持续增长",
"sections": [7, 8, 9, 10],
}
},
},
"part-solo-appendix": {
"title": "附录",
"chapters": {
"ch-solo-appendix": {
"title": "术语手册",
"sections": ["glossary"],
}
},
},
}
# 章节文件映射
SECTION_FILES = {
1: "1、你越努力越穷，是因为你连方向都没找对.md",
2: "2、9个人干出千万业绩：一人公司到底怎么运转的.md",
3: "3、别再找项目了——你自己就是最值钱的产品.md",
4: "4、我从不追求原创，但我每篇内容都赚到了钱.md",
5: "5、30个账号每天自动来客户——你也能复制这套打法.md",
6: "6、你认识谁，比你会什么值钱10倍.md",
7: "7、搭一台你睡着了也在赚钱的机器.md",
8: "8、你赚多少钱，取决于你把自己当成什么人.md",
9: "9、一天重启你的人生：我用了10年验证的方法.md",
10: "10、别等准备好再开始——这游戏根本没有终点.md",
"glossary": "术语手册：这本书里的黑话全在这.md",
}
# 章节标题（去掉序号前缀）
SECTION_TITLES = {
1: "你越努力越穷，是因为你连方向都没找对",
2: "9个人干出千万业绩：一人公司到底怎么运转的",
3: "别再找项目了——你自己就是最值钱的产品",
4: "我从不追求原创，但我每篇内容都赚到了钱",
5: "30个账号每天自动来客户——你也能复制这套打法",
6: "你认识谁，比你会什么值钱10倍",
7: "搭一台你睡着了也在赚钱的机器",
8: "你赚多少钱，取决于你把自己当成什么人",
9: "一天重启你的人生：我用了10年验证的方法",
10: "别等准备好再开始——这游戏根本没有终点",
"glossary": "术语手册：这本书里的黑话全在这",
}
DB_CONFIG = {
"host": "56b4c23f6853c.gz.cdb.myqcloud.com",
"port": 14413,
"user": "cdb_outerroot",
"password": "Zhiqun1984",
"database": "soul_miniprogram",
}
def md_to_html(md_text: str) -> str:
"""将 Markdown 转为美观的 HTML（适配小程序 contentParser）"""
lines = md_text.splitlines()
# 去掉第一行 # 标题
if lines and lines[0].lstrip().startswith("# "):
lines = lines[1:]
html_parts: list[str] = []
i = 0
def inline_format(text: str) -> str:
"""处理行内格式：粗体、斜体、行内代码、链接"""
# 粗体+斜体
text = re.sub(r"\*\*\*(.+?)\*\*\*", r"<strong><em>\1</em></strong>", text)
# 粗体
text = re.sub(r"\*\*(.+?)\*\*", r"<strong>\1</strong>", text)
# 斜体
text = re.sub(r"(?<!\*)\*(?!\*)(.+?)(?<!\*)\*(?!\*)", r"<em>\1</em>", text)
# 行内代码
text = re.sub(r"`(.+?)`", r"<code>\1</code>", text)
# 链接
text = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r'<a href="\2">\1</a>', text)
return text
while i < len(lines):
line = lines[i]
stripped = line.strip()
# 空行跳过
if not stripped:
i += 1
continue
# 图片 → 跳过（小程序不支持外链图片）
if re.match(r"^!\[.*\]\(.*\)$", stripped):
i += 1
continue
# 水平线
if stripped == "---":
html_parts.append("<hr>")
i += 1
continue
# h2
if stripped.startswith("## "):
heading_text = inline_format(stripped[3:].strip())
html_parts.append(f"<h2>{heading_text}</h2>")
i += 1
continue
# h3
if stripped.startswith("### "):
heading_text = inline_format(stripped[4:].strip())
html_parts.append(f"<h3>{heading_text}</h3>")
i += 1
continue
# 引用块
if stripped.startswith("> "):
quote_lines = []
while i < len(lines) and lines[i].strip().startswith("> "):
quote_lines.append(inline_format(lines[i].strip()[2:].strip()))
i += 1
html_parts.append(f"<blockquote>{'<br>'.join(quote_lines)}</blockquote>")
continue
# 待办列表 - [ ] / - [x]
if re.match(r"^- \[[ x]\] ", stripped):
items = []
while i < len(lines) and re.match(r"^- \[[ x]\] ", lines[i].strip()):
s = lines[i].strip()
checked = s[3] == "x"
text = inline_format(s[6:].strip())
check_mark = "✅" if checked else "☐"
items.append(f"<li>{check_mark} {text}</li>")
i += 1
html_parts.append(f"<ul>{''.join(items)}</ul>")
continue
# 有序列表
if re.match(r"^\d+\.\s+", stripped):
items = []
while i < len(lines):
s = lines[i].strip()
m = re.match(r"^\d+\.\s+(.+)", s)
if not m:
break
text = inline_format(m.group(1))
items.append(f"<li>{text}</li>")
i += 1
# 如果下一行是缩进的续行，合并进当前 li
while i < len(lines) and lines[i].strip() and not re.match(r"^\d+\.\s+", lines[i].strip()) and not lines[i].strip().startswith("- ") and not lines[i].strip().startswith("#") and not lines[i].strip().startswith(">") and lines[i].startswith(" "):
items[-1] = items[-1].replace("</li>", f"<br>{inline_format(lines[i].strip())}</li>")
i += 1
html_parts.append(f"<ol>{''.join(items)}</ol>")
continue
# 无序列表（含多级缩进 *）
if re.match(r"^[-*]\s+", stripped) and not re.match(r"^- \[[ x]\] ", stripped):
items = []
while i < len(lines):
s = lines[i].strip()
m = re.match(r"^[-*]\s+(.+)", s)
if not m:
break
text = inline_format(m.group(1))
items.append(f"<li>{text}</li>")
i += 1
# 缩进续行
while i < len(lines) and lines[i].strip() and not re.match(r"^[-*]\s+", lines[i].strip()) and not re.match(r"^\d+\.\s+", lines[i].strip()) and not lines[i].strip().startswith("#") and not lines[i].strip().startswith(">"):
if lines[i].startswith(" ") or lines[i].startswith("\t"):
sub = lines[i].strip()
sub_m = re.match(r"^[-*]\s+(.+)", sub)
if sub_m:
items[-1] = items[-1].replace("</li>", f"<br>• {inline_format(sub_m.group(1))}</li>")
else:
items[-1] = items[-1].replace("</li>", f"<br>{inline_format(sub)}</li>")
i += 1
else:
break
html_parts.append(f"<ul>{''.join(items)}</ul>")
continue
# 普通段落
para_lines = [inline_format(stripped)]
i += 1
html_parts.append(f"<p>{para_lines[0]}</p>")
continue
return "\n".join(html_parts)
def find_section_location(section_key) -> tuple[str, str, str, str]:
"""根据 section key 找到对应的 part/chapter 信息"""
for part_id, part_info in PARTS.items():
for ch_id, ch_info in part_info["chapters"].items():
if section_key in ch_info["sections"]:
return part_id, part_info["title"], ch_id, ch_info["title"]
return "part-solo-1", "启程·方向", "chapter-solo-1", "找准方向，别再瞎忙"
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_free = 0.0
price_paid = 1.0
conn = pymysql.connect(**DB_CONFIG)
cur = conn.cursor()
# ── Step 1: 确保书籍存在 ──
cur.execute("SELECT id FROM books WHERE book_key = %s", (BOOK_KEY,))
row = cur.fetchone()
if row:
book_id = row[0]
print(f"📖 书籍已存在: id={book_id}, key={BOOK_KEY}")
if not dry_run:
cur.execute(
"UPDATE books SET title=%s, subtitle=%s, icon_emoji=%s, sort_order=%s, status='published' WHERE id=%s",
(BOOK_TITLE, BOOK_SUBTITLE, BOOK_EMOJI, BOOK_SORT_ORDER, book_id),
)
conn.commit()
else:
if dry_run:
print(f"📖 [dry] 将创建书籍: {BOOK_KEY} - {BOOK_TITLE}")
book_id = 999
else:
cur.execute(
"INSERT INTO books (book_key, title, subtitle, icon_emoji, sort_order, status) VALUES (%s, %s, %s, %s, %s, 'published')",
(BOOK_KEY, BOOK_TITLE, BOOK_SUBTITLE, BOOK_EMOJI, BOOK_SORT_ORDER),
)
conn.commit()
book_id = cur.lastrowid
print(f"📖 已创建书籍: id={book_id}, key={BOOK_KEY}")
# ── Step 2: 上传各章节 ──
total_ok = 0
total_err = 0
sort_order = 0
for section_key in list(range(1, 11)) + ["glossary"]:
filename = SECTION_FILES[section_key]
title = SECTION_TITLES[section_key]
section_id = f"solo-{section_key}" if isinstance(section_key, int) else "solo-glossary"
part_id, part_title, chapter_id, chapter_title = find_section_location(section_key)
filepath = BOOK_DIR / filename
if not filepath.exists():
print(f" ⚠ 文件不存在: {filepath}")
total_err += 1
continue
raw = filepath.read_text(encoding="utf-8")
html_content = md_to_html(raw)
word_count = len(re.sub(r"<[^>]+>", "", html_content))
is_free = section_key in [1, "glossary"]
price = price_free if is_free else price_paid
if dry_run:
preview = html_content[:80].replace("\n", " ")
print(f" [dry] {section_id:16s} | {title[:35]:35s} | {word_count:5d}字 | {'免费' if is_free else '付费'} | {preview}…")
total_ok += 1
sort_order += 1
continue
try:
cur.execute("SELECT mid FROM chapters WHERE id = %s", (section_id,))
existing = cur.fetchone()
if existing:
cur.execute(
"""UPDATE chapters SET
book_id=%s, section_title=%s, content=%s, word_count=%s,
part_id=%s, part_title=%s, chapter_id=%s, chapter_title=%s,
price=%s, is_free=%s, sort_order=%s, status='published',
edition_standard=1, edition_premium=0
WHERE id=%s""",
(book_id, title, html_content, word_count,
part_id, part_title, chapter_id, chapter_title,
price, is_free, sort_order, section_id),
)
conn.commit()
print(f" ✓ [updated] {section_id:16s} | {title[:35]:35s} | {word_count:5d}字")
else:
cur.execute(
"""INSERT INTO chapters
(id, book_id, section_title, content, word_count,
part_id, part_title, chapter_id, chapter_title,
price, is_free, sort_order, status, edition_standard, edition_premium)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'published', 1, 0)""",
(section_id, book_id, title, html_content, word_count,
part_id, part_title, chapter_id, chapter_title,
price, is_free, sort_order),
)
conn.commit()
print(f" ✓ [created] {section_id:16s} | {title[:35]:35s} | {word_count:5d}字")
total_ok += 1
except Exception as e:
conn.rollback()
print(f" ✗ 失败 {section_id}: {e}")
total_err += 1
sort_order += 1
conn.close()
print(f"\n{'='*60}")
print(f"完成：成功 {total_ok} | 失败 {total_err}")
if dry_run:
print("（dry-run 模式，未实际写库）")
else:
print("✅ 上传完毕！请在管理端或小程序验证。")
if __name__ == "__main__":
main()