101 lines
3.0 KiB
Python
101 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import signal
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
STOP_REQUESTED = False
|
|
|
|
|
|
def handle_signal(signum: int, frame: object) -> None:
|
|
del signum, frame
|
|
global STOP_REQUESTED
|
|
STOP_REQUESTED = True
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Poll /api/network/latest/ and append JSONL snapshots.")
|
|
parser.add_argument("--url", required=True, help="HTTP endpoint that returns the network summary JSON.")
|
|
parser.add_argument("--output", required=True, help="Output JSONL path.")
|
|
parser.add_argument(
|
|
"--interval-ms",
|
|
type=int,
|
|
default=2000,
|
|
help="Polling interval in milliseconds. Default: 2000.",
|
|
)
|
|
parser.add_argument(
|
|
"--request-timeout-sec",
|
|
type=float,
|
|
default=3.0,
|
|
help="Single request timeout in seconds. Default: 3.0.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def sleep_with_stop(seconds: float) -> None:
|
|
deadline = time.monotonic() + max(0.0, seconds)
|
|
while not STOP_REQUESTED:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0.0:
|
|
return
|
|
time.sleep(min(remaining, 0.2))
|
|
|
|
|
|
def fetch_json(url: str, timeout_sec: float) -> str:
|
|
request = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Cache-Control": "no-cache",
|
|
},
|
|
method="GET",
|
|
)
|
|
with urllib.request.urlopen(request, timeout=timeout_sec) as response:
|
|
charset = response.headers.get_content_charset("utf-8")
|
|
payload = response.read().decode(charset)
|
|
parsed = json.loads(payload)
|
|
return json.dumps(parsed, separators=(",", ":"), ensure_ascii=False)
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
interval_sec = max(args.interval_ms, 200) / 1000.0
|
|
output_path = Path(args.output)
|
|
last_error_log_monotonic = 0.0
|
|
|
|
signal.signal(signal.SIGINT, handle_signal)
|
|
signal.signal(signal.SIGTERM, handle_signal)
|
|
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with output_path.open("a", encoding="utf-8") as output_file:
|
|
while not STOP_REQUESTED:
|
|
started = time.monotonic()
|
|
try:
|
|
line = fetch_json(args.url, args.request_timeout_sec)
|
|
except (TimeoutError, urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError) as error:
|
|
now = time.monotonic()
|
|
if now - last_error_log_monotonic >= 10.0:
|
|
print(f"[network-summary] poll failed: {error}", file=sys.stderr)
|
|
last_error_log_monotonic = now
|
|
else:
|
|
output_file.write(line)
|
|
output_file.write("\n")
|
|
output_file.flush()
|
|
|
|
elapsed = time.monotonic() - started
|
|
sleep_with_stop(max(0.0, interval_sec - elapsed))
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|