49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
"""简单的文件锁标记工具。"""
|
|
|
|
from os import PathLike
|
|
from pathlib import Path
|
|
import json
|
|
import os
|
|
from tempfile import NamedTemporaryFile
|
|
|
|
|
|
def is_lock(file_path: str | PathLike[str]) -> bool:
|
|
"""判断指定的锁文件是否存在。"""
|
|
return Path(file_path).is_file()
|
|
|
|
|
|
def write_lockfile(file_path: str | PathLike[str]) -> None:
|
|
"""创建锁文件；父目录不存在时自动创建。"""
|
|
path = Path(file_path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("LOCK", encoding="utf-8")
|
|
|
|
|
|
def claim_json(path: Path, record: dict) -> bool:
|
|
"""原子占位并落盘，成功后才允许提交；异常留下记录等待核对。"""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
stream = path.open('x', encoding='utf-8')
|
|
except FileExistsError:
|
|
return False
|
|
with stream:
|
|
json.dump(record, stream, ensure_ascii=False)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
return True
|
|
|
|
|
|
def replace_json(path: Path, record: dict) -> None:
|
|
"""原子替换核对结果，避免其他任务读取到半条记录。"""
|
|
temporary = None
|
|
try:
|
|
with NamedTemporaryFile(mode='w', encoding='utf-8', dir=path.parent, delete=False) as stream:
|
|
temporary = Path(stream.name)
|
|
json.dump(record, stream, ensure_ascii=False)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
if temporary is not None:
|
|
temporary.unlink(missing_ok=True)
|