from fastapi import FastAPI, Header, HTTPException, Request from fastapi.responses import StreamingResponse from fastapi.staticfiles import StaticFiles import httpx from typing import Annotated import json import time from datetime import datetime, timezone 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" # Centralized backend policy for table-related derived fields. DELETE_POLICY = { "status": "suspended", "min_days": 14, "op_name_patterns": ("suspend", "приост", "замороз"), "op_time_fields": ("dtFinish", "dtStart", "dtCreated", "dtState"), "fallback_time_fields": ("dtState", "instanceConfigDtUpdated"), } 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 _parse_ts(value) -> float | None: if not value: return None if isinstance(value, (int, float)): return float(value) if not isinstance(value, str): return None raw = value.strip() if not raw: return None try: if raw.endswith("Z"): raw = raw[:-1] + "+00:00" dt = datetime.fromisoformat(raw) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.timestamp() except ValueError: return None def _extract_suspend_entered_ts(detail: dict | None) -> float | None: if not isinstance(detail, dict): return None latest_ts: float | None = None operations = detail.get("operations") if isinstance(operations, list): for op in operations: if not isinstance(op, dict): continue op_name = str(op.get("operation") or op.get("type") or op.get("action") or "").strip().lower() if not any(pattern in op_name for pattern in DELETE_POLICY["op_name_patterns"]): continue for field in DELETE_POLICY["op_time_fields"]: ts = _parse_ts(op.get(field)) if ts is None: continue if latest_ts is None or ts > latest_ts: latest_ts = ts break if latest_ts is not None: return latest_ts for field in DELETE_POLICY["fallback_time_fields"]: ts = _parse_ts(detail.get(field)) if ts is not None: return ts return None def build_stream_row_meta(row: dict, detail: dict | None) -> dict: status = str(row.get("explainedStatus") or "").strip().lower() meta = { "deleteCandidate": { "status": status, "eligible": False, "daysInSuspend": 0, "minDays": DELETE_POLICY["min_days"], "suspendEnteredAt": "", } } if status != DELETE_POLICY["status"]: return meta entered_ts = _extract_suspend_entered_ts(detail) if entered_ts is None: return meta days = max(0, int((time.time() - entered_ts) // 86400)) meta["deleteCandidate"]["daysInSuspend"] = days meta["deleteCandidate"]["eligible"] = days >= DELETE_POLICY["min_days"] meta["deleteCandidate"]["suspendEnteredAt"] = datetime.fromtimestamp(entered_ts, tz=timezone.utc).isoformat().replace("+00:00", "Z") return meta 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", []) if not isinstance(results, list): results = [] if not results: break all_results.extend(results) 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/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(): total = len(uid_to_inst) loaded = 0 started_at = time.time() async with httpx.AsyncClient(timeout=25) as shared_client: for uid in uid_to_inst: detail = None try: r = await shared_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: detail = r.json().get("instance") except Exception: detail = None loaded += 1 row = dict(uid_to_inst.get(uid, {})) row["_progress"] = {"loaded": loaded, "total": total} row["_meta"] = build_stream_row_meta(row, detail) if detail: row["_detail"] = detail yield json.dumps(row, ensure_ascii=False) + "\n" duration_ms = int((time.time() - started_at) * 1000) yield json.dumps({"_done": True, "total": total, "durationMs": duration_ms}, ensure_ascii=False) + "\n" return StreamingResponse(generate(), media_type="application/x-ndjson") @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")