From 41153a778295fb7a05e29995bb4cea6bd3fcb17b Mon Sep 17 00:00:00 2001 From: Kenichiro NOGI Date: Sun, 2 Aug 2026 10:08:48 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=E3=83=95=E3=82=A1=E3=82=A4=E3=83=AB?= =?UTF-8?q?=E5=90=8D=E8=A1=9D=E7=AA=81=E5=9B=9E=E9=81=BF=E3=82=B3=E3=83=94?= =?UTF-8?q?=E3=83=BC=E3=83=BB=E7=A7=BB=E5=8B=95=E3=81=A8=E3=82=B5=E3=82=A4?= =?UTF-8?q?=E3=82=BA=E5=AE=89=E5=AE=9A=E5=BE=85=E3=81=A1=E3=82=92=E8=BF=BD?= =?UTF-8?q?=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 5 --- PythonProj/ScanOCR/スクリプト/file_ops.py | 57 ++++++++++++ .../ScanOCR/スクリプト/tests/test_file_ops.py | 86 +++++++++++++++++++ 2 files changed, 143 insertions(+) create mode 100644 PythonProj/ScanOCR/スクリプト/file_ops.py create mode 100644 PythonProj/ScanOCR/スクリプト/tests/test_file_ops.py diff --git a/PythonProj/ScanOCR/スクリプト/file_ops.py b/PythonProj/ScanOCR/スクリプト/file_ops.py new file mode 100644 index 00000000..5bf953c3 --- /dev/null +++ b/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 diff --git a/PythonProj/ScanOCR/スクリプト/tests/test_file_ops.py b/PythonProj/ScanOCR/スクリプト/tests/test_file_ops.py new file mode 100644 index 00000000..dd479432 --- /dev/null +++ b/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