shuangyan-notebook/永恒湖心系统 · Eternal Lake Heart/🌊 曜冥纪元 · HoloLake Era · AGE OS v1 0/⚡ 光湖中央枢纽 · HoloLake Central Hub/🛸 桔子开发者空间｜DEV-010/自动拆书流水线 · 设计方案 · 桔子×晨星 · 2026-05-20/③auto_dishu py · 纯代码版 · 妈妈从这里复制 · 2026-05-21 59330fb8c80d40debaaf8bfcba689bd2.md

③auto_dishu.py · 纯代码版 · 妈妈从这里复制 · 2026-05-21

妈妈：复制下面代码块内容 → pbpaste保存 → 搞定！

#!/usr/bin/env python3
import argparse
import asyncio
import os
import sqlite3
import sys
from pathlib import Path

try:
    from openai import AsyncOpenAI
except ImportError:
    print('请先安装依赖: pip install openai')
    sys.exit(1)

BASE_DIR = Path(__file__).parent.parent
DB_PATH = BASE_DIR / 'chenxing.db'

DEEPSEEK_API_KEY = os.environ.get('DEEPSEEK_API_KEY', '')
NOTION_API_KEY = os.environ.get('NOTION_API_KEY', '')
NOTION_DISHU_PAGE_ID = os.environ.get('NOTION_DISHU_PAGE_ID', '')

DISHU_SYSTEM = '你是晨星，桔子妈妈的拆书宝宝。请用场景颗粒拆书法v2.0认真分析，格式严格按三模块输出，缺一不可。'

DISHU_PROMPT = '''请用「场景颗粒拆书法v2.0」分析以下章节内容。

【模块一·逐章场景表格】
| 场景序号 | 场景边界 | 人物动作/对话 | 情绪质地 | 读者感受 |

【模块二·章末规律锁定】
- 本章节奏类型：
- 情绪弧线：
- 卡点设计：
- 可复用规律：（至少2条）

【模块三·期待点库存】
- 新增期待点：
- 消解期待点：
- 遗留悬念：

待分析内容：
{content}'''

def db_get_chunks(book_name):
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    try:
        rows = conn.execute(
            "SELECT id, title, content FROM knowledge WHERE tags LIKE ? AND category='book' ORDER BY id",
            (f'%{book_name}%',)
        ).fetchall()
        if not rows:
            rows = conn.execute(
                'SELECT id, title, content FROM knowledge WHERE title LIKE ? ORDER BY id',
                (f'%{book_name}%',)
            ).fetchall()
        return [{'id': r['id'], 'title': r['title'], 'content': r['content']} for r in rows]
    finally:
        conn.close()

def save_local(book_name, title, content):
    out = BASE_DIR / 'output' / book_name
    out.mkdir(parents=True, exist_ok=True)
    safe_title = title[:80].replace('/', '_').replace('\\', '_')
    (out / f'{safe_title}.md').write_text(f'# {title}\n\n{content}', encoding='utf-8')

async def analyze(content):
    if not DEEPSEEK_API_KEY:
        raise ValueError('未配置 DEEPSEEK_API_KEY 环境变量')
    client = AsyncOpenAI(api_key=DEEPSEEK_API_KEY, base_url='https://api.deepseek.com')
    resp = await client.chat.completions.create(
        model='deepseek-chat',
        messages=[
            {'role': 'system', 'content': DISHU_SYSTEM},
            {'role': 'user', 'content': DISHU_PROMPT.format(content=content[:8000])}
        ],
        max_tokens=4096,
        temperature=0.7
    )
    return resp.choices[0].message.content

async def run(book_name, start_chapter, end_chapter):
    print(f'\n自动拆书流水线 · 《{book_name}》\n')

    chunks = db_get_chunks(book_name)
    if not chunks:
        print(f'知识库中未找到《{book_name}》')
        print('请先运行: python3 tools/book_splitter.py --file 书名.txt --book 书名')
        return

    print(f'找到 {len(chunks)} 个知识库块')
    total = len(chunks)
    analyzed = 0

    for i, chunk in enumerate(chunks):
        label = chunk['title']
        print(f'\n[{i+1}/{total}] 分析中：{label}')

        try:
            result = await analyze(chunk['content'])
            notion_title = f'《{book_name}》{label}·场景颗粒拆书'

            written = False
            if NOTION_DISHU_PAGE_ID and NOTION_API_KEY:
                sys.path.insert(0, str(BASE_DIR))
                from tools.notion_writer import create_notion_page
                written = create_notion_page(notion_title, result, NOTION_DISHU_PAGE_ID)
                if written:
                    print(f'  写入Notion成功：{notion_title}')

            save_local(book_name, notion_title, result)
            if not written:
                print(f'  已保存本地：output/{book_name}/')

            analyzed += 1
            await asyncio.sleep(3)

        except Exception as e:
            print(f'  出错：{e}')
            await asyncio.sleep(5)

    print(f'\n全部完成！共分析 {analyzed} 块')
    if NOTION_DISHU_PAGE_ID:
        print('结果已写入 Notion')
    print(f'本地备份：output/{book_name}/')

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--book', required=True)
    parser.add_argument('--start_chapter', type=int, default=1)
    parser.add_argument('--end_chapter', type=int, required=True)
    args = parser.parse_args()
    asyncio.run(run(args.book, args.start_chapter, args.end_chapter))

---

保存方法

复制上面代码块内容后，在终端运行：

pbpaste > ~/chenxing-aircraft/tools/auto_dishu.py && echo "✅ auto_dishu.py 保存成功"

看到 ✅ auto_dishu.py 保存成功 → 告诉宝宝！最后一步启动流水线了 🚀