| |
| """Mirror a pane stream to stdout and a small fixed set of rotated files.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import os |
| from pathlib import Path |
| import sys |
|
|
|
|
| def _rotate(path: Path, backups: int) -> None: |
| oldest = path.with_name(f"{path.name}.{backups}") |
| try: |
| oldest.unlink() |
| except FileNotFoundError: |
| pass |
| for index in range(backups - 1, 0, -1): |
| source = path.with_name(f"{path.name}.{index}") |
| destination = path.with_name(f"{path.name}.{index + 1}") |
| try: |
| source.replace(destination) |
| except FileNotFoundError: |
| pass |
| try: |
| path.replace(path.with_name(f"{path.name}.1")) |
| except FileNotFoundError: |
| pass |
|
|
|
|
| def _warn(message: str) -> None: |
| try: |
| os.write(2, f"[bounded_tee warning] {message}\n".encode("utf-8", errors="replace")) |
| except OSError: |
| pass |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--path", type=Path, required=True) |
| parser.add_argument("--max-mib", type=float, default=8.0) |
| parser.add_argument("--backups", type=int, default=3) |
| args = parser.parse_args() |
| if not 1.0 <= args.max_mib <= 1024.0 or not 1 <= args.backups <= 20: |
| parser.error("--max-mib must be in [1,1024] and --backups in [1,20]") |
|
|
| path = args.path.expanduser().resolve() |
| max_bytes = int(args.max_mib * 1024 * 1024) |
| sink = None |
| try: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| sink = path.open("ab", buffering=0) |
| except OSError as error: |
| _warn(f"file logging disabled for {path}: {error}") |
| try: |
| while True: |
| chunk = sys.stdin.buffer.read1(65536) |
| if not chunk: |
| break |
| try: |
| sys.stdout.buffer.write(chunk) |
| sys.stdout.buffer.flush() |
| except BrokenPipeError: |
| pass |
| if sink is None: |
| continue |
| try: |
| sink.write(chunk) |
| if sink.tell() >= max_bytes: |
| sink.close() |
| sink = None |
| _rotate(path, args.backups) |
| sink = path.open("ab", buffering=0) |
| except OSError as error: |
| _warn(f"file logging disabled for {path}: {error}") |
| if sink is not None: |
| try: |
| sink.close() |
| except OSError: |
| pass |
| sink = None |
| finally: |
| if sink is not None: |
| try: |
| sink.close() |
| except OSError: |
| pass |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
