1138 lines
29 KiB
Python
Executable File
1138 lines
29 KiB
Python
Executable File
#!/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())
|