feat: 二重起動防止ロック管理モジュールを追加

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

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

@@ -0,0 +1,44 @@
+# -*- coding: utf-8 -*-
+"""二重起動防止ロックファイル管理モジュール（Windows専用）。"""
+from __future__ import annotations
+
+import ctypes
+import os
+from pathlib import Path
+from typing import Callable
+
+
+class LockAcquisitionError(Exception):
+    pass
+
+
+def _is_pid_running_windows(pid: int) -> bool:
+    PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
+    handle = ctypes.windll.kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
+    if handle:
+        ctypes.windll.kernel32.CloseHandle(handle)
+        return True
+    return False
+
+
+def acquire_lock(
+    lock_path: Path,
+    pid_checker: Callable[[int], bool] = _is_pid_running_windows,
+) -> None:
+    if lock_path.exists():
+        try:
+            recorded_pid = int(lock_path.read_text(encoding="utf-8").strip())
+        except ValueError:
+            recorded_pid = None
+
+        if recorded_pid is not None and pid_checker(recorded_pid):
+            raise LockAcquisitionError(f"既に実行中です（PID={recorded_pid}）")
+
+        lock_path.unlink()
+
+    lock_path.write_text(str(os.getpid()), encoding="utf-8")
+
+
+def release_lock(lock_path: Path) -> None:
+    if lock_path.exists():
+        lock_path.unlink()

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

@@ -0,0 +1,46 @@
+import os
+
+import pytest
+
+from lock_manager import LockAcquisitionError, acquire_lock, release_lock
+
+
+def test_acquire_lock_creates_file_with_pid(tmp_path):
+    lock_path = tmp_path / ".lock"
+
+    acquire_lock(lock_path, pid_checker=lambda pid: True)
+
+    assert lock_path.exists()
+    assert lock_path.read_text(encoding="utf-8").strip() == str(os.getpid())
+
+
+def test_acquire_lock_raises_when_existing_pid_running(tmp_path):
+    lock_path = tmp_path / ".lock"
+    lock_path.write_text("12345", encoding="utf-8")
+
+    with pytest.raises(LockAcquisitionError):
+        acquire_lock(lock_path, pid_checker=lambda pid: True)
+
+
+def test_acquire_lock_replaces_stale_lock(tmp_path):
+    lock_path = tmp_path / ".lock"
+    lock_path.write_text("12345", encoding="utf-8")
+
+    acquire_lock(lock_path, pid_checker=lambda pid: False)
+
+    assert lock_path.read_text(encoding="utf-8").strip() == str(os.getpid())
+
+
+def test_release_lock_removes_file(tmp_path):
+    lock_path = tmp_path / ".lock"
+    lock_path.write_text("12345", encoding="utf-8")
+
+    release_lock(lock_path)
+
+    assert not lock_path.exists()
+
+
+def test_release_lock_missing_file_no_error(tmp_path):
+    lock_path = tmp_path / ".lock"
+
+    release_lock(lock_path)  # 例外が出ないことを確認するだけ