from pathlib import Path path = Path("grafo_d_server.py") text = path.read_text(encoding="utf-8") marker = "# --- V55B API PROBE FALLBACK OVERRIDE ---" block = r''' # --- V55B API PROBE FALLBACK OVERRIDE --- # Robust backend probe for /api/apis/status. # It first tries urllib and falls back to local curl when urllib times out. def _v55b_probe_with_urllib(api, timeout=15): api = normalize_api_row(api) endpoint = endpoint_for_api(api) if not endpoint: raise ValueError("Missing endpoint") parsed = urlparse(endpoint) if parsed.scheme not in ("http", "https"): raise ValueError("Unsupported endpoint scheme") started = time.perf_counter() req = Request( endpoint, headers=request_headers_for_api(api), method=safe_api_method(api), ) with urlopen(req, timeout=timeout) as res: # Leer poco para forzar recepción real sin bloquear con respuestas grandes. try: res.read(4096) except Exception: pass latency = int((time.perf_counter() - started) * 1000) http_status = int(getattr(res, "status", 0) or res.getcode() or 0) return { "ok": 200 <= http_status < 300, "http_status": http_status, "latency_ms": latency, "error": None if 200 <= http_status < 300 else f"HTTP {http_status}", "transport": "urllib", } def _v55b_probe_with_curl(api, timeout=15): api = normalize_api_row(api) endpoint = endpoint_for_api(api) if not endpoint: raise ValueError("Missing endpoint") parsed = urlparse(endpoint) if parsed.scheme not in ("http", "https"): raise ValueError("Unsupported endpoint scheme") started = time.perf_counter() cmd = [ "curl", "-L", "-sS", "-o", "/dev/null", "-w", "%{http_code} %{time_total}", "--max-time", str(int(timeout)), endpoint, ] proc = subprocess.run( cmd, cwd=str(ROOT), text=True, capture_output=True, timeout=timeout + 3, ) latency = int((time.perf_counter() - started) * 1000) if proc.returncode != 0: err = (proc.stderr or proc.stdout or "").strip() raise RuntimeError(err or f"curl returned {proc.returncode}") parts = (proc.stdout or "").strip().split() http_status = int(parts[0]) if parts and parts[0].isdigit() else 0 return { "ok": 200 <= http_status < 300, "http_status": http_status, "latency_ms": latency, "error": None if 200 <= http_status < 300 else f"HTTP {http_status}", "transport": "curl", } def check_api_connectivity(api, timeout=15): """ V55B robust real API connectivity check. Fixes false inactive state when urllib times out but curl succeeds locally. """ api = normalize_api_row(api) api_id = str(api.get("id") or api.get("name") or "api") name = str(api.get("name") or api_id) endpoint = endpoint_for_api(api) base = { "id": api_id, "name": name, "enabled": bool(api.get("enabled")), "target_graph": str(api.get("target_graph") or api.get("graph") or api.get("row") or ""), "endpoint": endpoint, "method": safe_api_method(api), "status": "unknown", "runtime_status": "unknown", "http_status": None, "latency_ms": None, "last_checked": now_iso(), "error": None, "transport": None, } if not api.get("enabled"): base["status"] = "unknown" base["runtime_status"] = "unknown" base["error"] = "API disabled" return base if not endpoint: base["status"] = "unknown" base["runtime_status"] = "unknown" base["error"] = "Missing endpoint" return base errors = [] probe = None try: probe = _v55b_probe_with_urllib(api, timeout=timeout) except Exception as exc: errors.append("urllib: " + str(exc)) try: probe = _v55b_probe_with_curl(api, timeout=timeout) except Exception as exc2: errors.append("curl: " + str(exc2)) if probe: base["http_status"] = probe.get("http_status") base["latency_ms"] = probe.get("latency_ms") base["transport"] = probe.get("transport") if probe.get("ok"): base["status"] = "active" base["runtime_status"] = "active" base["error"] = None else: status_code = int(probe.get("http_status") or 0) if 300 <= status_code < 400: base["status"] = "degraded" base["runtime_status"] = "degraded" base["error"] = probe.get("error") or f"HTTP {status_code}" else: base["status"] = "inactive" base["runtime_status"] = "inactive" base["error"] = probe.get("error") or f"HTTP {status_code}" else: base["status"] = "inactive" base["runtime_status"] = "inactive" base["error"] = " | ".join(errors) if errors else "probe failed" try: set_api_status( api_id, status=base["status"], last_status=base["http_status"], last_latency_ms=base["latency_ms"], last_run_at=base["last_checked"], last_error=base["error"] or "", last_checked=base["last_checked"], runtime_status=base["runtime_status"], ) except Exception: pass return base def api_status_sql(graph=None): """ V55B endpoint backend: GET /api/apis/status GET /api/apis/status?graph=ISS """ ensure_runtime_schema() apis = [ normalize_api_row(api) for api in list_apis() ] selected = [ api for api in apis if api_matches_graph(api, graph) ] results = [ check_api_connectivity(api, timeout=15) for api in selected ] if not results: overall = "unknown" elif any(x["status"] == "inactive" for x in results): overall = "inactive" elif any(x["status"] == "degraded" for x in results): overall = "degraded" elif any(x["status"] == "unknown" for x in results): overall = "unknown" else: overall = "active" return { "ok": True, "graph": graph, "status": overall, "checked_at": now_iso(), "count": len(results), "apis": results, } # --- END V55B API PROBE FALLBACK OVERRIDE --- ''' if marker in text: raise SystemExit("V55B ya estaba aplicado. No se duplicó.") main_marker = '\nif __name__ == "__main__":' if main_marker not in text: raise SystemExit("ERROR: no encontré if __name__ == \"__main__\".") text = text.replace(main_marker, "\n" + block + "\n" + main_marker, 1) path.write_text(text, encoding="utf-8") print("V55B API probe fallback aplicado.")