lingniu-vehicle-ingest/tools/go_native_prod_smoke.py

#!/usr/bin/env python3
"""HTTP smoke checks for the Go native production vehicle ingest stack."""

from __future__ import annotations

import argparse
import datetime as dt
import json
import sys
import urllib.parse
import urllib.request
from dataclasses import asdict, dataclass
from typing import Any


SHANGHAI = dt.timezone(dt.timedelta(hours=8))
DEFAULT_BASE_URL = "http://115.29.187.205:20210"


@dataclass(frozen=True)
class CheckSpec:
    name: str
    path: str
    params: dict[str, Any]
    minimum: int
    kind: str = "total"
    max_age_minutes: float | None = None


@dataclass(frozen=True)
class Check:
    name: str
    status: str
    count: int
    minimum: int
    message: str


def shanghai_day_window(now: dt.datetime | None = None) -> tuple[str, str]:
    now = now or dt.datetime.now(tz=SHANGHAI)
    local = now.astimezone(SHANGHAI)
    start = dt.datetime(local.year, local.month, local.day, tzinfo=SHANGHAI)
    end = start + dt.timedelta(days=1)
    return start.isoformat(timespec="seconds"), end.isoformat(timespec="seconds")


def api_url(base_url: str, path: str, params: dict[str, Any]) -> str:
    url = base_url.rstrip("/") + path
    if not params:
        return url
    return url + "?" + urllib.parse.urlencode(params)


def query_json(base_url: str, path: str, params: dict[str, Any], timeout: float) -> dict[str, Any]:
    request = urllib.request.Request(
        api_url(base_url, path, params),
        headers={"Accept": "application/json"},
    )
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return json.loads(response.read().decode("utf-8"))


def check_total(
    name: str,
    payload: dict[str, Any],
    minimum: int,
    max_age_minutes: float | None = None,
    now: dt.datetime | None = None,
) -> Check:
    count = int(payload.get("total") or 0)
    items = payload.get("items") or []
    sample = sample_summary(items)
    age_message = latest_age_message(items, now)
    if count >= minimum:
        if max_age_minutes is not None:
            latest_age = latest_age_minutes(items, now)
            if latest_age is None:
                return Check(name, "fail", count, minimum, f"count={count}; missing latest ts; sample={sample}")
            if latest_age > max_age_minutes:
                return Check(
                    name,
                    "fail",
                    count,
                    minimum,
                    f"count={count}; latest_age_minutes={latest_age:.1f}; "
                    f"expected <= {max_age_minutes:.1f}; sample={sample}",
                )
        return Check(name, "pass", count, minimum, f"count={count}; {age_message}; sample={sample}")
    return Check(name, "fail", count, minimum, f"count={count}; expected >= {minimum}; {age_message}; sample={sample}")


def parse_tdengine_utc_timestamp(value: str) -> dt.datetime:
    value = str(value or "").strip()
    for layout in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
        try:
            return dt.datetime.strptime(value[:19], layout).replace(tzinfo=dt.timezone.utc)
        except ValueError:
            continue
    raise ValueError(f"unsupported timestamp: {value}")


def latest_age_minutes(items: list[Any], now: dt.datetime | None = None) -> float | None:
    if not items or not isinstance(items[0], dict) or not items[0].get("ts"):
        return None
    current = now or dt.datetime.now(tz=dt.timezone.utc)
    if current.tzinfo is None:
        current = current.replace(tzinfo=dt.timezone.utc)
    latest = parse_tdengine_utc_timestamp(str(items[0]["ts"]))
    return max(0.0, (current.astimezone(dt.timezone.utc) - latest).total_seconds() / 60)


def latest_age_message(items: list[Any], now: dt.datetime | None = None) -> str:
    latest_age = latest_age_minutes(items, now)
    if latest_age is None:
        return "latest_age_minutes=unknown"
    return f"latest_age_minutes={latest_age:.1f}"


def check_realtime(name: str, payload: dict[str, Any]) -> Check:
    if payload.get("online") is False:
        return Check(name, "fail", 0, 1, "online=false")
    identifier = payload.get("vin") or payload.get("vehicle_key") or ""
    protocols = payload.get("protocols") or []
    protocol = payload.get("protocol") or ""
    fields = payload.get("fields") or {}
    message = json.dumps(
        {
            key: value
            for key, value in {
                "identifier": identifier,
                "protocol": protocol,
                "protocols": protocols,
                "field_count": len(fields) if isinstance(fields, dict) else 0,
                "online": payload.get("online"),
            }.items()
            if value not in (None, "", [])
        },
        ensure_ascii=False,
        sort_keys=True,
    )
    return Check(name, "pass", 1, 1, message)


def sample_summary(items: list[Any]) -> str:
    if not items:
        return "{}"
    first = items[0]
    if not isinstance(first, dict):
        return json.dumps(first, ensure_ascii=False, sort_keys=True)
    keep = {
        key: first.get(key)
        for key in (
            "ts",
            "stat_date",
            "protocol",
            "vehicle_key",
            "vin",
            "phone",
            "message_id_hex",
            "metric_key",
            "metric_value",
            "total_mileage_km",
            "latest_total_mileage_km",
        )
        if first.get(key) not in (None, "")
    }
    return json.dumps(keep, ensure_ascii=False, sort_keys=True)


def overall_status(checks: list[Check]) -> str:
    return "fail" if any(check.status == "fail" for check in checks) else "pass"


def build_check_specs(
    *,
    date_from: str,
    date_to: str,
    stat_date: str,
    min_raw: int,
    min_history: int,
    min_stat: int,
    max_raw_age_minutes: float | None = None,
) -> list[CheckSpec]:
    raw_params = {"dateFrom": date_from, "dateTo": date_to, "limit": 1}
    history_params = {"protocol": "JT808", "dateFrom": date_from, "dateTo": date_to, "limit": 1}
    return [
        CheckSpec(
            "gb32960.raw",
            "/api/history/raw-frames",
            {**raw_params, "protocol": "GB32960"},
            min_raw,
            max_age_minutes=max_raw_age_minutes,
        ),
        CheckSpec(
            "jt808.raw",
            "/api/history/raw-frames",
            {**raw_params, "protocol": "JT808"},
            min_raw,
            max_age_minutes=max_raw_age_minutes,
        ),
        CheckSpec(
            "yutong_mqtt.raw",
            "/api/history/raw-frames",
            {**raw_params, "protocol": "YUTONG_MQTT"},
            min_raw,
            max_age_minutes=max_raw_age_minutes,
        ),
        CheckSpec("jt808.locations", "/api/history/locations", history_params, min_history),
        CheckSpec("jt808.mileage_points", "/api/history/mileage-points", history_params, min_history),
        CheckSpec(
            "gb32960.daily_mileage",
            "/api/stats/daily-metrics",
            {
                "protocol": "GB32960",
                "metricKey": "daily_mileage_km",
                "dateFrom": stat_date,
                "dateTo": stat_date,
                "limit": 1,
            },
            min_stat,
        ),
        CheckSpec(
            "gb32960.daily_total_mileage",
            "/api/stats/daily-metrics",
            {
                "protocol": "GB32960",
                "metricKey": "daily_total_mileage_km",
                "dateFrom": stat_date,
                "dateTo": stat_date,
                "limit": 1,
            },
            min_stat,
        ),
        CheckSpec(
            "jt808.daily_mileage",
            "/api/stats/daily-metrics",
            {
                "protocol": "JT808",
                "metricKey": "daily_mileage_km",
                "dateFrom": stat_date,
                "dateTo": stat_date,
                "limit": 1,
            },
            min_stat,
        ),
        CheckSpec(
            "jt808.daily_total_mileage",
            "/api/stats/daily-metrics",
            {
                "protocol": "JT808",
                "metricKey": "daily_total_mileage_km",
                "dateFrom": stat_date,
                "dateTo": stat_date,
                "limit": 1,
            },
            min_stat,
        ),
    ]


def vehicle_identifier(item: dict[str, Any]) -> str:
    for key in ("vin", "vehicle_key"):
        value = str(item.get(key) or "").strip()
        if value and value.lower() != "unknown":
            return value
    return ""


def build_realtime_specs(payloads: dict[str, dict[str, Any]]) -> list[CheckSpec]:
    configs = [
        ("gb32960", "gb32960.raw", "GB32960"),
        ("jt808", "jt808.raw", "JT808"),
        ("yutong_mqtt", "yutong_mqtt.raw", "YUTONG_MQTT"),
    ]
    specs: list[CheckSpec] = []
    for prefix, source_name, protocol in configs:
        items = payloads.get(source_name, {}).get("items") or []
        if not items or not isinstance(items[0], dict):
            continue
        identifier = vehicle_identifier(items[0])
        if not identifier:
            continue
        encoded = urllib.parse.quote(identifier, safe="")
        base_path = "/api/realtime/vehicles/" + encoded
        specs.extend([
            CheckSpec(prefix + ".realtime", base_path, {}, 1, "realtime"),
            CheckSpec(prefix + ".realtime_online", base_path + "/online", {}, 1, "realtime"),
            CheckSpec(prefix + ".realtime_protocol", base_path + "/protocols/" + protocol, {}, 1, "realtime"),
        ])
    return specs


def run_checks(base_url: str, specs: list[CheckSpec], timeout: float) -> list[Check]:
    checks: list[Check] = []
    for spec in specs:
        try:
            payload = query_json(base_url, spec.path, spec.params, timeout)
            if spec.kind == "realtime":
                checks.append(check_realtime(spec.name, payload))
            else:
                checks.append(check_total(spec.name, payload, spec.minimum, spec.max_age_minutes))
        except Exception as exc:  # pragma: no cover - exercised by live failures.
            checks.append(Check(spec.name, "fail", 0, spec.minimum, f"request failed: {exc}"))
    return checks


def query_payloads(base_url: str, specs: list[CheckSpec], timeout: float) -> tuple[list[Check], dict[str, dict[str, Any]]]:
    checks: list[Check] = []
    payloads: dict[str, dict[str, Any]] = {}
    for spec in specs:
        try:
            payload = query_json(base_url, spec.path, spec.params, timeout)
            payloads[spec.name] = payload
            checks.append(check_total(spec.name, payload, spec.minimum, spec.max_age_minutes))
        except Exception as exc:
            checks.append(Check(spec.name, "fail", 0, spec.minimum, f"request failed: {exc}"))
    return checks, payloads


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
    parser.add_argument("--date", help="Shanghai local date, YYYY-MM-DD. Defaults to today.")
    parser.add_argument("--timeout", type=float, default=5.0)
    parser.add_argument("--min-raw", type=int, default=1)
    parser.add_argument("--min-history", type=int, default=1)
    parser.add_argument("--min-stat", type=int, default=1)
    parser.add_argument(
        "--max-raw-age-minutes",
        type=float,
        default=15.0,
        help="Require latest RAW sample age to be at most this many minutes when checking today's date.",
    )
    return parser.parse_args(argv)


def main(argv: list[str]) -> int:
    args = parse_args(argv)
    if args.date:
        day = dt.date.fromisoformat(args.date)
        start = dt.datetime(day.year, day.month, day.day, tzinfo=SHANGHAI)
        date_from, date_to = shanghai_day_window(start)
        stat_date = args.date
        max_raw_age_minutes = None
    else:
        date_from, date_to = shanghai_day_window()
        stat_date = date_from[:10]
        max_raw_age_minutes = args.max_raw_age_minutes
    specs = build_check_specs(
        date_from=date_from,
        date_to=date_to,
        stat_date=stat_date,
        min_raw=args.min_raw,
        min_history=args.min_history,
        min_stat=args.min_stat,
        max_raw_age_minutes=max_raw_age_minutes,
    )
    checks, payloads = query_payloads(args.base_url, specs, args.timeout)
    checks.extend(run_checks(args.base_url, build_realtime_specs(payloads), args.timeout))
    status = overall_status(checks)
    print(json.dumps({
        "status": status,
        "baseUrl": args.base_url,
        "dateFrom": date_from,
        "dateTo": date_to,
        "checks": [asdict(check) for check in checks],
    }, ensure_ascii=False, indent=2))
    return 0 if status == "pass" else 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))