streamtop 1.1.0

HLS/DASH/IPTV stream diagnostics TUI and CLI
Documentation 
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

#!/usr/bin/env python3
"""Validate streamtop summary JSON against schemas/summary.v1.json."""

from __future__ import annotations

import json
import sys
from pathlib import Path


def basic_validate(doc: dict) -> list[str]:
    errs: list[str] = []
    required = [
        "schema",
        "schema_version",
        "verdict",
        "ok",
        "health_score",
        "health_label",
        "status",
        "latency",
        "cdn",
        "origin_stalls",
        "critical_rfc_errors",
        "url",
        "errors",
        "saw_segment",
    ]
    for key in required:
        if key not in doc:
            errs.append(f"missing required field: {key}")
    if doc.get("schema") != "streamtop.summary.v1":
        errs.append("schema must be streamtop.summary.v1")
    if doc.get("verdict") not in ("PASS", "FAIL"):
        errs.append("invalid verdict")
    if doc.get("status") not in ("LIVE", "DEGRADED", "ERROR"):
        errs.append("invalid status")
    score = doc.get("health_score")
    if not isinstance(score, int) or not (0 <= score <= 100):
        errs.append("health_score out of range")
    return errs


def jsonschema_validate(doc: dict, schema_path: Path) -> list[str]:
    try:
        import jsonschema  # type: ignore
    except ImportError:
        return basic_validate(doc)
    schema = json.loads(schema_path.read_text(encoding="utf-8"))
    validator = jsonschema.Draft202012Validator(schema)
    return [e.message for e in validator.iter_errors(doc)]


def main() -> int:
    if len(sys.argv) < 2:
        print("usage: validate_summary.py <summary.json>", file=sys.stderr)
        return 2
    path = Path(sys.argv[1])
    doc = json.loads(path.read_text(encoding="utf-8"))
    schema = Path(__file__).resolve().parents[2] / "schemas" / "summary.v1.json"
    errs = jsonschema_validate(doc, schema)
    if errs:
        for e in errs:
            print(f"FAIL: {e}", file=sys.stderr)
        return 1
    print("PASS: summary matches schema")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())