1c1d47e3fa
意図的な停止操作をエラー扱いすると、ログ確認時に異常発生と誤認するため専用カテゴリに分離。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
313 lines
10 KiB
Python
313 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
テンプレート(赤枠=氏名, 青枠=日付)を基準に、"スキャン"フォルダ内のPDFから
|
||
氏名・日付をOCR抽出し、"アウトプット/氏名/氏名_YYYYMMDD.pdf" にコピー、
|
||
元ファイルは"成功"（または"失敗"）フォルダへ移動する。
|
||
|
||
起動は OCR仕分け実行.bat から行う想定（config.txt の固定フォルダ構成に依存）。
|
||
|
||
必要ライブラリ: opencv-python, pdf2image, pytesseract, pillow, numpy
|
||
必要な外部ツール: tesseract-ocr, tesseract-ocr-jpn, poppler-utils(pdftoppm)
|
||
"""
|
||
|
||
import importlib
|
||
import re
|
||
import time
|
||
import traceback
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
import pytesseract
|
||
from pdf2image import convert_from_path
|
||
from PIL import Image
|
||
|
||
from box_detector import detect_template_fields, to_ratio_box
|
||
from config_loader import load_config
|
||
from file_ops import copy_with_unique_name, move_with_unique_name, wait_until_stable
|
||
from lock_manager import LockAcquisitionError, acquire_lock, release_lock
|
||
from logger import append_log
|
||
|
||
DPI = 300 # PDF→画像変換の解像度。テンプレートも同じDPIで作成/スキャンしてください
|
||
|
||
TEMPLATE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".bmp"}
|
||
|
||
|
||
def select_template_file(template_dir: Path) -> Path:
|
||
candidates = sorted(
|
||
p for p in template_dir.iterdir()
|
||
if p.is_file() and p.suffix.lower() in TEMPLATE_EXTENSIONS
|
||
)
|
||
if not candidates:
|
||
raise FileNotFoundError(
|
||
f"テンプレートフォルダに画像ファイルが見つかりません: {template_dir}"
|
||
)
|
||
return candidates[0]
|
||
|
||
|
||
REQUIRED_MODULES = ("cv2", "numpy", "PIL", "pytesseract", "pdf2image")
|
||
|
||
|
||
def check_dependencies(modules: tuple[str, ...] = REQUIRED_MODULES) -> list[str]:
|
||
missing = []
|
||
for mod in modules:
|
||
try:
|
||
importlib.import_module(mod)
|
||
except ImportError:
|
||
missing.append(mod)
|
||
return missing
|
||
|
||
|
||
def check_external_tools(tesseract_exe: Path, poppler_exe: Path) -> list[str]:
|
||
missing = []
|
||
if not tesseract_exe.exists():
|
||
missing.append(str(tesseract_exe))
|
||
if not poppler_exe.exists():
|
||
missing.append(str(poppler_exe))
|
||
return missing
|
||
|
||
|
||
def key_pressed() -> bool:
|
||
import msvcrt
|
||
|
||
return msvcrt.kbhit()
|
||
|
||
|
||
@dataclass
|
||
class FieldBox:
|
||
label: str
|
||
x_ratio: float
|
||
y_ratio: float
|
||
w_ratio: float
|
||
h_ratio: float
|
||
|
||
def to_pixel_box(self, img_w: int, img_h: int, margin: float) -> tuple[int, int, int, int]:
|
||
"""比率座標を対象画像の実ピクセルに変換し、上下左右にmargin(比率)だけ広げる"""
|
||
x = self.x_ratio * img_w
|
||
y = self.y_ratio * img_h
|
||
w = self.w_ratio * img_w
|
||
h = self.h_ratio * img_h
|
||
|
||
mx = w * margin
|
||
my = h * margin
|
||
|
||
x0 = max(0, int(x - mx))
|
||
y0 = max(0, int(y - my))
|
||
x1 = min(img_w, int(x + w + mx))
|
||
y1 = min(img_h, int(y + h + my))
|
||
return x0, y0, x1, y1
|
||
|
||
|
||
def load_field_boxes(template_path: str) -> list[FieldBox]:
|
||
fields = detect_template_fields(template_path)
|
||
image_size = fields.pop("_image_size")
|
||
|
||
boxes = []
|
||
for label, box in fields.items():
|
||
xr, yr, wr, hr = to_ratio_box(box, image_size)
|
||
boxes.append(FieldBox(label, xr, yr, wr, hr))
|
||
|
||
if not boxes:
|
||
raise ValueError("テンプレートから赤枠・青枠が検出できませんでした。枠の色・太さを確認してください。")
|
||
return boxes
|
||
|
||
|
||
def ocr_region(page_img: Image.Image, box: tuple[int, int, int, int], lang: str = "jpn") -> str:
|
||
x0, y0, x1, y1 = box
|
||
cropped = page_img.crop((x0, y0, x1, y1))
|
||
text = pytesseract.image_to_string(cropped, lang=lang)
|
||
return text.strip()
|
||
|
||
|
||
def clean_name(raw: str) -> str:
|
||
"""OCR結果から氏名らしき文字列を抽出（改行・空白・記号ノイズを除去）"""
|
||
text = re.sub(r"[\s ]+", "", raw)
|
||
text = re.sub(r"[^\w一-龠ぁ-んァ-ヶー]", "", text)
|
||
return text
|
||
|
||
|
||
def clean_date(raw: str) -> str | None:
|
||
"""OCR結果から日付を検出し YYYYMMDD 形式で返す。見つからなければNone。"""
|
||
text = raw.replace(" ", "").replace(" ", "")
|
||
|
||
patterns = [
|
||
r"(\d{4})[年/\-\.](\d{1,2})[月/\-\.](\d{1,2})", # 2026年08月02日 / 2026/08/02 等
|
||
r"(\d{2})[年/\-\.](\d{1,2})[月/\-\.](\d{1,2})", # 26/08/02 のような2桁年
|
||
]
|
||
for pat in patterns:
|
||
m = re.search(pat, text)
|
||
if m:
|
||
y, mo, d = m.groups()
|
||
if len(y) == 2:
|
||
y = "20" + y # 2桁年は20XX年と仮定。運用に応じて要調整
|
||
return f"{int(y):04d}{int(mo):02d}{int(d):02d}"
|
||
return None
|
||
|
||
|
||
def process_pdf(pdf_path: Path, boxes: list[FieldBox], margin: float) -> tuple[str | None, str | None]:
|
||
"""PDFの1ページ目を画像化し、氏名・日付を抽出して返す"""
|
||
pages = convert_from_path(str(pdf_path), dpi=DPI, first_page=1, last_page=1)
|
||
if not pages:
|
||
return None, None
|
||
page_img = pages[0]
|
||
img_w, img_h = page_img.size
|
||
|
||
name_text, date_text = None, None
|
||
for box in boxes:
|
||
pixel_box = box.to_pixel_box(img_w, img_h, margin)
|
||
raw = ocr_region(page_img, pixel_box, lang="jpn")
|
||
if box.label == "name":
|
||
name_text = clean_name(raw)
|
||
elif box.label == "date":
|
||
date_text = clean_date(raw)
|
||
|
||
return name_text, date_text
|
||
|
||
|
||
def run_queue(
|
||
scan_dir: Path,
|
||
output_dir: Path,
|
||
success_dir: Path,
|
||
failed_dir: Path,
|
||
log_dir: Path,
|
||
boxes: list[FieldBox],
|
||
margin: float,
|
||
stable_wait_sec: float,
|
||
stable_retries: int,
|
||
process_pdf_func=process_pdf,
|
||
key_check=key_pressed,
|
||
sleep_func=time.sleep,
|
||
) -> None:
|
||
while True:
|
||
queue = sorted(scan_dir.glob("*.pdf"))
|
||
if not queue:
|
||
break
|
||
|
||
pdf_path = queue[0]
|
||
|
||
if key_check():
|
||
append_log(
|
||
log_dir, "手動停止", pdf_path.name, None, None,
|
||
"ユーザー操作により停止しました",
|
||
)
|
||
break
|
||
|
||
if not pdf_path.exists():
|
||
continue
|
||
|
||
if not wait_until_stable(pdf_path, stable_wait_sec, stable_retries, sleep_func):
|
||
append_log(
|
||
log_dir, "スキップ", pdf_path.name, None, None,
|
||
"サイズ不安定のためスキップ",
|
||
)
|
||
continue
|
||
|
||
try:
|
||
name, date = process_pdf_func(pdf_path, boxes, margin)
|
||
except Exception as e:
|
||
append_log(
|
||
log_dir, "エラー", pdf_path.name, None, None,
|
||
f"{type(e).__name__}: {e}\n{traceback.format_exc()}",
|
||
)
|
||
move_with_unique_name(pdf_path, failed_dir, pdf_path.stem, pdf_path.suffix)
|
||
continue
|
||
|
||
if not name or not date:
|
||
append_log(log_dir, "失敗", pdf_path.name, name, date, "")
|
||
move_with_unique_name(pdf_path, failed_dir, pdf_path.stem, pdf_path.suffix)
|
||
continue
|
||
|
||
copy_with_unique_name(pdf_path, output_dir / name, f"{name}_{date}", ".pdf")
|
||
move_with_unique_name(pdf_path, success_dir, pdf_path.stem, pdf_path.suffix)
|
||
append_log(log_dir, "成功", pdf_path.name, name, date, "")
|
||
|
||
|
||
HOME_DIR = Path(__file__).resolve().parent.parent
|
||
RUNTIME_FOLDER_KEYS = (
|
||
"テンプレートフォルダ",
|
||
"スキャンフォルダ",
|
||
"アウトプットフォルダ",
|
||
"成功フォルダ",
|
||
"失敗フォルダ",
|
||
"ログフォルダ",
|
||
)
|
||
|
||
|
||
def main() -> None:
|
||
config_path = HOME_DIR / "config.txt"
|
||
try:
|
||
config = load_config(config_path)
|
||
except Exception as e:
|
||
print(f"設定エラー: {e}")
|
||
input("何かキーを押すと終了します...")
|
||
return
|
||
|
||
folders = {key: HOME_DIR / config[key] for key in RUNTIME_FOLDER_KEYS}
|
||
for folder in folders.values():
|
||
folder.mkdir(parents=True, exist_ok=True)
|
||
|
||
missing_modules = check_dependencies()
|
||
if missing_modules:
|
||
print(f"配布パッケージが壊れています。不足モジュール: {', '.join(missing_modules)}")
|
||
input("何かキーを押すと終了します...")
|
||
return
|
||
|
||
script_dir = HOME_DIR / config["スクリプトフォルダ"]
|
||
tesseract_exe = script_dir / "tools" / "tesseract" / "tesseract.exe"
|
||
poppler_exe = script_dir / "tools" / "poppler" / "pdftoppm.exe"
|
||
missing_tools = check_external_tools(tesseract_exe, poppler_exe)
|
||
if missing_tools:
|
||
print(f"配布パッケージが壊れています。不足ファイル: {', '.join(missing_tools)}")
|
||
input("何かキーを押すと終了します...")
|
||
return
|
||
|
||
lock_path = HOME_DIR / ".lock"
|
||
try:
|
||
acquire_lock(lock_path)
|
||
except LockAcquisitionError as e:
|
||
print(str(e))
|
||
input("何かキーを押すと終了します...")
|
||
return
|
||
|
||
try:
|
||
try:
|
||
template_path = select_template_file(folders["テンプレートフォルダ"])
|
||
except FileNotFoundError as e:
|
||
print(str(e))
|
||
return
|
||
|
||
print(f"テンプレート採用: {template_path.name}")
|
||
boxes = load_field_boxes(str(template_path))
|
||
print(f"検出フィールド: {[b.label for b in boxes]}")
|
||
|
||
margin = float(config["margin"])
|
||
stable_wait_sec = float(config["ファイル安定待ち秒"])
|
||
stable_retries = int(config["ファイル安定待ちリトライ回数"])
|
||
|
||
try:
|
||
run_queue(
|
||
folders["スキャンフォルダ"],
|
||
folders["アウトプットフォルダ"],
|
||
folders["成功フォルダ"],
|
||
folders["失敗フォルダ"],
|
||
folders["ログフォルダ"],
|
||
boxes,
|
||
margin,
|
||
stable_wait_sec,
|
||
stable_retries,
|
||
)
|
||
except Exception as e:
|
||
append_log(
|
||
folders["ログフォルダ"], "エラー", "-", None, None,
|
||
f"予期しないエラーで処理を中断しました: {type(e).__name__}: {e}",
|
||
)
|
||
print(f"予期しないエラーが発生しました: {e}")
|
||
finally:
|
||
release_lock(lock_path)
|
||
|
||
print("処理完了。")
|
||
input("何かキーを押すと終了します...")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|