#!/usr/bin/env python3 from __future__ import annotations import json from collections import Counter, defaultdict from pathlib import Path from typing import Any ROOT = Path("/Users/lc/coding/pingcap-kb-source") ENTRYPOINT_CATALOG = ROOT / "knowledge/deployment/public-entrypoint-catalog.json" OUTPUT_JSON = ROOT / "knowledge/deployment/public-service-catalog.json" OUTPUT_MD = ROOT / "knowledge/deployment/public-service-catalog.md" OBJECT_OVERRIDES: dict[str, dict[str, Any]] = { "dedicated-cluster-ng-service": { "source_owner": "tidbcloud/cluster-service-ng", "interface_artifacts": [ "knowledge/repos/cluster-service-ng/interfaces/cmd-cluster_ng_server.json", ], "host_family_hint": "regional-api.*.tidbcloud.com via shared Kong gateway", }, "tcidm": { "source_owner": "tidbcloud/tcidm", "interface_artifacts": [ "knowledge/repos/tcidm/interfaces/cmd-regional_tcidm_service.json", ], "host_family_hint": "regional-api.*.tidbcloud.com via shared Kong gateway", }, "dedicated-cluster-service": { "source_owner": "tidbcloud/infra-provider (cluster-server) + tiworkload agent bundle", "interface_artifacts": [ "knowledge/repos/infra-provider/interfaces/cluster-server.json", ], "host_family_hint": "regional-api.*.tidbcloud.com via shared Kong gateway", }, "dedicated-ticdc-service": { "source_owner": "tidbcloud/dataflow-service", "interface_artifacts": [ "knowledge/repos/dataflow-service/interfaces/cmd-dataflow-ng.json", ], "host_family_hint": "regional-api.*.tidbcloud.com via shared Kong gateway", }, "config-center-regional": { "host_family_hint": "regional-api.*.tidbcloud.com via shared Kong gateway", }, "dedicated-regional-server": { "interface_artifacts": [ "knowledge/repos/infra-provider/interfaces/regional-server.json", ], }, } def load_json(path: Path) -> Any: with path.open() as f: return json.load(f) def is_noise_record(record: dict[str, Any]) -> bool: source_file = (record.get("source_file") or "").lower() backend = (record.get("backend_service") or "").lower() manifest_name = (record.get("manifest_name") or "").lower() noisy_terms = ("cleanup", "readonly", "qa", "clinic") return any(term in source_file or term in backend or term in manifest_name for term in noisy_terms) def classify_audience(records: list[dict[str, Any]]) -> str: hosts = " ".join(filter(None, (record.get("host") for record in records))).lower() paths = " ".join(filter(None, (record.get("path") for record in records))).lower() deployment_object = records[0]["deployment_object"] if "investibook" in hosts: return "preview_web" if "o11y" in hosts or "monitoringpipelines" in paths or "metrics" in paths or "diagnosis" in paths: return "tenant_observability" if "/dataservice" in paths: return "tenant_data_api" if "/ops" in paths: return "operator_api" if "regional-api" in hosts or "regionalsvc" in hosts: return "regional_api" if any(token in paths for token in ["/api/premium", "/dataplane", "/cluster-server", "/tiworkload", "/config-center", "/api/v1/tcidm", "/ticdc-service"]): return "shared_gateway_api" if deployment_object in {"account-svc", "authserver", "tidb-mgmt-portal"}: return "unresolved_public_candidate" return "unknown" def classify_status(records: list[dict[str, Any]]) -> str: reachability = {record["reachability"] for record in records} if "public_dns" in reachability or "internet_facing_lb" in reachability: return "confirmed_external_entrypoint" if "gateway_routed_unknown_host" in reachability: return "shared_gateway_exposure" return "deployment_exposed_but_host_unresolved" def classify_signal(records: list[dict[str, Any]], audience: str) -> str: reachability = {record["reachability"] for record in records} if "public_dns" in reachability: return "high" if "internet_facing_lb" in reachability: return "medium" if "gateway_routed_unknown_host" in reachability and audience in {"shared_gateway_api", "regional_api"}: return "medium" return "low" def interface_summary(interface_files: list[str]) -> list[dict[str, Any]]: summaries: list[dict[str, Any]] = [] for rel in interface_files[:8]: path = ROOT / rel try: data = load_json(path) except Exception: continue services = data.get("services", []) summaries.append( { "file": rel, "binary": data.get("binary"), "type": data.get("type"), "service_count": len(services), "sample_services": [ { "name": svc.get("name"), "type": svc.get("type") or svc.get("transport"), "method_count": len(svc.get("methods", [])), } for svc in services[:6] ], } ) return summaries def main() -> None: catalog = load_json(ENTRYPOINT_CATALOG) grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) unresolved_candidates: list[dict[str, Any]] = [] for record in catalog["records"]: if is_noise_record(record): continue if record["reachability"] in {"public_dns", "internet_facing_lb", "gateway_routed_unknown_host"}: grouped[record["deployment_object"]].append(record) elif record["deployment_object"] in {"account-svc", "authserver", "tidb-mgmt-portal"}: unresolved_candidates.append(record) service_entries: list[dict[str, Any]] = [] for deployment_object, records in sorted(grouped.items()): override = OBJECT_OVERRIDES.get(deployment_object, {}) source_owner = override.get("source_owner") or records[0].get("source_repo") interface_files = sorted({item for record in records for item in (record.get("interface_artifacts") or [])}) if override.get("interface_artifacts"): interface_files = override["interface_artifacts"] audience = classify_audience(records) entry = { "service_id": f"public-service:{deployment_object}", "deployment_object": deployment_object, "source_owner": source_owner, "exposure_status": classify_status(records), "public_signal": classify_signal(records, audience), "audience": audience, "host_patterns": sorted({record.get("host") for record in records if record.get("host")}), "path_prefixes": sorted({record.get("path") for record in records if record.get("path")}), "backend_services": sorted({record.get("backend_service") for record in records if record.get("backend_service")}), "exposure_surfaces": sorted({record.get("exposure_surface") for record in records if record.get("exposure_surface")}), "host_family_hint": override.get("host_family_hint"), "interface_artifacts": interface_files, "interface_summaries": interface_summary(interface_files), "example_evidence": [ { "source_file": record["source_file"], "reachability": record["reachability"], "host": record.get("host"), "path": record.get("path"), "backend_service": record.get("backend_service"), } for record in records[:6] ], "record_count": len(records), } service_entries.append(entry) unresolved_by_object: dict[str, list[dict[str, Any]]] = defaultdict(list) for record in unresolved_candidates: unresolved_by_object[record["deployment_object"]].append(record) unresolved_entries: list[dict[str, Any]] = [] for deployment_object, records in sorted(unresolved_by_object.items()): override = OBJECT_OVERRIDES.get(deployment_object, {}) source_owner = override.get("source_owner") or records[0].get("source_repo") interface_files = sorted({item for record in records for item in (record.get("interface_artifacts") or [])}) if override.get("interface_artifacts"): interface_files = override["interface_artifacts"] unresolved_entries.append( { "service_id": f"public-candidate:{deployment_object}", "deployment_object": deployment_object, "source_owner": source_owner, "status": "deployment_exposed_but_host_unresolved", "reason": "nodeport_or_internal_service_without_resolved_external_host", "backend_services": sorted({record.get("backend_service") for record in records if record.get("backend_service")}), "host_family_hint": override.get("host_family_hint"), "interface_artifacts": interface_files, "interface_summaries": interface_summary(interface_files), "example_evidence": [ { "source_file": record["source_file"], "reachability": record["reachability"], "backend_service": record.get("backend_service"), } for record in records[:6] ], } ) summary = { "catalog_entry_count": len(service_entries), "confirmed_external_entrypoint_count": sum(1 for entry in service_entries if entry["exposure_status"] == "confirmed_external_entrypoint"), "shared_gateway_exposure_count": sum(1 for entry in service_entries if entry["exposure_status"] == "shared_gateway_exposure"), "deployment_exposed_candidate_count": len(unresolved_entries), "audience_counts": dict(Counter(entry["audience"] for entry in service_entries)), "public_signal_counts": dict(Counter(entry["public_signal"] for entry in service_entries)), } result = { "date": "2026-03-30", "generator": "construction/derive-public-service-catalog.py", "source_documents": [ "knowledge/deployment/public-entrypoint-catalog.json", "knowledge/repos/*/interfaces/*.json", ], "scope_notes": [ "catalog is derived from deployment-side public entrypoint scan plus repo interface artifacts", "catalog filters obvious cleanup and readonly noise", "public status is heuristic and should not be treated as final product taxonomy", "gateway-routed entries may still need host resolution through shared gateway config joins", ], "summary": summary, "services": sorted(service_entries, key=lambda item: (-item["record_count"], item["deployment_object"])), "unresolved_candidates": unresolved_entries, } OUTPUT_JSON.write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n") lines: list[str] = [] lines.append("# Public Service Catalog") lines.append("") lines.append("Date: 2026-03-30") lines.append("") lines.append("## Purpose") lines.append("") lines.append("This catalog refines the public entrypoint scan into a smaller set of public or externally reachable service candidates.") lines.append("") lines.append("It is intended to answer which deployment objects most likely represent public services, how they are exposed, and where their interface artifacts live.") lines.append("") lines.append("## Summary") lines.append("") lines.append(f"- catalog entries: `{summary['catalog_entry_count']}`") lines.append(f"- confirmed external entrypoints: `{summary['confirmed_external_entrypoint_count']}`") lines.append(f"- shared-gateway exposures: `{summary['shared_gateway_exposure_count']}`") lines.append(f"- deployment-exposed unresolved candidates: `{summary['deployment_exposed_candidate_count']}`") lines.append("") lines.append("Audience counts:") for key, value in sorted(summary["audience_counts"].items()): lines.append(f"- `{key}`: `{value}`") lines.append("") lines.append("## Highest-Signal Services") lines.append("") for entry in sorted(result["services"], key=lambda item: (item["public_signal"] != "high", -item["record_count"], item["deployment_object"]))[:15]: sample_host = entry["host_patterns"][0] if entry["host_patterns"] else None sample_path = entry["path_prefixes"][0] if entry["path_prefixes"] else None host_text = sample_host or entry.get("host_family_hint") lines.append( f"- `{entry['deployment_object']}`: status `{entry['exposure_status']}`, audience `{entry['audience']}`, source `{entry['source_owner']}`, sample host `{host_text}`, sample path `{sample_path}`" ) lines.append("") lines.append("## Unresolved But Important Candidates") lines.append("") for entry in result["unresolved_candidates"]: lines.append( f"- `{entry['deployment_object']}`: source `{entry['source_owner']}`, deployment exposure exists but no resolved public host yet" ) lines.append("") lines.append("## Machine-Readable Output") lines.append("") lines.append("- `knowledge/deployment/public-service-catalog.json`") OUTPUT_MD.write_text("\n".join(lines) + "\n") if __name__ == "__main__": main()