#!/usr/bin/env python3
from __future__ import annotations

import argparse
import base64
import json
import os
import sys
import time
from pathlib import Path
from typing import Any


DEFAULT_RESOURCE_META = {
    "mq": {
        "backup_status": "云产品托管，平台不执行恢复动作",
        "name": "Tencent Cloud MQ",
        "owner": "后端平台",
        "topology": "托管消息队列",
        "type": "mq",
    },
    "mysql": {
        "backup_status": "云产品自动备份，平台不执行恢复动作",
        "name": "TencentDB MySQL",
        "owner": "后端平台",
        "topology": "托管主从 MySQL",
        "type": "mysql",
    },
    "redis": {
        "backup_status": "缓存资源不作为恢复事实源",
        "name": "TencentDB Redis",
        "owner": "后端平台",
        "topology": "托管 Redis",
        "type": "redis",
    },
}


def load_env_file(path: str) -> None:
    if not path:
        return
    env_path = Path(path).expanduser().resolve()
    if not env_path.exists():
        raise RuntimeError(f"env file not found: {env_path}")
    for raw_line in env_path.read_text(encoding="utf-8", errors="replace").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip().strip('"').strip("'")
        if key and key not in os.environ:
            os.environ[key] = value


def load_inventory(path: str) -> dict[str, Any]:
    inventory_path = Path(path).expanduser().resolve()
    with inventory_path.open("r", encoding="utf-8") as handle:
        inventory = json.load(handle)
    validate_inventory(inventory)
    return inventory


def validate_inventory(inventory: dict[str, Any]) -> None:
    if not isinstance(inventory.get("deploy_server"), dict):
        raise RuntimeError("inventory.deploy_server is required")
    if not str(inventory["deploy_server"].get("instance_id") or "").strip():
        raise RuntimeError("inventory.deploy_server.instance_id is required")
    if not isinstance(inventory.get("hosts"), dict) or not inventory["hosts"]:
        raise RuntimeError("inventory.hosts is required")
    if not isinstance(inventory.get("services"), dict) or not inventory["services"]:
        raise RuntimeError("inventory.services is required")


def secret_id() -> str:
    return os.environ.get("TENCENTCLOUD_SECRET_ID") or os.environ.get("TENCENT_SECRET_ID") or ""


def secret_key() -> str:
    return os.environ.get("TENCENTCLOUD_SECRET_KEY") or os.environ.get("TENCENT_SECRET_KEY") or ""


def region(inventory: dict[str, Any]) -> str:
    return os.environ.get("TENCENTCLOUD_REGION") or os.environ.get("DEPLOY_REGION") or str(inventory.get("region") or "")


def require_tencent_credentials(inventory: dict[str, Any]) -> tuple[str, str, str]:
    sid = secret_id()
    skey = secret_key()
    reg = region(inventory)
    missing = []
    if not sid:
        missing.append("TENCENTCLOUD_SECRET_ID")
    if not skey:
        missing.append("TENCENTCLOUD_SECRET_KEY")
    if not reg:
        missing.append("TENCENTCLOUD_REGION")
    if missing:
        raise RuntimeError("missing Tencent Cloud credentials: " + ", ".join(missing))
    return sid, skey, reg


def build_tat_client(inventory: dict[str, Any]) -> Any:
    sid, skey, reg = require_tencent_credentials(inventory)
    try:
        from tencentcloud.common import credential
        from tencentcloud.common.profile.client_profile import ClientProfile
        from tencentcloud.common.profile.http_profile import HttpProfile
        from tencentcloud.tat.v20201028 import tat_client
    except Exception as exc:  # noqa: BLE001
        raise RuntimeError(f"tencentcloud sdk unavailable: {exc}") from exc
    return tat_client.TatClient(
        credential.Credential(sid, skey),
        reg,
        ClientProfile(httpProfile=HttpProfile(endpoint="tat.tencentcloudapi.com")),
    )


def invoke_tencent(action: Any, label: str, retries: int = 3) -> Any:
    try:
        from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
    except Exception:  # noqa: BLE001
        TencentCloudSDKException = Exception  # type: ignore[assignment]
    last_error: Exception | None = None
    for attempt in range(1, retries + 1):
        try:
            return action()
        except TencentCloudSDKException as exc:  # type: ignore[misc]
            last_error = exc
            if attempt == retries:
                break
            time.sleep(2 * attempt)
    raise RuntimeError(f"{label} failed: {last_error}") from last_error


def decode_tat_output(output: str) -> str:
    if not output:
        return ""
    return base64.b64decode(output).decode("utf-8", errors="replace")


def run_tat_shell(
    client: Any,
    *,
    instance_id: str,
    script: str,
    command_name: str,
    timeout_seconds: int,
    poll_seconds: int,
) -> dict[str, Any]:
    from tencentcloud.tat.v20201028 import models as tat_models

    request = tat_models.RunCommandRequest()
    request.from_json_string(
        json.dumps(
            {
                "CommandName": command_name,
                "CommandType": "SHELL",
                "Content": base64.b64encode(script.encode("utf-8")).decode("ascii"),
                "InstanceIds": [instance_id],
                "Timeout": timeout_seconds,
                "WorkingDirectory": "/root",
            }
        )
    )
    response = json.loads(invoke_tencent(lambda: client.RunCommand(request), "RunCommand").to_json_string())
    invocation_id = response["InvocationId"]
    deadline = time.time() + timeout_seconds + 60
    while time.time() < deadline:
        time.sleep(max(poll_seconds, 1))
        query = tat_models.DescribeInvocationTasksRequest()
        query.from_json_string(
            json.dumps(
                {
                    "HideOutput": False,
                    "Limit": 50,
                    "Filters": [{"Name": "invocation-id", "Values": [invocation_id]}],
                }
            )
        )
        payload = json.loads(invoke_tencent(lambda: client.DescribeInvocationTasks(query), "DescribeInvocationTasks").to_json_string())
        for task in payload.get("InvocationTaskSet") or []:
            if str(task.get("InstanceId") or "") != instance_id:
                continue
            status = str(task.get("TaskStatus") or "")
            if status in {"SUCCESS", "FAILED", "TIMEOUT", "CANCELLED", "START_FAILED"}:
                return {
                    "status": status,
                    "output": decode_tat_output(str((task.get("TaskResult") or {}).get("Output") or "")),
                }
    return {"status": "TIMEOUT", "output": ""}


def normalize_data_resources(inventory: dict[str, Any]) -> list[dict[str, Any]]:
    raw_resources = inventory.get("data_resources")
    resources: list[dict[str, Any]] = []
    if isinstance(raw_resources, dict) and raw_resources:
        for resource_id, raw in raw_resources.items():
            if isinstance(raw, dict):
                resource = dict(raw)
            else:
                resource = {"description": str(raw)}
            resource["id"] = str(resource.get("id") or resource_id)
            resources.append(resource)
    else:
        managed_dependencies = inventory.get("managed_dependencies") or {}
        if isinstance(managed_dependencies, dict):
            for resource_id, description in managed_dependencies.items():
                resource = {"description": str(description), "id": str(resource_id)}
                resources.append(resource)

    service_names = sorted(str(name) for name in inventory.get("services", {}).keys())
    normalized = []
    for resource in resources:
        resource_id = str(resource.get("id") or "").strip()
        defaults = DEFAULT_RESOURCE_META.get(resource_id, {})
        resource_type = str(resource.get("type") or defaults.get("type") or resource_id or "unknown")
        related_services = resource.get("related_services") or resource.get("relatedServices") or service_names
        normalized.append(
            {
                "backupStatus": str(resource.get("backup_status") or resource.get("backupStatus") or defaults.get("backup_status") or "未采集"),
                "capacityUsedPercent": resource.get("capacity_used_percent") or resource.get("capacityUsedPercent"),
                "connections": resource.get("connections"),
                "description": str(resource.get("description") or ""),
                "endpoints": resource.get("endpoints") or ([resource.get("endpoint")] if resource.get("endpoint") else []),
                "id": resource_id,
                "name": str(resource.get("name") or defaults.get("name") or resource_id),
                "owner": str(resource.get("owner") or defaults.get("owner") or "后端平台"),
                "relatedServices": [str(item) for item in related_services if str(item).strip()],
                "slowQueryCount": resource.get("slow_query_count") or resource.get("slowQueryCount"),
                "topology": str(resource.get("topology") or defaults.get("topology") or "托管资源"),
                "type": resource_type,
            }
        )
    return normalized


def build_remote_payload(inventory: dict[str, Any], *, max_bytes: int) -> dict[str, Any]:
    services = []
    for service_name, service in inventory["services"].items():
        services.append(
            {
                "configPath": service.get("config_path") or f"/etc/hyapp/{service_name}/config.yaml",
                "envFile": service.get("env_file") or "",
                "hosts": list(service.get("hosts") or []),
                "name": service_name,
                "unit": service.get("unit") or "",
            }
        )
    return {
        "dataResources": normalize_data_resources(inventory),
        "hosts": inventory["hosts"],
        "maxBytes": max_bytes,
        "region": region(inventory),
        "services": services,
        "sshUser": os.environ.get("DEPLOY_DATA_SSH_USER", "root"),
        "tatFallbackEnvFiles": [
            item.strip()
            for item in os.environ.get(
                "DEPLOY_DATA_TAT_FALLBACK_ENV_FILES",
                "/opt/hy-app-monitor/.env,/etc/hyapp/deploy-platform/docker.env",
            ).split(",")
            if item.strip()
        ],
        "tatFallbackTimeout": int(inventory.get("tat_data_fallback_timeout_seconds") or 120),
        "tatPollSeconds": int(inventory.get("tat_poll_seconds") or 2),
    }


def remote_collect_script(payload: dict[str, Any]) -> str:
    blob = base64.b64encode(json.dumps(payload).encode("utf-8")).decode("ascii")
    return f"""set -euo pipefail
python3 - <<'PY'
import base64
import json
import os
import re
import socket
import subprocess
import sys
import time

payload = json.loads(base64.b64decode({blob!r}).decode("utf-8"))
services = payload["services"]
hosts = payload["hosts"]
data_resources = payload["dataResources"]
ssh_user = payload.get("sshUser") or "root"
max_bytes = int(payload.get("maxBytes") or 24000)
region_name = payload.get("region") or ""
tat_fallback_env_files = payload.get("tatFallbackEnvFiles") or []
tat_fallback_timeout = int(payload.get("tatFallbackTimeout") or 120)
tat_poll_seconds = int(payload.get("tatPollSeconds") or 2)

ssh_options = [
    "-o", "BatchMode=yes",
    "-o", "ConnectTimeout=5",
    "-o", "ServerAliveInterval=5",
    "-o", "ServerAliveCountMax=1",
    "-o", "StrictHostKeyChecking=accept-new",
]

def clean_env_value(value):
    return str(value or "").strip().replace("export ", "", 1).strip().strip('"').strip("'")

def load_env_file(path):
    if not path:
        return
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as handle:
            lines = handle.read().splitlines()
    except Exception:
        return
    for raw_line in lines:
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip().replace("export ", "", 1)
        if key and key not in os.environ:
            os.environ[key] = clean_env_value(value)
    if "TENCENTCLOUD_SECRET_ID" not in os.environ and os.environ.get("TENCENT_SECRET_ID"):
        os.environ["TENCENTCLOUD_SECRET_ID"] = os.environ["TENCENT_SECRET_ID"]
    if "TENCENTCLOUD_SECRET_KEY" not in os.environ and os.environ.get("TENCENT_SECRET_KEY"):
        os.environ["TENCENTCLOUD_SECRET_KEY"] = os.environ["TENCENT_SECRET_KEY"]

def cloud_secret_id():
    return os.environ.get("TENCENTCLOUD_SECRET_ID") or os.environ.get("TENCENT_SECRET_ID") or ""

def cloud_secret_key():
    return os.environ.get("TENCENTCLOUD_SECRET_KEY") or os.environ.get("TENCENT_SECRET_KEY") or ""

def build_tat_client():
    for env_file in tat_fallback_env_files:
        load_env_file(env_file)
    sid = cloud_secret_id()
    skey = cloud_secret_key()
    missing = []
    if not sid:
        missing.append("TENCENTCLOUD_SECRET_ID")
    if not skey:
        missing.append("TENCENTCLOUD_SECRET_KEY")
    if not region_name:
        missing.append("TENCENTCLOUD_REGION")
    if missing:
        raise RuntimeError("missing Tencent Cloud credentials on deploy machine: " + ", ".join(missing))
    from tencentcloud.common import credential
    from tencentcloud.common.profile.client_profile import ClientProfile
    from tencentcloud.common.profile.http_profile import HttpProfile
    from tencentcloud.tat.v20201028 import tat_client
    return tat_client.TatClient(
        credential.Credential(sid, skey),
        region_name,
        ClientProfile(httpProfile=HttpProfile(endpoint="tat.tencentcloudapi.com")),
    )

def invoke_tencent(action, label, retries=3):
    try:
        from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
    except Exception:
        TencentCloudSDKException = Exception
    last_error = None
    for attempt in range(1, retries + 1):
        try:
            return action()
        except TencentCloudSDKException as exc:
            last_error = exc
            if attempt == retries:
                break
            time.sleep(2 * attempt)
    raise RuntimeError(f"{{label}} failed: {{last_error}}") from last_error

def decode_tat_output(output):
    if not output:
        return ""
    return base64.b64decode(output).decode("utf-8", errors="replace")

def run_tat_shell(instance_id, script, command_name):
    from tencentcloud.tat.v20201028 import models as tat_models

    client = build_tat_client()
    request = tat_models.RunCommandRequest()
    request.from_json_string(
        json.dumps(
            {{
                "CommandName": command_name,
                "CommandType": "SHELL",
                "Content": base64.b64encode(script.encode("utf-8")).decode("ascii"),
                "InstanceIds": [instance_id],
                "Timeout": tat_fallback_timeout,
                "WorkingDirectory": "/root",
            }}
        )
    )
    response = json.loads(invoke_tencent(lambda: client.RunCommand(request), "RunCommand").to_json_string())
    invocation_id = response["InvocationId"]
    deadline = time.time() + tat_fallback_timeout + 60
    while time.time() < deadline:
        time.sleep(max(tat_poll_seconds, 1))
        query = tat_models.DescribeInvocationTasksRequest()
        query.from_json_string(
            json.dumps(
                {{
                    "HideOutput": False,
                    "Limit": 50,
                    "Filters": [{{"Name": "invocation-id", "Values": [invocation_id]}}],
                }}
            )
        )
        payload = json.loads(invoke_tencent(lambda: client.DescribeInvocationTasks(query), "DescribeInvocationTasks").to_json_string())
        for task in payload.get("InvocationTaskSet") or []:
            if str(task.get("InstanceId") or "") != instance_id:
                continue
            status = str(task.get("TaskStatus") or "")
            if status in {{"SUCCESS", "FAILED", "TIMEOUT", "CANCELLED", "START_FAILED"}}:
                return {{
                    "status": status,
                    "output": decode_tat_output(str((task.get("TaskResult") or {{}}).get("Output") or "")),
                }}
    return {{"status": "TIMEOUT", "output": ""}}

remote_reader = r'''
import json
import os
import subprocess
import sys

payload = json.load(sys.stdin)
max_bytes = int(payload.get("maxBytes") or 24000)

def read_path(path):
    item = {{"content": "", "exists": False, "path": path, "truncated": False}}
    if not path:
        return item
    item["exists"] = os.path.exists(path)
    if item["exists"]:
        with open(path, "r", encoding="utf-8", errors="replace") as handle:
            content = handle.read(max_bytes + 1)
        item["truncated"] = len(content) > max_bytes
        item["content"] = content[:max_bytes]
    return item

def command(args):
    try:
        return subprocess.check_output(args, text=True, stderr=subprocess.STDOUT, timeout=5).strip()
    except Exception as exc:
        return str(exc)

items = []
for service in payload.get("services") or []:
    seen = []
    paths = []
    for path in [service.get("configPath") or "", service.get("envFile") or ""]:
        if path and path not in seen:
            seen.append(path)
            paths.append(read_path(path))
    items.append({{
        "name": service.get("name") or "",
        "paths": paths,
        "unit": service.get("unit") or "",
        "unitState": command(["systemctl", "is-active", service.get("unit") or ""]) if service.get("unit") else "",
    }})

print(json.dumps({{"ok": True, "services": items}}, ensure_ascii=False))
'''

def run_tat_fallback(host_name, host, host_services, reason):
    instance_id = str(host.get("instance_id") or "").strip()
    private_ip = str(host.get("private_ip") or "").strip()
    if not instance_id:
        return {{
            "error": "ssh failed and instance_id missing: " + reason,
            "host": host_name,
            "privateIp": private_ip,
            "services": [],
            "status": "TAT_FALLBACK_UNAVAILABLE",
            "transport": "tat-fallback",
        }}
    remote_payload = {{
        "maxBytes": max_bytes,
        "services": host_services,
    }}
    reader_blob = base64.b64encode(remote_reader.encode("utf-8")).decode("ascii")
    payload_blob = base64.b64encode(json.dumps(remote_payload).encode("utf-8")).decode("ascii")
    script = '''set -euo pipefail
python3 - <<'FALLBACKPY'
import base64
import subprocess
import sys

code = base64.b64decode(%r).decode("utf-8")
payload_text = base64.b64decode(%r).decode("utf-8")
completed = subprocess.run(
    [sys.executable, "-c", code],
    input=payload_text,
    text=True,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    timeout=30,
    check=False,
)
sys.stdout.write(completed.stdout)
raise SystemExit(completed.returncode)
FALLBACKPY
''' % (reader_blob, payload_blob)
    try:
        result = run_tat_shell(instance_id, script, "hyapp-data-resource-fallback-" + host_name)
    except Exception as exc:
        return {{
            "error": "ssh failed: " + reason + "; tat fallback failed: " + str(exc),
            "host": host_name,
            "privateIp": private_ip,
            "services": [],
            "status": "TAT_FALLBACK_ERROR",
            "transport": "tat-fallback",
        }}
    if result["status"] != "SUCCESS":
        return {{
            "error": "ssh failed: " + reason + "; tat fallback status: " + result["status"],
            "host": host_name,
            "privateIp": private_ip,
            "services": [],
            "status": "TAT_FALLBACK_ERROR",
            "transport": "tat-fallback",
        }}
    try:
        decoded = json.loads(result.get("output") or "{{}}")
    except Exception as exc:
        return {{
            "error": "ssh failed: " + reason + "; invalid tat fallback json: " + str(exc),
            "host": host_name,
            "privateIp": private_ip,
            "services": [],
            "status": "TAT_FALLBACK_ERROR",
            "transport": "tat-fallback",
        }}
    return {{
        "fallbackReason": reason[-500:],
        "host": host_name,
        "privateIp": private_ip,
        "services": decoded.get("services") or [],
        "status": "TAT_FALLBACK_SUCCESS",
        "transport": "tat-fallback",
    }}

def run_ssh(host_name, host, host_services):
    private_ip = str(host.get("private_ip") or "").strip()
    if not private_ip:
        return {{
            "error": "private_ip missing",
            "host": host_name,
            "privateIp": "",
            "services": [],
            "status": "ERROR",
            "transport": "ssh",
        }}
    ssh_host = str(host.get("ssh_user") or ssh_user) + "@" + private_ip
    remote_payload = {{
        "maxBytes": max_bytes,
        "services": host_services,
    }}
    try:
        completed = subprocess.run(
            ["ssh", *ssh_options, ssh_host, "python3", "-c", remote_reader],
            input=json.dumps(remote_payload),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            timeout=20,
            check=False,
        )
    except Exception as exc:
        return run_tat_fallback(host_name, host, host_services, str(exc))
    if completed.returncode != 0:
        return run_tat_fallback(host_name, host, host_services, completed.stdout[-1000:])
    try:
        decoded = json.loads(completed.stdout)
    except Exception as exc:
        return run_tat_fallback(host_name, host, host_services, f"invalid ssh json: {{exc}}")
    return {{
        "host": host_name,
        "privateIp": private_ip,
        "services": decoded.get("services") or [],
        "status": "SUCCESS",
        "transport": "ssh",
    }}

def endpoint_from_host_port(host, port):
    host = str(host or "").strip().strip('"').strip("'")
    port = str(port or "").strip()
    if not host:
        return ""
    return host + ((":" + port) if port else "")

def normalize_endpoint(value, default_port=""):
    text = str(value or "").strip().strip('"').strip("'")
    if not text:
        return ""
    tcp_match = re.search(r"@tcp\\(([^)]+)\\)", text)
    if tcp_match:
        text = tcp_match.group(1)
    if "://" in text:
        text = text.split("://", 1)[1].split("/", 1)[0]
    text = text.split(",", 1)[0].strip()
    if not text:
        return ""
    if ":" not in text and default_port:
        text = text + ":" + default_port
    return text

def endpoint_candidates(line, default_port=""):
    found = []
    for host, port in re.findall(r"((?:\\d{{1,3}}\\.){{3}}\\d{{1,3}}|[A-Za-z0-9][A-Za-z0-9.-]*):([0-9]{{2,5}})", line):
        endpoint = endpoint_from_host_port(host, port)
        if endpoint:
            found.append(endpoint)
    if not found:
        endpoint = normalize_endpoint(line.split("=", 1)[-1].split(":", 1)[-1], default_port)
        if endpoint and re.search(r"[A-Za-z0-9.:-]", endpoint):
            found.append(endpoint)
    return found

def extract_config_observations(text):
    observations = {{
        "mq": {{"endpoints": set(), "services": set(), "values": set()}},
        "mysql": {{"databases": set(), "endpoints": set(), "services": set(), "values": set()}},
        "redis": {{"endpoints": set(), "services": set(), "values": set()}},
    }}

    for match in re.finditer(r"@tcp\\(([^)]+)\\)/([^?\\s\\\"']+)", text):
        endpoint = normalize_endpoint(match.group(1), "3306")
        if endpoint:
            observations["mysql"]["endpoints"].add(endpoint)
            observations["mysql"]["databases"].add(match.group(2))
            observations["mysql"]["values"].add(endpoint + "/" + match.group(2))

    section = ""
    for raw_line in text.splitlines():
        indent = len(raw_line) - len(raw_line.lstrip())
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        section_match = re.match(r"^([A-Za-z0-9_.-]+):\\s*$", line)
        if section_match:
            key = section_match.group(1).lower()
            if indent == 0 or key in ("redis", "rocketmq"):
                section = key
            continue
        lower = line.lower()
        if "mysql" in lower and ("dsn" in lower or "@tcp(" in lower):
            endpoint = normalize_endpoint(line, "3306")
            if endpoint:
                observations["mysql"]["endpoints"].add(endpoint)
        if "redis" in lower and ("addr" in lower or "host" in lower or "endpoint" in lower):
            for endpoint in endpoint_candidates(line, "6379"):
                observations["redis"]["endpoints"].add(endpoint)
        if section == "redis" and re.match(r"^(addr|host|endpoint)\\s*:", lower):
            for endpoint in endpoint_candidates(line, "6379"):
                observations["redis"]["endpoints"].add(endpoint)
        if "rocketmq" in lower or "name_server" in lower or "nameserver" in lower or section == "rocketmq":
            if re.search(r"((?:\\d{{1,3}}\\.){{3}}\\d{{1,3}}|[A-Za-z0-9][A-Za-z0-9.-]*):([0-9]{{2,5}})", line) or any(
                key in lower for key in ("addr", "endpoint", "host", "name_server", "nameserver")
            ):
                for endpoint in endpoint_candidates(line, "9876"):
                    observations["mq"]["endpoints"].add(endpoint)
            if "topic" in lower or "group" in lower:
                observations["mq"]["values"].add(line[:160])

    return observations

def merge_observation(target, source, service_name):
    for resource_id, fields in source.items():
        target[resource_id]["services"].add(service_name)
        for key, values in fields.items():
            if key == "services":
                continue
            target[resource_id].setdefault(key, set()).update(values)

def tcp_check(endpoint):
    host, sep, raw_port = endpoint.rpartition(":")
    if not sep:
        return {{"detail": "missing port", "endpoint": endpoint, "status": "unknown"}}
    try:
        port = int(raw_port)
    except ValueError:
        return {{"detail": "invalid port", "endpoint": endpoint, "status": "unknown"}}
    started = time.time()
    try:
        with socket.create_connection((host, port), timeout=4):
            pass
        return {{
            "detail": "connect ok",
            "endpoint": endpoint,
            "latencyMs": round((time.time() - started) * 1000, 1),
            "status": "success",
        }}
    except Exception as exc:
        return {{
            "detail": str(exc),
            "endpoint": endpoint,
            "status": "failed",
        }}

services_by_host = {{}}
for service in services:
    for host_name in service.get("hosts") or []:
        services_by_host.setdefault(host_name, []).append(service)

host_results = []
for host_name in services_by_host:
    host = hosts.get(host_name) or {{}}
    host_results.append(run_ssh(host_name, host, services_by_host[host_name]))

observations = {{
    "mq": {{"endpoints": set(), "services": set(), "values": set()}},
    "mysql": {{"databases": set(), "endpoints": set(), "services": set(), "values": set()}},
    "redis": {{"endpoints": set(), "services": set(), "values": set()}},
}}

for host_result in host_results:
    if host_result["status"] not in ("SUCCESS", "TAT_FALLBACK_SUCCESS"):
        continue
    for service in host_result.get("services") or []:
        text_parts = []
        for path in service.get("paths") or []:
            if path.get("exists"):
                text_parts.append(str(path.get("content") or ""))
        service_observations = extract_config_observations("\\n".join(text_parts))
        merge_observation(observations, service_observations, str(service.get("name") or ""))

resources = []
successful_sources = [item for item in host_results if item["status"] in ("SUCCESS", "TAT_FALLBACK_SUCCESS")]
for resource in data_resources:
    resource_id = str(resource.get("id") or "")
    resource_type = str(resource.get("type") or resource_id)
    observation_key = "mq" if resource_type in ("mq", "rocketmq", "kafka") else resource_type
    observed = observations.get(observation_key, {{"endpoints": set(), "services": set(), "values": set()}})
    endpoints = set(str(item) for item in observed.get("endpoints", set()) if str(item).strip())
    for endpoint in resource.get("endpoints") or []:
        normalized = normalize_endpoint(endpoint)
        if normalized:
            endpoints.add(normalized)
    checks = []
    for endpoint in sorted(endpoints)[:12]:
        checks.append(tcp_check(endpoint))
    if not checks and resource.get("endpoints"):
        checks.append({{"detail": "endpoint 格式无法识别", "status": "unknown"}})

    if not successful_sources:
        status = "critical"
    elif checks and all(check["status"] == "success" for check in checks):
        status = "healthy"
    elif any(check["status"] == "success" for check in checks):
        status = "warning"
    elif checks:
        status = "critical"
    else:
        status = "unknown"

    related_services = sorted(observed.get("services", set())) or resource.get("relatedServices") or []
    databases = sorted(observed.get("databases", set())) if observation_key == "mysql" else []
    resources.append({{
        "accessPath": "TAT -> deploy；服务器配置 SSH/TAT 兜底",
        "backupStatus": resource.get("backupStatus") or "未采集",
        "capacityUsedPercent": resource.get("capacityUsedPercent"),
        "checks": checks,
        "connections": resource.get("connections"),
        "description": resource.get("description") or "",
        "endpoints": sorted(endpoints),
        "id": resource_id,
        "name": resource.get("name") or resource_id,
        "observedDatabases": databases,
        "owner": resource.get("owner") or "后端平台",
        "relatedServices": related_services,
        "slowQueryCount": resource.get("slowQueryCount"),
        "sourceHosts": [
            {{
                "error": item.get("error") or "",
                "fallbackReason": item.get("fallbackReason") or "",
                "host": item["host"],
                "privateIp": item.get("privateIp") or "",
                "serviceCount": len(item.get("services") or []),
                "status": item["status"],
                "transport": item.get("transport") or "",
            }}
            for item in host_results
        ],
        "status": status,
        "topology": resource.get("topology") or "托管资源",
        "type": resource_type,
    }})

summary = {{
    "critical": len([item for item in resources if item["status"] == "critical"]),
    "healthy": len([item for item in resources if item["status"] == "healthy"]),
    "unknown": len([item for item in resources if item["status"] == "unknown"]),
    "warning": len([item for item in resources if item["status"] == "warning"]),
}}
print(json.dumps({{
    "lastSyncedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    "ok": True,
    "resources": resources,
    "sourceHosts": [
        {{
            "error": item.get("error") or "",
            "fallbackReason": item.get("fallbackReason") or "",
            "host": item["host"],
            "privateIp": item.get("privateIp") or "",
            "serviceCount": len(item.get("services") or []),
            "status": item["status"],
            "transport": item.get("transport") or "",
        }}
        for item in host_results
    ],
    "summary": summary,
}}, ensure_ascii=False))
PY
"""


def target_payload(inventory: dict[str, Any]) -> dict[str, Any]:
    deploy_server = inventory["deploy_server"]
    return {
        "instanceId": str(deploy_server.get("instance_id") or ""),
        "name": str(deploy_server.get("name") or "deploy"),
        "privateIp": str(deploy_server.get("private_ip") or ""),
        "publicIp": str(deploy_server.get("public_ip") or ""),
    }


def dry_run_payload(inventory: dict[str, Any], *, max_bytes: int) -> dict[str, Any]:
    return {
        "dryRun": True,
        "ok": True,
        "resources": [
            {
                "accessPath": "TAT -> deploy；服务器配置 SSH/TAT 兜底",
                "backupStatus": resource["backupStatus"],
                "capacityUsedPercent": resource["capacityUsedPercent"],
                "checks": [],
                "connections": resource["connections"],
                "description": resource["description"],
                "endpoints": resource["endpoints"],
                "id": resource["id"],
                "name": resource["name"],
                "observedDatabases": [],
                "owner": resource["owner"],
                "relatedServices": resource["relatedServices"],
                "slowQueryCount": resource["slowQueryCount"],
                "sourceHosts": [],
                "status": "unknown",
                "topology": resource["topology"],
                "type": resource["type"],
            }
            for resource in normalize_data_resources(inventory)
        ],
        "sourceHosts": [],
        "summary": {"critical": 0, "healthy": 0, "unknown": len(normalize_data_resources(inventory)), "warning": 0},
        "target": target_payload(inventory),
        "tat": {"status": "DRY_RUN"},
        "maxBytes": max_bytes,
    }


def read_resources(args: argparse.Namespace) -> dict[str, Any]:
    inventory = load_inventory(args.inventory)
    if args.dry_run:
        return dry_run_payload(inventory, max_bytes=args.max_bytes)

    deploy_server = inventory["deploy_server"]
    deploy_instance_id = str(deploy_server.get("instance_id") or "").strip()
    client = build_tat_client(inventory)
    result = run_tat_shell(
        client,
        instance_id=deploy_instance_id,
        script=remote_collect_script(build_remote_payload(inventory, max_bytes=args.max_bytes)),
        command_name="hyapp-data-resources",
        timeout_seconds=args.timeout_seconds,
        poll_seconds=args.poll_seconds,
    )
    tat = {"instanceId": deploy_instance_id, "status": result["status"], "target": target_payload(inventory)}
    if result["status"] != "SUCCESS":
        return {"error": "TAT command failed", "ok": False, "output": result.get("output") or "", "resources": [], "tat": tat}
    try:
        payload = json.loads(str(result.get("output") or "{}"))
    except json.JSONDecodeError:
        return {"error": "invalid TAT json output", "ok": False, "output": result.get("output") or "", "resources": [], "tat": tat}
    payload["target"] = target_payload(inventory)
    payload["tat"] = tat
    return payload


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Read HYApp data resource status through Tencent TAT on deploy machine.")
    parser.add_argument("action", choices=("resources",))
    parser.add_argument("--dry-run", action="store_true", help="Print resources configured in inventory without calling Tencent APIs.")
    parser.add_argument("--env-file", default="", help="Optional .env file with Tencent Cloud credentials.")
    parser.add_argument("--inventory", default="deploy/tencent_tat/hyapp-services.inventory.prod.example.json")
    parser.add_argument("--max-bytes", type=int, default=24_000)
    parser.add_argument("--poll-seconds", type=int, default=2)
    parser.add_argument("--timeout-seconds", type=int, default=180)
    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()
    try:
        load_env_file(args.env_file)
        payload = read_resources(args)
        print(json.dumps(payload, ensure_ascii=False, indent=2))
        return 0 if payload.get("ok") else 1
    except Exception as exc:  # noqa: BLE001
        print(json.dumps({"error": str(exc), "ok": False}, ensure_ascii=False), file=sys.stderr)
        return 1


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