from __future__ import annotations import argparse import json import re import sys from collections import defaultdict from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen SOURCE = "clinic_health_check" SOURCE_TYPE = "inspection" DETECTOR_CODE = "clinic.health_check.sql_digest_report" DETECTOR_VERSION = "clinic-health-check-ui-extract-v1" METRIC_SCHEMA_VERSION = "clinic.health_check.sql_digest_report/v1" CREATED_BY = "clinic-health-check-adapter" GOVERNANCE_SIGNAL_ANOMALY = "anomaly" GOVERNANCE_SIGNAL_RANKING_OBSERVATION = "ranking_observation" SEVERITY_RANK = { "info": 0, "low": 1, "medium": 2, "high": 3, "critical": 4, } TOP_DIMENSIONS = { "Count": ("top_by_count", "slow_query_frequency", "slowlog_slow_count"), "Total Latency": ( "top_by_total_latency", "high_total_latency", "slowlog_total_latency_s", ), "SUM Total Keys": ( "top_by_total_keys", "large_key_scan", "slowlog_total_keys", ), "SUM Process Keys": ( "top_by_process_keys", "large_key_scan", "slowlog_process_keys", ), "SUM Request Count": ( "top_by_request_count", "high_kv_request_count", "slowlog_request_count", ), } TOP_METRIC_COLUMNS = { "Slow Count": "slowlog_slow_count", "Total Latency (s)": "slowlog_total_latency_s", "Total Keys": "slowlog_total_keys", "Process Keys": "slowlog_process_keys", "Request Count": "slowlog_request_count", "SUM RU": "slowlog_sum_ru", "Max Latency (s)": "slowlog_max_latency_s", } def compact_json(value: Any) -> bytes: return json.dumps( value, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") def clean_text(value: Any) -> str: return str(value or "").strip() def parse_number(value: Any) -> int | float | None: text = clean_text(value).replace(",", "") match = re.search(r"-?\d+(?:\.\d+)?", text) if not match: return None number = float(match.group(0)) return int(number) if number.is_integer() else number def parse_memory_bytes(value: Any) -> int | None: text = clean_text(value).replace(",", "") match = re.search( r"(-?\d+(?:\.\d+)?)\s*(B|KB|KiB|MB|MiB|GB|GiB|TB|TiB)?", text, re.IGNORECASE, ) if not match: return None number = float(match.group(1)) unit = (match.group(2) or "B").lower() multipliers = { "b": 1, "kb": 1_000, "kib": 1_024, "mb": 1_000_000, "mib": 1_048_576, "gb": 1_000_000_000, "gib": 1_073_741_824, "tb": 1_000_000_000_000, "tib": 1_099_511_627_776, } return round(number * multipliers[unit]) def labeled_number(value: Any, label: str) -> int | float | None: text = clean_text(value) patterns = ( rf"{re.escape(label)}\s*:?\s*(-?\d[\d,]*(?:\.\d+)?)", rf"(-?\d[\d,]*(?:\.\d+)?)\s*{re.escape(label)}", ) for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: return parse_number(match.group(1)) return None def normalize_hotspot_severity(value: Any) -> str: severity = clean_text(value).lower() return { "critical": "critical", "major": "high", "high": "high", "warning": "medium", "medium": "medium", "low": "low", "info": "info", }.get(severity, "medium") def table_rows(table: dict[str, Any]) -> list[dict[str, dict[str, Any]]]: headers = [clean_text(value) for value in table.get("headers") or []] rows: list[dict[str, dict[str, Any]]] = [] for raw_row in table.get("rows") or []: if not isinstance(raw_row, list): continue row: dict[str, dict[str, Any]] = {} for index, header in enumerate(headers): raw_cell = raw_row[index] if index < len(raw_row) else {} if isinstance(raw_cell, dict): row[header] = { "text": clean_text(raw_cell.get("text")), "links": [ link for link in raw_cell.get("links") or [] if isinstance(link, dict) and clean_text(link.get("href")) ], } else: row[header] = {"text": clean_text(raw_cell), "links": []} rows.append(row) return rows def cell_text(row: dict[str, dict[str, Any]], name: str) -> str: return clean_text((row.get(name) or {}).get("text")) def row_links(row: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: links: list[dict[str, Any]] = [] seen: set[str] = set() for cell in row.values(): for link in cell.get("links") or []: url = clean_text(link.get("href")) if not url or url in seen: continue seen.add(url) links.append( { "type": "clinic_link", "name": clean_text(link.get("text")) or "Clinic evidence", "url": url, } ) return links def serializable_row(row: dict[str, dict[str, Any]]) -> dict[str, Any]: return { header: { "text": cell.get("text") or "", "links": cell.get("links") or [], } for header, cell in row.items() } def max_metric(metrics: dict[str, Any], key: str, value: int | float | None) -> None: if value is None: return current = metrics.get(key) metrics[key] = value if not isinstance(current, (int, float)) else max(current, value) def add_evidence(target: list[dict[str, Any]], evidence: dict[str, Any]) -> None: fingerprint = json.dumps(evidence, ensure_ascii=False, sort_keys=True) if all( json.dumps(existing, ensure_ascii=False, sort_keys=True) != fingerprint for existing in target ): target.append(evidence) def report_created_at(report: dict[str, Any]) -> str: created_at = clean_text(report.get("created_at") or report.get("to")) if not created_at: raise ValueError(f"report {report.get('id') or ''} has no creation time") return created_at def new_event(report: dict[str, Any], context: dict[str, Any], digest: str) -> dict[str, Any]: report_url = clean_text(report.get("source_url")) evidence_refs: list[dict[str, Any]] = [] if report_url: evidence_refs.append( { "type": "clinic_health_check_report", "report_id": report["id"], "url": report_url, } ) return { "source": SOURCE, "source_type": SOURCE_TYPE, "source_event_id": f"clinic:healthcheck:{report['id']}:sql_digest:{digest}", "source_run_id": report["id"], "org_id": context["org_id"], "project_id": context.get("project_id"), "cluster_id": context["cluster_id"], "sql_digest": digest, "normalized_sql": "", "reported_at": report_created_at(report), "observed_at": report["to"], "time_window_start": report["from"], "time_window_end": report["to"], "detector_code": DETECTOR_CODE, "detector_version": DETECTOR_VERSION, "metric_schema_version": METRIC_SCHEMA_VERSION, "metrics": {}, "dimensions": { "check_names": [], "top_digest_ranks": {}, "cluster": { "name": context.get("cluster_name"), "org_name": context.get("org_name"), "dbaas_org_id": context.get("dbaas_org_id"), "provider": context.get("provider"), "region": context.get("region"), }, "report": context["report"], }, "evidence_refs": evidence_refs, "raw_payload": { "report": context["report"], "extraction": { "source": context["extraction_source"], "synthetic": False, }, "checks": {}, }, "created_by": CREATED_BY, "_symptoms": {}, } def ensure_event( events: dict[str, dict[str, Any]], report: dict[str, Any], context: dict[str, Any], digest: str, ) -> dict[str, Any]: if digest not in events: events[digest] = new_event(report, context, digest) return events[digest] def add_check_name(event: dict[str, Any], check_name: str) -> None: check_names = event["dimensions"]["check_names"] if check_name not in check_names: check_names.append(check_name) def set_normalized_sql(event: dict[str, Any], sql: str) -> None: if sql and not event["normalized_sql"]: event["normalized_sql"] = sql def add_raw_check( event: dict[str, Any], check_name: str, prefix: str, row: dict[str, dict[str, Any]], **metadata: Any, ) -> None: check = event["raw_payload"]["checks"].setdefault( check_name, {"report_summary": prefix, "rows": []}, ) check["rows"].append({"data": serializable_row(row), **metadata}) def add_symptom( event: dict[str, Any], *, code: str, severity: str, signal_code: str, metric_keys: list[str], assessment: dict[str, Any], evidence: dict[str, Any], ) -> None: symptoms = event["_symptoms"] symptom = symptoms.get(code) if symptom is None: symptom = { "code": code, "severity": severity, "signal_codes": [], "metric_keys": [], "assessment": assessment, "evidence_refs": [], } symptoms[code] = symptom elif SEVERITY_RANK[severity] > SEVERITY_RANK[symptom["severity"]]: symptom["severity"] = severity symptom["assessment"] = assessment if signal_code not in symptom["signal_codes"]: symptom["signal_codes"].append(signal_code) for key in metric_keys: if key in event["metrics"] and key not in symptom["metric_keys"]: symptom["metric_keys"].append(key) add_evidence(symptom["evidence_refs"], evidence) def add_extreme_slow_query( events: dict[str, dict[str, Any]], report: dict[str, Any], context: dict[str, Any], ) -> None: section = report["sections"]["extreme_slow_query"] check_name = "slow_query_huge_resource_usage" for row in table_rows(section["table"]): digest = cell_text(row, "Digest") if not digest: continue event = ensure_event(events, report, context, digest) add_check_name(event, check_name) set_normalized_sql(event, cell_text(row, "Sample Query")) values = { "extreme_hit_records": parse_number(cell_text(row, "Hit Records")), "extreme_avg_query_time_s": parse_number( cell_text(row, "Avg Query Time (s)") ), "extreme_max_query_time_s": parse_number( cell_text(row, "Max Query Time (s)") ), "extreme_hit_query_time_records": parse_number( cell_text(row, "Hit Time Records") ), "extreme_avg_max_mem_bytes": parse_memory_bytes( cell_text(row, "Avg Max Mem") ), "extreme_max_max_mem_bytes": parse_memory_bytes( cell_text(row, "Max Max Mem") ), "extreme_hit_mem_records": parse_number( cell_text(row, "Hit Mem Records") ), } for key, value in values.items(): max_metric(event["metrics"], key, value) evidence = { "type": "clinic_health_check_result", "check_name": check_name, "hit_records": values["extreme_hit_records"], "hit_query_time_records": values["extreme_hit_query_time_records"], "hit_mem_records": values["extreme_hit_mem_records"], } for link in row_links(row): add_evidence(event["evidence_refs"], link) add_symptom( event, code="high_total_latency", severity="high", signal_code="slow_query.extreme_threshold_hit", metric_keys=list(values), assessment={ "method": "external", "summary": ( "Clinic returned this digest from the Extreme Slow Query check. " "The adapter maps a threshold hit to High; it does not promote it " "to Critical without an explicit source severity." ), "rule_code": "clinic.slow_query_huge_resource_usage", "rule_version": DETECTOR_VERSION, "policy_id": "clinic-health-check-report", "parameters": { "source_severity": "not_provided", "adapter_mapping": "extreme_hit_to_high", }, }, evidence=evidence, ) add_raw_check(event, check_name, section.get("prefix") or "", row) def add_table_full_scan( events: dict[str, dict[str, Any]], report: dict[str, Any], context: dict[str, Any], ) -> None: section = report["sections"]["table_full_scan"] check_name = "tablefullscan_slow_query" for row in table_rows(section["table"]): digest = cell_text(row, "Digest") if not digest: continue event = ensure_event(events, report, context, digest) add_check_name(event, check_name) set_normalized_sql(event, cell_text(row, "Sample Query")) values = { "table_full_total_scan_rows": parse_number( cell_text(row, "Total Scan Rows") ), "table_full_slow_count": parse_number(cell_text(row, "Slow Count")), "table_full_max_scanned_rows": parse_number( cell_text(row, "Max Scanned Rows") ), "table_full_p50_scanned_rows": parse_number( cell_text(row, "P50 Scanned Rows") ), "table_full_p95_scanned_rows": parse_number( cell_text(row, "P95 Scanned Rows") ), "table_full_p50_latency_s": parse_number( cell_text(row, "P50 Latency (s)") ), "table_full_p95_latency_s": parse_number( cell_text(row, "P95 Latency (s)") ), "table_full_total_latency_s": parse_number( cell_text(row, "Total Latency (s)") ), } for key, value in values.items(): max_metric(event["metrics"], key, value) evidence = { "type": "clinic_health_check_result", "check_name": check_name, "total_scan_rows": values["table_full_total_scan_rows"], "slow_count": values["table_full_slow_count"], } for link in row_links(row): add_evidence(event["evidence_refs"], link) add_symptom( event, code="large_key_scan", severity="high", signal_code="slow_query.table_full_scan", metric_keys=list(values), assessment={ "method": "external", "summary": ( "Clinic returned this digest among the TableFullScan anomaly " "results. The adapter records the detector hit as High." ), "rule_code": "clinic.tablefullscan_slow_query", "rule_version": DETECTOR_VERSION, "policy_id": "clinic-health-check-report", "parameters": { "source_severity": "not_provided", "adapter_mapping": "table_full_scan_hit_to_high", }, }, evidence=evidence, ) add_raw_check(event, check_name, section.get("prefix") or "", row) def add_top_digest( events: dict[str, dict[str, Any]], report: dict[str, Any], context: dict[str, Any], ) -> None: section = report["sections"]["slow_log_top_digest"] check_name = "slowlog_top_digest" for tab_name, table in (section.get("tabs") or {}).items(): mapping = TOP_DIMENSIONS.get(tab_name) if not mapping: continue dimension, symptom_code, primary_metric = mapping for row in table_rows(table): digest = cell_text(row, "Digest") if not digest: continue event = ensure_event(events, report, context, digest) add_check_name(event, check_name) set_normalized_sql(event, cell_text(row, "Sample Query")) rank = parse_number(cell_text(row, "Rank")) event["dimensions"]["top_digest_ranks"][dimension] = rank metric_keys: list[str] = [] for column, metric_key in TOP_METRIC_COLUMNS.items(): value = parse_number(cell_text(row, column)) max_metric(event["metrics"], metric_key, value) if value is not None: metric_keys.append(metric_key) evidence = { "type": "clinic_health_check_result", "check_name": check_name, "dimension": dimension, "rank": rank, } for link in row_links(row): add_evidence(event["evidence_refs"], link) add_symptom( event, code=symptom_code, severity="medium", signal_code=f"slowlog.{dimension}", metric_keys=[primary_metric, *metric_keys], assessment={ "method": "external", "summary": ( "Clinic returned this digest in a configured Top-N dimension. " "Because the report does not expose a severity for ranking-only " "signals, the adapter records Medium for review." ), "rule_code": f"clinic.slowlog_top_digest.{dimension}", "rule_version": DETECTOR_VERSION, "policy_id": "clinic-health-check-report", "parameters": { "source_severity": "not_provided", "adapter_mapping": "top_n_hit_to_medium", "dimension": dimension, }, }, evidence=evidence, ) add_raw_check( event, check_name, section.get("prefix") or "", row, dimension=dimension, rank=rank, ) def add_hotspot( events: dict[str, dict[str, Any]], report: dict[str, Any], context: dict[str, Any], section_key: str, ) -> None: section = report["sections"][section_key] summary_tables = [ table for table in section.get("tables") or [] if table.get("kind") == "digest_summary" ] if not summary_tables: return check_name = ( "sql_tikv_high_cpu" if section_key == "sql_tikv_high_cpu" else "sql_tikv_high_duration" ) symptom_code = ( "tikv_cpu_hotspot" if section_key == "sql_tikv_high_cpu" else "high_total_latency" ) window_rows = [ row for table in section.get("tables") or [] if table.get("kind") == "windows" for row in table_rows(table) ] for table in summary_tables: for row in table_rows(table): digest = cell_text(row, "SQL Digest") if not digest: continue event = ensure_event(events, report, context, digest) add_check_name(event, check_name) affected_text = cell_text(row, "Affected") share_text = cell_text(row, "TopSQL Share") max_value_column = "Max CPU" if section_key == "sql_tikv_high_cpu" else "Max Duration" values = { f"{section_key}_affected_windows": labeled_number( affected_text, "windows" ) or parse_number(affected_text), f"{section_key}_affected_instances": labeled_number( affected_text, "instances" ), f"{section_key}_avg_topsql_share_percent": labeled_number( share_text, "avg" ), f"{section_key}_max_topsql_share_percent": labeled_number( share_text, "max" ), ( "tikv_cpu_max_percent" if section_key == "sql_tikv_high_cpu" else "tikv_duration_max_ms" ): parse_number(cell_text(row, max_value_column)), f"{section_key}_slow_log_count": parse_number( cell_text(row, "Slow Log") ), } matching_windows = [ window for window in window_rows if digest in cell_text(window, "Top SQL") ] values[f"{section_key}_matching_window_count"] = len(matching_windows) for key, value in values.items(): max_metric(event["metrics"], key, value) source_severity = cell_text(row, "Max Severity") severity = normalize_hotspot_severity(source_severity) evidence = { "type": "clinic_health_check_result", "check_name": check_name, "source_severity": source_severity, "affected_windows": values[f"{section_key}_affected_windows"], "affected_instances": values[f"{section_key}_affected_instances"], } for link in row_links(row): add_evidence(event["evidence_refs"], link) add_symptom( event, code=symptom_code, severity=severity, signal_code=( "tikv.high_cpu.topsql_attribution" if section_key == "sql_tikv_high_cpu" else "tikv.high_duration.topsql_attribution" ), metric_keys=list(values), assessment={ "method": "external", "summary": ( f"Severity is mapped directly from Clinic {check_name} " f"Max Severity: {source_severity or 'not provided'}." ), "rule_code": f"clinic.{check_name}", "rule_version": DETECTOR_VERSION, "policy_id": "clinic-health-check-report", "parameters": {"source_severity": source_severity}, }, evidence=evidence, ) add_raw_check( event, check_name, section.get("prefix") or "", row, matching_windows=[serializable_row(value) for value in matching_windows], ) def report_context( provenance: dict[str, Any], report: dict[str, Any], ) -> dict[str, Any]: org_id = clean_text(provenance.get("portal_org_id")) cluster_id = clean_text(provenance.get("cluster_id")) if not org_id or not cluster_id: raise ValueError("provenance.portal_org_id and provenance.cluster_id are required") return { "org_id": org_id, "dbaas_org_id": clean_text(provenance.get("dbaas_org_id")) or None, "org_name": clean_text(provenance.get("org_name")) or None, "project_id": clean_text(provenance.get("project_id")) or None, "cluster_id": cluster_id, "cluster_name": clean_text(provenance.get("cluster_name")) or None, "provider": clean_text(provenance.get("provider")) or None, "region": clean_text(provenance.get("region")) or None, "extraction_source": clean_text(provenance.get("source")), "report": { "result_uuid": report["id"], "report_id": report["id"], "org_id": org_id, "dbaas_org_id": clean_text(provenance.get("dbaas_org_id")) or None, "cluster_id": cluster_id, "time_range_a_start": report["from"], "time_range_a_end": report["to"], "created_at": report_created_at(report), "source_url": report.get("source_url"), "status": clean_text(report.get("status")) or "success", }, } def build_report_findings( provenance: dict[str, Any], report: dict[str, Any], ) -> list[dict[str, Any]]: context = report_context(provenance, report) events: dict[str, dict[str, Any]] = {} add_extreme_slow_query(events, report, context) add_table_full_scan(events, report, context) add_top_digest(events, report, context) add_hotspot(events, report, context, "sql_tikv_high_duration") add_hotspot(events, report, context, "sql_tikv_high_cpu") findings: list[dict[str, Any]] = [] for digest in sorted(events): event = events[digest] event["dimensions"]["check_names"].sort() event["dimensions"]["governance_signal_class"] = ( GOVERNANCE_SIGNAL_RANKING_OBSERVATION if event["dimensions"]["check_names"] == ["slowlog_top_digest"] else GOVERNANCE_SIGNAL_ANOMALY ) event["symptoms"] = sorted( event.pop("_symptoms").values(), key=lambda value: ( -SEVERITY_RANK[value["severity"]], value["code"], ), ) if not event["symptoms"]: raise ValueError(f"report {report['id']} digest {digest} has no symptoms") findings.append(event) return findings def build_manifest(data: dict[str, Any], input_path: Path) -> dict[str, Any]: provenance = data.get("provenance") or {} reports = data.get("reports") or [] if not isinstance(reports, list) or not reports: raise ValueError("reports must be a non-empty array") manifest_runs: list[dict[str, Any]] = [] seen_events: set[str] = set() for report in sorted(reports, key=lambda value: clean_text(value.get("from"))): created_at = report_created_at(report) findings = build_report_findings(provenance, report) for finding in findings: event_id = finding["source_event_id"] if event_id in seen_events: raise ValueError(f"duplicate source_event_id: {event_id}") seen_events.add(event_id) manifest_runs.append( { "run": { "source": SOURCE, "source_type": SOURCE_TYPE, "source_run_id": report["id"], "org_id": provenance["portal_org_id"], "project_id": provenance.get("project_id"), "cluster_id": provenance["cluster_id"], "time_window_start": report["from"], "time_window_end": report["to"], "status": clean_text(report.get("status")) or "success", "detector_version": DETECTOR_VERSION, "expected_detector_count": 5, "completed_detector_count": 5, "started_at": report["from"], "completed_at": report["to"], "metadata": { "result_uuid": report["id"], "report_created_at": created_at, "source_url": report.get("source_url"), "source_file": str(input_path), "source_format": clean_text(provenance.get("source_format")) or "clinic-rendered-report-ui-extract/v1", "synthetic": False, }, }, "replay_at": created_at, "findings": findings, } ) return { "source": SOURCE, "source_type": SOURCE_TYPE, "provenance": provenance, "runs": manifest_runs, } def chunk_findings( run_id: str, findings: list[dict[str, Any]], max_request_bytes: int, ) -> list[list[dict[str, Any]]]: chunks: list[list[dict[str, Any]]] = [] current: list[dict[str, Any]] = [] for finding in findings: candidate = [*current, finding] body = { "source": SOURCE, "source_type": SOURCE_TYPE, "source_batch_id": f"{run_id}:part-{len(chunks) + 1}", "findings": candidate, "created_by": CREATED_BY, } if len(compact_json(body)) <= max_request_bytes: current = candidate continue if not current: raise ValueError( f"one Finding exceeds the request limit: {finding['source_event_id']}" ) chunks.append(current) current = [finding] if current: chunks.append(current) return chunks def post_json( api_base: str, path: str, payload: dict[str, Any], *, replay_at: str | None = None, ) -> dict[str, Any]: headers = { "Accept": "application/json", "Content-Type": "application/json", } if replay_at: headers["X-SQL-Catalog-Replay-At"] = replay_at request = Request( f"{api_base.rstrip('/')}/{path.lstrip('/')}", data=compact_json(payload), headers=headers, method="POST", ) try: with urlopen(request, timeout=120) as response: body = json.loads(response.read().decode("utf-8")) if not 200 <= response.status < 300: raise RuntimeError(f"{path} returned {response.status}: {body}") return body except HTTPError as exc: error_body = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"{path} returned {exc.code}: {error_body}") from exc except URLError as exc: raise RuntimeError(f"cannot reach SQL Catalog API at {api_base}: {exc}") from exc def manifest_summary( manifest: dict[str, Any], max_request_bytes: int, ) -> dict[str, Any]: digest_runs: defaultdict[str, int] = defaultdict(int) severity_counts: defaultdict[str, int] = defaultdict(int) symptom_counts: defaultdict[str, int] = defaultdict(int) report_summaries: list[dict[str, Any]] = [] total_batches = 0 for entry in manifest["runs"]: findings = entry["findings"] chunks = chunk_findings( entry["run"]["source_run_id"], findings, max_request_bytes, ) total_batches += len(chunks) for finding in findings: digest_runs[finding["sql_digest"]] += 1 severity = max( finding["symptoms"], key=lambda value: SEVERITY_RANK[value["severity"]], )["severity"] severity_counts[severity] += 1 for symptom in finding["symptoms"]: symptom_counts[symptom["code"]] += 1 report_summaries.append( { "source_run_id": entry["run"]["source_run_id"], "time_window_start": entry["run"]["time_window_start"], "report_created_at": entry["replay_at"], "findings": len(findings), "batches": len(chunks), } ) return { "runs": len(manifest["runs"]), "batches": total_batches, "findings": sum(len(entry["findings"]) for entry in manifest["runs"]), "catalog_digests": len(digest_runs), "digests_present_in_all_runs": sum( count == len(manifest["runs"]) for count in digest_runs.values() ), "severity_counts": dict(sorted(severity_counts.items())), "symptom_counts": dict(sorted(symptom_counts.items())), "reports": report_summaries, } def import_manifest( manifest: dict[str, Any], api_base: str, max_request_bytes: int, ) -> dict[str, Any]: created = 0 duplicate = 0 imported_runs: list[dict[str, Any]] = [] for entry in manifest["runs"]: run_payload = entry["run"] replay_at = entry["replay_at"] run_response = post_json( api_base, "finding-runs", run_payload, replay_at=replay_at, ) chunks = chunk_findings( run_payload["source_run_id"], entry["findings"], max_request_bytes, ) run_created = 0 run_duplicate = 0 batch_ids: list[str] = [] for index, chunk in enumerate(chunks, start=1): batch_id = f"{run_payload['source_run_id']}:part-{index}" batch_response = post_json( api_base, "finding-batches", { "source": SOURCE, "source_type": SOURCE_TYPE, "source_batch_id": batch_id, "findings": chunk, "created_by": CREATED_BY, }, replay_at=replay_at, ) summary = batch_response.get("summary") or {} if summary.get("rejected"): raise RuntimeError( f"batch {batch_id} rejected {summary['rejected']} Findings" ) run_created += int(summary.get("created") or 0) run_duplicate += int(summary.get("duplicate") or 0) batch_ids.append(batch_id) created += run_created duplicate += run_duplicate imported_runs.append( { "source_run_id": run_payload["source_run_id"], "finding_run_id": run_response["run"]["id"], "run_created": bool(run_response.get("created")), "created": run_created, "duplicate": run_duplicate, "batch_ids": batch_ids, } ) return { "runs": imported_runs, "created": created, "duplicate": duplicate, } def main() -> None: parser = argparse.ArgumentParser( description=( "Import real SQL signals extracted from authenticated Clinic Health " "Check report pages through the SQL Catalog HTTP API." ) ) parser.add_argument("input", type=Path) parser.add_argument( "--api-base", default="http://127.0.0.1:8765/api/v1", ) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--write-manifest", type=Path) parser.add_argument( "--max-request-bytes", type=int, default=1_750_000, ) args = parser.parse_args() data = json.loads(args.input.read_text(encoding="utf-8")) manifest = build_manifest(data, args.input) summary = manifest_summary(manifest, args.max_request_bytes) if args.write_manifest: args.write_manifest.parent.mkdir(parents=True, exist_ok=True) args.write_manifest.write_text( json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) result: dict[str, Any] = {"input": str(args.input), "summary": summary} if not args.dry_run: result["import"] = import_manifest( manifest, args.api_base, args.max_request_bytes, ) print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True)) if __name__ == "__main__": try: main() except (RuntimeError, ValueError, KeyError) as exc: print(str(exc), file=sys.stderr) raise SystemExit(1) from exc