#!/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 require_parsed_json: bool = False require_frame_id: bool = False @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, require_parsed_json: bool = False, require_frame_id: bool = False, ) -> 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: parsed_message = "" 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}", ) if require_parsed_json: parsed_message = parsed_json_message(items) if not parsed_message.startswith("parsed_json=ok"): return Check(name, "fail", count, minimum, f"count={count}; {parsed_message}; sample={sample}") frame_message = "" if require_frame_id: frame_message = frame_id_message(items) if not frame_message.startswith("frame_id=ok"): return Check(name, "fail", count, minimum, f"count={count}; {frame_message}; sample={sample}") details = [f"count={count}", age_message] if parsed_message: details.append(parsed_message) if frame_message: details.append(frame_message) return Check(name, "pass", count, minimum, "; ".join(details) + f"; 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 parsed_json_message(items: list[Any]) -> str: if not items or not isinstance(items[0], dict): return "missing parsed_json" raw = items[0].get("parsed_json") if isinstance(raw, dict) and raw: return "parsed_json=ok" if not isinstance(raw, str) or not raw.strip() or raw.strip().lower() == "null": return "missing parsed_json" try: parsed = json.loads(raw) except json.JSONDecodeError: return "invalid parsed_json" if isinstance(parsed, dict) and parsed: return "parsed_json=ok" return "invalid parsed_json" def frame_id_message(items: list[Any]) -> str: if not items or not isinstance(items[0], dict): return "missing frame_id" frame_id = str(items[0].get("frame_id") or "").strip() if frame_id: return "frame_id=ok" return "missing frame_id" 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} gb32960_history_params = {"protocol": "GB32960", "dateFrom": date_from, "dateTo": date_to, "limit": 1} jt808_history_params = {"protocol": "JT808", "dateFrom": date_from, "dateTo": date_to, "limit": 1} yutong_mqtt_history_params = {"protocol": "YUTONG_MQTT", "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, require_parsed_json=True, ), CheckSpec( "jt808.raw", "/api/history/raw-frames", {**raw_params, "protocol": "JT808"}, min_raw, max_age_minutes=max_raw_age_minutes, require_parsed_json=True, ), CheckSpec( "yutong_mqtt.raw", "/api/history/raw-frames", {**raw_params, "protocol": "YUTONG_MQTT"}, min_raw, max_age_minutes=max_raw_age_minutes, require_parsed_json=True, ), CheckSpec("gb32960.locations", "/api/history/locations", gb32960_history_params, min_history, require_frame_id=True), CheckSpec("gb32960.mileage_points", "/api/history/mileage-points", gb32960_history_params, min_history, require_frame_id=True), CheckSpec("jt808.locations", "/api/history/locations", jt808_history_params, min_history, require_frame_id=True), CheckSpec("jt808.mileage_points", "/api/history/mileage-points", jt808_history_params, min_history, require_frame_id=True), CheckSpec("yutong_mqtt.locations", "/api/history/locations", yutong_mqtt_history_params, min_history, require_frame_id=True), CheckSpec("yutong_mqtt.mileage_points", "/api/history/mileage-points", yutong_mqtt_history_params, min_history, require_frame_id=True), 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, require_parsed_json=spec.require_parsed_json, require_frame_id=spec.require_frame_id, )) 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, require_parsed_json=spec.require_parsed_json, require_frame_id=spec.require_frame_id, )) 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:]))