mirror of
https://github.com/alexgetmancom/miband-bot.git
synced 2026-07-16 12:23:15 +03:00
Alex
69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
# Copyright (C) 2026 Alexey
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
|
|
# Попытка импорта fcntl (для Unix-систем) или msvcrt (для Windows)
|
|
try:
|
|
import fcntl
|
|
_HAS_FCNTL = True
|
|
except ImportError:
|
|
_HAS_FCNTL = False
|
|
try:
|
|
import msvcrt
|
|
_HAS_MSVCRT = True
|
|
except ImportError:
|
|
_HAS_MSVCRT = False
|
|
|
|
|
|
class LockUnavailable(RuntimeError):
|
|
"""Raised when another process already owns the sync lock."""
|
|
|
|
|
|
@contextmanager
|
|
def exclusive_file_lock(path: Path) -> Iterator[None]:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("a+", encoding="utf-8") as lock_file:
|
|
fd = lock_file.fileno()
|
|
|
|
if _HAS_FCNTL:
|
|
try:
|
|
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError as exc:
|
|
raise LockUnavailable(f"Lock is already held: {path}") from exc
|
|
elif _HAS_MSVCRT:
|
|
try:
|
|
lock_file.seek(0)
|
|
# Блокируем первые 64 байта файла без ожидания (non-blocking)
|
|
msvcrt.locking(fd, msvcrt.LK_NBLCK, 64)
|
|
except OSError as exc:
|
|
raise LockUnavailable(f"Lock is already held: {path}") from exc
|
|
else:
|
|
# Резервный вариант, если блокировки недоступны на платформе
|
|
pass
|
|
|
|
try:
|
|
lock_file.seek(0)
|
|
lock_file.truncate()
|
|
lock_file.write(f"pid={os.getpid()} time={int(time.time())}\n")
|
|
lock_file.flush()
|
|
yield
|
|
finally:
|
|
if _HAS_FCNTL:
|
|
try:
|
|
fcntl.flock(fd, fcntl.LOCK_UN)
|
|
except Exception:
|
|
pass
|
|
elif _HAS_MSVCRT:
|
|
try:
|
|
lock_file.seek(0)
|
|
msvcrt.locking(fd, msvcrt.LK_UNLCK, 64)
|
|
except Exception:
|
|
pass
|