276 lines
9.0 KiB
Python
276 lines
9.0 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""
|
|||
|
|
Soul 内容上传接口
|
|||
|
|
可从 Cursor Skill / 命令行直接调用，将新内容写入数据库
|
|||
|
|
|
|||
|
|
用法:
|
|||
|
|
python3 content_upload.py --title "标题" --price 1.0 --content "正文" \
|
|||
|
|
--part part-1 --chapter chapter-1 --format markdown
|
|||
|
|
|
|||
|
|
python3 content_upload.py --json '{
|
|||
|
|
"title": "标题",
|
|||
|
|
"price": 1.0,
|
|||
|
|
"content": "正文内容...",
|
|||
|
|
"part_id": "part-1",
|
|||
|
|
"chapter_id": "chapter-1",
|
|||
|
|
"format": "markdown",
|
|||
|
|
"images": ["https://xxx.com/img1.png"]
|
|||
|
|
}'
|
|||
|
|
|
|||
|
|
python3 content_upload.py --list-structure # 查看篇章结构
|
|||
|
|
|
|||
|
|
环境依赖: pip install pymysql
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import argparse
|
|||
|
|
import json
|
|||
|
|
import sys
|
|||
|
|
import re
|
|||
|
|
from datetime import datetime
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
import pymysql
|
|||
|
|
except ImportError:
|
|||
|
|
print("需要安装 pymysql: pip3 install pymysql")
|
|||
|
|
sys.exit(1)
|
|||
|
|
|
|||
|
|
DB_CONFIG = {
|
|||
|
|
"host": "56b4c23f6853c.gz.cdb.myqcloud.com",
|
|||
|
|
"port": 14413,
|
|||
|
|
"user": "cdb_outerroot",
|
|||
|
|
"password": "Zhiqun1984",
|
|||
|
|
"database": "soul_miniprogram",
|
|||
|
|
"charset": "utf8mb4",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
PART_MAP = {
|
|||
|
|
"part-1": "第一篇｜真实的人",
|
|||
|
|
"part-2": "第二篇｜真实的行业",
|
|||
|
|
"part-3": "第三篇｜真实的错误",
|
|||
|
|
"part-4": "第四篇｜真实的赚钱",
|
|||
|
|
"part-5": "第五篇｜真实的社会",
|
|||
|
|
"appendix": "附录",
|
|||
|
|
"intro": "序言",
|
|||
|
|
"outro": "尾声",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
CHAPTER_MAP = {
|
|||
|
|
"chapter-1": "第1章｜人与人之间的底层逻辑",
|
|||
|
|
"chapter-2": "第2章｜人性困境案例",
|
|||
|
|
"chapter-3": "第3章｜电商篇",
|
|||
|
|
"chapter-4": "第4章｜内容商业篇",
|
|||
|
|
"chapter-5": "第5章｜传统行业篇",
|
|||
|
|
"chapter-6": "第6章｜我人生错过的4件大钱",
|
|||
|
|
"chapter-7": "第7章｜别人犯的错误",
|
|||
|
|
"chapter-8": "第8章｜底层结构",
|
|||
|
|
"chapter-9": "第9章｜我在Soul上亲访的赚钱案例",
|
|||
|
|
"chapter-10": "第10章｜未来职业的变化趋势",
|
|||
|
|
"chapter-11": "第11章｜中国社会商业生态的未来",
|
|||
|
|
"appendix": "附录",
|
|||
|
|
"preface": "序言",
|
|||
|
|
"epilogue": "尾声",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_connection():
|
|||
|
|
return pymysql.connect(**DB_CONFIG)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def list_structure():
|
|||
|
|
conn = get_connection()
|
|||
|
|
cur = conn.cursor()
|
|||
|
|
cur.execute("""
|
|||
|
|
SELECT part_id, part_title, chapter_id, chapter_title, COUNT(*) as sections
|
|||
|
|
FROM chapters
|
|||
|
|
GROUP BY part_id, part_title, chapter_id, chapter_title
|
|||
|
|
ORDER BY part_id, chapter_id
|
|||
|
|
""")
|
|||
|
|
rows = cur.fetchall()
|
|||
|
|
print("篇章结构:")
|
|||
|
|
for part_id, part_title, ch_id, ch_title, cnt in rows:
|
|||
|
|
print(f" {part_id} ({part_title}) / {ch_id} ({ch_title}) - {cnt}节")
|
|||
|
|
|
|||
|
|
cur.execute("SELECT COUNT(*) FROM chapters")
|
|||
|
|
total = cur.fetchone()[0]
|
|||
|
|
print(f"\n总计: {total} 节")
|
|||
|
|
conn.close()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def generate_section_id(cur, chapter_id):
|
|||
|
|
"""根据 chapter 编号自动生成下一个 section id"""
|
|||
|
|
ch_num = re.search(r"\d+", chapter_id)
|
|||
|
|
if not ch_num:
|
|||
|
|
cur.execute("SELECT MAX(CAST(REPLACE(id, '.', '') AS UNSIGNED)) FROM chapters")
|
|||
|
|
max_id = cur.fetchone()[0] or 0
|
|||
|
|
return str(max_id + 1)
|
|||
|
|
|
|||
|
|
prefix = ch_num.group()
|
|||
|
|
cur.execute(
|
|||
|
|
"SELECT id FROM chapters WHERE id LIKE %s ORDER BY CAST(SUBSTRING_INDEX(id, '.', -1) AS UNSIGNED) DESC LIMIT 1",
|
|||
|
|
(f"{prefix}.%",),
|
|||
|
|
)
|
|||
|
|
row = cur.fetchone()
|
|||
|
|
if row:
|
|||
|
|
last_num = int(row[0].split(".")[-1])
|
|||
|
|
return f"{prefix}.{last_num + 1}"
|
|||
|
|
return f"{prefix}.1"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def upload_content(data):
|
|||
|
|
title = data.get("title", "").strip()
|
|||
|
|
if not title:
|
|||
|
|
print("错误: 标题不能为空")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
content = data.get("content", "").strip()
|
|||
|
|
if not content:
|
|||
|
|
print("错误: 内容不能为空")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
price = float(data.get("price", 1.0))
|
|||
|
|
is_free = 1 if price == 0 else 0
|
|||
|
|
part_id = data.get("part_id", "part-1")
|
|||
|
|
chapter_id = data.get("chapter_id", "chapter-1")
|
|||
|
|
fmt = data.get("format", "markdown")
|
|||
|
|
images = data.get("images", [])
|
|||
|
|
section_id = data.get("id", "")
|
|||
|
|
|
|||
|
|
if images:
|
|||
|
|
for i, img_url in enumerate(images):
|
|||
|
|
placeholder = f"{{{{image_{i+1}}}}}"
|
|||
|
|
if placeholder in content:
|
|||
|
|
if fmt == "markdown":
|
|||
|
|
content = content.replace(placeholder, f"")
|
|||
|
|
else:
|
|||
|
|
content = content.replace(placeholder, img_url)
|
|||
|
|
|
|||
|
|
word_count = len(re.sub(r"\s+", "", content))
|
|||
|
|
|
|||
|
|
part_title = PART_MAP.get(part_id, part_id)
|
|||
|
|
chapter_title = CHAPTER_MAP.get(chapter_id, chapter_id)
|
|||
|
|
|
|||
|
|
conn = get_connection()
|
|||
|
|
cur = conn.cursor()
|
|||
|
|
|
|||
|
|
if not section_id:
|
|||
|
|
section_id = generate_section_id(cur, chapter_id)
|
|||
|
|
|
|||
|
|
cur.execute("SELECT mid FROM chapters WHERE id = %s", (section_id,))
|
|||
|
|
existing = cur.fetchone()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
if existing:
|
|||
|
|
cur.execute("""
|
|||
|
|
UPDATE chapters SET
|
|||
|
|
section_title = %s, content = %s, word_count = %s,
|
|||
|
|
is_free = %s, price = %s, part_id = %s, part_title = %s,
|
|||
|
|
chapter_id = %s, chapter_title = %s, status = 'published'
|
|||
|
|
WHERE id = %s
|
|||
|
|
""", (title, content, word_count, is_free, price, part_id, part_title,
|
|||
|
|
chapter_id, chapter_title, section_id))
|
|||
|
|
action = "更新"
|
|||
|
|
else:
|
|||
|
|
cur.execute("SELECT COALESCE(MAX(sort_order), 0) + 1 FROM chapters")
|
|||
|
|
next_order = cur.fetchone()[0]
|
|||
|
|
|
|||
|
|
cur.execute("""
|
|||
|
|
INSERT INTO chapters (id, part_id, part_title, chapter_id, chapter_title,
|
|||
|
|
section_title, content, word_count, is_free, price, sort_order, status)
|
|||
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'published')
|
|||
|
|
""", (section_id, part_id, part_title, chapter_id, chapter_title,
|
|||
|
|
title, content, word_count, is_free, price, next_order))
|
|||
|
|
action = "创建"
|
|||
|
|
|
|||
|
|
conn.commit()
|
|||
|
|
|
|||
|
|
result = {
|
|||
|
|
"success": True,
|
|||
|
|
"action": action,
|
|||
|
|
"data": {
|
|||
|
|
"id": section_id,
|
|||
|
|
"title": title,
|
|||
|
|
"part": f"{part_id} ({part_title})",
|
|||
|
|
"chapter": f"{chapter_id} ({chapter_title})",
|
|||
|
|
"price": price,
|
|||
|
|
"is_free": bool(is_free),
|
|||
|
|
"word_count": word_count,
|
|||
|
|
"format": fmt,
|
|||
|
|
"images_count": len(images),
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
except pymysql.err.IntegrityError as e:
|
|||
|
|
print(json.dumps({"success": False, "error": f"ID冲突: {e}"}, ensure_ascii=False))
|
|||
|
|
return False
|
|||
|
|
except Exception as e:
|
|||
|
|
conn.rollback()
|
|||
|
|
print(json.dumps({"success": False, "error": str(e)}, ensure_ascii=False))
|
|||
|
|
return False
|
|||
|
|
finally:
|
|||
|
|
conn.close()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
parser = argparse.ArgumentParser(description="Soul 内容上传接口")
|
|||
|
|
parser.add_argument("--json", help="JSON格式的完整数据")
|
|||
|
|
parser.add_argument("--title", help="标题")
|
|||
|
|
parser.add_argument("--price", type=float, default=1.0, help="定价(0=免费)")
|
|||
|
|
parser.add_argument("--content", help="内容正文")
|
|||
|
|
parser.add_argument("--content-file", help="从文件读取内容")
|
|||
|
|
parser.add_argument("--format", default="markdown", choices=["markdown", "text", "html"])
|
|||
|
|
parser.add_argument("--part", default="part-1", help="所属篇 (part-1 ~ part-5)")
|
|||
|
|
parser.add_argument("--chapter", default="chapter-1", help="所属章 (chapter-1 ~ chapter-11)")
|
|||
|
|
parser.add_argument("--id", help="指定 section ID (如 1.6)，不指定则自动生成")
|
|||
|
|
parser.add_argument("--images", nargs="*", help="图片URL列表")
|
|||
|
|
parser.add_argument("--list-structure", action="store_true", help="查看篇章结构")
|
|||
|
|
parser.add_argument("--list-chapters", action="store_true", help="列出所有章节")
|
|||
|
|
|
|||
|
|
args = parser.parse_args()
|
|||
|
|
|
|||
|
|
if args.list_structure:
|
|||
|
|
list_structure()
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
if args.list_chapters:
|
|||
|
|
conn = get_connection()
|
|||
|
|
cur = conn.cursor()
|
|||
|
|
cur.execute("SELECT id, section_title, is_free, price FROM chapters ORDER BY sort_order")
|
|||
|
|
for row in cur.fetchall():
|
|||
|
|
free_tag = "[免费]" if row[2] else f"[¥{row[3]}]"
|
|||
|
|
print(f" {row[0]} {row[1]} {free_tag}")
|
|||
|
|
conn.close()
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
if args.json:
|
|||
|
|
data = json.loads(args.json)
|
|||
|
|
else:
|
|||
|
|
if not args.title or (not args.content and not args.content_file):
|
|||
|
|
parser.print_help()
|
|||
|
|
print("\n错误: 需要 --title 和 --content (或 --content-file)")
|
|||
|
|
sys.exit(1)
|
|||
|
|
|
|||
|
|
content = args.content
|
|||
|
|
if args.content_file:
|
|||
|
|
with open(args.content_file, "r", encoding="utf-8") as f:
|
|||
|
|
content = f.read()
|
|||
|
|
|
|||
|
|
data = {
|
|||
|
|
"title": args.title,
|
|||
|
|
"price": args.price,
|
|||
|
|
"content": content,
|
|||
|
|
"format": args.format,
|
|||
|
|
"part_id": args.part,
|
|||
|
|
"chapter_id": args.chapter,
|
|||
|
|
"images": args.images or [],
|
|||
|
|
}
|
|||
|
|
if args.id:
|
|||
|
|
data["id"] = args.id
|
|||
|
|
|
|||
|
|
upload_content(data)
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
main()
|