feat: ファイル名衝突回避コピー・移動とサイズ安定待ちを追加

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

PythonProj/ScanOCR/スクリプト/file_ops.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+"""ファイル名衝突回避コピー・移動、ファイルサイズ安定待ちモジュール。"""
+from __future__ import annotations
+
+import shutil
+import time
+from pathlib import Path
+from typing import Callable
+
+
+def resolve_unique_path(dest_dir: Path, stem: str, suffix: str) -> Path:
+    candidate = dest_dir / f"{stem}{suffix}"
+    if not candidate.exists():
+        return candidate
+
+    counter = 2
+    while True:
+        candidate = dest_dir / f"{stem}({counter}){suffix}"
+        if not candidate.exists():
+            return candidate
+        counter += 1
+
+
+def copy_with_unique_name(src: Path, dest_dir: Path, stem: str, suffix: str) -> Path:
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    dest_path = resolve_unique_path(dest_dir, stem, suffix)
+    shutil.copy2(str(src), str(dest_path))
+    return dest_path
+
+
+def move_with_unique_name(src: Path, dest_dir: Path, stem: str, suffix: str) -> Path:
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    dest_path = resolve_unique_path(dest_dir, stem, suffix)
+    shutil.move(str(src), str(dest_path))
+    return dest_path
+
+
+def wait_until_stable(
+    path: Path,
+    interval_sec: float,
+    retries: int,
+    sleep_func: Callable[[float], None] = time.sleep,
+) -> bool:
+    if not path.exists():
+        return False
+
+    previous_size = path.stat().st_size
+    for _ in range(retries):
+        sleep_func(interval_sec)
+        if not path.exists():
+            return False
+        current_size = path.stat().st_size
+        if current_size == previous_size:
+            return True
+        previous_size = current_size
+
+    return False

PythonProj/ScanOCR/スクリプト/tests/test_file_ops.py

@@ -0,0 +1,86 @@
+from pathlib import Path
+
+from file_ops import (
+    copy_with_unique_name,
+    move_with_unique_name,
+    resolve_unique_path,
+    wait_until_stable,
+)
+
+
+def test_resolve_unique_path_no_collision(tmp_path):
+    result = resolve_unique_path(tmp_path, "氏名A_20260101", ".pdf")
+    assert result == tmp_path / "氏名A_20260101.pdf"
+
+
+def test_resolve_unique_path_with_collision(tmp_path):
+    (tmp_path / "氏名A_20260101.pdf").write_bytes(b"x")
+
+    result = resolve_unique_path(tmp_path, "氏名A_20260101", ".pdf")
+
+    assert result == tmp_path / "氏名A_20260101(2).pdf"
+
+
+def test_resolve_unique_path_with_multiple_collisions(tmp_path):
+    (tmp_path / "氏名A_20260101.pdf").write_bytes(b"x")
+    (tmp_path / "氏名A_20260101(2).pdf").write_bytes(b"x")
+
+    result = resolve_unique_path(tmp_path, "氏名A_20260101", ".pdf")
+
+    assert result == tmp_path / "氏名A_20260101(3).pdf"
+
+
+def test_copy_with_unique_name_creates_dest_dir(tmp_path):
+    src = tmp_path / "src.pdf"
+    src.write_bytes(b"content")
+    dest_dir = tmp_path / "output" / "氏名A"
+
+    result = copy_with_unique_name(src, dest_dir, "氏名A_20260101", ".pdf")
+
+    assert result == dest_dir / "氏名A_20260101.pdf"
+    assert result.read_bytes() == b"content"
+    assert src.exists()  # コピーなので元ファイルは残る
+
+
+def test_move_with_unique_name_moves_source(tmp_path):
+    src = tmp_path / "src.pdf"
+    src.write_bytes(b"content")
+    dest_dir = tmp_path / "success"
+
+    result = move_with_unique_name(src, dest_dir, "src", ".pdf")
+
+    assert result == dest_dir / "src.pdf"
+    assert result.read_bytes() == b"content"
+    assert not src.exists()  # 移動なので元ファイルは消える
+
+
+def test_wait_until_stable_missing_file_returns_false(tmp_path):
+    missing = tmp_path / "missing.pdf"
+
+    result = wait_until_stable(missing, interval_sec=0, retries=3, sleep_func=lambda s: None)
+
+    assert result is False
+
+
+def test_wait_until_stable_stable_file_returns_true(tmp_path):
+    path = tmp_path / "stable.pdf"
+    path.write_bytes(b"1234")
+
+    result = wait_until_stable(path, interval_sec=0, retries=3, sleep_func=lambda s: None)
+
+    assert result is True
+
+
+def test_wait_until_stable_growing_file_returns_false(tmp_path):
+    path = tmp_path / "growing.pdf"
+    path.write_bytes(b"1")
+
+    call_count = {"n": 0}
+
+    def fake_sleep(_seconds):
+        call_count["n"] += 1
+        path.write_bytes(b"1" * (call_count["n"] + 1))
+
+    result = wait_until_stable(path, interval_sec=0, retries=3, sleep_func=fake_sleep)
+
+    assert result is False