"""
Asset Scanner Job — ADR-090 §4.1 資產盤點 cron
================================================
每 1 小時掃描 K8s + 寫入 asset_inventory + asset_discovery_run + asset_coverage_snapshot.

職責邊界:
  ✅ K8s API 列出全部 namespace 的 pods/deployments/services (shallow scan)
  ✅ UPSERT asset_inventory (asset_key 為 UNIQUE)
  ✅ 為每個 active asset 寫 7 維 asset_coverage_snapshot (預設 unknown,後續 service 補)
  ✅ 完成時寫 automation_operation_log(asset_discovered)
  ❌ 不掃 Prometheus targets / Gitea repos / Docker compose (留下一階段)
  ❌ 不算 capacity 欄位 (留 capacity_scanner_job)

設計鐵律 (參考 ADR-090 §3.4):
  - 同一個 asset 跨 run 沿用同 asset_id (asset_key 為自然鍵)
  - 上次出現但這次沒出現的 asset → lifecycle_state='deprecated' + decommissioned_at
  - run_id 串連 inventory 與 coverage_snapshot,提供完整稽核

排程:
  - 預設每 3600s (1 小時) 跑一次,首次延遲 60s 等其他 service init
  - 由 main.py lifespan asyncio.create_task() 啟動

2026-04-19 ogt + Claude Opus 4.7 (1M context) Asia/Taipei
ADR-090 監控盲區治理 § Phase 7 Asset Inventory Foundation
"""
from __future__ import annotations

import asyncio
import json as _json
import time as _time
from typing import Any

import structlog

logger = structlog.get_logger(__name__)

# ============================================================================
# 排程參數
# ============================================================================
_SCAN_INTERVAL_SEC = 3600       # 每 1 小時
_FIRST_DELAY_SEC = 60           # 啟動後等 60s 再首掃 (其他 service init)
_KUBECTL_TIMEOUT_SEC = 30
_LOOP_BACKOFF_SEC = 300         # 異常時 backoff 5 分鐘

# 7 個自動化覆蓋維度 (ADR-090 §3.5)
_COVERAGE_DIMENSIONS = (
    "auto_monitoring", "auto_alerting", "auto_rule_creation",
    "auto_rule_matching", "auto_playbook", "auto_remediation", "auto_km_creation",
)

# K8s asset_type 對應
_K8S_RESOURCE_TO_ASSET_TYPE = {
    "Pod": "container",
    "Deployment": "k8s_workload",
    "StatefulSet": "k8s_workload",
    "DaemonSet": "k8s_workload",
    "Service": "k8s_resource",
    "ConfigMap": "k8s_resource",
    "Secret": "secret",
}


# ============================================================================
# Public entry — main.py lifespan 呼叫
# ============================================================================

async def run_asset_scanner_loop() -> None:
    """
    永久迴圈：每 _SCAN_INTERVAL_SEC 秒做一次資產盤點。

    錯誤策略:
    - 單次 scan 異常 → backoff 5 分鐘再試,不 crash
    - 連續 5 次失敗 → 寫 ai_governance_event (Phase 6 自我治理) — TODO 後續實作
    """
    logger.info("asset_scanner_loop_started", interval_sec=_SCAN_INTERVAL_SEC)
    await asyncio.sleep(_FIRST_DELAY_SEC)

    while True:
        try:
            await scan_once(triggered_by="cron")
        except Exception as e:
            logger.exception("asset_scanner_loop_error", error=str(e))
            await asyncio.sleep(_LOOP_BACKOFF_SEC)
            continue
        await asyncio.sleep(_SCAN_INTERVAL_SEC)


async def scan_once(
    triggered_by: str = "cron",
    scope: tuple[str, ...] = ("k8s",),
    scan_depth: str = "shallow",
) -> str | None:
    """
    執行一次資產盤點。

    Args:
        triggered_by: cron / ai / human / incident
        scope:        本次掃描範圍標籤 (寫入 asset_discovery_run.scope)
        scan_depth:   shallow (僅 list) / deep (含 describe) / full

    Returns:
        run_id (UUID 字串) 或 None (寫 header 失敗時)
    """
    started_ms = _time.time()
    run_id = await _start_discovery_run(triggered_by, list(scope), scan_depth)
    if not run_id:
        return None

    new_count = 0
    modified_count = 0
    total_count = 0
    error_msg: str | None = None

    try:
        # 2026-04-19 v3 擴充: 多資源類型掃描 + relationship 提取
        # 資源類型: pods (container), deployments/statefulsets/daemonsets (k8s_workload),
        #          services (k8s_resource), nodes (host), configmaps (k8s_resource)
        # 跳過: secrets (awoooi-executor RBAC 不允許 list)
        k8s_assets, relationships = await _collect_all_k8s_assets()
        total_count = len(k8s_assets)

        # UPSERT inventory
        new_count, modified_count = await _upsert_assets(k8s_assets, run_id)

        # 建立 asset_relationship (OwnerReference + Service selector + Pod volumes)
        rel_written = await _upsert_relationships(relationships)

        # 為每個 active asset 寫 7 維 coverage (預設 unknown,後續其他 service 升級為 green/yellow/red)
        await _write_coverage_snapshots(run_id)

        logger.info("asset_scan_relationships_written", run_id=run_id, relationships=rel_written)

    except Exception as e:
        error_msg = f"{type(e).__name__}: {e}"[:1000]
        logger.exception("asset_scan_once_failed", run_id=run_id, error=error_msg)

    duration_ms = int((_time.time() - started_ms) * 1000)
    final_status = "failed" if error_msg else "success"

    await _finish_discovery_run(
        run_id=run_id,
        status=final_status,
        total_assets=total_count,
        new_assets=new_count,
        modified_assets=modified_count,
        duration_ms=duration_ms,
        error=error_msg,
    )

    # ADR-090 § aol 留痕 — asset_discovered 是合法 op_type
    await _log_aol_asset_discovered(
        run_id=run_id,
        triggered_by=triggered_by,
        total=total_count,
        new=new_count,
        modified=modified_count,
        duration_ms=duration_ms,
        status=final_status,
        error=error_msg,
    )

    logger.info(
        "asset_scan_once_done",
        run_id=run_id,
        status=final_status,
        total=total_count,
        new=new_count,
        modified=modified_count,
        duration_ms=duration_ms,
    )
    return run_id


# ============================================================================
# K8s 資產收集
# ============================================================================

async def _fetch_kubectl_json(resource: str, all_namespaces: bool = True) -> dict[str, Any]:
    """
    subprocess 執行 kubectl get <resource> --all-namespaces -o json (或 nodes 不帶 ns).
    回傳 parse 後的 payload dict ({'items': [...]}).
    """
    cmd = ["kubectl", "get", resource, "-o", "json"]
    if all_namespaces:
        cmd.insert(3, "--all-namespaces")
    proc = await asyncio.wait_for(
        asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        ),
        timeout=_KUBECTL_TIMEOUT_SEC,
    )
    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=_KUBECTL_TIMEOUT_SEC)
    if proc.returncode != 0:
        raise RuntimeError(f"kubectl {resource} failed rc={proc.returncode}: {stderr.decode('utf-8', errors='replace')[:300]}")
    try:
        return _json.loads(stdout.decode("utf-8", errors="replace"))
    except _json.JSONDecodeError as e:
        raise RuntimeError(f"kubectl {resource} JSON parse failed: {e}") from e


def _build_pod_asset(item: dict[str, Any]) -> dict[str, Any]:
    """Pod → asset_inventory row (asset_type='container')."""
    meta = item.get("metadata", {}) or {}
    spec = item.get("spec", {}) or {}
    ns = meta.get("namespace") or "default"
    name = meta.get("name") or "unknown"
    node = spec.get("nodeName") or ""
    labels = meta.get("labels", {}) or {}

    tags = []
    for k in ("app", "environment", "system"):
        if labels.get(k):
            tags.append(f"{k if k != 'environment' else 'env'}:{labels[k]}")

    return {
        "asset_key": f"k8s/pod/{ns}/{name}",
        "asset_type": "container",
        "host": node or None,
        "namespace": ns,
        "name": name,
        "metadata": {
            "owner_references": meta.get("ownerReferences", []),
            "labels": labels,
            "phase": (item.get("status", {}) or {}).get("phase", ""),
            "node": node,
            "volumes": [
                {
                    "name": v.get("name"),
                    "configMap": v.get("configMap", {}).get("name"),
                    "secret": v.get("secret", {}).get("secretName"),
                }
                for v in (spec.get("volumes") or [])
                if v.get("configMap") or v.get("secret")
            ],
        },
        "tags": tags,
    }


def _build_workload_asset(item: dict[str, Any], kind: str) -> dict[str, Any]:
    """Deployment/StatefulSet/DaemonSet → asset_inventory row (asset_type='k8s_workload')."""
    meta = item.get("metadata", {}) or {}
    ns = meta.get("namespace") or "default"
    name = meta.get("name") or "unknown"
    labels = meta.get("labels", {}) or {}
    spec = item.get("spec", {}) or {}
    status = item.get("status", {}) or {}

    return {
        "asset_key": f"k8s/{kind.lower()}/{ns}/{name}",
        "asset_type": "k8s_workload",
        "host": None,
        "namespace": ns,
        "name": name,
        "metadata": {
            "kind": kind,
            "labels": labels,
            "replicas": spec.get("replicas"),
            "ready_replicas": status.get("readyReplicas"),
            "selector": (spec.get("selector", {}) or {}).get("matchLabels", {}),
        },
        "tags": [f"kind:{kind}"] + [f"app:{labels['app']}"] if labels.get("app") else [f"kind:{kind}"],
    }


def _build_service_asset(item: dict[str, Any]) -> dict[str, Any]:
    """Service → asset_inventory row (asset_type='k8s_resource')."""
    meta = item.get("metadata", {}) or {}
    ns = meta.get("namespace") or "default"
    name = meta.get("name") or "unknown"
    spec = item.get("spec", {}) or {}

    return {
        "asset_key": f"k8s/service/{ns}/{name}",
        "asset_type": "k8s_resource",
        "host": None,
        "namespace": ns,
        "name": name,
        "metadata": {
            "kind": "Service",
            "type": spec.get("type"),
            "cluster_ip": spec.get("clusterIP"),
            "selector": spec.get("selector", {}),
            "ports": spec.get("ports", []),
        },
        "tags": [f"svc_type:{spec.get('type', '')}"],
    }


def _build_node_asset(item: dict[str, Any]) -> dict[str, Any]:
    """Node → asset_inventory row (asset_type='host')."""
    meta = item.get("metadata", {}) or {}
    name = meta.get("name") or "unknown"
    labels = meta.get("labels", {}) or {}
    status = item.get("status", {}) or {}
    addresses = status.get("addresses", []) or []
    internal_ip = next((a["address"] for a in addresses if a.get("type") == "InternalIP"), "")

    return {
        "asset_key": f"k8s/node/{name}",
        "asset_type": "host",
        "host": name,
        "namespace": None,
        "name": name,
        "metadata": {
            "kind": "Node",
            "internal_ip": internal_ip,
            "labels": labels,
            "capacity": status.get("capacity", {}),
            "conditions": [
                {"type": c.get("type"), "status": c.get("status")}
                for c in status.get("conditions", [])
            ],
        },
        "tags": [f"role:{labels.get('kubernetes.io/role', 'worker')}"],
    }


def _build_configmap_asset(item: dict[str, Any]) -> dict[str, Any]:
    """ConfigMap → asset_inventory row (asset_type='k8s_resource')."""
    meta = item.get("metadata", {}) or {}
    ns = meta.get("namespace") or "default"
    name = meta.get("name") or "unknown"

    return {
        "asset_key": f"k8s/configmap/{ns}/{name}",
        "asset_type": "k8s_resource",
        "host": None,
        "namespace": ns,
        "name": name,
        "metadata": {
            "kind": "ConfigMap",
            "labels": meta.get("labels", {}),
            "keys": list((item.get("data") or {}).keys()),
            "creationTimestamp": meta.get("creationTimestamp"),
        },
        "tags": ["kind:ConfigMap"],
    }


async def _collect_all_k8s_assets() -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
    """
    掃多種 K8s 資源類型 + 提取 relationship.

    Relationships:
      - Pod ─ depends_on ─> Deployment (via ReplicaSet 橋接 owner chain)
      - Pod ─ depends_on ─> StatefulSet/DaemonSet (ownerReferences 直連)
      - Service ─ routes_to ─> Pod (via spec.selector 匹配 Pod.labels)
      - Pod ─ depends_on ─> ConfigMap (via spec.volumes[].configMap.name)

    2026-04-19 ogt + Claude Opus 4.7 v3 bug fix:
      Pod.ownerReferences[0].kind 99% 是 ReplicaSet (Deployment 管 ReplicaSet 管 Pod),
      原 code 跳過 ReplicaSet → Pod→Deployment 全部漏掉.
      修: 先掃 ReplicaSet 建 rs_to_deployment map,Pod 用 rs_name 反查 Deployment.

    回傳 (assets, relationships) tuple.
    """
    assets: list[dict[str, Any]] = []
    relationships: list[dict[str, str]] = []

    # 0. ReplicaSets — 僅作為 Pod→Deployment 橋樑,不寫入 asset_inventory
    rs_to_deployment: dict[str, str] = {}  # "ns/rs_name" -> "deployment_name"
    try:
        payload = await _fetch_kubectl_json("replicasets")
        for item in payload.get("items", []) or []:
            meta = item.get("metadata", {}) or {}
            rs_ns = meta.get("namespace") or "default"
            rs_name = meta.get("name") or ""
            for ref in meta.get("ownerReferences", []) or []:
                if ref.get("kind", "").lower() == "deployment":
                    rs_to_deployment[f"{rs_ns}/{rs_name}"] = ref.get("name", "")
    except Exception as e:
        logger.warning("collect_replicasets_failed", error=str(e))

    # 1. Nodes (不帶 ns)
    try:
        payload = await _fetch_kubectl_json("nodes", all_namespaces=False)
        for item in payload.get("items", []) or []:
            assets.append(_build_node_asset(item))
    except Exception as e:
        logger.warning("collect_nodes_failed", error=str(e))

    # 2. Pods — 主體 + 從 ownerReferences 建 relationship
    pod_by_key: dict[str, dict[str, Any]] = {}
    try:
        payload = await _fetch_kubectl_json("pods")
        for item in payload.get("items", []) or []:
            a = _build_pod_asset(item)
            assets.append(a)
            pod_by_key[a["asset_key"]] = item

            meta = item.get("metadata", {}) or {}
            ns = meta.get("namespace") or "default"
            for ref in meta.get("ownerReferences", []) or []:
                owner_kind = ref.get("kind", "").lower()
                owner_name = ref.get("name", "")
                if not owner_name:
                    continue
                # StatefulSet/DaemonSet 直接 owner Pod,直接建 relationship
                if owner_kind in ("statefulset", "daemonset"):
                    relationships.append({
                        "from_key": a["asset_key"],
                        "to_key": f"k8s/{owner_kind}/{ns}/{owner_name}",
                        "relationship_type": "depends_on",
                    })
                # ReplicaSet 中介: 用 rs_to_deployment map 反查 Deployment
                elif owner_kind == "replicaset":
                    deploy_name = rs_to_deployment.get(f"{ns}/{owner_name}")
                    if deploy_name:
                        relationships.append({
                            "from_key": a["asset_key"],
                            "to_key": f"k8s/deployment/{ns}/{deploy_name}",
                            "relationship_type": "depends_on",
                        })
                # 極少數直接是 Deployment owner (舊版 K8s)
                elif owner_kind == "deployment":
                    relationships.append({
                        "from_key": a["asset_key"],
                        "to_key": f"k8s/deployment/{ns}/{owner_name}",
                        "relationship_type": "depends_on",
                    })

            # Pod volumes → ConfigMap relationship
            for v in (item.get("spec", {}) or {}).get("volumes", []) or []:
                cm = (v.get("configMap") or {}).get("name")
                if cm:
                    relationships.append({
                        "from_key": a["asset_key"],
                        "to_key": f"k8s/configmap/{ns}/{cm}",
                        "relationship_type": "depends_on",
                    })
    except Exception as e:
        logger.warning("collect_pods_failed", error=str(e))

    # 3. Deployments / StatefulSets / DaemonSets
    for kind, resource in (("Deployment", "deployments"), ("StatefulSet", "statefulsets"), ("DaemonSet", "daemonsets")):
        try:
            payload = await _fetch_kubectl_json(resource)
            for item in payload.get("items", []) or []:
                assets.append(_build_workload_asset(item, kind))
        except Exception as e:
            logger.warning(f"collect_{resource}_failed", error=str(e))

    # 4. Services + routes_to Pod (via selector match)
    try:
        payload = await _fetch_kubectl_json("services")
        for item in payload.get("items", []) or []:
            svc = _build_service_asset(item)
            assets.append(svc)

            # 為該 Service 找出匹配的 Pod
            selector = (item.get("spec", {}) or {}).get("selector") or {}
            if not selector:
                continue
            svc_ns = (item.get("metadata", {}) or {}).get("namespace") or "default"
            for pod_key, pod_item in pod_by_key.items():
                if not pod_key.startswith(f"k8s/pod/{svc_ns}/"):
                    continue
                pod_labels = (pod_item.get("metadata", {}) or {}).get("labels", {}) or {}
                # selector 所有 kv 必須都在 pod labels 內
                if all(pod_labels.get(k) == v for k, v in selector.items()):
                    relationships.append({
                        "from_key": svc["asset_key"],
                        "to_key": pod_key,
                        "relationship_type": "routes_to",
                    })
    except Exception as e:
        logger.warning("collect_services_failed", error=str(e))

    # 5. ConfigMaps
    try:
        payload = await _fetch_kubectl_json("configmaps")
        for item in payload.get("items", []) or []:
            assets.append(_build_configmap_asset(item))
    except Exception as e:
        logger.warning("collect_configmaps_failed", error=str(e))

    # 6. Prometheus targets — 補齊 host-install services (110/112/188/125 等非 K8s)
    #    Gap 1 修補 (2026-04-19 audit): 原本 asset_inventory 只涵蓋 K8s,
    #    110 Harbor/Gitea/監控 + 188 PostgreSQL/Redis host-install 全漏
    #    用 Prometheus /api/v1/targets 自動發現全節點服務
    try:
        prom_assets, host_relationships = await _collect_prometheus_targets()
        assets.extend(prom_assets)
        relationships.extend(host_relationships)
    except Exception as e:
        logger.warning("collect_prometheus_targets_failed", error=str(e))

    return assets, relationships


def _is_valid_ipv4(s: str) -> bool:
    """嚴格 IPv4 判斷: 4 段 + 每段 0-255 整數.

    避免 '125' (短名) / 'cadvisor-110' (hostname) 被誤判為 IP.
    """
    if not s or s.count(".") != 3:
        return False
    parts = s.split(".")
    if len(parts) != 4:
        return False
    for p in parts:
        if not p or not p.isdigit():
            return False
        try:
            n = int(p)
        except ValueError:
            return False
        if n < 0 or n > 255:
            return False
    return True


async def _collect_prometheus_targets() -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
    """
    從 Prometheus /api/v1/targets 發現所有被監控的 host-install service + 主機.

    每個 target 建 third_party_service / host_service asset.
    每個 unique IP 建 host asset (若尚未存在).
    target → host 建 depends_on relationship.
    """
    import httpx
    from src.core.config import settings

    assets: list[dict[str, Any]] = []
    relationships: list[dict[str, str]] = []
    seen_hosts: set[str] = set()

    url = f"{settings.PROMETHEUS_URL.rstrip('/')}/api/v1/targets"
    try:
        async with httpx.AsyncClient(timeout=10.0, trust_env=False) as client:
            resp = await client.get(url, params={"state": "active"})
            resp.raise_for_status()
            data = resp.json()
    except Exception as e:
        logger.warning("prometheus_targets_fetch_failed", error=str(e))
        return assets, relationships

    for t in (data.get("data", {}) or {}).get("activeTargets", []) or []:
        labels = t.get("labels", {}) or {}
        instance = labels.get("instance", "")
        job = labels.get("job", "")
        if not instance or not job:
            continue

        # 2026-04-19 Audit 1 修: 嚴格 IPv4 判斷
        # 原 code bug: labels.host="125" (短名) 被 "125".replace(".","").isdigit()=True 誤判 IP
        # 修: 優先從 instance 抽 IP (IP:port 形式或純 IP 無 port),嚴格 4 段 0-255 驗證
        # labels.host 可能是短名不可靠,只信 instance
        instance_host = instance.split(":")[0] if ":" in instance else instance
        host_ip = instance_host if _is_valid_ipv4(instance_host) else ""

        if not host_ip:
            # target instance 不是 IP 形式 → 建 third_party_service asset 但 host 留空
            asset_key = f"prometheus_target/{job}/{instance}"
            assets.append({
                "asset_key": asset_key,
                "asset_type": "third_party_service",
                "host": None,
                "namespace": None,
                "name": f"{job}:{instance}",
                "metadata": {
                    "job": job,
                    "instance": instance,
                    "scrape_url": t.get("scrapeUrl"),
                    "health": t.get("health"),
                    "labels": labels,
                },
                "tags": [f"job:{job}", "source:prometheus_target"],
            })
            continue

        # IP 形式 target — 用 'monitoring_target' (asset_inventory CHECK 允許列表)
        # host_service 不在 ADR-090 asset_type CHECK 內,之前 1 筆 125:32334 scan 拋
        # CheckViolationError (constraint asset_inventory_type_valid)
        asset_key = f"prometheus_target/{job}/{instance}"
        assets.append({
            "asset_key": asset_key,
            "asset_type": "monitoring_target",
            "host": host_ip,
            "namespace": None,
            "name": f"{job}@{host_ip}",
            "metadata": {
                "job": job,
                "instance": instance,
                "scrape_url": t.get("scrapeUrl"),
                "health": t.get("health"),
                "labels": labels,
            },
            "tags": [f"job:{job}", f"host:{host_ip}", "source:prometheus_target"],
        })

        # 對每個 IP 建 host asset (若尚未)
        if host_ip not in seen_hosts:
            seen_hosts.add(host_ip)
            host_key = f"host/{host_ip}"
            assets.append({
                "asset_key": host_key,
                "asset_type": "host",
                "host": host_ip,
                "namespace": None,
                "name": host_ip,
                "metadata": {
                    "discovered_by": "prometheus_targets",
                    "source": "blackbox_icmp_or_node_exporter",
                },
                "tags": [f"ip:{host_ip}", "source:prometheus"],
            })

        # 建 target → host 的 depends_on relationship
        relationships.append({
            "from_key": asset_key,
            "to_key": f"host/{host_ip}",
            "relationship_type": "depends_on",
        })

    logger.info("prometheus_targets_collected", count=len(assets), hosts=len(seen_hosts))
    return assets, relationships


# ============================================================================
# DB 寫入
# ============================================================================

async def _start_discovery_run(
    triggered_by: str,
    scope: list[str],
    scan_depth: str,
) -> str | None:
    """
    寫 asset_discovery_run header (status='running'), 回傳 run_id。

    失敗 → log warning + 回 None,主流程靜默跳過本次 scan。
    """
    try:
        from sqlalchemy import text as _sql
        from src.db.base import get_db_context

        async with get_db_context() as db:
            row = await db.execute(
                _sql("""
                    INSERT INTO asset_discovery_run (
                        triggered_by, scope, scan_depth, status,
                        new_assets, modified_assets, disappeared_assets,
                        tools_used
                    ) VALUES (
                        :tb, :scope, :sd, 'running',
                        0, 0, 0,
                        CAST(:tools AS jsonb)
                    )
                    RETURNING run_id
                """),
                {
                    "tb": triggered_by,
                    "scope": scope,
                    "sd": scan_depth,
                    "tools": _json.dumps({"k8s": "kubectl_get pods --all-namespaces"}),
                },
            )
            run_id = row.scalar()
            return str(run_id) if run_id else None
    except Exception as e:
        logger.warning("asset_discovery_run_start_failed", error=str(e))
        return None


async def _finish_discovery_run(
    run_id: str,
    status: str,
    total_assets: int,
    new_assets: int,
    modified_assets: int,
    duration_ms: int,
    error: str | None,
) -> None:
    try:
        from sqlalchemy import text as _sql
        from src.db.base import get_db_context

        async with get_db_context() as db:
            await db.execute(
                _sql("""
                    UPDATE asset_discovery_run
                    SET status = :st,
                        ended_at = NOW(),
                        total_assets = :total,
                        new_assets = :new,
                        modified_assets = :mod,
                        duration_ms = :dur,
                        error = :err
                    WHERE run_id = CAST(:rid AS uuid)
                """),
                {
                    "st": status,
                    "total": total_assets,
                    "new": new_assets,
                    "mod": modified_assets,
                    "dur": duration_ms,
                    "err": error,
                    "rid": run_id,
                },
            )
    except Exception as e:
        logger.warning("asset_discovery_run_finish_failed", run_id=run_id, error=str(e))


async def _upsert_assets(
    assets: list[dict[str, Any]],
    run_id: str,
) -> tuple[int, int]:
    """
    UPSERT asset_inventory,回傳 (new_count, modified_count)。

    用 ON CONFLICT (asset_key) DO UPDATE 確保 idempotent。
    """
    if not assets:
        return 0, 0

    from sqlalchemy import text as _sql
    from src.db.base import get_db_context

    new_count = 0
    modified_count = 0

    try:
        async with get_db_context() as db:
            for a in assets:
                # xmax = 0 表示 INSERT (新),否則表示 UPDATE (修改)
                row = await db.execute(
                    _sql("""
                        INSERT INTO asset_inventory (
                            asset_key, asset_type, host, namespace, name,
                            metadata, tags, environment, lifecycle_state,
                            first_seen_at, last_seen_at
                        ) VALUES (
                            :ak, :at, :host, :ns, :name,
                            CAST(:md AS jsonb), :tags, 'prod', 'active',
                            NOW(), NOW()
                        )
                        ON CONFLICT (asset_key) DO UPDATE
                          SET last_seen_at = NOW(),
                              host = EXCLUDED.host,
                              metadata = EXCLUDED.metadata,
                              tags = EXCLUDED.tags,
                              lifecycle_state = 'active',
                              updated_at = NOW(),
                              decommissioned_at = NULL
                        RETURNING asset_id, (xmax = 0) AS inserted
                    """),
                    {
                        "ak": a["asset_key"],
                        "at": a["asset_type"],
                        "host": a["host"],
                        "ns": a["namespace"],
                        "name": a["name"],
                        "md": _json.dumps(a["metadata"], ensure_ascii=False),
                        "tags": a["tags"],
                    },
                )
                _, inserted = row.one()
                if inserted:
                    new_count += 1
                else:
                    modified_count += 1
    except Exception as e:
        logger.exception("asset_upsert_failed", run_id=run_id, error=str(e))

    return new_count, modified_count


async def _upsert_relationships(relationships: list[dict[str, str]]) -> int:
    """
    UPSERT asset_relationship (from_asset_id/to_asset_id/relationship_type 為 UNIQUE).

    用 asset_key 查 asset_id 後 INSERT,忽略 asset 不存在的 relationship.
    回傳實際寫入 (新建/更新) 筆數.
    """
    if not relationships:
        return 0

    from sqlalchemy import text as _sql
    from src.db.base import get_db_context

    written = 0
    try:
        async with get_db_context() as db:
            for rel in relationships:
                try:
                    # 用 asset_key → asset_id 解析,同時 UPSERT
                    await db.execute(
                        _sql("""
                            INSERT INTO asset_relationship (
                                from_asset_id, to_asset_id, relationship_type,
                                first_detected_at, last_verified_at, is_active
                            )
                            SELECT a1.asset_id, a2.asset_id, :rt, NOW(), NOW(), true
                            FROM asset_inventory a1, asset_inventory a2
                            WHERE a1.asset_key = :from_key AND a2.asset_key = :to_key
                              AND a1.asset_id <> a2.asset_id
                            ON CONFLICT (from_asset_id, to_asset_id, relationship_type) DO UPDATE
                              SET last_verified_at = NOW(),
                                  is_active = true
                        """),
                        {
                            "from_key": rel["from_key"],
                            "to_key": rel["to_key"],
                            "rt": rel["relationship_type"],
                        },
                    )
                    written += 1
                except Exception as e:
                    logger.debug("relationship_upsert_skipped",
                                 from_key=rel["from_key"], to_key=rel["to_key"], error=str(e))
    except Exception as e:
        logger.warning("relationship_upsert_failed", error=str(e))
    return written


async def _write_coverage_snapshots(run_id: str) -> None:
    """
    為本次 run 中的所有 active asset 寫 7 維 coverage_snapshot (預設 unknown)。

    後續 service (rule_catalog / playbook_extractor / km_writer) 會 UPDATE 對應維度。
    """
    try:
        from sqlalchemy import text as _sql
        from src.db.base import get_db_context

        async with get_db_context() as db:
            # 一次性 INSERT: 取所有 active asset × 7 dimensions
            await db.execute(
                _sql("""
                    INSERT INTO asset_coverage_snapshot (
                        run_id, asset_id, dimension, coverage_status,
                        evidence, detected_by
                    )
                    SELECT
                        CAST(:rid AS uuid),
                        ai.asset_id,
                        d.dimension,
                        'unknown' AS coverage_status,
                        '{}'::jsonb,
                        'asset_scanner' AS detected_by
                    FROM asset_inventory ai
                    CROSS JOIN (
                        VALUES ('auto_monitoring'),('auto_alerting'),('auto_rule_creation'),
                               ('auto_rule_matching'),('auto_playbook'),('auto_remediation'),
                               ('auto_km_creation')
                    ) AS d(dimension)
                    WHERE ai.lifecycle_state = 'active'
                    ON CONFLICT (run_id, asset_id, dimension) DO NOTHING
                """),
                {"rid": run_id},
            )
    except Exception as e:
        logger.warning("asset_coverage_write_failed", run_id=run_id, error=str(e))


async def _log_aol_asset_discovered(
    run_id: str,
    triggered_by: str,
    total: int,
    new: int,
    modified: int,
    duration_ms: int,
    status: str,
    error: str | None,
) -> None:
    """寫 automation_operation_log(asset_discovered)。失敗只 log 不阻塞。"""
    try:
        from sqlalchemy import text as _sql
        from src.db.base import get_db_context

        aol_status = "success" if status == "success" else "failed"
        input_payload = {
            "run_id": run_id,
            "triggered_by": triggered_by,
            "scope": ["k8s"],
            "scan_depth": "shallow",
        }
        output_payload = {
            "run_id": run_id,
            "total_assets": total,
            "new_assets": new,
            "modified_assets": modified,
        }

        async with get_db_context() as db:
            await db.execute(
                _sql("""
                    INSERT INTO automation_operation_log (
                        operation_type, actor, status,
                        input, output, run_id, duration_ms, error, tags
                    ) VALUES (
                        'asset_discovered',
                        'asset_scanner',
                        :st,
                        CAST(:input AS jsonb),
                        CAST(:output AS jsonb),
                        CAST(:rid AS uuid),
                        :dur, :err,
                        :tags
                    )
                """),
                {
                    "st": aol_status,
                    "input": _json.dumps(input_payload, ensure_ascii=False),
                    "output": _json.dumps(output_payload, ensure_ascii=False),
                    "rid": run_id,
                    "dur": duration_ms,
                    "err": (error or "")[:2000] if error else None,
                    "tags": ["asset_scanner", "discovery", "k8s"],
                },
            )
    except Exception as e:
        logger.warning("asset_scanner_aol_write_failed", run_id=run_id, error=str(e))
