diff --git a/PythonProj/ScanOCR/スクリプト/extract_and_rename.py b/PythonProj/ScanOCR/スクリプト/extract_and_rename.py index 208a8d70..8328e960 100644 --- a/PythonProj/ScanOCR/スクリプト/extract_and_rename.py +++ b/PythonProj/ScanOCR/スクリプト/extract_and_rename.py @@ -69,6 +69,19 @@ def check_external_tools(tesseract_exe: Path, poppler_exe: Path) -> list[str]: return missing +import time +import traceback + +from file_ops import copy_with_unique_name, move_with_unique_name, wait_until_stable +from logger import append_log + + +def key_pressed() -> bool: + import msvcrt + + return msvcrt.kbhit() + + @dataclass class FieldBox: label: str @@ -160,6 +173,64 @@ def process_pdf(pdf_path: Path, boxes: list[FieldBox], margin: float) -> tuple[s 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, "") + + def safe_move(src: Path, dest_dir: Path, filename: str) -> Path: dest_dir.mkdir(parents=True, exist_ok=True) dest_path = dest_dir / filename diff --git a/PythonProj/ScanOCR/スクリプト/tests/test_extract_and_rename.py b/PythonProj/ScanOCR/スクリプト/tests/test_extract_and_rename.py index 000751b1..8a6aec85 100644 --- a/PythonProj/ScanOCR/スクリプト/tests/test_extract_and_rename.py +++ b/PythonProj/ScanOCR/スクリプト/tests/test_extract_and_rename.py @@ -3,6 +3,7 @@ from pathlib import Path from extract_and_rename import ( check_dependencies, check_external_tools, + run_queue, select_template_file, ) @@ -56,3 +57,162 @@ def test_check_external_tools_all_present(tmp_path): result = check_external_tools(tesseract, poppler) assert result == [] + + +def _make_dirs(tmp_path): + dirs = { + "scan": tmp_path / "スキャン", + "output": tmp_path / "アウトプット", + "success": tmp_path / "成功", + "failed": tmp_path / "失敗", + "log": tmp_path / "ログ", + } + dirs["scan"].mkdir() + return dirs + + +def test_run_queue_success_moves_and_copies(tmp_path): + dirs = _make_dirs(tmp_path) + pdf = dirs["scan"] / "scan001.pdf" + pdf.write_bytes(b"dummy") + + def fake_process_pdf(path, boxes, margin): + return ("氏名A", "20260802") + + run_queue( + dirs["scan"], dirs["output"], dirs["success"], dirs["failed"], dirs["log"], + boxes=[], margin=0.1, stable_wait_sec=0, stable_retries=1, + process_pdf_func=fake_process_pdf, + key_check=lambda: False, + sleep_func=lambda s: None, + ) + + assert (dirs["output"] / "氏名A" / "氏名A_20260802.pdf").exists() + assert (dirs["success"] / "scan001.pdf").exists() + assert not pdf.exists() + + from datetime import datetime + log_content = (dirs["log"] / f"{datetime.now().strftime('%Y-%m-%d')}.log").read_text(encoding="utf-8-sig") + assert "結果=成功" in log_content + + +def test_run_queue_missing_name_moves_to_failed(tmp_path): + dirs = _make_dirs(tmp_path) + pdf = dirs["scan"] / "scan002.pdf" + pdf.write_bytes(b"dummy") + + def fake_process_pdf(path, boxes, margin): + return (None, "20260802") + + run_queue( + dirs["scan"], dirs["output"], dirs["success"], dirs["failed"], dirs["log"], + boxes=[], margin=0.1, stable_wait_sec=0, stable_retries=1, + process_pdf_func=fake_process_pdf, + key_check=lambda: False, + sleep_func=lambda s: None, + ) + + assert (dirs["failed"] / "scan002.pdf").exists() + assert not pdf.exists() + + +def test_run_queue_exception_moves_to_failed(tmp_path): + dirs = _make_dirs(tmp_path) + pdf = dirs["scan"] / "scan003.pdf" + pdf.write_bytes(b"dummy") + + def fake_process_pdf(path, boxes, margin): + raise RuntimeError("OCRエラー") + + run_queue( + dirs["scan"], dirs["output"], dirs["success"], dirs["failed"], dirs["log"], + boxes=[], margin=0.1, stable_wait_sec=0, stable_retries=1, + process_pdf_func=fake_process_pdf, + key_check=lambda: False, + sleep_func=lambda s: None, + ) + + assert (dirs["failed"] / "scan003.pdf").exists() + + +def test_run_queue_stops_on_key_press(tmp_path): + dirs = _make_dirs(tmp_path) + pdf = dirs["scan"] / "scan004.pdf" + pdf.write_bytes(b"dummy") + + calls = [] + + def fake_process_pdf(path, boxes, margin): + calls.append(path) + return ("氏名A", "20260802") + + run_queue( + dirs["scan"], dirs["output"], dirs["success"], dirs["failed"], dirs["log"], + boxes=[], margin=0.1, stable_wait_sec=0, stable_retries=1, + process_pdf_func=fake_process_pdf, + key_check=lambda: True, # 常に押下扱い + sleep_func=lambda s: None, + ) + + assert calls == [] + assert pdf.exists() # 処理されず残る + + +def test_run_queue_skips_file_deleted_before_processing(tmp_path): + dirs = _make_dirs(tmp_path) + pdf = dirs["scan"] / "scan005.pdf" + pdf.write_bytes(b"dummy") + + calls = [] + + def fake_process_pdf(path, boxes, margin): + calls.append(path) + return ("氏名A", "20260802") + + def key_check_and_delete(): + if pdf.exists(): + pdf.unlink() + return False + + run_queue( + dirs["scan"], dirs["output"], dirs["success"], dirs["failed"], dirs["log"], + boxes=[], margin=0.1, stable_wait_sec=0, stable_retries=1, + process_pdf_func=fake_process_pdf, + key_check=key_check_and_delete, + sleep_func=lambda s: None, + ) + + assert calls == [] + + +def test_run_queue_empty_scan_dir_returns_immediately(tmp_path): + dirs = _make_dirs(tmp_path) + + run_queue( + dirs["scan"], dirs["output"], dirs["success"], dirs["failed"], dirs["log"], + boxes=[], margin=0.1, stable_wait_sec=0, stable_retries=1, + process_pdf_func=lambda p, b, m: ("氏名A", "20260802"), + key_check=lambda: False, + sleep_func=lambda s: None, + ) + # 例外が出ずに戻ることのみ確認 + + +def test_run_queue_name_collision_gets_suffix(tmp_path): + dirs = _make_dirs(tmp_path) + existing_output_dir = dirs["output"] / "氏名A" + existing_output_dir.mkdir(parents=True) + (existing_output_dir / "氏名A_20260802.pdf").write_bytes(b"old") + + pdf = dirs["scan"] / "scan006.pdf" + pdf.write_bytes(b"dummy") + + run_queue( + dirs["scan"], dirs["output"], dirs["success"], dirs["failed"], dirs["log"], + boxes=[], margin=0.1, stable_wait_sec=0, stable_retries=1, + process_pdf_func=lambda p, b, m: ("氏名A", "20260802"), + key_check=lambda: False, + sleep_func=lambda s: None, + ) + + assert (existing_output_dir / "氏名A_20260802(2).pdf").exists()