sync: save local state before v9.7 work

main.py

@@ -1,8 +1,10 @@
 from fastapi import FastAPI, Header, HTTPException, Request
+from fastapi.responses import StreamingResponse
 from fastapi.staticfiles import StaticFiles
 import httpx
 from typing import Annotated
 import asyncio
+import json
 import time
 
 app = FastAPI()
@@ -28,6 +30,53 @@ def resolve_deck_api(env_raw: str | None) -> tuple[str, str]:
     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 = []
@@ -103,11 +152,11 @@ async def get_graph(
         }
         allowed_uids = set(uid_to_inst.keys())
 
-        sem = asyncio.Semaphore(16)
+        sem = asyncio.Semaphore(20)
 
-        async def fetch_detail(uid: str):
+        async def fetch_detail(uid: str, client: httpx.AsyncClient):
             async with sem:
-                async with httpx.AsyncClient(timeout=20) as client:
+                try:
                     r = await client.get(
                         f"{deck_api}/index.cfm/instances/{uid}",
                         headers=auth_headers(x_deck_token),
@@ -117,38 +166,86 @@ async def get_graph(
                     if r.status_code >= 400:
                         return uid, None
                     return uid, r.json().get("instance")
+                except httpx.TimeoutException:
+                    return uid, None
 
-        details = await asyncio.gather(*[fetch_detail(uid) for uid in allowed_uids])
+        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}
 
-        edge_set = set()
+        # 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 dep_uid(dep_obj: dict):
-            return dep_obj.get("uid") or dep_obj.get("instanceUid")
+        def _is_platform_svc(svc_name: str) -> bool:
+            s = svc_name.strip().lower()
+            return any(p in s for p in _PLATFORM_PATTERNS)
 
-        for uid, inst in detail_map.items():
-            for dep in inst.get("dependencies", []) or []:
-                src = dep_uid(dep)
-                if src and src in allowed_uids:
-                    edge_set.add((src, uid))
-            for dep_on in inst.get("dependentInstances", []) or []:
-                dst = dep_uid(dep_on)
-                if dst and dst in allowed_uids:
-                    edge_set.add((uid, dst))
+        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
+
+        # Build directed edges from dependencies/dependentInstances
+        # Arrow points TO the dependency (what you depend on)
+        edge_set: set[tuple[str, str]] = set()
+        for uid_d, inst_d in detail_map.items():
+            if uid_d not in allowed_uids:
+                continue
+            for dep in inst_d.get("dependencies", []) or []:
+                src = dep.get("uid") or dep.get("instanceUid") or ""
+                if src and src in allowed_uids and src != uid_d:
+                    edge_set.add((uid_d, src))  # uid_d depends on src, arrow uid_d→src
+            for dep_on in inst_d.get("dependentInstances", []) or []:
+                dst = dep_on.get("uid") or dep_on.get("instanceUid") or ""
+                if dst and dst in allowed_uids and dst != uid_d:
+                    edge_set.add((dst, uid_d))  # dst depends on uid_d, arrow dst→uid_d
 
         nodes = []
-        for uid in allowed_uids:
-            i = uid_to_inst.get(uid, {})
+        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,
-                    "label": i.get("displayName") or uid,
+                    "id": uid_i,
+                    "label": i.get("displayName") or uid_i,
                     "status": i.get("explainedStatus") or "unknown",
-                    "svc": i.get("svc") or "",
+                    "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)}, edges: {len(edge_set)}", flush=True)
+        for _n in nodes:
+            if _n["lane"] in ("platform", "singleton"):
+                print(f"  {_n['lane'].upper()}: {_n['label']} | svc={_n['svc']}", flush=True)
+
         edges = [{"source": s, "target": t} for s, t in sorted(edge_set)]
         base_graph = {"nodes": nodes, "edges": edges}
         GRAPH_CACHE[cache_key] = (now, base_graph)
@@ -197,6 +294,57 @@ async def get_graph(
     }
 
 
+@app.get("/api/instances/stream")
+async def stream_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)
+
+    all_instances = await fetch_all_instances(x_deck_token, deck_api)
+    uid_to_inst = {
+        i.get("instanceUid"): i
+        for i in all_instances
+        if i.get("instanceUid")
+    }
+
+    async def generate():
+        sem = asyncio.Semaphore(20)
+        queue: asyncio.Queue = asyncio.Queue()
+        total = len(uid_to_inst)
+
+        async def fetch_one(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),
+                    )
+                    detail = None if r.status_code >= 400 else r.json().get("instance")
+                except Exception:
+                    detail = None
+                await queue.put((uid, detail))
+
+        async with httpx.AsyncClient(timeout=25) as shared_client:
+            tasks = [
+                asyncio.create_task(fetch_one(uid, shared_client))
+                for uid in uid_to_inst
+            ]
+            for _ in range(total):
+                uid, detail = await queue.get()
+                row = dict(uid_to_inst.get(uid, {}))
+                if detail:
+                    row["_detail"] = detail
+                yield json.dumps(row, ensure_ascii=False) + "\n"
+
+        yield json.dumps({"_done": True, "total": total}, ensure_ascii=False) + "\n"
+
+    return StreamingResponse(generate(), media_type="application/x-ndjson")
+
+
 @app.get("/api/instances/{uid}")
 async def get_instance(
     request: Request,