#!/usr/bin/env python3 """ 未登録PDFを検出し、先頭ページを段階的にテキスト化する。 通常は先頭2ページだけを抽出する。 抽出文字数が --min-text-length 未満の場合は、 --fallback-pages まで再抽出する。 既存PDFと正規化本文のSHA256が一致した場合は、 重複PDFとして paper_files テーブルへ代替版登録し、 AI解析用テキストを生成しない。 既存PDFの本文ハッシュは paper_files テーブルへ保存する。 初回実行時は既存PDFのハッシュ作成に時間がかかるが、 2回目以降はファイルのSHA256が変わらない限り再利用する。 例: python3 scripts/extract_text.py \ --database papers.db \ --pdf-root "/path/to/PDF_ROOT" \ --output-dir tmp/text \ --limit 3 \ --pages 2 \ --fallback-pages 5 \ --min-text-length 1500 \ --scan-dir "大窪" \ --scan-dir "JAEA三ツ井" 終了コード: 0 1件以上のテキストを生成、または重複PDFを記録 10 未登録PDFなし 1 エラー """ from __future__ import annotations import argparse import hashlib import re import shutil import sqlite3 import subprocess import sys import tempfile import unicodedata from dataclasses import dataclass from pathlib import Path from typing import Iterable from common import CONFIG MIN_DUPLICATE_TEXT_LENGTH = 500 @dataclass(frozen=True) class RegisteredPaper: paper_id: int pdf_path: str normalized_path: str @dataclass(frozen=True) class TextHashRecord: paper_id: int pdf_path: str file_sha256: str text_sha256: str text_length: int @dataclass(frozen=True) class ProcessResult: generated: bool duplicate: bool text_length: int insufficient: bool fallback_used: bool def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "papers.dbに未登録のPDFを段階的にテキスト化する" ) ) parser.add_argument( "--database", default=CONFIG.path.database, type=Path, help="papers.db のパス", ) parser.add_argument( "--pdf-root", default=CONFIG.path.pdf_root, type=Path, help="PDF_ROOT のパス", ) parser.add_argument( "--output-dir", default=CONFIG.path.text_dir, type=Path, help="抽出テキストの出力先", ) parser.add_argument( "--limit", type=int, default=CONFIG.processing.max_papers, help="AI解析用テキストを生成する最大件数", ) parser.add_argument( "--pages", type=int, default=CONFIG.processing.initial_pages, help="最初に抽出する先頭ページ数", ) parser.add_argument( "--fallback-pages", type=int, default=CONFIG.processing.fallback_pages, help=( "文字数不足時に再抽出する先頭ページ数。" "0で再抽出しない" ), ) parser.add_argument( "--min-text-length", type=int, default=CONFIG.processing.min_text_length, help=( "再抽出を判断する最小文字数。" "これ未満ならfallback-pagesまで再抽出する" ), ) parser.add_argument( "--scan-dir", action="append", default=None, help=( "PDF_ROOTからの相対探索ディレクトリ。" "複数指定可能。未指定時は設定値を使用" ), ) parser.add_argument( "--rebuild-text-hashes", action="store_true", help="既存PDFの本文ハッシュキャッシュを再作成する", ) return parser.parse_args() def validate_args(args: argparse.Namespace) -> None: if not args.database.is_file(): raise FileNotFoundError( f"データベースがありません: {args.database}" ) if not args.pdf_root.is_dir(): raise NotADirectoryError( f"PDF_ROOTがありません: {args.pdf_root}" ) if args.limit < 1: raise ValueError("--limit は1以上にしてください") if args.pages < 1: raise ValueError("--pages は1以上にしてください") if args.fallback_pages < 0: raise ValueError("--fallback-pages は0以上にしてください") if ( args.fallback_pages > 0 and args.fallback_pages < args.pages ): raise ValueError( "--fallback-pages は --pages 以上にしてください" ) if args.min_text_length < 0: raise ValueError( "--min-text-length は0以上にしてください" ) if shutil.which("pdftotext") is None: raise RuntimeError( "pdftotext が見つかりません。" "macOSでは `sudo port install poppler` " "または `brew install poppler` で導入できます。" ) def normalize_pdf_path(value: str) -> str: """PDFパスを比較用に正規化する。""" normalized = unicodedata.normalize("NFC", value) normalized = normalized.replace("\\", "/") normalized = normalized.lstrip("./") return normalized.casefold() def normalize_text(value: str) -> str: """本文を重複判定用に正規化する。""" normalized = unicodedata.normalize("NFKC", value) normalized = normalized.replace("\x00", "") normalized = re.sub(r"\s+", " ", normalized) return normalized.strip() def calculate_text_sha256(value: str) -> str: """正規化本文のSHA256を返す。""" normalized = normalize_text(value) return hashlib.sha256( normalized.encode("utf-8") ).hexdigest() def ensure_paper_files_table(conn: sqlite3.Connection) -> None: """paper_filesを作成し、旧データを自動移行する。""" conn.execute( """ CREATE TABLE IF NOT EXISTS paper_files ( id INTEGER PRIMARY KEY, paper_id INTEGER NOT NULL, pdf_path TEXT NOT NULL UNIQUE, file_sha256 TEXT, text_sha256 TEXT, text_length INTEGER, version_type TEXT NOT NULL DEFAULT 'unknown' CHECK (version_type IN ( 'publisher', 'author', 'repository', 'scan', 'translated', 'duplicate', 'unknown' )), is_primary INTEGER NOT NULL DEFAULT 0 CHECK (is_primary IN (0, 1)), notes TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (paper_id) REFERENCES papers(id) ON DELETE CASCADE ) """ ) conn.execute( """ CREATE UNIQUE INDEX IF NOT EXISTS idx_paper_files_one_primary ON paper_files(paper_id) WHERE is_primary = 1 """ ) conn.execute( """ CREATE INDEX IF NOT EXISTS idx_paper_files_text_sha256 ON paper_files(text_sha256) """ ) conn.execute( """ CREATE INDEX IF NOT EXISTS idx_paper_files_file_sha256 ON paper_files(file_sha256) """ ) # papers.pdf_pathを主PDFとして移行する。 conn.execute( """ INSERT OR IGNORE INTO paper_files ( paper_id, pdf_path, file_sha256, version_type, is_primary, notes ) SELECT id, pdf_path, file_sha256, 'unknown', 1, 'papers.pdf_pathから移行' FROM papers WHERE pdf_path IS NOT NULL AND TRIM(pdf_path) <> '' """ ) # 旧本文ハッシュキャッシュを取り込む。 if table_exists(conn, "pdf_text_hashes"): conn.execute( """ UPDATE paper_files SET file_sha256 = COALESCE( ( SELECT h.file_sha256 FROM pdf_text_hashes AS h WHERE h.pdf_path = paper_files.pdf_path ), file_sha256 ), text_sha256 = COALESCE( ( SELECT h.text_sha256 FROM pdf_text_hashes AS h WHERE h.pdf_path = paper_files.pdf_path ), text_sha256 ), text_length = COALESCE( ( SELECT h.text_length FROM pdf_text_hashes AS h WHERE h.pdf_path = paper_files.pdf_path ), text_length ) WHERE EXISTS ( SELECT 1 FROM pdf_text_hashes AS h WHERE h.pdf_path = paper_files.pdf_path ) """ ) # 旧duplicate_filesを代替PDFとして移行する。 if table_exists(conn, "duplicate_files"): conn.execute( """ INSERT OR IGNORE INTO paper_files ( paper_id, pdf_path, file_sha256, text_sha256, version_type, is_primary, notes ) SELECT paper_id, pdf_path, file_sha256, text_sha256, 'duplicate', 0, '旧duplicate_filesから移行; matched_pdf_path=' || matched_pdf_path FROM duplicate_files """ ) def table_exists( conn: sqlite3.Connection, table_name: str, ) -> bool: row = conn.execute( """ SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ? """, (table_name,), ).fetchone() return row is not None def load_registered_papers( conn: sqlite3.Connection, ) -> list[RegisteredPaper]: """paper_filesに登録されたすべてのPDFを返す。""" rows = conn.execute( """ SELECT paper_id, pdf_path FROM paper_files WHERE pdf_path IS NOT NULL AND TRIM(pdf_path) <> '' ORDER BY paper_id, is_primary DESC, id """ ).fetchall() return [ RegisteredPaper( paper_id=int(row[0]), pdf_path=str(row[1]), normalized_path=normalize_pdf_path(str(row[1])), ) for row in rows ] def load_known_paths( conn: sqlite3.Connection, ) -> set[str]: rows = conn.execute( """ SELECT pdf_path FROM paper_files WHERE pdf_path IS NOT NULL AND TRIM(pdf_path) <> '' """ ).fetchall() return { normalize_pdf_path(str(row[0])) for row in rows if row[0] is not None } def load_cached_text_hash( conn: sqlite3.Connection, pdf_path: str, ) -> TextHashRecord | None: row = conn.execute( """ SELECT paper_id, pdf_path, file_sha256, text_sha256, text_length FROM paper_files WHERE pdf_path = ? AND file_sha256 IS NOT NULL AND text_sha256 IS NOT NULL AND text_length IS NOT NULL """, (pdf_path,), ).fetchone() if row is None: return None return TextHashRecord( paper_id=int(row[0]), pdf_path=str(row[1]), file_sha256=str(row[2]), text_sha256=str(row[3]), text_length=int(row[4]), ) def save_text_hash( conn: sqlite3.Connection, record: TextHashRecord, ) -> None: conn.execute( """ UPDATE paper_files SET file_sha256 = ?, text_sha256 = ?, text_length = ?, updated_at = CURRENT_TIMESTAMP WHERE pdf_path = ? """, ( record.file_sha256, record.text_sha256, record.text_length, record.pdf_path, ), ) def save_duplicate_file( conn: sqlite3.Connection, paper_id: int, pdf_path: str, file_sha256: str, text_sha256: str, matched_pdf_path: str, ) -> None: """重複PDFを同じ論文の代替ファイルとして登録する。""" conn.execute( """ INSERT INTO paper_files ( paper_id, pdf_path, file_sha256, text_sha256, version_type, is_primary, notes, created_at, updated_at ) VALUES ( ?, ?, ?, ?, 'duplicate', 0, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP ) ON CONFLICT(pdf_path) DO UPDATE SET paper_id = excluded.paper_id, file_sha256 = excluded.file_sha256, text_sha256 = excluded.text_sha256, version_type = excluded.version_type, is_primary = 0, notes = excluded.notes, updated_at = CURRENT_TIMESTAMP """, ( paper_id, pdf_path, file_sha256, text_sha256, f"本文一致: {matched_pdf_path}", ), ) def iter_scan_roots( pdf_root: Path, scan_dirs: list[str], ) -> list[Path]: if not scan_dirs: return [pdf_root] roots: list[Path] = [] for value in scan_dirs: relative = Path(value) if relative.is_absolute(): raise ValueError( "--scan-dir はPDF_ROOTからの相対パスに" f"してください: {value}" ) root = (pdf_root / relative).resolve() try: root.relative_to(pdf_root.resolve()) except ValueError as exc: raise ValueError( f"PDF_ROOT外のディレクトリは探索できません: " f"{value}" ) from exc if not root.is_dir(): raise NotADirectoryError( f"探索ディレクトリがありません: {root}" ) roots.append(root) return roots def iter_pdf_files( scan_roots: Iterable[Path], ) -> Iterable[Path]: found: set[Path] = set() for root in scan_roots: for pdf in root.rglob("*"): if not pdf.is_file(): continue if pdf.suffix.lower() != ".pdf": continue resolved = pdf.resolve() if resolved in found: continue found.add(resolved) yield resolved def calculate_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as file: for chunk in iter( lambda: file.read(1024 * 1024), b"", ): digest.update(chunk) return digest.hexdigest() def read_text(path: Path) -> str: return path.read_text( encoding="utf-8", errors="replace", ).strip() def run_pdftotext( pdf_path: Path, output_path: Path, pages: int, ) -> None: command = [ "pdftotext", "-enc", "UTF-8", "-f", "1", "-l", str(pages), str(pdf_path), str(output_path), ] completed = subprocess.run( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) if completed.returncode != 0: error = ( completed.stderr.strip() or "不明なpdftotextエラー" ) raise RuntimeError(error) def extract_pdf_text( pdf_path: Path, output_path: Path, pages: int, fallback_pages: int, min_text_length: int, ) -> tuple[str, int, bool]: run_pdftotext( pdf_path=pdf_path, output_path=output_path, pages=pages, ) body = read_text(output_path) used_pages = pages fallback_used = False should_retry = ( fallback_pages > pages and len(body) < min_text_length ) if should_retry: run_pdftotext( pdf_path=pdf_path, output_path=output_path, pages=fallback_pages, ) body = read_text(output_path) used_pages = fallback_pages fallback_used = True return body, used_pages, fallback_used def extract_text_to_memory( pdf_path: Path, pages: int, fallback_pages: int, min_text_length: int, ) -> tuple[str, int, bool]: with tempfile.TemporaryDirectory() as tmp_dir: output_path = Path(tmp_dir) / "text.txt" return extract_pdf_text( pdf_path=pdf_path, output_path=output_path, pages=pages, fallback_pages=fallback_pages, min_text_length=min_text_length, ) def write_text_with_metadata( text_path: Path, body: str, relative_pdf_path: str, file_sha256: str, text_sha256: str, file_size: int, requested_pages: int, extracted_pages: int, fallback_used: bool, min_text_length: int, ) -> int: text_length = len(body) pdf_text_empty = "yes" if text_length == 0 else "no" text_insufficient = ( "yes" if text_length < min_text_length else "no" ) fallback_value = "yes" if fallback_used else "no" header = ( f"PDF_PATH: {relative_pdf_path}\n" f"FILE_SHA256: {file_sha256}\n" f"TEXT_SHA256: {text_sha256}\n" f"PDF_FILE_SIZE: {file_size}\n" f"REQUESTED_PAGES: {requested_pages}\n" f"EXTRACTED_PAGES: {extracted_pages}\n" f"FALLBACK_USED: {fallback_value}\n" f"TEXT_LENGTH: {text_length}\n" f"PDF_TEXT_EMPTY: {pdf_text_empty}\n" f"TEXT_INSUFFICIENT: {text_insufficient}\n" "\n" "----- BEGIN PDF TEXT -----\n" "\n" ) footer = "\n\n----- END PDF TEXT -----\n" text_path.write_text( header + body + footer, encoding="utf-8", ) return text_length def collect_candidates( pdf_root: Path, scan_roots: Iterable[Path], known_paths: set[str], ) -> list[tuple[str, Path]]: candidates: list[tuple[str, Path]] = [] for pdf_path in iter_pdf_files(scan_roots): relative_path = pdf_path.relative_to( pdf_root ).as_posix() if normalize_pdf_path(relative_path) in known_paths: continue candidates.append( (relative_path, pdf_path) ) candidates.sort( key=lambda item: normalize_pdf_path(item[0]) ) return candidates def build_registered_hash_index( conn: sqlite3.Connection, pdf_root: Path, papers: list[RegisteredPaper], args: argparse.Namespace, ) -> dict[str, TextHashRecord]: index: dict[str, TextHashRecord] = {} updated = 0 missing = 0 failed = 0 if args.rebuild_text_hashes: conn.execute( """ UPDATE paper_files SET file_sha256 = NULL, text_sha256 = NULL, text_length = NULL, updated_at = CURRENT_TIMESTAMP """ ) for paper in papers: pdf_path = pdf_root / paper.pdf_path if not pdf_path.is_file(): missing += 1 continue try: file_sha256 = calculate_sha256(pdf_path) cached = load_cached_text_hash( conn, paper.pdf_path, ) if ( cached is not None and cached.file_sha256 == file_sha256 ): record = cached else: body, _, _ = extract_text_to_memory( pdf_path=pdf_path, pages=args.pages, fallback_pages=args.fallback_pages, min_text_length=args.min_text_length, ) record = TextHashRecord( paper_id=paper.paper_id, pdf_path=paper.pdf_path, file_sha256=file_sha256, text_sha256=calculate_text_sha256(body), text_length=len(body), ) save_text_hash(conn, record) updated += 1 if record.text_length < MIN_DUPLICATE_TEXT_LENGTH: continue index.setdefault( record.text_sha256, record, ) except Exception as exc: failed += 1 print( "[警告] 既存PDFの本文ハッシュを" f"作成できません: {paper.pdf_path}: {exc}", file=sys.stderr, ) conn.commit() if updated or missing or failed: print( "既存PDF本文ハッシュ: " f"更新 {updated} 件, " f"missing {missing} 件, " f"失敗 {failed} 件" ) return index def find_duplicate( body: str, text_sha256: str, registered_hashes: dict[str, TextHashRecord], ) -> TextHashRecord | None: if len(body) < MIN_DUPLICATE_TEXT_LENGTH: return None return registered_hashes.get(text_sha256) def report_generated( index: int, relative_path: str, text_path: Path, text_length: int, used_pages: int, fallback_used: bool, min_text_length: int, ) -> None: if text_length == 0: print( "[警告] 文字を抽出できませんでした: " f"{relative_path}", file=sys.stderr, ) return if text_length < min_text_length: print( "[警告] 文字数が不足しています: " f"{relative_path} ({text_length}文字)", file=sys.stderr, ) return fallback_note = ( f", fallback={used_pages}ページ" if fallback_used else "" ) print( f"[{index:03d}] {relative_path} " f"-> {text_path.name} " f"({text_length}文字{fallback_note})" ) def process_candidate( output_index: int, relative_path: str, pdf_path: Path, output_dir: Path, args: argparse.Namespace, conn: sqlite3.Connection, registered_hashes: dict[str, TextHashRecord], ) -> ProcessResult: temp_path = output_dir / ".extracting.txt" body, used_pages, fallback_used = extract_pdf_text( pdf_path=pdf_path, output_path=temp_path, pages=args.pages, fallback_pages=args.fallback_pages, min_text_length=args.min_text_length, ) temp_path.unlink(missing_ok=True) file_sha256 = calculate_sha256(pdf_path) text_sha256 = calculate_text_sha256(body) duplicate = find_duplicate( body=body, text_sha256=text_sha256, registered_hashes=registered_hashes, ) if duplicate is not None: save_duplicate_file( conn=conn, paper_id=duplicate.paper_id, pdf_path=relative_path, file_sha256=file_sha256, text_sha256=text_sha256, matched_pdf_path=duplicate.pdf_path, ) conn.commit() print( f"[重複] {relative_path}\n" f" 既存論文ID: {duplicate.paper_id}\n" f" 登録済PDF: {duplicate.pdf_path}\n" " AI解析: 省略" ) return ProcessResult( generated=False, duplicate=True, text_length=len(body), insufficient=False, fallback_used=fallback_used, ) text_path = output_dir / f"paper-{output_index:03d}.txt" text_length = write_text_with_metadata( text_path=text_path, body=body, relative_pdf_path=relative_path, file_sha256=file_sha256, text_sha256=text_sha256, file_size=pdf_path.stat().st_size, requested_pages=args.pages, extracted_pages=used_pages, fallback_used=fallback_used, min_text_length=args.min_text_length, ) insufficient = text_length < args.min_text_length report_generated( index=output_index, relative_path=relative_path, text_path=text_path, text_length=text_length, used_pages=used_pages, fallback_used=fallback_used, min_text_length=args.min_text_length, ) return ProcessResult( generated=True, duplicate=False, text_length=text_length, insufficient=insufficient, fallback_used=fallback_used, ) def clear_previous_text_files(output_dir: Path) -> None: for text_path in output_dir.glob("paper-*.txt"): text_path.unlink() def main() -> int: args = parse_args() try: validate_args(args) pdf_root = args.pdf_root.expanduser().resolve() database = args.database.expanduser().resolve() output_dir = args.output_dir.expanduser().resolve() output_dir.mkdir( parents=True, exist_ok=True, ) clear_previous_text_files(output_dir) with sqlite3.connect(database) as conn: conn.execute("PRAGMA foreign_keys = ON") ensure_paper_files_table(conn) conn.commit() registered_papers = load_registered_papers(conn) known_paths = load_known_paths(conn) registered_hashes = build_registered_hash_index( conn=conn, pdf_root=pdf_root, papers=registered_papers, args=args, ) scan_dirs = ( args.scan_dir if args.scan_dir is not None else list( CONFIG.processing.scan_directories ) ) scan_roots = iter_scan_roots( pdf_root, scan_dirs, ) candidates = collect_candidates( pdf_root=pdf_root, scan_roots=scan_roots, known_paths=known_paths, ) if not candidates: print( "処理対象の未登録PDFはありません。" ) return 10 generated = 0 examined = 0 duplicates = 0 failed = 0 empty = 0 insufficient = 0 fallback_count = 0 for relative_path, pdf_path in candidates: if generated >= args.limit: break examined += 1 try: result = process_candidate( output_index=generated + 1, relative_path=relative_path, pdf_path=pdf_path, output_dir=output_dir, args=args, conn=conn, registered_hashes=registered_hashes, ) if result.duplicate: duplicates += 1 continue if result.generated: generated += 1 if result.text_length == 0: empty += 1 if result.insufficient: insufficient += 1 if result.fallback_used: fallback_count += 1 except Exception as exc: failed += 1 print( f"[エラー] {relative_path}: {exc}", file=sys.stderr, ) print() print( f"未登録PDF候補: {len(candidates)} 件" ) print( f"確認件数: {examined} 件" ) print( f"重複PDF: {duplicates} 件" ) print( f"AI解析対象: {generated} 件" ) print( f"テキスト生成: {generated} 件" ) print( f"追加ページ抽出: {fallback_count} 件" ) print( f"文字数不足: {insufficient} 件" ) print( f"空テキスト: {empty} 件" ) print( f"失敗: {failed} 件" ) print( f"出力先: {output_dir}" ) if generated > 0 or duplicates > 0: return 0 print( "テキストを1件も生成できませんでした。", file=sys.stderr, ) return 1 except Exception as exc: print( f"extract_text.py: {exc}", file=sys.stderr, ) return 1 if __name__ == "__main__": raise SystemExit(main())