X-Financial/server/src/app/services/risk_rule_execution_trace.py

from __future__ import annotations

from typing import Any


def build_risk_rule_execution_trace(
    manifest: dict[str, Any],
    *,
    result: dict[str, Any] | None,
) -> dict[str, Any]:
    evidence = result.get("evidence") if isinstance(result, dict) else {}
    if not isinstance(evidence, dict):
        evidence = {}
    matched = isinstance(result, dict)
    severity = _risk_severity(manifest) if matched else "none"
    steps = _build_condition_steps(manifest, evidence)
    if not steps:
        steps = [_generic_step(manifest, evidence, matched)]
    path_node_ids = ["start", "evidence", *[step["node_id"] for step in steps]]
    path_node_ids.append("hit" if matched else "pass")
    return {
        "matched": matched,
        "risk_level": severity,
        "risk_score": _risk_score(manifest),
        "path_node_ids": _dedupe(path_node_ids),
        "steps": steps,
    }


def _build_condition_steps(manifest: dict[str, Any], evidence: dict[str, Any]) -> list[dict[str, Any]]:
    steps: list[dict[str, Any]] = []
    condition_results = evidence.get("condition_results")
    condition_evidence = evidence.get("conditions")
    if isinstance(condition_results, dict):
        evidence_by_id = {
            str(item.get("id") or ""): item
            for item in condition_evidence
            if isinstance(item, dict)
        } if isinstance(condition_evidence, list) else {}
        for condition_id, passed in condition_results.items():
            item = evidence_by_id.get(str(condition_id), {})
            steps.append(
                {
                    "node_id": str(condition_id),
                    "title": _condition_title(manifest, str(condition_id)),
                    "result": bool(passed),
                    "operator": str(item.get("operator") or ""),
                    "inputs": _compact_inputs(item),
                }
            )
        return steps

    city_consistency = evidence.get("city_consistency")
    if isinstance(city_consistency, dict):
        steps.append(
            {
                "node_id": "city_consistency",
                "title": "城市一致性判断",
                "result": bool(
                    city_consistency.get("unexpected_route_cities")
                    or not _has_overlap(
                        city_consistency.get("attachment_values"),
                        city_consistency.get("reference_values"),
                    )
                ),
                "operator": "route_city_consistency",
                "inputs": {
                    "application_reference_values": city_consistency.get("application_reference_values") or [],
                    "claim_reference_values": city_consistency.get("claim_reference_values") or [],
                    "attachment_values": city_consistency.get("attachment_values") or [],
                    "reference_values": city_consistency.get("reference_values") or [],
                    "unexpected_route_cities": city_consistency.get("unexpected_route_cities") or [],
                    "explanation_hits": city_consistency.get("explanation_hits") or [],
                },
            }
        )
        return steps

    failed_conditions = evidence.get("failed_conditions")
    if isinstance(failed_conditions, list):
        for index, item in enumerate(failed_conditions, start=1):
            if not isinstance(item, dict):
                continue
            steps.append(
                {
                    "node_id": str(item.get("id") or f"condition_{index}"),
                    "title": _condition_title(manifest, str(item.get("id") or f"condition_{index}")),
                    "result": True,
                    "operator": str(item.get("operator") or ""),
                    "inputs": _compact_inputs(item),
                }
            )
    return steps


def _generic_step(
    manifest: dict[str, Any],
    evidence: dict[str, Any],
    matched: bool,
) -> dict[str, Any]:
    params = manifest.get("params") if isinstance(manifest.get("params"), dict) else {}
    return {
        "node_id": "decision",
        "title": "规则判断",
        "result": matched,
        "operator": str(params.get("template_key") or manifest.get("template_key") or ""),
        "inputs": {
            "condition_summary": evidence.get("condition_summary") or params.get("condition_summary") or "",
            "missing_fields": evidence.get("missing_fields") or [],
            "keyword_hits": evidence.get("keyword_hits") or [],
        },
    }


def _condition_title(manifest: dict[str, Any], condition_id: str) -> str:
    params = manifest.get("params") if isinstance(manifest.get("params"), dict) else {}
    conditions = params.get("conditions") if isinstance(params.get("conditions"), list) else []
    for index, condition in enumerate(conditions, start=1):
        if not isinstance(condition, dict):
            continue
        current_id = str(condition.get("id") or f"condition_{index}")
        if current_id == condition_id:
            return str(condition.get("title") or condition.get("operator") or condition_id)
    return condition_id


def _compact_inputs(item: dict[str, Any]) -> dict[str, Any]:
    keys = (
        "fields",
        "left_fields",
        "right_fields",
        "left_values",
        "right_values",
        "values",
        "missing_fields",
        "keyword_hits",
        "dates",
        "range_start",
        "range_end",
        "outside_dates",
    )
    return {key: item.get(key) for key in keys if item.get(key) not in (None, "", [])}


def _risk_severity(manifest: dict[str, Any]) -> str:
    outcomes = manifest.get("outcomes") if isinstance(manifest.get("outcomes"), dict) else {}
    fail = outcomes.get("fail") if isinstance(outcomes.get("fail"), dict) else {}
    return str(fail.get("severity") or "medium")


def _risk_score(manifest: dict[str, Any]) -> int | None:
    metadata = manifest.get("metadata") if isinstance(manifest.get("metadata"), dict) else {}
    outcomes = manifest.get("outcomes") if isinstance(manifest.get("outcomes"), dict) else {}
    fail = outcomes.get("fail") if isinstance(outcomes.get("fail"), dict) else {}
    for value in (fail.get("risk_score"), metadata.get("risk_score")):
        try:
            return int(value)
        except (TypeError, ValueError):
            continue
    return None


def _has_overlap(left: Any, right: Any) -> bool:
    left_set = {str(item).strip().lower() for item in left or [] if str(item).strip()}
    right_set = {str(item).strip().lower() for item in right or [] if str(item).strip()}
    return bool(left_set & right_set)


def _dedupe(values: list[str]) -> list[str]:
    rows: list[str] = []
    for value in values:
        if value and value not in rows:
            rows.append(value)
    return rows