from fastapi import FastAPI, Header, HTTPException, Request from fastapi.staticfiles import StaticFiles import httpx from typing import Annotated import asyncio import time app = FastAPI() DECK_APIS = { "prod": "https://deck-api.ngcloud.ru/api/v1", "dev": "https://deck-api-dev.ngcloud.ru/api/v1", "test": "https://deck-api-test.ngcloud.ru/api/v1", } DEFAULT_DECK_ENV = "test" CACHE_TTL_SECONDS = 180 GRAPH_CACHE: dict[str, tuple[float, dict]] = {} def auth_headers(token: str): return {"Authorization": f"Bearer {token}"} def resolve_deck_api(env_raw: str | None) -> tuple[str, str]: env = (env_raw or DEFAULT_DECK_ENV).strip().lower() if env not in DECK_APIS: env = DEFAULT_DECK_ENV return env, DECK_APIS[env] def _as_text(value) -> str: if value is None: return "" if isinstance(value, str): return value.strip() if isinstance(value, (int, float, bool)): return str(value).strip() if isinstance(value, dict): for key in ("value", "displayValue", "name", "label", "id", "text"): nested = _as_text(value.get(key)) if nested: return nested return "" def extract_platform_name(inst: dict | None) -> str: if not isinstance(inst, dict): return "" sources = [ inst.get("parameters"), inst.get("params"), inst.get("instanceParameters"), inst.get("inputParameters"), ] needles = ("platform", "kuber", "k8s", "cluster", "shturval", "штурвал") for src in sources: if isinstance(src, dict): for key, value in src.items(): key_s = str(key).strip().lower() if any(n in key_s for n in needles): txt = _as_text(value) if txt: return txt elif isinstance(src, list): for item in src: if not isinstance(item, dict): continue key_s = str(item.get("name") or item.get("key") or item.get("code") or "").strip().lower() if any(n in key_s for n in needles): txt = _as_text(item.get("value") or item.get("displayValue") or item.get("text")) if txt: return txt return "" async def fetch_all_instances(token: str, deck_api: str): params: dict = {"page": 1, "size": 200} all_results = [] async with httpx.AsyncClient(timeout=20) as client: while True: r = await client.get( f"{deck_api}/index.cfm/instances", headers=auth_headers(token), params=params, ) if r.status_code == 401: raise HTTPException(status_code=401, detail="Invalid token") data = r.json() results = data.get("results", []) all_results.extend(results) if len(results) < 200: break params["page"] += 1 return all_results @app.get("/api/instances") async def list_instances( request: Request, x_deck_token: Annotated[str | None, Header()] = None, x_deck_env: Annotated[str | None, Header()] = None, ): if not x_deck_token: raise HTTPException(status_code=401, detail="Token required") _, deck_api = resolve_deck_api(x_deck_env) # parse status from query string manually (FastAPI list param can fail behind ingress rewrite) status_list = request.query_params.getlist("status") all_results = await fetch_all_instances(x_deck_token, deck_api) if status_list: all_results = [i for i in all_results if i.get("explainedStatus") in status_list] return {"instances": all_results, "total": len(all_results)} @app.get("/api/graph") async def get_graph( request: Request, x_deck_token: Annotated[str | None, Header()] = None, x_deck_env: Annotated[str | None, Header()] = None, ): if not x_deck_token: raise HTTPException(status_code=401, detail="Token required") env, deck_api = resolve_deck_api(x_deck_env) status_list = request.query_params.getlist("status") root_uid = request.query_params.get("root_uid") depth_raw = request.query_params.get("depth", "2") try: depth = max(1, min(5, int(depth_raw))) except ValueError: depth = 2 status_key = ",".join(sorted(status_list)) if status_list else "__all__" cache_key = f"{x_deck_token[:24]}::{env}::{status_key}" now = time.time() cached = GRAPH_CACHE.get(cache_key) if cached and (now - cached[0] < CACHE_TTL_SECONDS): base_graph = cached[1] else: all_instances = await fetch_all_instances(x_deck_token, deck_api) if status_list: all_instances = [i for i in all_instances if i.get("explainedStatus") in status_list] uid_to_inst = { i.get("instanceUid"): i for i in all_instances if i.get("instanceUid") } allowed_uids = set(uid_to_inst.keys()) sem = asyncio.Semaphore(20) async def fetch_detail(uid: str, client: httpx.AsyncClient): async with sem: try: r = await client.get( f"{deck_api}/index.cfm/instances/{uid}", headers=auth_headers(x_deck_token), ) if r.status_code == 401: raise HTTPException(status_code=401, detail="Invalid token") if r.status_code >= 400: return uid, None return uid, r.json().get("instance") except httpx.TimeoutException: return uid, None async with httpx.AsyncClient(timeout=25) as shared_client: details = await asyncio.gather(*[fetch_detail(uid, shared_client) for uid in allowed_uids]) detail_map = {uid: inst for uid, inst in details if inst} # Categorise singletons by service name _PLATFORM_PATTERNS = ("виртуальный датацентр", "vdc", "virtual data center", "cloud director", "kubernetes", "k8s") _SINGLETON_PATTERNS = ("object storage", "s3", "edge", "шлюз периметра", "dns", "container registry", "реестр контейнеров") def _is_platform_svc(svc_name: str) -> bool: s = svc_name.strip().lower() return any(p in s for p in _PLATFORM_PATTERNS) def _is_singleton_svc(svc_name: str) -> bool: s = svc_name.strip().lower() return any(p in s for p in _SINGLETON_PATTERNS) # Build realm→platform_uid map: instances whose resourceRealmCnt > 0 ARE platforms realm_to_platform: dict[str, str] = {} for uid_d, inst_d in detail_map.items(): cnt = inst_d.get("resourceRealmCnt") or 0 realm_name = inst_d.get("resourceRealm") or "" if cnt and realm_name: realm_to_platform[realm_name] = uid_d nodes = [] for uid_i in allowed_uids: i = uid_to_inst.get(uid_i, {}) detail = detail_map.get(uid_i, {}) svc = i.get("svc") or "" realm_cnt = detail.get("resourceRealmCnt") or 0 is_aux = bool(detail.get("isAuxiliary")) # Determine lane by service name if _is_platform_svc(svc): lane = "platform" elif _is_singleton_svc(svc): lane = "singleton" else: lane = "regular" nodes.append( { "id": uid_i, "label": i.get("displayName") or uid_i, "status": i.get("explainedStatus") or "unknown", "svc": svc, "realm": i.get("resourceRealm") or "", "lane": lane, "isAuxiliary": is_aux, "created": i.get("instanceConfigDtCreated") or i.get("dtCreated") or "", } ) # Debug: log lane distribution import sys _lane_counts = {} for _n in nodes: _l = _n.get("lane", "unknown") _lane_counts[_l] = _lane_counts.get(_l, 0) + 1 print(f"[GRAPH] Lanes: {_lane_counts}, nodes: {len(nodes)}", flush=True) for _n in nodes: if _n["lane"] in ("platform", "singleton"): print(f" {_n['lane'].upper()}: {_n['label']} | svc={_n['svc']}", flush=True) base_graph = {"nodes": nodes, "edges": []} GRAPH_CACHE[cache_key] = (now, base_graph) # Return full graph fast if not root_uid: return { "nodes": base_graph["nodes"], "edges": [], "total_nodes": len(base_graph["nodes"]), "total_edges": 0, "cached": bool(cached), } node_ids = {n["id"] for n in base_graph["nodes"]} if root_uid not in node_ids: return {"nodes": [], "edges": [], "total_nodes": 0, "total_edges": 0, "cached": bool(cached)} sub_nodes = [n for n in base_graph["nodes"] if n["id"] == root_uid] return { "nodes": sub_nodes, "edges": [], "total_nodes": len(sub_nodes), "total_edges": 0, "cached": bool(cached), } @app.get("/api/instances/{uid}") async def get_instance( request: Request, uid: str, x_deck_token: Annotated[str | None, Header()] = None, x_deck_env: Annotated[str | None, Header()] = None, ): if not x_deck_token: raise HTTPException(status_code=401, detail="Token required") _, deck_api = resolve_deck_api(x_deck_env) async with httpx.AsyncClient(timeout=20) as client: r = await client.get( f"{deck_api}/index.cfm/instances/{uid}", headers=auth_headers(x_deck_token), ) if r.status_code == 401: raise HTTPException(status_code=401, detail="Invalid token") return r.json() app.mount("/", StaticFiles(directory="/app/static", html=True), name="static")