45 lines
1.2 KiB
Python
45 lines
1.2 KiB
Python
# -*- 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()
|