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