| """跨平台文件锁:Unix 用 fcntl.flock,Windows 用 msvcrt.locking(骨架原为 Linux-only)。 | |
| 目标运行环境仍是 Linux/Docker(fcntl 路径),Windows 兼容仅保证本地开发/测试可用。 | |
| """ | |
| from __future__ import annotations | |
| try: # Unix | |
| import fcntl | |
| def flock_ex(fd: int) -> None: | |
| fcntl.flock(fd, fcntl.LOCK_EX) | |
| def flock_sh(fd: int) -> None: | |
| fcntl.flock(fd, fcntl.LOCK_SH) | |
| except ImportError: # Windows | |
| import msvcrt | |
| def flock_ex(fd: int) -> None: | |
| # msvcrt.locking 锁定文件当前位置起 N 字节;先回到文件头锁 1 字节 | |
| try: | |
| import os | |
| os.lseek(fd, 0, os.SEEK_SET) | |
| msvcrt.locking(fd, msvcrt.LK_LOCK, 1) | |
| except OSError: | |
| pass # 空文件/单进程场景可忽略;生产在 Linux 上 | |
| def flock_sh(fd: int) -> None: | |
| pass # msvcrt 无共享锁;读场景不加锁 | |