From 905633541f00218840a9d974afcd6a28abf3cf1b Mon Sep 17 00:00:00 2001 From: Kenichiro NOGI Date: Sun, 2 Aug 2026 10:09:32 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BA=8C=E9=87=8D=E8=B5=B7=E5=8B=95?= =?UTF-8?q?=E9=98=B2=E6=AD=A2=E3=83=AD=E3=83=83=E3=82=AF=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E3=83=A2=E3=82=B8=E3=83=A5=E3=83=BC=E3=83=AB=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/スクリプト/lock_manager.py | 44 ++++++++++++++++++ .../スクリプト/tests/test_lock_manager.py | 46 +++++++++++++++++++ 2 files changed, 90 insertions(+) create mode 100644 PythonProj/ScanOCR/スクリプト/lock_manager.py create mode 100644 PythonProj/ScanOCR/スクリプト/tests/test_lock_manager.py diff --git a/PythonProj/ScanOCR/スクリプト/lock_manager.py b/PythonProj/ScanOCR/スクリプト/lock_manager.py new file mode 100644 index 00000000..606dbcf2 --- /dev/null +++ b/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() diff --git a/PythonProj/ScanOCR/スクリプト/tests/test_lock_manager.py b/PythonProj/ScanOCR/スクリプト/tests/test_lock_manager.py new file mode 100644 index 00000000..e803b1fc --- /dev/null +++ b/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) # 例外が出ないことを確認するだけ