qihongkun
a41f61bcf6
将仓库目录从 ai_auto_test 迁至 quality-inspection-platform，统一质量检测平台命名；MCP 环境变量新增 QIP_* 并兼容 AI_TEST_*。 Co-authored-by: Cursor <cursoragent@cursor.com>
127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
"""Build General / Grafana Loki explore links from workflow run context."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any
|
||
from urllib.parse import quote
|
||
|
||
|
||
def _parse_run_time(value: str | None) -> datetime:
|
||
if not value:
|
||
return datetime.now(timezone.utc)
|
||
text = str(value).strip()
|
||
if text.endswith("Z"):
|
||
text = text[:-1] + "+00:00"
|
||
try:
|
||
parsed = datetime.fromisoformat(text)
|
||
except ValueError:
|
||
return datetime.now(timezone.utc)
|
||
if parsed.tzinfo is None:
|
||
parsed = parsed.replace(tzinfo=timezone.utc)
|
||
return parsed.astimezone(timezone.utc)
|
||
|
||
|
||
def _url_path_hint(url: str) -> str:
|
||
text = str(url or "").strip()
|
||
if not text:
|
||
return ""
|
||
if "://" in text:
|
||
try:
|
||
from urllib.parse import urlparse
|
||
|
||
parsed = urlparse(text)
|
||
return parsed.path or text
|
||
except Exception:
|
||
return text
|
||
return text.split("?")[0]
|
||
|
||
|
||
def _extract_http_request(node_result: dict[str, Any] | None) -> dict[str, str]:
|
||
if not isinstance(node_result, dict):
|
||
return {}
|
||
output = node_result.get("output") or {}
|
||
if not isinstance(output, dict):
|
||
return {}
|
||
request = output.get("request") or {}
|
||
if not isinstance(request, dict):
|
||
return {}
|
||
method = str(request.get("method") or "GET").upper()
|
||
url = str(request.get("url") or "")
|
||
return {"method": method, "url": url, "path": _url_path_hint(url)}
|
||
|
||
|
||
def _build_logql(method: str, url_path: str) -> str:
|
||
service_label = os.getenv("LOKI_LABEL_SELECTOR", "").strip()
|
||
if service_label:
|
||
base = service_label
|
||
else:
|
||
service_key = os.getenv("LOKI_SERVICE_LABEL", "app").strip() or "app"
|
||
service_value = os.getenv("LOKI_SERVICE_VALUE", "").strip() or "quality-inspection-platform"
|
||
base = f'{{{service_key}="{service_value}"}}'
|
||
|
||
filters: list[str] = [base]
|
||
if url_path:
|
||
escaped = url_path.replace('"', '\\"')
|
||
filters.append(f'|= "{escaped}"')
|
||
if method and method != "GET":
|
||
filters.append(f'|= "{method}"')
|
||
return " ".join(filters)
|
||
|
||
|
||
def _build_grafana_explore_url(logql: str, time_from: datetime, time_to: datetime) -> str:
|
||
base = (os.getenv("GENERAL_LOKI_EXPLORE_URL") or os.getenv("LOKI_EXPLORE_URL") or "").strip().rstrip("/")
|
||
if not base:
|
||
return ""
|
||
|
||
datasource = os.getenv("LOKI_DATASOURCE", "Loki").strip() or "Loki"
|
||
org_id = os.getenv("LOKI_ORG_ID", "1").strip() or "1"
|
||
from_ms = int(time_from.timestamp() * 1000)
|
||
to_ms = int(time_to.timestamp() * 1000)
|
||
left = json.dumps(
|
||
[str(from_ms), str(to_ms), datasource, {"expr": logql, "refId": "A"}],
|
||
ensure_ascii=False,
|
||
)
|
||
return f"{base}?orgId={quote(org_id)}&left={quote(left)}"
|
||
|
||
|
||
def build_run_loki_link(
|
||
*,
|
||
run_created_at: str | None,
|
||
duration_ms: int = 0,
|
||
node_result: dict[str, Any] | None = None,
|
||
padding_seconds: int | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Return explore URL + LogQL for a single HTTP node execution."""
|
||
|
||
padding = int(padding_seconds or os.getenv("LOKI_TIME_PADDING_SECONDS", "120") or 120)
|
||
started = _parse_run_time(run_created_at)
|
||
ended = started + timedelta(milliseconds=max(duration_ms, 0))
|
||
time_from = started - timedelta(seconds=padding)
|
||
time_to = ended + timedelta(seconds=padding)
|
||
|
||
http = _extract_http_request(node_result)
|
||
method = http.get("method", "")
|
||
url = http.get("url", "")
|
||
path = http.get("path", "")
|
||
logql = _build_logql(method, path)
|
||
explore_url = _build_grafana_explore_url(logql, time_from, time_to)
|
||
enabled = bool(explore_url)
|
||
|
||
hint = "已生成 General Loki 查询链接，将按执行时间窗口与接口路径过滤日志。"
|
||
if not enabled:
|
||
hint = "未配置 GENERAL_LOKI_EXPLORE_URL / LOKI_EXPLORE_URL，请在环境变量中设置 General 探索页地址。"
|
||
|
||
return {
|
||
"enabled": enabled,
|
||
"explore_url": explore_url,
|
||
"logql": logql,
|
||
"time_from": time_from.isoformat(),
|
||
"time_to": time_to.isoformat(),
|
||
"method": method,
|
||
"url": url,
|
||
"hint": hint,
|
||
}
|