shuangyan-notebook/永恒湖心系统 · Eternal Lake Heart/🌊 曜冥纪元 · HoloLake Era · AGE OS v1 0/⚡ 光湖中央枢纽 · HoloLake Central Hub/🛸 桔子开发者空间｜DEV-010/自动拆书流水线 · 设计方案 · 桔子×晨星 · 2026-05-20/①book_splitter py · 切书脚本 · 2026-05-21 a32e997c7f1745ed8e62132462db3a7b.md

①book_splitter.py · 切书脚本 · 2026-05-21

HLDP://juzi/tools/book_splitter/2026-05-21
├── 作用: 把 txt 文件按章节切块存入知识库
├── 操作: 整段复制粘贴到终端，回车
└── 看到「✅ book_splitter.py 创建完成」就成功了

---

复制这整段到终端，回车

cat > ~/chenxing-aircraft/tools/book_splitter.py << 'ENDOFFILE'
#!/usr/bin/env python3
"""
书籍分块工具 · book_splitter.py
把 txt 文件按章节切块存入知识库
用法: python3 tools/book_splitter.py --file 折春漪.txt --book 折春漪
"""
import argparse
import re
import sqlite3
import sys
from pathlib import Path

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

def split_by_chapters(text: str, chapters_per_chunk: int = 2) -> list:
    """按章节标题切割"""
    patterns = [
        r'第[零一二三四五六七八九十百千0-9]+章[\s\u3000]',
        r'第[零一二三四五六七八九十百千0-9]+章$',
        r'Chapter\s+\d+',
    ]
    combined = '|'.join(f'(?:{p})' for p in patterns)
    positions = [(m.start(), m.group().strip()) for m in re.finditer(combined, text, re.MULTILINE)]

    if not positions:
        print("⚠️  未找到章节标题，按5000字切割")
        chunks = []
        for i, start in enumerate(range(0, len(text), 5000)):
            chunks.append({"title": f"第{i+1}块", "content": text[start:start+5000]})
        return chunks

    chapters = []
    for i, (pos, title) in enumerate(positions):
        end = positions[i+1][0] if i+1 < len(positions) else len(text)
        chapters.append({"title": title, "content": text[pos:end]})

    chunks = []
    for i in range(0, len(chapters), chapters_per_chunk):
        batch = chapters[i:i+chapters_per_chunk]
        if len(batch) > 1:
            combined_title = f"{batch[0]['title']}-{batch[-1]['title']}"
        else:
            combined_title = batch[0]['title']
        chunks.append({
            "title": combined_title,
            "content": "\n\n".join(c["content"] for c in batch)
        })
    return chunks

def import_to_knowledge(book_name: str, chunks: list, user_id: int = 1):
    """导入到知识库"""
    conn = sqlite3.connect(str(DB_PATH))
    imported = skipped = 0
    try:
        for chunk in chunks:
            title = f"《{book_name}》{chunk['title']}"
            existing = conn.execute(
                "SELECT id FROM knowledge WHERE title=? AND user_id=?", (title, user_id)
            ).fetchone()
            if existing:
                skipped += 1
                continue
            conn.execute(
                """INSERT INTO knowledge (user_id,title,content,category,tags,source)
                   VALUES (?,?,?,'book',?,?)""",
                (user_id, title, chunk["content"], book_name, f"书籍分块·{book_name}")
            )
            imported += 1
        conn.commit()
        print(f"✅ 导入完成：新增 {imported} 块，跳过 {skipped} 块（已存在）")
        return imported
    finally:
        conn.close()

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="书籍分块导入知识库")
    parser.add_argument("--file", required=True, help="书籍 txt 文件路径")
    parser.add_argument("--book", required=True, help="书名（不含《》）")
    parser.add_argument("--chunks", type=int, default=2, help="每块几章（默认2）")
    parser.add_argument("--user_id", type=int, default=1, help="用户ID（默认1）")
    args = parser.parse_args()

    fp = Path(args.file)
    if not fp.exists():
        print(f"❌ 文件不存在：{args.file}")
        sys.exit(1)

    text = fp.read_text(encoding="utf-8", errors="ignore")
    print(f"📚 书名：《{args.book}》")
    print(f"📏 总字数：{len(text):,}")

    chunks = split_by_chapters(text, args.chunks)
    print(f"✂️  共切割为 {len(chunks)} 块（每块约 {args.chunks} 章）")

    import_to_knowledge(args.book, chunks, args.user_id)
ENDOFFILE
echo "✅ book_splitter.py 创建完成"

---

看到 ✅ book_splitter.py 创建完成 → 告诉宝宝，进行下一步 💜