book-to-skill extraction pipeline 抽取流水线
把一本技术书或文档文件夹转成 Claude Code skill 的工具。值得拆的不是 README 里的承诺,而是它怎样处理多格式抽取、fallback、Unicode 清洗、输出目录权限和生成 skill 的扫描门禁。
抽取流水线的入口契约
抽取链从 sources/book-to-skill/book_to_skill/utils.py:250 这一处之后,sources/book-to-skill/book_to_skill/utils.py:252 这一处之后,sources/book-to-skill/book_to_skill/utils.py:256 这一处之后,sources/book-to-skill/book_to_skill/utils.py:262 这一处之后,sources/book-to-skill/book_to_skill/utils.py:265 开始:先确认路径、扩展名和 magic bytes,再把 PDF、EPUB、DOCX 这类可能伪装扩展的文件纠正到真实格式。随后 sources/book-to-skill/book_to_skill/utils.py:281 这一处之后,sources/book-to-skill/book_to_skill/utils.py:282 这一处之后,sources/book-to-skill/book_to_skill/utils.py:288 这一处之后,sources/book-to-skill/book_to_skill/utils.py:305 这一处之后,sources/book-to-skill/book_to_skill/utils.py:344 把依赖准备和格式分支拆开,避免一个坏格式拖垮所有输入。
250def extract_single_file(input_path: Path, extraction_mode: str, install_mode: str) -> dict:\n251 input_str = str(input_path)\n252 if not input_path.exists():\n253 raise ExtractionError(f'File not found: {input_str}')\n254 ext = input_path.suffix.lower()\n255 document_format = ext.lstrip('.')\n256 if ext not in SUPPORTED_EXTENSIONS:\n257 try:\n258 with open(input_str, 'rb') as f:\n259 header = f.read(8)\n260 except OSError as exc:\n261 raise ExtractionError(f'Could not read {input_path.name}: {exc.strerror or exc}') from exc\n262 if header[:4] == b'%PDF':\n263 ext = '.pdf'\n264 document_format = 'pdf'\n265 elif header[:2] == b'PK':\n266 try:\n267 with zipfile.ZipFile(input_str) as zf:\n268 names = set(zf.namelist())\n269 if 'mimetype' in names and zf.read('mimetype').startswith(b'application/epub'):\n270 ext = '.epub'\n271 document_format = 'epub'\n272 elif 'word/document.xml' in names:\n273 ext = '.docx'\n274 document_format = 'docx'\n275 else:\n276 raise ExtractionError(f"Unsupported ZIP-based format '{input_path.name}'. Supported: {supported_formats_message()}")\n277 except (zipfile.BadZipFile, KeyError, OSError):\n278 raise ExtractionError(f"Unsupported ZIP-based format '{input_path.name}'. Supported: {supported_formats_message()}")\n279 else:\n280 raise ExtractionError(f"Unsupported format '{ext or '<none>'}'. Supported: {supported_formats_message()}")\n281 prepare_dependencies(ext, extraction_mode, install_mode)\n282 if ext in CALIBRE_EBOOK_EXTENSIONS and (not shutil.which('ebook-convert')):\n283 raise ExtractionError("MOBI/AZW/AZW3 extraction requires Calibre's ebook-convert command. Install Calibre and ensure ebook-convert is on PATH, then rerun this command.")\n284 text = ''\n285 method = ''\n286 pages = 0
PDF、EPUB、DOCX 的降级路径
▸PDF 先判断 image-only,再按 technical/text 降级
PDF 是最复杂分支:sources/book-to-skill/book_to_skill/utils.py:305 这一处之后,sources/book-to-skill/book_to_skill/utils.py:307 这一处之后,sources/book-to-skill/book_to_skill/utils.py:309 这一处之后,sources/book-to-skill/book_to_skill/utils.py:311 这一处之后,sources/book-to-skill/book_to_skill/utils.py:316 先尝试 Docling technical extraction,失败后转回 text mode;sources/book-to-skill/book_to_skill/utils.py:319 这一处之后,sources/book-to-skill/book_to_skill/utils.py:321 这一处之后,sources/book-to-skill/book_to_skill/utils.py:327 这一处之后,sources/book-to-skill/book_to_skill/utils.py:335 这一处之后,sources/book-to-skill/book_to_skill/utils.py:341 依次尝试 pdftotext、pypdf、pdfminer,全部失败才抛出安装建议。
305 elif ext == '.pdf':\n306 print(f'Extracting PDF: {input_str}')\n307 if looks_image_only(input_str):\n308 raise ExtractionError(f'{input_path.name} looks like a scanned (image-only) PDF: its first pages contain no extractable text, only images.\nRun OCR on it first, then retry:\n ocrmypdf input.pdf output.pdf')\n309 if extraction_mode == 'technical':\n310 print('Mode: technical — using Docling (layout-aware)...', end=' ', flush=True)\n311 text = extract_with_docling(input_str)\n312 if text and text.strip():\n313 method = 'docling'\n314 print('OK')\n315 else:\n316 print('not available, falling back to pdftotext')\n317 extraction_mode = 'text'\n318 if extraction_mode == 'text' or not text:\n319 print('Mode: text — using pdftotext...')\n320 print('Trying pdftotext...', end=' ', flush=True)\n321 text = extract_with_pdftotext(input_str)\n322 if text and text.strip():\n323 method = 'pdftotext'\n324 print('OK')\n325 else:\n326 print('not available')\n327 print('Trying pypdf...', end=' ', flush=True)\n328 text = extract_with_pypdf(input_str)\n329 if text and text.strip():\n330 method = 'pypdf'\n331 print('OK')\n332 else:\n333 print('not available')\n334 print('Trying pdfminer.six...', end=' ', flush=True)\n335 text = extract_with_pdfminer(input_str)\n336 if text and text.strip():\n337 method = 'pdfminer'\n338 print('OK')\n339 else:\n340 print('FAILED')\n341 raise ExtractionError('Could not extract text from PDF.\nInstall one of: poppler-utils (pdftotext), pypdf, or pdfminer.six\n sudo apt install poppler-utils\n pip3 install pypdf\n pip3 install pdfminer.six')\n342 pages = count_pages(input_str)
▸EPUB fallback 尊重 OPF spine,而不是纯文件名排序
EPUB stdlib 分支通过 sources/book-to-skill/book_to_skill/parsers/epub.py:40 这一处之后,sources/book-to-skill/book_to_skill/parsers/epub.py:45 这一处之后,sources/book-to-skill/book_to_skill/parsers/epub.py:47 这一处之后,sources/book-to-skill/book_to_skill/parsers/epub.py:54 这一处之后,sources/book-to-skill/book_to_skill/parsers/epub.py:57 建立 spine order,再用 sources/book-to-skill/book_to_skill/parsers/epub.py:63 这一处之后,sources/book-to-skill/book_to_skill/parsers/epub.py:67 这一处之后,sources/book-to-skill/book_to_skill/parsers/epub.py:69 这一处之后,sources/book-to-skill/book_to_skill/parsers/epub.py:70 这一处之后,sources/book-to-skill/book_to_skill/parsers/epub.py:72 这一处之后,sources/book-to-skill/book_to_skill/parsers/epub.py:75 读取 HTML 文档。这个顺序保证多章节书不会因为 zip 内部路径顺序而重排。
36def extract_with_zipfile(epub_path: str) -> str | None:\n37 try:\n38 with zipfile.ZipFile(epub_path) as zf:\n39 names = zf.namelist()\n40 opf_path = _find_opf_path(zf)\n41 opf_dir = posixpath.dirname(opf_path) if opf_path else ''\n42 spine_order: list[str] = []\n43 seen: set[str] = set()\n44 if opf_path:\n45 opf_text = zf.read(opf_path).decode('utf-8', errors='replace')\n46 manifest: dict[str, str] = {}\n47 for item_tag in re.findall('<item\\b[^>]*?/?>', opf_text):\n48 id_m = re.search('\\bid=["\\\']([^"\\\']+)["\\\']', item_tag)\n49 href_m = re.search('\\bhref=["\\\']([^"\\\']+)["\\\']', item_tag)\n50 if id_m and href_m:\n51 href = href_m.group(1)\n52 resolved = posixpath.normpath(posixpath.join(opf_dir, href)) if opf_dir else href\n53 manifest[id_m.group(1)] = resolved\n54 for idref in re.findall('<itemref\\b[^>]*?\\bidref=["\\\']([^"\\\']+)["\\\']', opf_text):\n55 href = manifest.get(idref)\n56 if href and href not in seen:\n57 spine_order.append(href)\n58 seen.add(href)\n59 for href in manifest.values():\n60 if href.endswith(('.html', '.xhtml')) and href not in seen:\n61 spine_order.append(href)\n62 seen.add(href)\n63 html_files = spine_order or sorted((n for n in names if n.endswith(('.html', '.xhtml'))))\n64 if not html_files:\n65 return None\n66 parts = []\n67 for name in html_files:\n68 try:\n69 raw = zf.read(name).decode('utf-8', errors='replace')\n70 parser = _HTMLTextExtractor()\n71 parser.feed(raw)\n72 parts.append(parser.get_text())\n73 except Exception:\n74 continue\n75 return '\n\n'.join(parts) if parts else None\n76 except Exception as e:\n77 print(f' [warn] extract_with_zipfile failed: {type(e).__name__}: {e}', file=sys.stderr)\n78 return None
所有格式最后都回到统一 metadata
▸抽取结果统一成 source dict
无论前面走哪条后端,最终都经过 sources/book-to-skill/book_to_skill/utils.py:378 这一处之后,sources/book-to-skill/book_to_skill/utils.py:383 这一处之后,sources/book-to-skill/book_to_skill/utils.py:384 这一处之后,sources/book-to-skill/book_to_skill/utils.py:385 这一处之后,sources/book-to-skill/book_to_skill/utils.py:386 输出同一个 dict shape。主循环再用 sources/book-to-skill/book_to_skill/utils.py:443 这一处之后,sources/book-to-skill/book_to_skill/utils.py:444 这一处之后,sources/book-to-skill/book_to_skill/utils.py:451 这一处之后,sources/book-to-skill/book_to_skill/utils.py:452 这一处之后,sources/book-to-skill/book_to_skill/utils.py:461 这一处之后,sources/book-to-skill/book_to_skill/utils.py:462 合并为 full_text.txt 和 metadata.json,这就是后续 skill 生成能保持简单的原因。
378 text, removed_invisible = sanitize_extracted_text(text)\n379 if removed_invisible:\n380 print(f' [security] removed {removed_invisible} invisible Unicode code point(s) from {input_path.name}', file=sys.stderr)\n381 if not text.strip():\n382 raise ExtractionError(f'Extracted text from {input_path.name} contained no visible content after Unicode sanitization.')\n383 tokens = estimate_tokens(text)\n384 structure = detect_structure(text)\n385 file_size_mb = os.path.getsize(input_str) / (1024 * 1024)\n386 return {'source_file': str(input_path.resolve()), 'filename': input_path.name, 'format': document_format, 'extraction_method': method, 'file_size_mb': round(file_size_mb, 2), pages_label: pages, 'pages_label': pages_label, 'pages': pages, 'chars': len(text), 'words': len(text.split()), 'estimated_tokens': tokens, 'text': text, **structure}
边界与失败模式
- PDF image-only 会在
sources/book-to-skill/book_to_skill/utils.py:307被拒绝,避免把空文本传给后续生成。 - MOBI/AZW 依赖 Calibre,
sources/book-to-skill/book_to_skill/utils.py:282到sources/book-to-skill/book_to_skill/utils.py:283是硬失败,不是 fallback。 - 所有来源失败时,
sources/book-to-skill/book_to_skill/utils.py:446到sources/book-to-skill/book_to_skill/utils.py:450会退出 1,避免生成空 skill。
关联模块
sources/book-to-skill/book_to_skill/dependencies.py:82 按扩展名选择提示策略。sources/book-to-skill/book_to_skill/utils.py:170 把标题和目录信号写入 metadata。