File size: 2,784 Bytes
d8290eb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | #!/usr/bin/env python3
"""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())
|