#!/usr/bin/env python3 from __future__ import annotations import argparse import json import math import re from collections import Counter from dataclasses import dataclass from datetime import date, datetime, time, timedelta, timezone from pathlib import Path from typing import Any, Iterable from zoneinfo import ZoneInfo import pymysql # Flow-audit error payloads carry record ids / numbers that explode one defect # into many distinct strings. Fold them into a signature before clustering. # These rules mirror scripts/daily_prod_backend_log_report.py (kept as an # independent copy so the two reports stay decoupled). SALESFORCE_ID_RE = re.compile(r"\b[0-9A-Za-z]{15,18}\b") UUID_RE = re.compile(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b") QUOTED_STRING_RE = re.compile(r"'(?:''|[^'])*'") NUMBER_RE = re.compile(r"\b\d+(?:\.\d+)?\b") WHITESPACE_RE = re.compile(r"\s+") FLOW_ERROR_STATUSES = ("error", "failed", "failure", "panic") REPORT_WINDOW_END_TIME = time(hour=15) @dataclass(frozen=True) class ReportWindow: report_date: date timezone_name: str local_start: datetime local_end: datetime utc_start: datetime utc_end: datetime def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Generate production access report from TiDB.") parser.add_argument("--db-host", required=True) parser.add_argument("--db-port", type=int, default=4000) parser.add_argument("--db-user", required=True) parser.add_argument("--db-password", required=True) parser.add_argument("--db-database", required=True) parser.add_argument("--timezone", default="Asia/Shanghai") parser.add_argument("--report-date", default="", help="Asia/Shanghai report window end date in YYYY-MM-DD.") parser.add_argument("--output-json", required=True) parser.add_argument("--output-markdown", required=True) parser.add_argument("--output-feishu-text", required=True) return parser.parse_args() def resolve_report_window(report_date_raw: str, timezone_name: str) -> ReportWindow: tz = ZoneInfo(timezone_name) if report_date_raw: report_date = date.fromisoformat(report_date_raw) else: now = datetime.now(tz) report_date = now.date() if now.timetz() >= REPORT_WINDOW_END_TIME else (now - timedelta(days=1)).date() local_end = datetime.combine(report_date, REPORT_WINDOW_END_TIME, tzinfo=tz) local_start = local_end - timedelta(days=1) utc_start = local_start.astimezone(timezone.utc).replace(tzinfo=None) utc_end = local_end.astimezone(timezone.utc).replace(tzinfo=None) return ReportWindow( report_date=report_date, timezone_name=timezone_name, local_start=local_start, local_end=local_end, utc_start=utc_start, utc_end=utc_end, ) def connect_db(args: argparse.Namespace) -> pymysql.connections.Connection: return pymysql.connect( host=args.db_host, port=args.db_port, user=args.db_user, password=args.db_password, database=args.db_database, ssl={"ssl": {}}, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=True, ) def fetch_one(cursor: pymysql.cursors.Cursor, sql: str, params: Iterable[Any]) -> dict[str, Any]: cursor.execute(sql, tuple(params)) row = cursor.fetchone() return dict(row or {}) def fetch_all(cursor: pymysql.cursors.Cursor, sql: str, params: Iterable[Any]) -> list[dict[str, Any]]: cursor.execute(sql, tuple(params)) return [dict(row) for row in cursor.fetchall()] def compute_percentile(values: list[float], ratio: float) -> float: if not values: return 0.0 ordered = sorted(values) index = max(0, math.ceil(ratio * len(ordered)) - 1) return float(ordered[index]) def safe_int(value: Any) -> int: if value is None: return 0 return int(value) def safe_float(value: Any) -> float: if value is None: return 0.0 return float(value) def build_login_metrics(cursor: pymysql.cursors.Cursor, window: ReportWindow) -> dict[str, Any]: row = fetch_one( cursor, """ SELECT COUNT(*) AS total_count, SUM(CASE WHEN COALESCE(LOWER(`Status`), '') IN ('success', 'succeeded') THEN 1 ELSE 0 END) AS success_count, SUM(CASE WHEN COALESCE(LOWER(`Status`), '') IN ('success', 'succeeded') THEN 0 ELSE 1 END) AS failed_count, COUNT(DISTINCT CASE WHEN COALESCE(LOWER(`Status`), '') IN ('success', 'succeeded') THEN `UserId` END) AS unique_success_users, COUNT(DISTINCT `SourceIp`) AS unique_source_ips FROM `_System_LoginHistory` WHERE `LoginTime` >= %s AND `LoginTime` < %s """, (window.utc_start, window.utc_end), ) top_failures = fetch_all( cursor, """ SELECT COALESCE(NULLIF(`Status`, ''), '(empty)') AS status, COUNT(*) AS count FROM `_System_LoginHistory` WHERE `LoginTime` >= %s AND `LoginTime` < %s AND COALESCE(LOWER(`Status`), '') NOT IN ('success', 'succeeded') GROUP BY COALESCE(NULLIF(`Status`, ''), '(empty)') ORDER BY count DESC, status ASC LIMIT 5 """, (window.utc_start, window.utc_end), ) total = safe_int(row.get("total_count")) failed = safe_int(row.get("failed_count")) return { "total_count": total, "success_count": safe_int(row.get("success_count")), "failed_count": failed, "failure_rate": round((failed / total) * 100, 2) if total else 0.0, "unique_success_users": safe_int(row.get("unique_success_users")), "unique_source_ips": safe_int(row.get("unique_source_ips")), "top_failure_statuses": [ {"status": str(item["status"]), "count": safe_int(item["count"])} for item in top_failures ], } def build_request_metrics(cursor: pymysql.cursors.Cursor, window: ReportWindow) -> dict[str, Any]: overview = fetch_one( cursor, """ SELECT COUNT(*) AS total_count, COUNT(DISTINCT NULLIF(`UserId`, '')) AS unique_users, COUNT(DISTINCT NULLIF(`ClientIp`, '')) AS unique_client_ips, SUM(CASE WHEN `ResponseStatus` < 400 THEN 1 ELSE 0 END) AS success_count, SUM(CASE WHEN `ResponseStatus` >= 400 THEN 1 ELSE 0 END) AS failed_count, SUM(CASE WHEN `ResponseStatus` >= 500 THEN 1 ELSE 0 END) AS error_5xx_count, AVG(`LatencyMs`) AS avg_latency_ms, MAX(`LatencyMs`) AS max_latency_ms FROM `_System_RequestCapture` WHERE `ReceivedAt` >= %s AND `ReceivedAt` < %s """, (window.utc_start, window.utc_end), ) by_entrypoint = fetch_all( cursor, """ SELECT COALESCE(NULLIF(`Entrypoint`, ''), '(empty)') AS entrypoint, COUNT(*) AS count, SUM(CASE WHEN `ResponseStatus` >= 400 THEN 1 ELSE 0 END) AS failed_count FROM `_System_RequestCapture` WHERE `ReceivedAt` >= %s AND `ReceivedAt` < %s GROUP BY COALESCE(NULLIF(`Entrypoint`, ''), '(empty)') ORDER BY count DESC, entrypoint ASC """, (window.utc_start, window.utc_end), ) top_paths = fetch_all( cursor, """ SELECT COALESCE(NULLIF(`Path`, ''), '(empty)') AS path, COUNT(*) AS count, SUM(CASE WHEN `ResponseStatus` >= 400 THEN 1 ELSE 0 END) AS failed_count FROM `_System_RequestCapture` WHERE `ReceivedAt` >= %s AND `ReceivedAt` < %s GROUP BY COALESCE(NULLIF(`Path`, ''), '(empty)') ORDER BY count DESC, path ASC LIMIT 10 """, (window.utc_start, window.utc_end), ) top_failing_paths = fetch_all( cursor, """ SELECT COALESCE(NULLIF(`Path`, ''), '(empty)') AS path, COUNT(*) AS failed_count, AVG(`LatencyMs`) AS avg_latency_ms, MAX(`ResponseStatus`) AS max_status FROM `_System_RequestCapture` WHERE `ReceivedAt` >= %s AND `ReceivedAt` < %s AND `ResponseStatus` >= 400 GROUP BY COALESCE(NULLIF(`Path`, ''), '(empty)') ORDER BY failed_count DESC, path ASC LIMIT 10 """, (window.utc_start, window.utc_end), ) latency_rows = fetch_all( cursor, """ SELECT `LatencyMs` AS latency_ms FROM `_System_RequestCapture` WHERE `ReceivedAt` >= %s AND `ReceivedAt` < %s AND `LatencyMs` IS NOT NULL """, (window.utc_start, window.utc_end), ) latencies = [safe_float(item["latency_ms"]) for item in latency_rows] total = safe_int(overview.get("total_count")) failed = safe_int(overview.get("failed_count")) capture_available = total > 0 capture_note = "" if capture_available else "No _System_RequestCapture rows found in the report window." return { "total_count": total, "capture_available": capture_available, "capture_note": capture_note, "unique_users": safe_int(overview.get("unique_users")), "unique_client_ips": safe_int(overview.get("unique_client_ips")), "success_count": safe_int(overview.get("success_count")), "failed_count": failed, "error_5xx_count": safe_int(overview.get("error_5xx_count")), "failure_rate": round((failed / total) * 100, 2) if total else 0.0, "avg_latency_ms": round(safe_float(overview.get("avg_latency_ms")), 2), "max_latency_ms": round(safe_float(overview.get("max_latency_ms")), 2), "p50_latency_ms": round(compute_percentile(latencies, 0.50), 2), "p95_latency_ms": round(compute_percentile(latencies, 0.95), 2), "entrypoints": [ { "entrypoint": str(item["entrypoint"]), "count": safe_int(item["count"]), "failed_count": safe_int(item["failed_count"]), } for item in by_entrypoint ], "top_paths": [ { "path": str(item["path"]), "count": safe_int(item["count"]), "failed_count": safe_int(item["failed_count"]), } for item in top_paths ], "top_failing_paths": [ { "path": str(item["path"]), "failed_count": safe_int(item["failed_count"]), "avg_latency_ms": round(safe_float(item["avg_latency_ms"]), 2), "max_status": safe_int(item["max_status"]), } for item in top_failing_paths ], } def format_rate(value: float) -> str: return f"{value:.2f}%" def format_latency(value: float) -> str: return f"{value:.2f} ms" def truncate(value: str, limit: int = 200) -> str: if len(value) <= limit: return value return value[: limit - 3] + "..." def normalize_payload(payload: str) -> str: normalized = QUOTED_STRING_RE.sub("'?'", payload or "") normalized = UUID_RE.sub("?", normalized) normalized = SALESFORCE_ID_RE.sub("?", normalized) normalized = NUMBER_RE.sub("?", normalized) normalized = WHITESPACE_RE.sub(" ", normalized).strip() return truncate(normalized, 200) # Deterministic ElementType -> subsystem mapping (Axis A in the triage spec). # ElementType is a flow-engine primitive recorded in the data, so this is a # lookup, not a guess. Unknown primitives fall back to the generic "flow". ELEMENT_TYPE_SUBSYSTEM = { "RecordLookup": "lookup-resolution", "RecordCreate": "persistence-dml", "RecordUpdate": "persistence-dml", "RecordDelete": "persistence-dml", "RecordRollback": "persistence-dml", } def subsystem_for_element_type(element_type: str) -> str: return ELEMENT_TYPE_SUBSYSTEM.get(element_type, "flow") def build_flow_error_metrics(cursor: pymysql.cursors.Cursor, window: ReportWindow) -> dict[str, Any]: """Cluster _System_FlowAuditLog error rows by a normalized signature. Flow / validation-rule / lookup / rollup errors never reach the login or request metrics above, which only see the HTTP layer. Degrades gracefully (available=False) if the table is missing or the query fails — this block must never fail the surrounding report. """ unavailable = { "available": False, "note": "", "flow_execution_count": 0, "error_row_count": 0, "error_rate": 0.0, "affected_records": 0, "affected_flow_defs": 0, "clusters": [], } status_in = ", ".join(["%s"] * len(FLOW_ERROR_STATUSES)) try: overview = fetch_one( cursor, f""" SELECT COUNT(*) AS total_count, SUM(CASE WHEN LOWER(`Status`) IN ({status_in}) OR COALESCE(`ErrorPayload`, '') <> '' THEN 1 ELSE 0 END) AS error_count FROM `_System_FlowAuditLog` WHERE `CreatedDate` >= %s AND `CreatedDate` < %s """, (*FLOW_ERROR_STATUSES, window.utc_start, window.utc_end), ) error_rows = fetch_all( cursor, f""" SELECT `ElementType`, `ErrorPayload`, `FlowDefinitionId`, `RecordId`, `CreatedDate` FROM `_System_FlowAuditLog` WHERE `CreatedDate` >= %s AND `CreatedDate` < %s AND (LOWER(`Status`) IN ({status_in}) OR COALESCE(`ErrorPayload`, '') <> '') """, (window.utc_start, window.utc_end, *FLOW_ERROR_STATUSES), ) except pymysql.Error as exc: note = str(exc).split("\n", 1)[0] return {**unavailable, "note": f"flow audit query failed: {note}"} total = safe_int(overview.get("total_count")) error_count = safe_int(overview.get("error_count")) clusters: dict[str, dict[str, Any]] = {} affected_records: set[str] = set() affected_flow_defs: set[str] = set() for row in error_rows: element_type = str(row.get("ElementType") or "(unknown)") payload = str(row.get("ErrorPayload") or "") signature = f"[{element_type}] {normalize_payload(payload)}" record_id = str(row.get("RecordId") or "") flow_def = str(row.get("FlowDefinitionId") or "") created = row.get("CreatedDate") if record_id: affected_records.add(record_id) if flow_def: affected_flow_defs.add(flow_def) cluster = clusters.get(signature) if cluster is None: cluster = { "signature": signature, "element_type": element_type, "count": 0, "record_ids": set(), "first_seen": created, "last_seen": created, "sample_payload": payload, "sample_record_id": record_id, } clusters[signature] = cluster cluster["count"] += 1 if record_id: cluster["record_ids"].add(record_id) if created is not None: if cluster["first_seen"] is None or created < cluster["first_seen"]: cluster["first_seen"] = created if cluster["last_seen"] is None or created > cluster["last_seen"]: cluster["last_seen"] = created def iso(value: Any) -> str: if isinstance(value, datetime): return value.isoformat() + "Z" return str(value) if value else "" cluster_rows = [ { "signature": cluster["signature"], "element_type": cluster["element_type"], "subsystem": subsystem_for_element_type(cluster["element_type"]), "count": cluster["count"], "affected_records": len(cluster["record_ids"]), "first_seen": iso(cluster["first_seen"]), "last_seen": iso(cluster["last_seen"]), "sample_payload": cluster["sample_payload"], "sample_record_id": cluster["sample_record_id"], } for cluster in clusters.values() ] cluster_rows.sort(key=lambda item: (-item["count"], item["signature"])) return { "available": True, "note": "", "flow_execution_count": total, "error_row_count": error_count, "error_rate": round((error_count / total) * 100, 2) if total else 0.0, "affected_records": len(affected_records), "affected_flow_defs": len(affected_flow_defs), "clusters": cluster_rows[:20], } def format_request_metric(request: dict[str, Any], key: str) -> str: if request["capture_available"]: return str(request[key]) return "N/A" def format_request_latency(request: dict[str, Any], key: str) -> str: if request["capture_available"]: return format_latency(request[key]) return "N/A" def build_markdown(report: dict[str, Any]) -> str: login = report["login_metrics"] request = report["request_metrics"] lines = [ "## Daily Production Access Report", "", f"- report_date: `{report['report_date']}`", f"- timezone: `{report['timezone']}`", f"- local_window: `{report['local_window_start']}` -> `{report['local_window_end']}`", f"- utc_window: `{report['utc_window_start']}` -> `{report['utc_window_end']}`", "", "### Login", "", "| Metric | Value |", "| --- | ---: |", f"| Total login events | {login['total_count']} |", f"| Successful logins | {login['success_count']} |", f"| Failed logins | {login['failed_count']} |", f"| Login failure rate | {format_rate(login['failure_rate'])} |", f"| Unique successful users | {login['unique_success_users']} |", f"| Unique source IPs | {login['unique_source_ips']} |", "", "### Requests", "", "| Metric | Value |", "| --- | ---: |", f"| Total captured requests | {request['total_count']} |", f"| Successful requests | {format_request_metric(request, 'success_count')} |", f"| Failed requests | {format_request_metric(request, 'failed_count')} |", f"| 5xx requests | {format_request_metric(request, 'error_5xx_count')} |", f"| Request failure rate | {format_rate(request['failure_rate']) if request['capture_available'] else 'N/A'} |", f"| Unique request users | {format_request_metric(request, 'unique_users')} |", f"| Unique client IPs | {format_request_metric(request, 'unique_client_ips')} |", f"| Avg latency | {format_request_latency(request, 'avg_latency_ms')} |", f"| P50 latency | {format_request_latency(request, 'p50_latency_ms')} |", f"| P95 latency | {format_request_latency(request, 'p95_latency_ms')} |", f"| Max latency | {format_request_latency(request, 'max_latency_ms')} |", ] if request["capture_note"]: lines.extend(["", f"> request_capture: {request['capture_note']}"]) if request["entrypoints"]: lines.extend( [ "", "### Entrypoints", "", "| Entrypoint | Requests | Failed |", "| --- | ---: | ---: |", ] ) for item in request["entrypoints"]: lines.append(f"| {item['entrypoint']} | {item['count']} | {item['failed_count']} |") if login["top_failure_statuses"]: lines.extend( [ "", "### Login Failure Statuses", "", "| Status | Count |", "| --- | ---: |", ] ) for item in login["top_failure_statuses"]: lines.append(f"| {item['status']} | {item['count']} |") if request["top_paths"]: lines.extend( [ "", "### Top Paths", "", "| Path | Requests | Failed |", "| --- | ---: | ---: |", ] ) for item in request["top_paths"]: lines.append(f"| `{item['path']}` | {item['count']} | {item['failed_count']} |") if request["top_failing_paths"]: lines.extend( [ "", "### Top Failing Paths", "", "| Path | Failed | Avg Latency | Max Status |", "| --- | ---: | ---: | ---: |", ] ) for item in request["top_failing_paths"]: lines.append( f"| `{item['path']}` | {item['failed_count']} | {format_latency(item['avg_latency_ms'])} | {item['max_status']} |" ) flow = report.get("flow_error_metrics") if flow is not None: lines.extend( [ "", "### Flow Errors", "", "| Metric | Value |", "| --- | ---: |", f"| Flow executions | {flow['flow_execution_count']} |", f"| Error rows | {flow['error_row_count']} |", f"| Error rate | {format_rate(flow['error_rate'])} |", f"| Affected records | {flow['affected_records']} |", f"| Affected flow definitions | {flow['affected_flow_defs']} |", ] ) if not flow["available"] and flow["note"]: lines.extend(["", f"> flow_audit: {flow['note']}"]) if flow["clusters"]: lines.extend( [ "", "#### Flow Error Clusters", "", "| Signature | Element | Count | Affected Records | First Seen | Last Seen |", "| --- | --- | ---: | ---: | --- | --- |", ] ) for item in flow["clusters"]: lines.append( f"| `{truncate(item['signature'], 160)}` | {item['element_type']} | {item['count']} " f"| {item['affected_records']} | {item['first_seen']} | {item['last_seen']} |" ) return "\n".join(lines) + "\n" def build_feishu_text(report: dict[str, Any]) -> str: login = report["login_metrics"] request = report["request_metrics"] lines = [ "NexusCRM daily prod access report", f"date: {report['report_date']} ({report['timezone']})", f"login_success_users: {login['unique_success_users']}", f"logins: total={login['total_count']} success={login['success_count']} failed={login['failed_count']} failure_rate={format_rate(login['failure_rate'])}", ] if request["capture_available"]: lines.extend( [ f"requests: total={request['total_count']} failed={request['failed_count']} 5xx={request['error_5xx_count']} failure_rate={format_rate(request['failure_rate'])}", f"request_users: {request['unique_users']}", f"latency: p50={format_latency(request['p50_latency_ms'])} p95={format_latency(request['p95_latency_ms'])} max={format_latency(request['max_latency_ms'])}", ] ) else: lines.extend( [ f"requests: total={request['total_count']} unavailable", f"request_capture: {request['capture_note']}", ] ) flow = report.get("flow_error_metrics") if flow is not None: if not flow["available"]: lines.append(f"flow_errors: unavailable ({flow['note']})") else: lines.append( f"flow_errors: total={flow['error_row_count']} rate={format_rate(flow['error_rate'])} " f"affected_records={flow['affected_records']}" ) if flow["clusters"]: lines.append("top_flow_errors:") for item in flow["clusters"][:5]: lines.append(f"- {truncate(item['signature'], 120)}: count={item['count']}") return "\n".join(lines) def ensure_parent(path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) def main() -> int: args = parse_args() output_json = Path(args.output_json).expanduser().resolve() output_markdown = Path(args.output_markdown).expanduser().resolve() output_feishu = Path(args.output_feishu_text).expanduser().resolve() window = resolve_report_window(args.report_date, args.timezone) with connect_db(args) as conn: with conn.cursor() as cursor: login_metrics = build_login_metrics(cursor, window) request_metrics = build_request_metrics(cursor, window) flow_error_metrics = build_flow_error_metrics(cursor, window) report = { "report_date": window.report_date.isoformat(), "timezone": window.timezone_name, "local_window_start": window.local_start.isoformat(), "local_window_end": window.local_end.isoformat(), "utc_window_start": window.utc_start.isoformat() + "Z", "utc_window_end": window.utc_end.isoformat() + "Z", "login_metrics": login_metrics, "request_metrics": request_metrics, "flow_error_metrics": flow_error_metrics, } markdown = build_markdown(report) feishu_text = build_feishu_text(report) for path in (output_json, output_markdown, output_feishu): ensure_parent(path) output_json.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") output_markdown.write_text(markdown, encoding="utf-8") output_feishu.write_text(feishu_text + "\n", encoding="utf-8") return 0 if __name__ == "__main__": raise SystemExit(main())