"""Собирает .archimate (нативный формат Archi) из профиля контура и матрицы связей. py build_archimate.py <профиль.json> <матрица.json> <выход.archimate> [ещё...] Необязательные аргументы распознаются по расширению: .json — граф межсервисных вызовов (scan_calls.py), .md — описания компонентов прикладного слоя. Файлов с описаниями может быть несколько, каждый следующий переопределяет предыдущий: в модель идёт короткая версия (app-summaries.md), развёрнутая (app-docs.md) остаётся справочником. Все имена, зависящие от конкретной инсталляции (кластер, узлы, домены, реестр), приходят из профиля — в самом скрипте их нет. См. contour-profile.example.json. """ import json, pathlib, sys, io, re from xml.sax.saxutils import escape, quoteattr sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8") if len(sys.argv) < 4: sys.exit(__doc__) PROFILE = json.loads(pathlib.Path(sys.argv[1]).read_text(encoding="utf-8")) DATA = json.loads(pathlib.Path(sys.argv[2]).read_text(encoding="utf-8")) OUT = pathlib.Path(sys.argv[3]) def parse_app_docs(path): """Разбирает app-docs.md: заголовок `## ` -> текст до следующего заголовка.""" blocks = re.split(r"^## ", path.read_text(encoding="utf-8"), flags=re.M)[1:] out = {} for b in blocks: head, _, body = b.partition("\n") out[head.strip()] = body.strip() return out CALLS, APPDOCS = None, {} for extra in sys.argv[4:]: p = pathlib.Path(extra) if p.suffix.lower() == ".md": APPDOCS.update(parse_app_docs(p)) else: CALLS = json.loads(p.read_text(encoding="utf-8")) elements, relations, _rc = [], [], [0] REL_INDEX = {} def plural(n, one, few, many): """Русское согласование числительного: 1 сервис, 3 сервиса, 5 сервисов.""" if n % 100 // 10 == 1: return many return {1: one, 2: few, 3: few, 4: few}.get(n % 10, many) def el(folder, eid, etype, name, doc=""): elements.append((folder, eid, etype, name, doc)); return eid def R(rtype, src, tgt): _rc[0] += 1 rid = f"rel-{_rc[0]:03d}" relations.append((rid, rtype, src, tgt)) REL_INDEX[(rtype, src, tgt)] = rid return rid # ---------------------------------------------------------------- технологический слой T = "technology" nd_cluster = el(T, "nd-cluster", "Node", PROFILE["cluster"]["name"], PROFILE["cluster"].get("doc", "")) for s in PROFILE["software"]: el(T, s["id"], "SystemSoftware", s["name"], s.get("doc", "")) R("Composition", nd_cluster, s["id"]) cam_ids = [] for n in PROFILE.get("camunda_sub", []): if isinstance(n, str): n = {"name": n} cid = "ss-cam-" + n["name"].lower().replace(" ", "-") el(T, cid, "SystemSoftware", n["name"], n.get("doc", "")) R("Composition", "ss-camunda", cid); cam_ids.append(cid) def serves_of(x): """serves — id обслуживаемого элемента, либо список: у каталога предприятия их несколько (провайдер идентификации, брокер, служба федерации).""" s = x.get("serves") return [] if not s else ([s] if isinstance(s, str) else list(s)) for x in PROFILE["external"]: el(T, x["id"], "Node", x["name"], x.get("doc", "")) for a in x.get("artifacts", []): el(T, a["id"], "Artifact", a["name"]) R("Composition", x["id"], a["id"]) for tgt in serves_of(x): R("Serving", x["id"], nd_cluster if tgt == "cluster" else tgt) fn = PROFILE["frontend_node"] el(T, fn["id"], "Node", fn["name"], fn.get("doc", "")) R("Assignment", fn["id"], fn["assigned_to"]) for ts in PROFILE["services"]: el(T, ts["id"], "TechnologyService", ts["name"], ts.get("doc", "")) for r in ts["realizers"]: R("Realization", r, ts["id"]) if ts.get("serves_software"): R("Serving", ts["id"], ts["serves_software"]) # ---------------------------------------------------------------- прикладной слой app_id = {a: "ac-" + a for a in DATA["apps"]} for a in DATA["apps"]: el("application", app_id[a], "ApplicationComponent", a, APPDOCS.get(a, f"apps/{a}//")) consumers = {} for ts in PROFILE["services"]: key = ts.get("key") if not key: consumers[ts["id"]] = []; continue if key == "istio": lst = [a for a in DATA["apps"] if a in DATA["exposed"]] else: lst = [a for a in DATA["apps"] if DATA["matrix"][a].get(key)] consumers[ts["id"]] = lst for a in lst: R("Serving", ts["id"], app_id[a]) # ---------------------------------------------------------------- представления views, _oc = [], [0] def view(name, doc=""): v = {"id": f"view-{len(views)+1}", "name": name, "doc": doc, "objects": []} views.append(v); return v def obj(v, eid, x, y, w, h, parent=None): _oc[0] += 1 o = {"id": f"do-{_oc[0]:04d}", "el": eid, "b": (x, y, w, h), "children": [], "src": [], "tgt": []} (parent["children"] if parent else v["objects"]).append(o) return o def note(v, text, x, y, w, h, parent=None): """Пояснительная надпись на схеме. Элементом модели не является: нужна легенде, где подписать надо не архитектуру, а саму нотацию.""" o = obj(v, None, x, y, w, h, parent) o["note"] = text return o def conn(rtype, so, to): key = (rtype, so["el"], to["el"]) if key not in REL_INDEX: raise KeyError(f"нет отношения в модели: {key}") _oc[0] += 1 cid = f"cn-{_oc[0]:04d}" so["src"].append({"id": cid, "rel": REL_INDEX[key], "tgt": to["id"]}) to["tgt"].append(cid) SERVICES = PROFILE["services"] SW = {s["id"]: s for s in PROFILE["software"]} MIN_LINKS = 3 # порог, с которого сервису отводится отдельное представление _tno = [2] # T1 и T2 названы явно, дальше нумерация сквозная по семейству def tnum(): _tno[0] += 1 return f"T{_tno[0]}" v1 = view("T1. Технологический слой", "Что развёрнуто в кластере и какие технологические сервисы это даёт. " "Приложения-потребители вынесены в отдельные представления T2 и далее.") o_ts = {ts["id"]: obj(v1, ts["id"], 20 + i * 165, 30, 150, 60) for i, ts in enumerate(SERVICES)} o_cluster = obj(v1, nd_cluster, 20, 140, 1270, 400) o_ss, ids = {}, [s["id"] for s in PROFILE["software"]] for i, sid in enumerate(ids[:5]): o_ss[sid] = obj(v1, sid, 20 + i * 240, 45, 220, 70, o_cluster) for i, sid in enumerate(ids[5:9]): o_ss[sid] = obj(v1, sid, 20 + i * 240, 145, 220, 70, o_cluster) if len(ids) > 9: o_ss[ids[9]] = obj(v1, ids[9], 20, 245, 1230, 130, o_cluster) for i, cid in enumerate(cam_ids): obj(v1, cid, 10 + i * 174, 50, 162, 60, o_ss[ids[9]]) o_ext, ext_y = {}, 140 for x in PROFILE["external"]: h = 130 if x.get("artifacts") else 70 o_ext[x["id"]] = obj(v1, x["id"], 1320, ext_y, 330, h) for j, a in enumerate(x.get("artifacts", [])): obj(v1, a["id"], 15 + j * 155, 50, 145, 60, o_ext[x["id"]]) ext_y += h + 20 o_front = obj(v1, fn["id"], 20, 570, 300, 60) for ts in SERVICES: for r in ts["realizers"]: conn("Realization", o_ss.get(r) or o_ext[r], o_ts[ts["id"]]) if ts.get("serves_software"): conn("Serving", o_ts[ts["id"]], o_ss[ts["serves_software"]]) conn("Assignment", o_front, o_ss[fn["assigned_to"]]) for x in PROFILE["external"]: for tgt in serves_of(x): to = o_cluster if tgt == "cluster" else (o_ss.get(tgt) or o_ext[tgt]) conn("Serving", o_ext[x["id"]], to) # --- представление «Развёртывание по узлам» GROUPS, EXTRA_USES = PROFILE.get("node_groups", []), {} if GROUPS: claimed = {a for g in GROUPS if isinstance(g.get("runs_apps"), list) for a in g["runs_apps"]} for g in GROUPS: el(T, g["id"], "Node", g["name"], g.get("doc", "")) g["_hosts"] = [] for hn in g.get("hosts", []): hid = "nd-host-" + hn el(T, hid, "Node", hn); R("Composition", g["id"], hid); g["_hosts"].append(hid) g["_apps"] = ([a for a in DATA["apps"] if a not in claimed] if g.get("runs_apps") == "*" else list(g.get("runs_apps") or [])) for a in g["_apps"]: R("Assignment", g["id"], app_id[a]) for s in g.get("runs_software", []): R("Aggregation", g["id"], s) # компоненты, которых нет в манифестах кластера (заданы профилем вручную) g["_extra"] = [] for c in g.get("components", []): el("application", c["id"], "ApplicationComponent", c["name"], APPDOCS.get(c["id"], c.get("doc", ""))) R("Assignment", g["id"], c["id"]); g["_extra"].append(c["id"]) for ts_id in c.get("uses", []): R("Serving", ts_id, c["id"]) EXTRA_USES.setdefault(ts_id, []).append(c["id"]) vd = view("T2. Развёртывание по узлам", "Какие группы узлов какие компоненты несут. Вложенность заменяет стрелки: " "хост внутри группы — composition, приложение — assignment, системное ПО — aggregation. " "Имена хостов заданы в профиле контура.") def host_row(v, g, parent, y): for i, hid in enumerate(g["_hosts"]): obj(v, hid, 14 + i * 125, y, 115, 44, parent) return y + (54 if g["_hosts"] else 0) # левая колонка — небольшие группы left = [g for g in GROUPS if g.get("runs_apps") != "*"] y = 20 for g in left: items = g["_apps"] + list(g.get("runs_software", [])) + g["_extra"] rows = (len(items) + 1) // 2 h = 40 + (54 if g["_hosts"] else 0) + rows * 62 + 10 og = obj(vd, g["id"], 20, y, 400, h) yy = host_row(vd, g, og, 34) for i, it in enumerate(items): eid = app_id.get(it, it) obj(vd, eid, 14 + (i % 2) * 195, yy + (i // 2) * 62, 180, 50, og) y += h + 20 # центр — группа прикладных серверов big = next((g for g in GROUPS if g.get("runs_apps") == "*"), None) if big: per, bw = 6, 180 rows = (len(big["_apps"]) + per - 1) // per h = 40 + (54 if big["_hosts"] else 0) + rows * 62 + 10 og = obj(vd, big["id"], 450, 20, per * bw + 28, max(h, y - 40)) yy = host_row(vd, big, og, 34) for i, a in enumerate(big["_apps"]): obj(vd, app_id[a], 14 + (i % per) * bw, yy + (i // per) * 62, bw - 15, 50, og) # --- представление «Внешние интеграции» IV = PROFILE.get("integration_view") if IV: vi = view(f"{tnum()}. {IV['name']}", IV.get("doc", "")) CW, CH = 320, 110 shown = {} for ci, col in enumerate(IV["columns"]): for ri, eid in enumerate(col): shown[eid] = obj(vi, eid, 20 + ci * (CW + 60), 20 + ri * CH, CW, 70) # связи не перечисляем: берём те, что уже есть в модели между показанными # элементами, — иначе представление начнёт расходиться с моделью for _rid, rtype, src, tgt in relations: if src in shown and tgt in shown: conn(rtype, shown[src], shown[tgt]) # --- прикладной слой: смысловые домены # Строится последним из прикладного, чтобы захватить и компоненты, заданные # в node_groups вручную: в матрице контура их нет. DOMAINS = PROFILE.get("domains", []) if DOMAINS: app_ids = {e[1] for e in elements if e[0] == "application"} claimed = {app_id.get(a, a) for g in DOMAINS if isinstance(g.get("apps"), list) for a in g["apps"]} for g in DOMAINS: want = ([e for e in app_ids if e not in claimed] if g.get("apps") == "*" else [app_id.get(a, a) for a in g.get("apps") or []]) missing = [e for e in want if e not in app_ids] if missing: sys.exit(f"домен {g['id']}: нет таких компонентов — {', '.join(missing)}") g["_apps"] = sorted(want) # пустая группа — не элемент модели, а незаполненная строка профиля DOMAINS = [g for g in DOMAINS if g["_apps"]] for g in DOMAINS: el("other", g["id"], "Grouping", g["name"], g.get("doc", "")) for eid in g["_apps"]: R("Aggregation", g["id"], eid) va = view("A0. Карта прикладного слоя", f"Из чего состоит прикладной слой: {len(app_ids)} " f"{plural(len(app_ids), 'компонент', 'компонента', 'компонентов')} по смысловым " f"группам. Связи здесь намеренно не показаны — на таком числе компонентов они " f"делают схему нечитаемой; для них есть представления A1 и далее. Вложенность " f"означает aggregation: группа объединяет компоненты, но не владеет ими.") PER = 7 y = 20 for g in DOMAINS: rows = (len(g["_apps"]) + PER - 1) // PER h = 34 + rows * 62 + 8 og = obj(va, g["id"], 20, y, PER * 180 + 28, h) for i, eid in enumerate(g["_apps"]): obj(va, eid, 14 + (i % PER) * 180, 34 + (i // PER) * 62, 165, 50, og) y += h + 20 # --- прикладной слой: межсервисные вызовы if CALLS: import math, collections known = set(DATA["apps"]) calls = [(e["from"], e["to"], e["src"]) for e in CALLS["edges"] if e["from"] in known and e["to"] in known] # если A зовёт B, то в ArchiMate B обслуживает A for a, b, _ in calls: R("Serving", app_id[b], app_id[a]) inbound = collections.Counter(b for _, b, _ in calls) outbound = collections.Counter(a for a, _, _ in calls) # A1: ядро — самые связанные сервисы и связи между ними, по кругу core = [a for a, _ in collections.Counter( {a: inbound[a] + outbound[a] for a in known}).most_common(8)] vc = view("A1. Ядро прикладного слоя", f"Восемь сервисов с наибольшим числом связей и вызовы между ними. " f"Стрелка идёт от вызываемого к вызывающему (serving): кого лишишься — " f"тот и сломается. Всего в контуре {len(calls)} межсервисных связей.") # координаты считаем от нуля, затем сдвигаем: обменный формат не допускает # отрицательных x/y, а круговая раскладка их легко даёт rad, pos = 300, {} for i, a in enumerate(core): ang = 2 * math.pi * i / len(core) - math.pi / 2 pos[a] = (int(rad * math.cos(ang) * 1.35) - 85, int(rad * math.sin(ang)) - 27) dx, dy = 20 - min(x for x, _ in pos.values()), 20 - min(y for _, y in pos.values()) core_obj = {a: obj(vc, app_id[a], x + dx, y + dy, 170, 55) for a, (x, y) in pos.items()} for a, b, _ in calls: if a in core_obj and b in core_obj: conn("Serving", core_obj[b], core_obj[a]) # A2…: по представлению на каждый сервис с тремя и более потребителями a_no, hub_view = 2, {} for hub, n in inbound.most_common(): if n < MIN_LINKS: break callers = sorted(a for a, b, _ in calls if b == hub) v = view(f"A{a_no}. Зависят от: {hub}", f"К {hub} обращаются {n} {plural(n, 'сервис', 'сервиса', 'сервисов')}.") hub_view[hub] = f"A{a_no}" a_no += 1 oh = obj(v, app_id[hub], 20, 20, 260, 60) for i, a in enumerate(callers): conn("Serving", oh, obj(v, app_id[a], 20 + (i % 6) * 190, 150 + (i // 6) * 80, 170, 55)) # P1…: зеркало к A-представлениям. Те отвечают «кто сломается без сервиса», # эти — «без чего не работает он сам»: технологические сервисы сверху, # вызываемые им приложения снизу. p_no = 1 for hub, n in outbound.most_common(): if n < MIN_LINKS: break callees = sorted({b for a, b, _ in calls if a == hub}) uses = [ts for ts in SERVICES if hub in consumers[ts["id"]]] back = (f"Кто зависит от самого {hub} — представление {hub_view[hub]}." if hub in hub_view else f"Входящих вызовов у {hub} меньше {MIN_LINKS}, отдельного представления для них нет.") m = len(callees) v = view(f"P{p_no}. Паспорт: {hub}", f"Что нужно, чтобы {hub} работал: {len(uses)} " f"{plural(len(uses), 'технологический сервис', 'технологических сервиса', 'технологических сервисов')} " f"и {m} {plural(m, 'прикладной сервис', 'прикладных сервиса', 'прикладных сервисов')}. " f"Стрелки, как и везде, идут от обслуживающего к обслуживаемому. {back}") p_no += 1 wide = min(6, max(len(uses), m)) oh = obj(v, app_id[hub], 20 + max(0, (wide * 190 - 260) // 2), 20 + 80 * ((len(uses) + 5) // 6), 260, 60) for i, ts in enumerate(uses): conn("Serving", obj(v, ts["id"], 20 + (i % 6) * 190, 20 + (i // 6) * 80, 170, 55), oh) base = oh["b"][1] + 130 for i, b in enumerate(callees): conn("Serving", obj(v, app_id[b], 20 + (i % 6) * 190, base + (i // 6) * 80, 170, 55), oh) PER_ROW = 6 for ts in SERVICES: lst = consumers[ts["id"]] extra = EXTRA_USES.get(ts["id"], []) if not lst and not extra: continue n = len(lst) v = view(f"{tnum()}. Потребители: {ts['name']}", f"Сервисом {plural(n, 'пользуется', 'пользуются', 'пользуются')} " f"{n} из {len(DATA['apps'])} приложений контура.") ots = obj(v, ts["id"], 20, 20, 260, 60) for i, r in enumerate(ts["realizers"]): conn("Realization", obj(v, r, 320 + i * 280, 20, 260, 60), ots) for i, eid in enumerate([app_id[a] for a in lst] + extra): conn("Serving", ots, obj(v, eid, 20 + (i % PER_ROW) * 190, 150 + (i // PER_ROW) * 80, 170, 55)) # --- легенда нотации # Собрана из настоящих элементов модели, а не из выдуманных ради неё: так легенда # заодно оказывается разобранным примером и не может разойтись с остальными схемами. LEG_TS = max((t for t in SERVICES if consumers[t["id"]] and t["realizers"][0] in SW), key=lambda t: len(consumers[t["id"]]), default=None) if LEG_TS and GROUPS and DOMAINS: l_ss = LEG_TS["realizers"][0] l_app = app_id[consumers[LEG_TS["id"]][0]] l_gsw = next((g for g in GROUPS if l_ss in (g.get("runs_software") or [])), None) l_gap = next((g for g in GROUPS if any(app_id[a] == l_app for a in g["_apps"])), None) l_dm = next((g for g in DOMAINS if l_app in g["_apps"]), None) l_art = next((e[1] for e in elements if e[2] == "Artifact"), None) vl = view("L1. Как читать схемы", "Легенда нотации. Показан настоящий фрагмент модели, подписанный по типам: " "каждый элемент и каждая линия здесь встречаются на остальных представлениях " "в том же начертании.") W, COL = 260, 320 note(vl, "Стрелка serving читается не как «кто кого вызывает», а как «кто без кого не " "работает»: она идёт от обслуживающего к обслуживаемому. Направление выбрано так " "потому, что от схемы обычно нужен ответ на вопрос «что откажет, если убрать этот " "элемент», — и он читается по стрелкам вперёд.", 20, 20, COL * 4 - 60, 66) row_a = [ (nd_cluster, "Узел (Node) — вычислительная среда: кластер, группа серверов или " "отдельный хост. Ромб на линии — composition: узел состоит из этого, " "и отдельно от узла содержимое не существует."), (l_ss, "Системное ПО (System Software) — платформенный компонент внутри узла. " "Пунктир с полым треугольником — realization: компонент реализует " "технологический сервис."), (LEG_TS["id"], "Технологический сервис (Technology Service) — что платформа даёт " "приложениям, без указания, чем именно это сделано. Приложение зависит " "от сервиса, а не от конкретного продукта под ним."), (l_app, "Прикладной компонент (Application Component) — сервис прикладного слоя. " "Цветом различаются слои: голубой — прикладной, зелёный — технологический."), ] o_leg = {} for i, (eid, txt) in enumerate(row_a): o_leg[eid] = obj(vl, eid, 20 + i * COL, 120, W, 60) note(vl, txt, 20 + i * COL, 190, W, 96) conn("Composition", o_leg[nd_cluster], o_leg[l_ss]) conn("Realization", o_leg[l_ss], o_leg[LEG_TS["id"]]) conn("Serving", o_leg[LEG_TS["id"]], o_leg[l_app]) row_b = [ (1, l_gsw and l_gsw["id"], l_ss, "Aggregation", "Линия с полым ромбом — aggregation: группа объединяет элемент, но тот может " "существовать и без неё. Так показано, на каких узлах работает системное ПО."), (2, l_gap and l_gap["id"], l_app, "Assignment", "Линия с шариком и закрашенной стрелкой — assignment: узел выделен под компонент, " "и компонент на нём исполняется."), (3, l_dm and l_dm["id"], l_app, "Aggregation", "Смысловая группа (Grouping) — домен прикладного слоя. Тоже aggregation: " "компонент принадлежит домену по смыслу, а не по месту размещения."), ] for i, src, tgt, rtype, txt in row_b: if not src: continue os_ = obj(vl, src, 20 + i * COL, 320, W, 60) note(vl, txt, 20 + i * COL, 390, W, 96) conn(rtype, os_, o_leg[tgt]) if l_art: obj(vl, l_art, 20, 320, W, 60) note(vl, "Артефакт (Artifact) — то, что физически лежит в хранилище: файл, " "образ контейнера, дистрибутив.", 20, 390, W, 96) # --- порядок представлений в файле: легенда, технологический слой, прикладной, паспорта ORDER = {"L": 0, "T": 1, "A": 2, "P": 3} views.sort(key=lambda v: (ORDER.get(v["name"][0], 9), int(re.match(r"[A-Z](\d+)", v["name"]).group(1)))) # ---------------------------------------------------------------- сериализация FOLDERS =[("Strategy", "strategy"), ("Business", "business"), ("Application", "application"), ("Technology & Physical", "technology"), ("Motivation", "motivation"), ("Implementation & Migration", "implementation_migration"), ("Other", "other"), ("Relations", "relations"), ("Views", "diagrams")] def emit_obj(o, ind): # Имена тегов — как их пишет сам Archi: , # соединение через атрибут relationship. Имена EClass из метамодели # (DiagramModelArchimateObject) в XML не используются. p = " " * ind tgt = f' targetConnections={quoteattr(" ".join(o["tgt"]))}' if o["tgt"] else "" x, y, w, h = o["b"] if "note" in o: return (f'{p}\n' f'{p} \n' f'{p} {escape(o["note"])}\n' f'{p}\n') s = f'{p}\n' s += f'{p} \n' for c in o["src"]: s += (f'{p} \n') for c in o["children"]: s += emit_obj(c, ind + 2) return s + f'{p}\n' out = ['', '', f' {escape(PROFILE["model"]["purpose"])}'] for fname, ftype in FOLDERS: inner = "" if ftype in ("technology", "application", "other"): for folder, eid, etype, name, doc in elements: if folder != ftype: continue if doc: inner += (f' \n' f' {escape(doc)}\n \n') else: inner += f' \n' elif ftype == "relations": for rid, rtype, src, tgt in relations: inner += (f' \n') elif ftype == "diagrams": for v in views: inner += f' \n' if v["doc"]: inner += f' {escape(v["doc"])}\n' for o in v["objects"]: inner += emit_obj(o, 6) inner += ' \n' out.append(f' \n{inner} ' if inner else f' ') out.append('') OUT.write_text("\n".join(out) + "\n", encoding="utf-8", newline="\n") print(f"записано: {OUT}") print(f"элементов: {len(elements)} отношений: {len(relations)} представлений: {len(views)}") for v in views: print(f" {v['name']}")