#!/usr/bin/env python3
"""GB32960 TCP ingest to history/archive smoke verifier."""

from __future__ import annotations

import argparse
import datetime as dt
import json
import os
import pathlib
import socket
import time
import urllib.parse
import urllib.request
from dataclasses import dataclass
from typing import Any

try:
    from tools import tdengine_smoke
except ModuleNotFoundError:
    import tdengine_smoke


SHANGHAI = dt.timezone(dt.timedelta(hours=8))
DEFAULT_SAMPLE = (
    "modules/protocols/protocol-gb32960/src/test/resources/samples/gb32960/realtime_001.hex"
)
DEFAULT_FIELD_KEYS = "VEHICLE.speedKmh,VEHICLE.totalMileageKm,POSITION_V2016.longitude,POSITION_V2016.latitude"


@dataclass(frozen=True)
class FrameInfo:
    vin: str
    command: int
    event_time: dt.datetime | None


def read_hex_frame(path: pathlib.Path) -> bytes:
    text = path.read_text(encoding="utf-8")
    return bytes.fromhex("".join(text.split()))


def parse_frame_info(frame: bytes) -> FrameInfo:
    if len(frame) < 25:
        raise ValueError(f"gb32960 frame too short: {len(frame)}")
    if frame[:2] not in (b"##", b"$$"):
        raise ValueError("gb32960 frame must start with ## or $$")
    data_len = int.from_bytes(frame[22:24], "big")
    expected_len = 24 + data_len + 1
    if len(frame) < expected_len:
        raise ValueError(f"gb32960 frame incomplete: len={len(frame)} expected={expected_len}")
    command = frame[2]
    vin = frame[4:21].decode("ascii", errors="ignore").strip()
    event_time = None
    if command in (0x01, 0x02, 0x03, 0x04, 0x05) and data_len >= 6:
        event_time = parse_collect_time(frame[24:30])
    return FrameInfo(vin=vin, command=command, event_time=event_time)


def parse_collect_time(raw: bytes) -> dt.datetime:
    if len(raw) != 6:
        raise ValueError("gb32960 collect time must be 6 bytes")
    year, month, day, hour, minute, second = raw
    return dt.datetime(2000 + year, month, day, hour, minute, second, tzinfo=SHANGHAI)


def query_range(event_time: dt.datetime | None, minutes: int) -> tuple[str, str]:
    center = event_time or dt.datetime.now(tz=SHANGHAI)
    center = center.astimezone(SHANGHAI)
    start = center - dt.timedelta(minutes=minutes)
    end = center + dt.timedelta(minutes=minutes)
    return start.isoformat(timespec="seconds"), end.isoformat(timespec="seconds")


def send_frame(host: str, port: int, frame: bytes, timeout: float) -> bytes:
    with socket.create_connection((host, port), timeout=timeout) as sock:
        sock.settimeout(timeout)
        sock.sendall(frame)
        try:
            return sock.recv(4096)
        except socket.timeout:
            return b""


def valid_ack(ack: bytes) -> bool:
    return len(ack) >= 2 and ack[:2] in (b"##", b"$$")


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


def wait_for_records(
    history_base_url: str,
    vin: str,
    date_from: str,
    date_to: str,
    minimum: int,
    timeout_seconds: float,
    request_timeout: float,
) -> list[dict[str, Any]]:
    deadline = time.monotonic() + timeout_seconds
    params = {
        "vin": vin,
        "dateFrom": date_from,
        "dateTo": date_to,
        "limit": max(minimum, 10),
        "order": "ASC",
    }
    last_items: list[dict[str, Any]] = []
    while time.monotonic() < deadline:
        response = query_json(history_base_url, "/api/event-history/gb32960/snapshots", params, request_timeout)
        last_items = response if isinstance(response, list) else []
        if len(last_items) >= minimum:
            return last_items
        time.sleep(1)
    return last_items


def wait_for_fields(
    history_base_url: str,
    vin: str,
    field_key: str,
    date_from: str,
    date_to: str,
    timeout_seconds: float,
    request_timeout: float,
) -> list[dict[str, Any]]:
    deadline = time.monotonic() + timeout_seconds
    params = {
        "vin": vin,
        "fields": field_key,
        "dateFrom": date_from,
        "dateTo": date_to,
        "limit": 10,
        "order": "ASC",
    }
    last_items: list[dict[str, Any]] = []
    while time.monotonic() < deadline:
        response = query_json(history_base_url, "/api/event-history/gb32960/snapshots/fields", params, request_timeout)
        last_items = response if isinstance(response, list) else []
        if any((item.get("fields") or {}).get(field_key) is not None for item in last_items):
            return last_items
        time.sleep(1)
    return last_items


def wait_for_tdengine_raw_frames(
    rest_url: str,
    username: str,
    password: str,
    sql: str,
    minimum: int,
    timeout_seconds: float,
    request_timeout: float,
) -> int:
    deadline = time.monotonic() + timeout_seconds
    last_count = 0
    while time.monotonic() < deadline:
        last_count = tdengine_smoke.query_count(rest_url, username, password, sql, request_timeout)
        if last_count >= minimum:
            return last_count
        time.sleep(1)
    return last_count


def parse_field_keys(value: str) -> list[str]:
    return [item.strip() for item in value.split(",") if item.strip()]


def archive_path(root: pathlib.Path, uri: str) -> pathlib.Path:
    prefix = "archive://"
    if not uri.startswith(prefix):
        raise ValueError(f"unsupported archive uri: {uri}")
    return root / uri[len(prefix):]


def check_archive(root: pathlib.Path, records: list[dict[str, Any]], limit: int) -> int:
    checked = 0
    seen: set[str] = set()
    for uri in raw_uris(records):
        if uri in seen:
            continue
        seen.add(uri)
        path = archive_path(root, uri)
        if not path.exists():
            raise RuntimeError(f"raw archive missing: {path}")
        checked += 1
        if checked >= limit:
            return checked
    return checked


def raw_frame_count_sql(info: FrameInfo, records: list[dict[str, Any]]) -> str:
    sqls = raw_frame_count_sqls(info, records)
    if not sqls:
        raise ValueError("no raw archive uri found in gb32960 records")
    return sqls[0]


def raw_frame_count_sqls(info: FrameInfo, records: list[dict[str, Any]]) -> list[str]:
    seen: set[str] = set()
    sqls: list[str] = []
    for uri in raw_uris(records):
        if uri in seen:
            continue
        seen.add(uri)
        sqls.append(tdengine_smoke.raw_frame_count_sql(
            protocol="GB32960",
            vehicle_key=info.vin,
            vin=info.vin,
            raw_uri=uri,
        ))
    return sqls


def verify_tdengine_raw_frames(
    rest_url: str,
    username: str,
    password: str,
    sqls: list[str],
    timeout_seconds: float,
    request_timeout: float,
) -> int:
    if not sqls:
        raise RuntimeError("no raw archive uri available for tdengine raw_frames verification")
    verified = 0
    for sql in sqls:
        count = wait_for_tdengine_raw_frames(
            rest_url, username, password, sql, 1, timeout_seconds, request_timeout)
        if count < 1:
            raise RuntimeError(f"tdengine raw_frames row not visible for query: {sql}")
        verified += 1
    return verified


def run(args: argparse.Namespace) -> dict[str, Any]:
    frame = read_hex_frame(pathlib.Path(args.frame_hex_file))
    info = parse_frame_info(frame)
    query_center = dt.datetime.now(tz=SHANGHAI)
    date_from, date_to = query_range(query_center, args.window_minutes)

    ack = send_frame(args.tcp_host, args.tcp_port, frame, args.tcp_timeout)
    if args.require_ack and not valid_ack(ack):
        raise RuntimeError(f"gb32960 ack missing or invalid: {ack.hex()}")

    records = wait_for_records(
        args.history_base_url,
        info.vin,
        date_from,
        date_to,
        args.expect_record_count,
        args.history_timeout,
        args.http_timeout,
    )
    if len(records) < args.expect_record_count:
        raise RuntimeError(f"gb32960 records not visible for {info.vin}: got {len(records)}")

    field_counts: dict[str, int] = {}
    for field_key in parse_field_keys(args.field_keys):
        field_items = wait_for_fields(
            args.history_base_url,
            info.vin,
            field_key,
            date_from,
            date_to,
            args.history_timeout,
            args.http_timeout,
        )
        field_counts[field_key] = len(field_items)
        if not field_items:
            raise RuntimeError(f"gb32960 telemetry field not visible for {info.vin}: {field_key}")

    archive_checked = 0
    if args.archive_root:
        archive_checked = check_archive(pathlib.Path(args.archive_root), records, args.archive_check_limit)

    tdengine_raw_frames = None
    if args.tdengine_rest_url:
        tdengine_raw_frames = verify_tdengine_raw_frames(
            args.tdengine_rest_url,
            args.tdengine_username,
            args.tdengine_password,
            raw_frame_count_sqls(info, records),
            args.history_timeout,
            args.http_timeout,
        )

    return {
        "tcpHost": args.tcp_host,
        "tcpPort": args.tcp_port,
        "vin": info.vin,
        "command": f"0x{info.command:02x}",
        "eventTime": info.event_time.isoformat() if info.event_time else None,
        "queryTimeSource": "received",
        "queryDateFrom": date_from,
        "queryDateTo": date_to,
        "ackBytesHex": ack.hex(),
        "records": len(records),
        "fieldCounts": field_counts,
        "tdengineRawFrames": tdengine_raw_frames,
        "archiveChecked": archive_checked,
        "firstRawUri": first_raw_uri(records),
    }


def first_raw_uri(records: list[dict[str, Any]]) -> str | None:
    uris = raw_uris(records)
    if uris:
        return uris[0]
    return None


def raw_uris(records: list[dict[str, Any]]) -> list[str]:
    out: list[str] = []
    for record in records:
        uri = record.get("rawArchiveUri")
        if isinstance(uri, str) and uri:
            out.append(uri)
        for frame in record.get("sourceFrames") or []:
            frame_uri = frame.get("rawArchiveUri") if isinstance(frame, dict) else ""
            if frame_uri:
                out.append(frame_uri)
    return out


def parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--tcp-host", default="127.0.0.1")
    p.add_argument("--tcp-port", type=int, default=32960)
    p.add_argument("--history-base-url", default="http://127.0.0.1:20200")
    p.add_argument("--frame-hex-file", default=DEFAULT_SAMPLE)
    p.add_argument("--field-keys", default=DEFAULT_FIELD_KEYS)
    p.add_argument("--window-minutes", type=int, default=10)
    p.add_argument("--expect-record-count", type=int, default=1)
    p.add_argument("--archive-root", default="")
    p.add_argument("--archive-check-limit", type=int, default=1)
    p.add_argument("--tdengine-rest-url", default=os.environ.get("TDENGINE_REST_URL", ""))
    p.add_argument("--tdengine-username", default=os.environ.get("TDENGINE_USERNAME", "root"))
    p.add_argument("--tdengine-password", default=os.environ.get("TDENGINE_PASSWORD", "taosdata"))
    p.add_argument("--require-ack", action="store_true", default=True)
    p.add_argument("--tcp-timeout", type=float, default=5)
    p.add_argument("--http-timeout", type=float, default=5)
    p.add_argument("--history-timeout", type=float, default=60)
    return p


def main() -> None:
    args = parser().parse_args()
    summary = run(args)
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()