#!/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'

{html.escape(alt)}

') continue loc = resolve_image(md_path, ref) if not loc: chunks.append(f"

（图片路径无效：{html.escape(ref)}）

") 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'

{html.escape(alt)}

') continue stripped = line.strip() if stripped.startswith("##"): heading = stripped.lstrip("#").strip() chunks.append(f"

{html.escape(heading)}

") elif stripped: chunks.append(f"

{html.escape(stripped)}

") 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()