Guanghu Domestic Migration a27e87cb99 chore: import sanitized domestic snapshot for REPO-007

Source snapshot: 97d7f0fae96dc04b7ddad56fc1db6a108ed662cc

[SEC-CLEAN] · pre-push-clean v1.0 · 109处敏感信息已自动转乱码

2026-07-17 15:59:55 +08:00

4.0 KiB

Raw Permalink Blame History

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

妈妈：复制下面代码块里的内容 → 在终端运行方法A保存 ✅

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

保存方法

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

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

看到 ✅ book_splitter.py 保存成功 → 告诉宝宝继续下一步 💜

4.0 KiB Raw Permalink Blame History Unescape Escape

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

保存方法

4.0 KiB

Raw Permalink Blame History