#!/usr/bin/env python3 """ Сравнивает сгенерированные YAML с живым API облака. Выявляет: новые/удалённые операции, новые/удалённые/изменённые параметры. Использование: python3 compare_yaml_vs_api.py --stand test --service 90,116,89,12,28 python3 compare_yaml_vs_api.py --stand dev --service 90 python3 compare_yaml_vs_api.py --stand prod --service 90,12 """ import argparse, json, os, sys, urllib.request, urllib.error, yaml, time from datetime import datetime, timezone # ── Конфигурация стендов ────────────────────────────────────────── STANDS = { "dev": { "api": "https://deck-api-dev.ngcloud.ru/api/v1/index.cfm", "token_file": "secrets/dev.token", "referer": "https://deck-dev.ngcloud.ru/", "yaml_dir": "generated/dev/resources_yaml", }, "test": { "api": "https://deck-api-test.ngcloud.ru/api/v1/index.cfm", "token_file": "secrets/test.token", "referer": "https://deck-test.ngcloud.ru/", "yaml_dir": "generated/test/resources_yaml", }, "prod": { "api": "https://deck-api.ngcloud.ru/api/v1/index.cfm", "token_file": "secrets/prod.token", "referer": "https://deck.ngcloud.ru/", "yaml_dir": "generated/prod/resources_yaml", }, } ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) def api_get(stand_cfg, path, timeout=15): """GET запрос к API стенда.""" url = f"{stand_cfg['api']}?endpoint={path}" token = open(os.path.join(ROOT, stand_cfg["token_file"])).read().strip() req = urllib.request.Request(url) req.add_header("Authorization", f"Bearer {token}") req.add_header("User-Agent", "Mozilla/5.0 (X11; Linux x86_64)") req.add_header("Referer", stand_cfg["referer"]) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as e: body = e.read().decode()[:200] return {"_error": f"HTTP {e.code}: {body}"} except Exception as e: return {"_error": str(e)} def get_service_info(stand_cfg, svc_id): """Получает список операций и MAN сервиса.""" data = api_get(stand_cfg, f"/services/{svc_id}") if "_error" in data: return data svc = data.get("svc", data) return { "svcId": svc.get("svcId"), "svc": svc.get("svc"), "operations": svc.get("operations", []), "man": svc.get("man", ""), } def get_operation_detail(stand_cfg, op_id): """Получает детали операции (kind, action, cfsParams).""" data = api_get(stand_cfg, f"/serviceOperation/{op_id}") if "_error" in data: return data op = data.get("svcOperation", {}) params = [] for p in op.get("cfsParams", []): params.append({ "id": p.get("svcOperationCfsParamId"), "code": p.get("svcOperationCfsParam"), # ← поле называется svcOperationCfsParam! "data_type": p.get("dataType"), "required": p.get("isRequired"), "default": str(p.get("defaultValue")) if p.get("defaultValue") is not None else None, "value_list": p.get("valueList"), "is_modifiable": p.get("isModifiable"), "sort": p.get("sort"), "func": p.get("func"), "ref_svc_id": p.get("refSvcId"), "sub_params": p.get("nestedRef"), # может содержать вложенные параметры }) return { "id": op.get("svcOperationId"), "name": op.get("operation"), "params": params, } def find_yaml(stand_cfg, svc_id): """Находит YAML-файл по service_id.""" yaml_dir = os.path.join(ROOT, stand_cfg["yaml_dir"]) for fname in os.listdir(yaml_dir): if not fname.endswith(".yaml"): continue fpath = os.path.join(yaml_dir, fname) with open(fpath) as f: data = yaml.safe_load(f) if data and data.get("service_id") == svc_id: return fname, data return None, None def compare_params(api_params, yaml_params): """Сравнивает два списка параметров.""" api_by_code = {} for p in api_params: code = p.get("code", "") if code: api_by_code[code] = p yaml_by_code = {} for p in yaml_params: code = p.get("code", "") if code: yaml_by_code[code] = p new_params = [] # есть в API, нет в YAML deleted_params = [] # есть в YAML, нет в API changed_params = [] # есть в обоих, но различаются for code, api_p in api_by_code.items(): if code not in yaml_by_code: new_params.append(api_p) else: yaml_p = yaml_by_code[code] diffs = [] # Сравниваем ключевые поля for field, api_key, yaml_key in [ ("type", "data_type", "data_type"), ("required", "required", "required"), ("default", "default", "default"), ("is_modifiable", "is_modifiable", "is_modifiable"), ("sort", "sort", "sort"), ("func", "func", "func"), ("ref_svc_id", "ref_svc_id", None), # в YAML может быть по-другому ]: api_val = api_p.get(api_key) yaml_val = yaml_p.get(yaml_key) if yaml_key else None if str(api_val) != str(yaml_val): diffs.append(f" {field}: API={api_val} YAML={yaml_val}") # value_list api_vl = api_p.get("value_list") yaml_vl = yaml_p.get("value_list") if api_vl or yaml_vl: api_vl_norm = sorted(api_vl) if api_vl else None yaml_vl_norm = sorted(yaml_vl) if yaml_vl else None if api_vl_norm != yaml_vl_norm: diffs.append(f" value_list: API={api_vl} YAML={yaml_vl}") if diffs: changed_params.append({"code": code, "diffs": diffs, "api": api_p, "yaml": yaml_p}) for code in yaml_by_code: if code not in api_by_code: deleted_params.append(yaml_by_code[code]) return new_params, deleted_params, changed_params def compare_service(stand_cfg, svc_id, svc_name): """Сравнивает один сервис: YAML vs API.""" print(f"\n{'='*70}") print(f" Сервис: {svc_name} (ID={svc_id})") print(f"{'='*70}") # 1. Получаем данные из API print(f" Запрос API /services/{svc_id} ...") info = get_service_info(stand_cfg, svc_id) if "_error" in info: print(f" ❌ Ошибка API: {info['_error']}") return {"error": info["_error"]} api_ops = {op["svcOperationId"]: op["operation"] for op in info["operations"]} print(f" API: {len(api_ops)} операций") # 2. Находим YAML yaml_fname, yaml_data = find_yaml(stand_cfg, svc_id) if not yaml_data: print(f" ⚠️ YAML не найден для service_id={svc_id}") return {"yaml_missing": True, "api_ops": api_ops} yaml_ops = {op["id"]: op["name"] for op in yaml_data.get("operations", [])} print(f" YAML ({yaml_fname}): {len(yaml_ops)} операций") # 3. Сравниваем операции result = {"service": svc_name, "service_id": svc_id, "yaml_file": yaml_fname, "ops_new": [], "ops_deleted": [], "ops_ok": [], "op_details": {}} for op_id, op_name in api_ops.items(): if op_id not in yaml_ops: result["ops_new"].append(f"{op_id}:{op_name}") for op_id, op_name in yaml_ops.items(): if op_id not in api_ops: result["ops_deleted"].append(f"{op_id}:{op_name}") # 4. Для совпадающих операций — сравниваем параметры common_ops = set(api_ops.keys()) & set(yaml_ops.keys()) print(f" Совпадающих операций: {len(common_ops)}") for op_id in common_ops: op_name = api_ops[op_id] detail = get_operation_detail(stand_cfg, op_id) if "_error" in detail: result["op_details"][op_name] = {"error": detail["_error"]} continue # Находим YAML-параметры для этой операции yaml_params = [] for op in yaml_data.get("operations", []): if op["id"] == op_id: yaml_params = op.get("params", []) break new_p, del_p, chg_p = compare_params(detail["params"], yaml_params) result["op_details"][op_name] = { "api_count": len(detail["params"]), "yaml_count": len(yaml_params), "new_params": new_p, "deleted_params": del_p, "changed_params": chg_p, } # 5. Вывод результата if result["ops_new"]: print(f" 🆕 НОВЫЕ операции в API (нет в YAML):") for o in result["ops_new"]: print(f" + {o}") if result["ops_deleted"]: print(f" 🗑 УДАЛЁННЫЕ операции (есть в YAML, нет в API):") for o in result["ops_deleted"]: print(f" - {o}") total_changes = 0 for op_name, detail in result["op_details"].items(): n = len(detail.get("new_params", [])) + len(detail.get("deleted_params", [])) + len(detail.get("changed_params", [])) if n > 0: total_changes += n print(f"\n 📋 {op_name}: API={detail['api_count']}params YAML={detail['yaml_count']}params") for p in detail["new_params"]: print(f" 🆕 + {p['code']} (type={p['data_type']}, required={p['required']})") for p in detail["deleted_params"]: print(f" 🗑 - {p['code']} (type={p.get('data_type','?')})") for p in detail["changed_params"]: print(f" ✏️ ~ {p['code']}:") for d in p["diffs"]: print(f" {d}") if total_changes == 0 and not result["ops_new"] and not result["ops_deleted"]: print(f" ✅ YAML полностью совпадает с API") return result def main(): parser = argparse.ArgumentParser(description="Сравнить YAML с живым API") parser.add_argument("--stand", required=True, choices=["dev", "test", "prod"]) parser.add_argument("--service", required=True, help="ID сервисов через запятую, например 90,116,89,12,28") args = parser.parse_args() stand_cfg = STANDS[args.stand] svc_ids = [int(x.strip()) for x in args.service.split(",") if x.strip()] print(f"Стенд: {args.stand.upper()} ({stand_cfg['api']})") print(f"Сервисов: {len(svc_ids)}") all_results = {} for svc_id in svc_ids: time.sleep(0.3) # щадим API # Определяем имя из YAML если есть _, yaml_data = find_yaml(stand_cfg, svc_id) svc_name = yaml_data.get("name", str(svc_id)) if yaml_data else str(svc_id) all_results[svc_id] = compare_service(stand_cfg, svc_id, svc_name) # ── Сводка ── print(f"\n\n{'='*70}") print(f" СВОДКА по стенду {args.stand.upper()}") print(f"{'='*70}") total_ok = 0 total_changes = 0 for svc_id, r in all_results.items(): if "error" in r: print(f" ❌ {r.get('service', svc_id)}: {r['error']}") continue if r.get("yaml_missing"): print(f" ⚠️ {r.get('service', svc_id)}: YAML отсутствует (API ops: {len(r.get('api_ops',{}))})") continue has_changes = bool(r["ops_new"] or r["ops_deleted"] or any(len(d.get("new_params",[])) + len(d.get("deleted_params",[])) + len(d.get("changed_params",[])) > 0 for d in r["op_details"].values())) if has_changes: total_changes += 1 print(f" ⚠️ {r['service']}: есть изменения") else: total_ok += 1 print(f" ✅ {r['service']}: OK") print(f"\n Итого: ✅ {total_ok} ⚠️ {total_changes}") if __name__ == "__main__": main()