#!/usr/bin/env python3 from __future__ import annotations import argparse import json from pathlib import Path from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen HEADER_TEMPLATE = "green" MAX_SOQL_ROWS = 5 MAX_CLUSTER_ROWS = 5 MAX_SOQL_CHARS = 60 MAX_SIGNATURE_CHARS = 60 FEISHU_API = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--app-id", required=True) parser.add_argument("--app-secret", required=True) parser.add_argument("--chat-id", required=True) parser.add_argument("--report-json", required=True) parser.add_argument("--run-url", required=True) return parser.parse_args() def post_json(url: str, payload: dict[str, object], headers: dict[str, str] | None = None) -> dict[str, object]: merged_headers = {"Content-Type": "application/json; charset=utf-8"} if headers: merged_headers.update(headers) request = Request( url, data=json.dumps(payload).encode("utf-8"), headers=merged_headers, method="POST", ) try: with urlopen(request, timeout=30) as response: return json.loads(response.read().decode("utf-8")) except HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"HTTP {exc.code} calling {url}: {body}") from exc except URLError as exc: raise RuntimeError(f"network error calling {url}: {exc}") from exc def fetch_tenant_access_token(app_id: str, app_secret: str) -> str: payload = post_json( "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal", {"app_id": app_id, "app_secret": app_secret}, ) token = str(payload.get("tenant_access_token", "")).strip() if not token: raise RuntimeError(f"missing tenant_access_token in Feishu response: {json.dumps(payload, ensure_ascii=True)}") return token def load_report(path: str) -> dict[str, object]: return json.loads(Path(path).read_text(encoding="utf-8")) def fmt_ms(value: object) -> str: try: return f"{float(value):.0f} ms" except (TypeError, ValueError): return "-" def fmt_pct(value: object) -> str: # report scripts already store rates as percentage numbers (e.g. 0.22 == 0.22%) try: return f"{float(value):.2f}%" except (TypeError, ValueError): return "-" def truncate(value: object, limit: int) -> str: text = str(value if value is not None else "").replace("\n", " ").replace("|", "/").strip() if len(text) > limit: return text[: limit - 1].rstrip() + "…" return text def short_field(key: str, value: object) -> dict[str, object]: return {"is_short": True, "text": {"tag": "lark_md", "content": f"**{key}**\n{value}"}} def render_lines(title: str, lines: list[str], empty_hint: str) -> dict[str, object]: body = "\n".join(lines) if lines else f"_{empty_hint}_" return {"tag": "div", "text": {"tag": "lark_md", "content": f"**{title}**\n{body}"}} def build_slow_log_card(report: dict[str, object], run_url: str) -> dict[str, object]: summary = report.get("summary", {}) if isinstance(report.get("summary"), dict) else {} report_date = report.get("report_date", "-") threshold = fmt_ms(summary.get("slow_threshold_ms")) parsed_total = summary.get("parsed_requests_in_window", 0) slow_total = summary.get("slow_request_count", 0) slow_total = "n/a" if slow_total is None else slow_total slow_rate = fmt_pct(summary.get("slow_request_rate")) inter_total = summary.get("interactive_request_count", 0) inter_slow = summary.get("interactive_slow_request_count", 0) inter_slow = "n/a" if inter_slow is None else inter_slow inter_rate = fmt_pct(summary.get("interactive_slow_request_rate")) avg_lat = fmt_ms(summary.get("avg_latency_ms")) p95 = fmt_ms(summary.get("p95_latency_ms")) max_lat = fmt_ms(summary.get("max_latency_ms")) elements: list[dict[str, object]] = [ { "tag": "div", "fields": [ short_field("报告日期", report_date), short_field("超时阈值", threshold), short_field("超时请求数", f"{slow_total} / {parsed_total}"), short_field("超时率", slow_rate), short_field("交互式超时", f"{inter_slow} / {inter_total}"), short_field("交互式超时率", inter_rate), short_field("平均延迟", avg_lat), short_field("P95 延迟", p95), short_field("最大延迟", max_lat), ], }, {"tag": "hr"}, ] coverage = report.get("source_coverage") if isinstance(coverage, dict): status_counts = summary.get("status_buckets", {}) elements.append(render_lines("HTTP 入口覆盖", [ f"**请求数**：{parsed_total}；**状态桶**：{json.dumps(status_counts, ensure_ascii=False)}", f"**覆盖状态**：{coverage.get('status')}；**耗时样本**：{summary.get('latency_sample_count', 0)} / {parsed_total}", f"**首末请求**：{summary.get('first_request_at')} / {summary.get('last_request_at')}", *[str(reason) for reason in coverage.get("reasons", [])], ], "覆盖不可用")) latency_incomplete = summary.get("latency_available") is False empty_slow = "耗时采集不完整，无法判定超时总量" if latency_incomplete else "本时段无超时请求" buckets = summary.get("slow_latency_buckets", {}) if isinstance(summary.get("slow_latency_buckets"), dict) else {} bucket_lines = [f"**{label}**：{buckets.get(label, 0)} 条" for label in ("5-10s", "10-30s", "30s+")] if latency_incomplete: bucket_lines.insert(0, "仅统计已测得耗时的样本；未测得的请求不计入以下分布。") elements.append(render_lines("已观测延迟分布（5s 起，10s 为超时阈值）", bucket_lines, empty_slow)) categories = summary.get("slow_request_categories", {}) if isinstance(summary.get("slow_request_categories"), dict) else {} cat_lines = [f"**{name}**：{count} 条" for name, count in sorted(categories.items(), key=lambda kv: (-kv[1], kv[0])) if count] elements.append(render_lines("已观测按类型超时", cat_lines, empty_slow)) permission = report.get("permission_latency_profile", {}) if isinstance(report.get("permission_latency_profile"), dict) else {} permission_gate = permission.get("gate", {}) if isinstance(permission.get("gate"), dict) else {} permission_status = str(permission_gate.get("status", "")).strip().upper() if permission: permission_lines = [ f"**Backend 窗口信号**：{permission_status or 'INCONCLUSIVE'}（不是发布验收）", f"**有效样本 / HTTP 对齐**：{permission.get('evaluated_sample_count', permission.get('sample_count', 0))} / " f"{permission.get('correlated_http_count', 0)}", ] for surface in permission.get("surfaces", []) if isinstance(permission.get("surfaces"), list) else []: if not isinstance(surface, dict): continue permission_lines.append( f"**{surface.get('surface', 'unknown')}**：" f"P95 {fmt_ms(surface.get('permission_p95_ms'))} · " f"P99 {fmt_ms(surface.get('permission_p99_ms'))} · " f"HTTP P99 {fmt_ms(surface.get('http_p99_ms'))} · " f"rows P99 {surface.get('field_bulk_rows_p99', 0)} · " f"rejected {surface.get('field_bulk_rejected', 0) if surface.get('field_bulk_rejected_available') else 'n/a'}" ) phases = permission.get("effective_permission_phases", {}) if isinstance(permission.get("effective_permission_phases"), dict) else {} permission_lines.append( f"**effective phases**：matched {phases.get('matched_request_count', 0)} / " f"{phases.get('expected_request_count', 0)} · incomplete {phases.get('incomplete_request_count', 0)} · " f"total P99 {fmt_ms(phases.get('total_p99_ms'))}" ) if permission_gate.get("insufficient_samples"): permission_lines.append(f"**样本不足**：{truncate(permission_gate['insufficient_samples'], MAX_SIGNATURE_CHARS)}") if permission_gate.get("missing_diagnostic_fields"): permission_lines.append(f"**字段缺失**：{truncate(permission_gate['missing_diagnostic_fields'], MAX_SIGNATURE_CHARS)}") for failure in permission_gate.get("failures", []) if isinstance(permission_gate.get("failures"), list) else []: permission_lines.append(f"**失败证据**：{truncate(failure, MAX_SIGNATURE_CHARS)}") elements.append(render_lines("权限读取证据", permission_lines, "无权限读取样本")) soql = report.get("slow_soql", []) if isinstance(report.get("slow_soql"), list) else [] if soql: soql_lines = [ f"{item.get('count', '-')} 次 · P95 {fmt_ms(item.get('p95_latency_ms'))} · {truncate(item.get('pattern') or item.get('soql'), MAX_SOQL_CHARS)}" for item in soql[:MAX_SOQL_ROWS] ] elements.append(render_lines("Top 慢 SOQL", soql_lines, "无")) elements.append({"tag": "div", "text": {"tag": "lark_md", "content": f"[查看完整报告]({run_url})"}}) window_warning = bool(summary.get("window_data_warning")) if window_warning: elements.insert( 0, { "tag": "div", "text": { "tag": "lark_md", "content": ( "⚠️ **窗口数据疑似不完整**：业务请求样本、耗时或日志来源覆盖不足，或存在解析失败。" "请查看报告中的来源和覆盖诊断；下方慢请求数据**不可作为健康判断依据**。" ), }, }, ) return { "config": {"wide_screen_mode": True}, "header": { "template": "red" if window_warning else HEADER_TEMPLATE, "title": {"tag": "plain_text", "content": f"NexusCRM 慢请求日报 {report_date}"}, }, "elements": elements, } def build_access_card(report: dict[str, object], run_url: str) -> dict[str, object]: login = report.get("login_metrics", {}) if isinstance(report.get("login_metrics"), dict) else {} request = report.get("request_metrics", {}) if isinstance(report.get("request_metrics"), dict) else {} flow = report.get("flow_error_metrics", {}) if isinstance(report.get("flow_error_metrics"), dict) else {} report_date = report.get("report_date", "-") capture_available = bool(request.get("capture_available")) req_failure = fmt_pct(request.get("failure_rate")) if capture_available else "request capture 不可用" req_p95 = fmt_ms(request.get("p95_latency_ms")) if capture_available else "-" elements: list[dict[str, object]] = [ { "tag": "div", "fields": [ short_field("报告日期", report_date), short_field("登录", f"{login.get('total_count', 0)}（成功 {login.get('success_count', 0)} / 失败 {login.get('failed_count', 0)}）"), short_field("登录失败率", fmt_pct(login.get("failure_rate"))), short_field("请求失败率", req_failure), short_field("请求 P95", req_p95), short_field("Flow 错误", f"{flow.get('error_row_count', 0)}（{fmt_pct(flow.get('error_rate'))}）"), ], } ] clusters = flow.get("clusters", []) if isinstance(flow.get("clusters"), list) else [] if clusters: cluster_lines = [ f"{c.get('count', '-')} 次 · {truncate(c.get('signature'), MAX_SIGNATURE_CHARS)}" for c in clusters[:MAX_CLUSTER_ROWS] ] elements.append({"tag": "hr"}) elements.append(render_lines(f"Flow 错误聚类（Top {MAX_CLUSTER_ROWS}）", cluster_lines, "无")) elements.append({"tag": "div", "text": {"tag": "lark_md", "content": f"[查看完整报告]({run_url})"}}) return { "config": {"wide_screen_mode": True}, "header": { "template": HEADER_TEMPLATE, "title": {"tag": "plain_text", "content": f"NexusCRM 访问日报 {report_date}"}, }, "elements": elements, } def build_card(report: dict[str, object], run_url: str) -> dict[str, object]: if "summary" in report and "slow_samples" in report: return build_slow_log_card(report, run_url) if "login_metrics" in report: return build_access_card(report, run_url) raise RuntimeError("unrecognized report JSON shape: missing both slow-log and access keys") def send_chat_message(token: str, chat_id: str, card: dict[str, object]) -> None: payload = { "receive_id": chat_id, "msg_type": "interactive", "content": json.dumps(card, ensure_ascii=False), } response = post_json(FEISHU_API, payload, headers={"Authorization": f"Bearer {token}"}) if response.get("code") not in (0, "0", None): raise RuntimeError(f"Feishu send message failed: {json.dumps(response, ensure_ascii=True)}") def main() -> int: args = parse_args() report = load_report(args.report_json) card = build_card(report, args.run_url) token = fetch_tenant_access_token(args.app_id, args.app_secret) send_chat_message(token, args.chat_id, card) return 0 if __name__ == "__main__": raise SystemExit(main())