OCR処理したので文章が間違っていたりします。
主に誤字脱字と改行場所の修正が必要です。
解決策＞languagetoolを使って修正する。 https://languagetool.org/
解決策＞languagetoolを使って修正する。 https://languagetool.org/
細かいところは直らないけど、ざっくり直る。
コードは下に。
JAVAが必要になりますのでインストールするか、zip解凍してパスを通す。
（Java17JREだとそれなりの容量です）
（Java17JREだとそれなりの容量です）
"""
OCR後の日本語テキストをローカルで整形・校正するパイプライン
- 前処理: 文字正規化 (neologdn) / 改行の再構成（OCR由来の不自然な改行を抑制）
- 校正: LanguageTool (ja-JP) を使った「安全寄り」自動修正
- 付帯: 修正候補ログ(JSONL) / 差分HTML生成
"""
from __future__ import annotations
import argparse
import html
import json
import os
import re
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import List, Tuple, Dict, Iterable
import neologdn
import regex # pip: regex
from tqdm import tqdm
from rapidfuzz.distance import Levenshtein
import language_tool_python
# ----------------------------
# 設定
# ----------------------------
@dataclass
class Config:
langtool_lang: str = "ja-JP"
# 改行整形の方針（安全寄り）
# - 行末が句点等で終わらない場合は、次の行を連結しやすい（OCRの行折返し対策）
join_if_line_not_end_with: str = r"[。．.!！\?？]$"
# ページ区切り行（OCR出力でよくある例）
keep_page_markers: bool = True
page_marker_regex: str = r"^\s*(=+)\s*Page\s*\d+\s*(=+)\s*$"
# LanguageTool適用を安全寄りにするルール
max_auto_apply_per_sentence: int = 6 # 1文あたり自動適用し過ぎない
max_repl_len: int = 12 # 置換候補が長すぎるものは自動適用しない
max_edit_distance: int = 3 # 元語と候補の編集距離が大きいものは自動適用しない
min_token_len_for_apply: int = 1
# 置換対象がこれにマッチする場合はスキップ（URLやコードっぽいもの、数値列など）
skip_pattern: re.Pattern = re.compile(
r"("
r"https?://\S+|"
r"www\.\S+|"
r"[A-Za-z0-9_\-]{20,}|" # 長い英数字（ハッシュ等）
r"[\d]{8,}|" # 長い数字列
r"`[^`]+`|" # インラインコード
r")"
)
# 文分割（雑に句点で切ると壊れるので、まず段落→簡易文分割）
sentence_split_regex: re.Pattern = re.compile(r"(?<=[。！？\?！])\s*")
# ----------------------------
# 文字正規化
# ----------------------------
def normalize_text(s: str) -> str:
# neologdn: 全半角・連続記号・濁点分離などを良い感じに正規化
s = neologdn.normalize(s)
# いくつか追加の軽い正規化
s = s.replace("\u00A0", " ") # NBSP
s = re.sub(r"[ \t]+", " ", s)
return s
# ----------------------------
# 改行整形（OCRの行折返し対策）
# ----------------------------
def fix_linebreaks(text: str, cfg: Config) -> str:
"""
OCR出力でありがちな「文中で改行されている」問題を緩和する。
方針:
- 空行は段落区切りとして維持
- ページマーカー行は維持（設定で）
- それ以外の行は、行末が句点/終端記号で終わらない場合は次行と連結
"""
lines = text.splitlines()
out: List[str] = []
def is_page_marker(line: str) -> bool:
return cfg.keep_page_markers and re.match(cfg.page_marker_regex, line) is not None
buffer = ""
for line in lines:
raw = line.rstrip("\n")
stripped = raw.strip()
if is_page_marker(raw):
# flush buffer
if buffer:
out.append(buffer.strip())
buffer = ""
out.append(raw)
continue
if stripped == "":
# 空行 = 段落区切り
if buffer:
out.append(buffer.strip())
buffer = ""
out.append("") # keep blank line
continue
# 連結対象判定
if not buffer:
buffer = stripped
else:
# 前行末が終端記号で終わらない → スペースで連結
if re.search(cfg.join_if_line_not_end_with, buffer) is None:
# 日本語では半角スペースより「そのまま連結」が自然な場合も多いが、
# OCR後は単語境界が曖昧なのでスペース挟みで安全寄り
buffer = buffer + " " + stripped
else:
# 文末っぽいなら段落内でも改行を段落内改行として維持（ただしここは好み）
buffer = buffer + "\n" + stripped
if buffer:
out.append(buffer.strip())
# 空行が多すぎる場合の軽い圧縮
result = "\n".join(out)
result = re.sub(r"\n{3,}", "\n\n", result)
return result
# ----------------------------
# 文・段落の分割
# ----------------------------
def split_into_paragraphs(text: str, cfg: Config) -> List[str]:
# ページマーカーは段落として独立扱い
paras: List[str] = []
buf: List[str] = []
for line in text.splitlines():
if cfg.keep_page_markers and re.match(cfg.page_marker_regex, line):
if buf:
paras.append("\n".join(buf).strip())
buf = []
paras.append(line.strip())
continue
if line.strip() == "":
if buf:
paras.append("\n".join(buf).strip())
buf = []
paras.append("") # paragraph separator
else:
buf.append(line)
if buf:
paras.append("\n".join(buf).strip())
# 空段落連続を抑制
cleaned: List[str] = []
prev_blank = False
for p in paras:
if p == "":
if not prev_blank:
cleaned.append("")
prev_blank = True
else:
cleaned.append(p)
prev_blank = False
return cleaned
def split_paragraph_into_sentences(p: str, cfg: Config) -> List[str]:
if p == "" or (cfg.keep_page_markers and re.match(cfg.page_marker_regex, p)):
return [p]
# 改行は文分割の妨げになるのでスペースへ（段落内の改行を保持したい場合は調整）
tmp = p.replace("\n", " ")
tmp = re.sub(r"\s{2,}", " ", tmp).strip()
if not tmp:
return [""]
sents = [s.strip() for s in cfg.sentence_split_regex.split(tmp) if s.strip()]
return sents if sents else [tmp]
# ----------------------------
# LanguageTool: 安全寄り自動適用
# ----------------------------
def apply_matches_safely(text: str, matches, cfg: Config) -> Tuple[str, List[Dict]]:
"""
LanguageToolのmatchesを受け取り、安全寄りに自動適用する。
- URL/コード/長い数字列などはスキップ
- replacementが1個のみ & 短い & 編集距離が小さいものを優先
- オフセットがずれるので、後ろから適用
"""
logs: List[Dict] = []
# 1文あたりの過剰適用抑制（matchesは文単位で呼ぶので単純に制限）
applied = 0
# offsetを使うので後ろから
for m in sorted(matches, key=lambda x: x.offset, reverse=True):
if applied >= cfg.max_auto_apply_per_sentence:
break
start = m.offset
# 互換対応
length = getattr(m, "errorLength", None)
if length is None:
length = m.error_length
end = start + length
frag = text[start:end]
# スキップ条件
if not frag or len(frag) < cfg.min_token_len_for_apply:
continue
if cfg.skip_pattern.search(frag):
continue
# 候補が無いならスキップ
repls = list(m.replacements) if m.replacements else []
if len(repls) != 1:
# 候補が複数ある場合は自動適用しない（安全寄り）
logs.append({
"action": "skip_multi_candidates",
"offset": start,
"length": m.errorLength,
"from": frag,
"candidates": repls[:10],
"ruleId": getattr(m, "ruleId", None),
"message": getattr(m, "message", None),
})
continue
repl = repls[0]
if not repl or len(repl) > cfg.max_repl_len:
logs.append({
"action": "skip_long_candidate",
"offset": start,
"length": m.errorLength,
"from": frag,
"candidate": repl,
"ruleId": getattr(m, "ruleId", None),
"message": getattr(m, "message", None),
})
continue
# 編集距離が大きいと誤訂正リスクが高い
dist = Levenshtein.distance(frag, repl)
if dist > cfg.max_edit_distance:
logs.append({
"action": "skip_far_edit",
"offset": start,
"length": m.errorLength,
"from": frag,
"candidate": repl,
"edit_distance": dist,
"ruleId": getattr(m, "ruleId", None),
"message": getattr(m, "message", None),
})
continue
# 適用
new_text = text[:start] + repl + text[end:]
logs.append({
"action": "apply",
"offset": start,
"length": m.errorLength,
"from": frag,
"to": repl,
"edit_distance": dist,
"ruleId": getattr(m, "ruleId", None),
"message": getattr(m, "message", None),
})
text = new_text
applied += 1
return text, logs
def proofread_text_with_languagetool(clean_text: str, cfg: Config) -> Tuple[str, List[Dict]]:
"""
段落→文単位でLanguageToolを適用し、校正済みテキストとログを返す
"""
tool = language_tool_python.LanguageTool(cfg.langtool_lang)
paras = split_into_paragraphs(clean_text, cfg)
out_paras: List[str] = []
all_logs: List[Dict] = []
for p in tqdm(paras, desc="LanguageTool proofreading", unit="para"):
if p == "":
out_paras.append("")
continue
if cfg.keep_page_markers and re.match(cfg.page_marker_regex, p):
out_paras.append(p)
continue
sents = split_paragraph_into_sentences(p, cfg)
fixed_sents: List[str] = []
for s in sents:
if not s:
continue
# 文単位でチェック（大量テキストでも安定）
matches = tool.check(s)
new_s, logs = apply_matches_safely(s, matches, cfg)
# 文脈を壊しやすい過剰スペースを軽く整形
new_s = re.sub(r"\s{2,}", " ", new_s).strip()
# ログに文情報を付与
for lg in logs:
lg["sentence"] = s
lg["sentence_fixed"] = new_s
all_logs.extend(logs)
fixed_sents.append(new_s)
# 文間はスペースで連結（段落内の自然さ優先）
out_paras.append(" ".join(fixed_sents).strip())
result = "\n".join(out_paras)
result = re.sub(r"\n{3,}", "\n\n", result)
return result, all_logs
# ----------------------------
# 差分HTML
# ----------------------------
def make_diff_html(before: str, after: str, title: str) -> str:
"""
最低限の差分表示（行単位）をHTMLで出す
※ もっと強力な差分が欲しければ difflib.HtmlDiff でもOK
"""
import difflib
before_lines = before.splitlines()
after_lines = after.splitlines()
diff = difflib.HtmlDiff(tabsize=2, wrapcolumn=120)
html_body = diff.make_file(before_lines, after_lines, fromdesc="before", todesc="after", context=True, numlines=3)
# タイトルだけ差し替え
html_body = html_body.replace("{html.escape(title)} ")
return html_body
# ----------------------------
# 入出力
# ----------------------------
def read_text(path: Path) -> str:
# まずUTF-8、失敗ならCP932を試す（OCR出力でありがち）
data = path.read_bytes()
for enc in ("utf-8", "utf-8-sig", "cp932"):
try:
return data.decode(enc)
except UnicodeDecodeError:
continue
# 最後はreplaceで
return data.decode("utf-8", errors="replace")
def write_text(path: Path, text: str) -> None:
path.write_text(text, encoding="utf-8", newline="\n")
def write_jsonl(path: Path, items: List[Dict]) -> None:
with path.open("w", encoding="utf-8", newline="\n") as f:
for it in items:
f.write(json.dumps(it, ensure_ascii=False) + "\n")
def iter_input_txt(input_path: Path) -> List[Path]:
if input_path.is_file():
return [input_path]
return sorted([p for p in input_path.rglob("*.txt") if p.is_file()])
# ----------------------------
# main
# ----------------------------
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--input", required=True, help="入力 .txt ファイル、または .txt を含むフォルダ")
parser.add_argument("--outdir", required=True, help="出力フォルダ")
parser.add_argument("--no-proofread", action="store_true", help="前処理のみ（LanguageTool校正なし）")
parser.add_argument("--keep-page-markers", action="store_true", help="Pageマーカー行を維持する")
args = parser.parse_args()
cfg = Config(keep_page_markers=bool(args.keep_page_markers))
in_path = Path(args.input)
outdir = Path(args.outdir)
outdir.mkdir(parents=True, exist_ok=True)
inputs = iter_input_txt(in_path)
if not inputs:
print("入力 .txt が見つかりませんでした。", file=sys.stderr)
sys.exit(1)
for p in inputs:
base = p.stem
raw = read_text(p)
# 1) 正規化
norm = normalize_text(raw)
# 2) 改行整形（OCR折返し対策）
clean = fix_linebreaks(norm, cfg)
clean_path = outdir / f"{base}_clean.txt"
write_text(clean_path, clean)
if args.no_proofread:
continue
# 3) LanguageTool 校正
proof, logs = proofread_text_with_languagetool(clean, cfg)
proof_path = outdir / f"{base}_proofread.txt"
write_text(proof_path, proof)
# 4) ログ
log_path = outdir / f"{base}_suggestions.jsonl"
write_jsonl(log_path, logs)
# 5) 差分HTML
diff_html = make_diff_html(clean, proof, title=f"diff: {base}")
diff_path = outdir / f"{base}_diff.html"
write_text(diff_path, diff_html)
print(f"[OK] {p.name}")
print(f" clean : {clean_path}")
print(f" proof : {proof_path}")
print(f" log : {log_path}")
print(f" diff : {diff_path}")
print("完了しました。")
if __name__ == "__main__":
main()
