95 lines
2.7 KiB
Python
95 lines
2.7 KiB
Python
|
"""Single instance lock management for Local Transcription application."""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
|
||
|
|
class InstanceLock:
|
||
|
|
"""Manages single instance lock using a PID file."""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
"""Initialize the instance lock."""
|
||
|
|
self.lock_dir = Path.home() / '.local-transcription'
|
||
|
|
self.lock_file = self.lock_dir / 'app.lock'
|
||
|
|
|
||
|
|
def acquire(self) -> bool:
|
||
|
|
"""
|
||
|
|
Try to acquire the instance lock.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if lock acquired (no other instance running),
|
||
|
|
False if another instance is already running.
|
||
|
|
"""
|
||
|
|
# Ensure lock directory exists
|
||
|
|
self.lock_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
if self.lock_file.exists():
|
||
|
|
try:
|
||
|
|
pid_str = self.lock_file.read_text().strip()
|
||
|
|
if pid_str:
|
||
|
|
pid = int(pid_str)
|
||
|
|
if self._is_process_running(pid):
|
||
|
|
return False
|
||
|
|
except (ValueError, OSError):
|
||
|
|
# Invalid PID file, we can overwrite it
|
||
|
|
pass
|
||
|
|
|
||
|
|
# Write our PID to the lock file
|
||
|
|
try:
|
||
|
|
self.lock_file.write_text(str(os.getpid()))
|
||
|
|
return True
|
||
|
|
except OSError:
|
||
|
|
return False
|
||
|
|
|
||
|
|
def release(self):
|
||
|
|
"""Release the instance lock."""
|
||
|
|
try:
|
||
|
|
if self.lock_file.exists():
|
||
|
|
# Only remove if it contains our PID
|
||
|
|
pid_str = self.lock_file.read_text().strip()
|
||
|
|
if pid_str and int(pid_str) == os.getpid():
|
||
|
|
self.lock_file.unlink()
|
||
|
|
except (ValueError, OSError):
|
||
|
|
pass
|
||
|
|
|
||
|
|
def _is_process_running(self, pid: int) -> bool:
|
||
|
|
"""
|
||
|
|
Check if a process with the given PID is running.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
pid: Process ID to check
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if process is running, False otherwise
|
||
|
|
"""
|
||
|
|
if sys.platform == 'win32':
|
||
|
|
# Windows
|
||
|
|
try:
|
||
|
|
import ctypes
|
||
|
|
kernel32 = ctypes.windll.kernel32
|
||
|
|
SYNCHRONIZE = 0x00100000
|
||
|
|
process = kernel32.OpenProcess(SYNCHRONIZE, False, pid)
|
||
|
|
if process:
|
||
|
|
kernel32.CloseHandle(process)
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
except Exception:
|
||
|
|
return False
|
||
|
|
else:
|
||
|
|
# Unix/Linux/macOS
|
||
|
|
try:
|
||
|
|
os.kill(pid, 0)
|
||
|
|
return True
|
||
|
|
except OSError:
|
||
|
|
return False
|
||
|
|
|
||
|
|
def __enter__(self):
|
||
|
|
"""Context manager entry."""
|
||
|
|
return self.acquire()
|
||
|
|
|
||
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
||
|
|
"""Context manager exit."""
|
||
|
|
self.release()
|
||
|
|
return False
|