script-build-and-framework-tmp/visualization/backend/scripts/validate_source_inspector.py

#!/usr/bin/env python3
"""Read-only validator for every visible data-source Inspector in real runs."""

from __future__ import annotations

import argparse
import hashlib
import json
import sys
import urllib.error
import urllib.parse
import urllib.request
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any


HIDDEN_BUSINESS_KEYS = {"technical", "changes"}


def request_json(base_url: str, path: str) -> tuple[int, Any]:
    url = f"{base_url.rstrip('/')}{path}"
    try:
        with urllib.request.urlopen(url, timeout=60) as response:
            return response.status, json.load(response)
    except urllib.error.HTTPError as exc:
        try:
            payload = json.load(exc)
        except Exception:
            payload = {"detail": str(exc)}
        return exc.code, payload


def quote_ref(value: str) -> str:
    return urllib.parse.quote(value, safe="")


def collect_detail_refs(value: Any) -> list[str]:
    found: list[str] = []

    def visit(node: Any) -> None:
        if isinstance(node, dict):
            for key, child in node.items():
                if key == "detailRef" and isinstance(child, str) and child:
                    found.append(child)
                visit(child)
        elif isinstance(node, list):
            for child in node:
                visit(child)

    visit(value)
    return list(dict.fromkeys(found))


def pointer_lookup(value: Any, pointer: str) -> tuple[bool, Any]:
    if not pointer:
        return True, value
    current = value
    for raw in pointer.lstrip("/").split("/"):
        part = raw.replace("~1", "/").replace("~0", "~")
        if isinstance(current, list):
            try:
                current = current[int(part)]
            except (ValueError, IndexError):
                return False, None
        elif isinstance(current, dict) and part in current:
            current = current[part]
        else:
            return False, None
    return True, current


def digest(value: Any) -> str:
    serialized = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
    return hashlib.sha256(serialized.encode("utf-8")).hexdigest()


def validate_binding(binding: dict[str, Any], sources: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    source_id = str(binding.get("sourceId") or "")
    selector = binding.get("selector") or {}
    resolution = binding.get("resolution")
    source = sources.get(source_id)
    if source is None:
        return [f"binding {binding.get('id')} 的 sourceId {source_id} 未返回"]
    if resolution == "resolved" and source.get("resultState") in {"missing", "unavailable"}:
        errors.append(
            f"binding {binding.get('id')} 标记 resolved，但来源状态为 {source.get('resultState')}"
        )
    if selector.get("kind") == "unresolved":
        code = str(selector.get("code") or "")
        reason = str(selector.get("reason") or "")
        if not reason:
            errors.append(f"binding {binding.get('id')} 的 unresolved 原因为空")
        if not code:
            errors.append(f"binding {binding.get('id')} 的 unresolved 缺少原因码")
        if code in {"binding-map-missing", "field-path-mismatch"}:
            errors.append(f"binding {binding.get('id')} 存在可视化映射错误 {code}")
        elif code not in {
            "source-record-not-saved",
            "event-body-not-saved",
            "unsafe-association",
            "reliable-log-anchor-unavailable",
        }:
            errors.append(f"binding {binding.get('id')} 使用了未允许的缺失原因码 {code}")
        if code in {"source-record-not-saved", "event-body-not-saved"} and source.get("resultState") not in {"missing", "unavailable"}:
            errors.append(
                f"binding {binding.get('id')} 声称原始记录未保存，"
                f"但来源状态为 {source.get('resultState')}"
            )
        return errors
    if selector.get("kind") == "json-pointer":
        found, _selected = pointer_lookup(source.get("rawRecord"), selector.get("path") or "")
        if not found:
            errors.append(f"binding {binding.get('id')} 无法反向定位 {selector.get('path')}")
    elif selector.get("kind") == "text-span":
        found, text = pointer_lookup(source.get("rawRecord"), selector.get("path") or "")
        if not found or not isinstance(text, str):
            errors.append(f"binding {binding.get('id')} 的原文字段不存在")
        else:
            characters = list(text)
            start = int(selector.get("startCodePoint") or 0)
            end = int(selector.get("endCodePoint") or 0)
            if "".join(characters[start:end]) != selector.get("exactText"):
                errors.append(f"binding {binding.get('id')} 的原文区间不匹配")
            if digest(text) != selector.get("sourceDigest"):
                errors.append(f"binding {binding.get('id')} 的 sourceDigest 不匹配")
    elif selector.get("kind") == "members":
        members = selector.get("members") or []
        if not members:
            errors.append(f"binding {binding.get('id')} 的聚合成员为空")
        for member in members:
            errors.extend(validate_binding({
                "id": f"{binding.get('id')}:member",
                "sourceId": member.get("sourceId"),
                "selector": member.get("selector"),
                "resolution": "resolved",
            }, sources))
    return errors


def selected_value(binding: dict[str, Any], source: dict[str, Any]) -> tuple[bool, Any]:
    selector = binding.get("selector") or {}
    kind = selector.get("kind")
    raw = source.get("rawRecord")
    if kind == "json-pointer":
        return pointer_lookup(raw, str(selector.get("path") or ""))
    if kind == "whole-record":
        return True, raw
    if kind == "text-span":
        found, text = pointer_lookup(raw, str(selector.get("path") or ""))
        if not found or not isinstance(text, str):
            return False, None
        start = int(selector.get("startCodePoint") or 0)
        end = int(selector.get("endCodePoint") or 0)
        return True, "".join(list(text)[start:end])
    return False, None


def equivalent(left: Any, right: Any) -> bool:
    return json.dumps(left, ensure_ascii=False, sort_keys=True, default=str) == json.dumps(
        right, ensure_ascii=False, sort_keys=True, default=str
    )


def validate_business_value(
    binding: dict[str, Any], source: dict[str, Any], business_value: Any
) -> list[str]:
    """Direct bindings must select the exact value shown in column one."""
    if (binding.get("transform") or {}).get("kind") != "direct":
        return []
    found, selected = selected_value(binding, source)
    if found and not equivalent(selected, business_value):
        return [
            f"binding {binding.get('id')} 声明直接展示，但 selector 值与业务项不一致"
        ]
    return []


def validate_empty_sources(sources: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    for source_id, source in sources.items():
        if source.get("resultState") != "empty":
            continue
        kind = source.get("kind")
        locator = source.get("locator") or {}
        raw = source.get("rawRecord")
        if kind == "database" and not (
            raw == [] and int(locator.get("resultCount") or 0) == 0
        ):
            errors.append(f"空结果来源 {source_id} 无法证明数据库查询返回 0 条")
        if kind == "runtime-event" and not isinstance(raw, dict):
            errors.append(f"空结果来源 {source_id} 没有保存运行 Event")
    return errors


def validate_view(payload: dict[str, Any]) -> tuple[Counter, list[str]]:
    counts: Counter = Counter()
    errors: list[str] = []
    if payload.get("schemaVersion") != "inspector-source-v2":
        return counts, ["schemaVersion 不是 inspector-source-v2"]
    business = payload.get("businessProjection") or {}
    sources = payload.get("sources") or {}
    for source in sources.values():
        counts[f"source:{source.get('resultState') or 'present'}"] += 1
    actual_selectors: list[str] = []
    for module in payload.get("modules") or []:
        bindings = {item.get("id"): item for item in module.get("bindings") or []}
        for row in module.get("rows") or []:
            counts["rows"] += 1
            business_selector = str(row.get("businessSelector") or "")
            actual_selectors.append(business_selector)
            found, business_value = pointer_lookup(business, business_selector)
            if not found:
                errors.append(f"业务行 {row.get('id')} 无法定位 {row.get('businessSelector')}")
            ids = row.get("bindingIds") or []
            if not ids:
                errors.append(f"业务行 {row.get('id')} 没有 binding")
            for binding_id in ids:
                binding = bindings.get(binding_id)
                if binding is None:
                    errors.append(f"业务行 {row.get('id')} 引用了不存在的 binding {binding_id}")
                    continue
                counts[str(binding.get("resolution") or "unknown")] += 1
                errors.extend(validate_binding(binding, sources))
                source = sources.get(str(binding.get("sourceId") or "")) or {}
                errors.extend(validate_business_value(binding, source, business_value))
    errors.extend(validate_log_sources(sources))
    errors.extend(validate_empty_sources(sources))
    expected_selectors = visible_business_selectors(business)
    missing_selectors = [value for value in expected_selectors if value not in actual_selectors]
    if missing_selectors:
        errors.append(f"旧业务 Inspector 的可见项没有生成对照行: {missing_selectors}")
    return counts, errors


def validate_log_sources(sources: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    for source_id, source in sources.items():
        if source.get("kind") != "log-anchor":
            continue
        locator = source.get("locator") or {}
        event_id = locator.get("eventId")
        if event_id is None:
            errors.append(f"日志来源 {source_id} 没有 eventId")
            continue
        event = sources.get(f"event:{event_id}") or {}
        event_raw = event.get("rawRecord") or {}
        expected_msg_id = str(event_raw.get("msg_id") or locator.get("msgId") or "")
        if not expected_msg_id:
            errors.append(f"Event {event_id} 没有 msg_id，不应附带日志来源")
            continue
        raw_log = source.get("rawRecord")
        rendered = raw_log if isinstance(raw_log, str) else json.dumps(raw_log, ensure_ascii=False)
        if f"msg_id={expected_msg_id}" not in rendered and f"msg_id:{expected_msg_id}" not in rendered:
            errors.append(f"日志来源 {source_id} 与 Event {event_id} 的 msg_id 不一致")
    return errors


def visible_business_selectors(projection: dict[str, Any]) -> list[str]:
    """Mirror the readable Inspector's visible leaves, including fixed empty notices."""
    selectors: list[str] = []
    if projection.get("question") not in (None, ""):
        selectors.append("/question")
    for section_index, section in enumerate(projection.get("businessSections") or []):
        if not isinstance(section, dict):
            continue
        root = f"/businessSections/{section_index}"
        if isinstance(section.get("items"), list):
            selectors.extend(f"{root}/items/{index}" for index, _ in enumerate(section["items"]))
        elif section.get("content") is not None:
            selectors.append(f"{root}/content")
    for block_index, block in enumerate(projection.get("blocks") or []):
        if not isinstance(block, dict):
            continue
        root = f"/blocks/{block_index}"
        kind = str(block.get("type") or "summary")
        if kind in {"summary", "notice"}:
            selectors.append(f"{root}/value")
        elif kind == "items":
            selectors.extend(f"{root}/items/{index}" for index, _ in enumerate(block.get("items") or []))
        elif kind == "implementation-plan":
            if block.get("mode") is not None:
                selectors.append(f"{root}/mode")
            for route_index, route in enumerate(block.get("routes") or []):
                if not isinstance(route, dict):
                    continue
                for key in ("target", "method", "action", "emphasis", "pathType"):
                    if route.get(key) is not None:
                        selectors.append(f"{root}/routes/{route_index}/{key}")
            if block.get("reasoning") is not None:
                selectors.append(f"{root}/reasoning")
        elif kind == "creative-process":
            for step_index, step in enumerate(block.get("steps") or []):
                if not isinstance(step, dict):
                    continue
                for key in ("action", "summary", "plan", "reasoning"):
                    if step.get(key) is not None:
                        selectors.append(f"{root}/steps/{step_index}/{key}")
        elif kind == "evaluation-branches":
            for branch_index, branch in enumerate(block.get("branches") or []):
                if not isinstance(branch, dict):
                    continue
                for key in ("conclusion", "achievements", "problems", "evidence"):
                    if branch.get(key) not in (None, [], ""):
                        selectors.append(f"{root}/branches/{branch_index}/{key}")
        elif kind in {"evaluation-matrix", "comparison"}:
            selectors.extend(f"{root}/rows/{index}" for index, _ in enumerate(block.get("rows") or []))
        elif kind == "artifact":
            selectors.extend(f"{root}/refs/{index}" for index, _ in enumerate(block.get("refs") or []))
        else:
            selectors.append(root)
    if projection.get("detailKind") == "retrieval-event":
        conditions = projection.get("queryConditions") or []
        selectors.extend(
            [f"/queryConditions/{index}/value" for index, _ in enumerate(conditions)]
            or ["/queryConditions"]
        )
        outcome = projection.get("outcome") or {}
        selectors.extend(
            f"/outcome/{key}"
            for key in ("message", "count", "errorKind")
            if outcome.get(key) is not None
        )
        selectors.extend(f"/items/{index}" for index, _ in enumerate(projection.get("items") or []))
        completeness = projection.get("completeness") or {}
        if completeness.get("bodyAvailable") is False:
            selectors.append("/completeness/bodyAvailable")
        if completeness.get("truncated"):
            selectors.append(
                "/completeness/omittedCharacters"
                if completeness.get("omittedCharacters") is not None
                else "/completeness/truncated"
            )
    if isinstance(projection.get("activities"), list):
        activities = projection.get("activities") or []
        selectors.extend(
            [f"/activities/{index}" for index, _ in enumerate(activities)]
            or ["/missingActivityNotice" if projection.get("missingActivityNotice") else "/activities"]
        )
    return selectors


def validate_detail(base_url: str, run_id: int, detail_ref: str) -> tuple[str, Counter, list[str], bool]:
    encoded = quote_ref(detail_ref)
    status, payload = request_json(
        base_url, f"/api/script-builds/{run_id}/inspector-view/{encoded}"
    )
    if status == 404:
        return detail_ref, Counter(), [], False
    if status != 200:
        return detail_ref, Counter(), [f"inspector-view HTTP {status}: {payload}"], True
    counts, errors = validate_view(payload)
    old_status, old = request_json(
        base_url, f"/api/script-builds/{run_id}/activities/{encoded}"
    )
    if old_status == 200:
        expected = {key: value for key, value in old.items() if key not in HIDDEN_BUSINESS_KEYS}
        if payload.get("businessProjection") != expected:
            errors.append("第一列与旧业务 Inspector 投影不一致")
    return detail_ref, counts, errors, True


def validate_run(base_url: str, run_id: int, *, workers: int) -> dict[str, Any]:
    status, execution = request_json(base_url, f"/api/script-builds/{run_id}/execution-view")
    if status != 200:
        return {"runId": run_id, "fatal": f"execution-view HTTP {status}: {execution}"}
    refs = collect_detail_refs(execution)
    totals: Counter = Counter()
    failures: list[dict[str, Any]] = []
    checked = 0
    with ThreadPoolExecutor(max_workers=max(1, workers)) as executor:
        futures = {
            executor.submit(validate_detail, base_url, run_id, detail_ref): detail_ref
            for detail_ref in refs
        }
        for future in as_completed(futures):
            detail_ref, counts, errors, did_check = future.result()
            checked += int(did_check)
            totals.update(counts)
            if errors:
                failures.append({"detailRef": detail_ref, "errors": errors})
    failures.sort(key=lambda item: refs.index(item["detailRef"]))
    return {
        "runId": run_id,
        "visibleRefs": len(refs),
        "checkedInspectors": checked,
        "counts": dict(totals),
        "failures": failures,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--base-url", default="http://127.0.0.1:8788")
    parser.add_argument("--runs", type=int, nargs="+", default=[422, 443, 444, 446])
    parser.add_argument("--output")
    parser.add_argument("--workers", type=int, default=8)
    args = parser.parse_args()
    results = [validate_run(args.base_url, run_id, workers=args.workers) for run_id in args.runs]
    report = {"baseUrl": args.base_url, "runs": results}
    rendered = json.dumps(report, ensure_ascii=False, indent=2)
    print(rendered)
    if args.output:
        with open(args.output, "w", encoding="utf-8") as handle:
            handle.write(rendered + "\n")
    return 1 if any(result.get("fatal") or result.get("failures") for result in results) else 0


if __name__ == "__main__":
    sys.exit(main())