emoon-backend/scripts/render_ai_platform_architecture.py

from PIL import Image, ImageDraw, ImageFont
from pathlib import Path
import textwrap


ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "docs/assets/ai-platform-module-architecture-map.png"

W, H = 5250, 3300
BG = "#F6FAFF"
INK = "#102A5C"
MUTED = "#536B90"
BLUE = "#0D63CE"
CYAN = "#00A7C7"
PURPLE = "#6B3FD6"
GREEN = "#10A37F"
ORANGE = "#F28C28"
RED = "#D94B4B"
LINE = "#B9D4F4"
CARD = "#FFFFFF"


def font(size, bold=False):
    candidates = [
        "/System/Library/Fonts/Hiragino Sans GB.ttc",
        "/System/Library/Fonts/STHeiti Medium.ttc" if bold else "/System/Library/Fonts/STHeiti Light.ttc",
        "/System/Library/Fonts/Supplemental/Arial Unicode.ttf",
    ]
    for p in candidates:
        if Path(p).exists():
            return ImageFont.truetype(p, size=size, index=0)
    return ImageFont.load_default()


F = {
    "title": font(74, True),
    "sub": font(34),
    "h1": font(42, True),
    "h2": font(34, True),
    "body": font(27),
    "small": font(23),
    "tiny": font(20),
    "mono": font(22),
}


img = Image.new("RGB", (W, H), BG)
draw = ImageDraw.Draw(img)


def text_size(s, ft):
    box = draw.textbbox((0, 0), s, font=ft)
    return box[2] - box[0], box[3] - box[1]


def wrap_text(text, ft, max_w):
    lines = []
    for para in text.split("\n"):
        current = ""
        for ch in para:
            if text_size(current + ch, ft)[0] <= max_w:
                current += ch
            else:
                if current:
                    lines.append(current)
                current = ch
        if current:
            lines.append(current)
    return lines


def rounded(x, y, w, h, fill=CARD, outline=LINE, radius=28, width=3):
    draw.rounded_rectangle([x, y, x + w, y + h], radius=radius, fill=fill, outline=outline, width=width)


def shadow_card(x, y, w, h, fill=CARD, outline=LINE, radius=28):
    draw.rounded_rectangle([x + 10, y + 12, x + w + 10, y + h + 12], radius=radius, fill="#D8E8FA")
    rounded(x, y, w, h, fill, outline, radius)


def label(text, x, y, fill=INK, ft=None, anchor=None):
    draw.text((x, y), text, font=ft or F["body"], fill=fill, anchor=anchor)


def pill(text, x, y, fill, fg="#FFFFFF", pad_x=22, pad_y=10, ft=None):
    ft = ft or F["small"]
    tw, th = text_size(text, ft)
    draw.rounded_rectangle([x, y, x + tw + pad_x * 2, y + th + pad_y * 2], radius=22, fill=fill)
    label(text, x + pad_x, y + pad_y - 2, fg, ft)
    return tw + pad_x * 2


def card(title, lines, x, y, w, h, accent=BLUE, tag=None, subtitle=None):
    shadow_card(x, y, w, h, CARD, "#A9C8EE", 22)
    draw.rounded_rectangle([x, y, x + w, y + 66], radius=22, fill=accent)
    draw.rectangle([x, y + 42, x + w, y + 66], fill=accent)
    label(title, x + 24, y + 14, "#FFFFFF", F["h2"])
    if tag:
        pill(tag, x + w - text_size(tag, F["tiny"])[0] - 84, y + 14, "#FFFFFF", accent, 16, 8, F["tiny"])
    ty = y + 88
    if subtitle:
        label(subtitle, x + 24, ty, accent, F["small"])
        ty += 38
    for ln in lines:
        for wrapped in wrap_text("• " + ln, F["small"], w - 56):
            label(wrapped, x + 28, ty, INK, F["small"])
            ty += 34
        ty += 2


def layer_band(idx, name, desc, y, h, color):
    draw.rounded_rectangle([80, y, W - 80, y + h], radius=30, fill="#EEF6FF", outline="#C8DDF5", width=2)
    draw.rounded_rectangle([96, y + 18, 310, y + h - 18], radius=24, fill=color)
    label(idx, 203, y + 48, "#FFFFFF", F["h1"], "ma")
    label(name, 340, y + 30, INK, F["h2"])
    label(desc, 340, y + 76, MUTED, F["small"])


def arrow(x1, y1, x2, y2, color=BLUE, width=8):
    draw.line([x1, y1, x2, y2], fill=color, width=width)
    import math
    ang = math.atan2(y2 - y1, x2 - x1)
    size = 24
    p1 = (x2, y2)
    p2 = (x2 - size * math.cos(ang - 0.45), y2 - size * math.sin(ang - 0.45))
    p3 = (x2 - size * math.cos(ang + 0.45), y2 - size * math.sin(ang + 0.45))
    draw.polygon([p1, p2, p3], fill=color)


# Title
label("医梦 AI 中台 Maven 模块演进与依赖关系图", W // 2, 70, INK, F["title"], "ma")
label("现状 emoon-mcp-api 先不动代码，仅按目标拆分为 ai-agent-api / ai-card-api / ai-mcp-api 进行架构设计", W // 2, 154, MUTED, F["sub"], "ma")
pill("大厂技术培训风", 350, 230, BLUE)
pill("高密度强层级", 660, 230, CYAN)
pill("现状 + 规划同屏", 965, 230, PURPLE)
pill("初级工程师可读", 1325, 230, GREEN)

shadow_card(1960, 215, 3160, 92, "#FFFFFF", "#8DB7E8", 22)
label("依赖红线", 2000, 238, RED, F["small"])
compact_rules = [
    "入口层只依赖 API 契约",
    "Card 不直连 MCP",
    "计量先落库再入账",
    "敏感数据不进外部日志/prompt 明文",
    "三开发优先模块化单体",
]
rx = 2200
for r in compact_rules:
    rx += pill(r, rx, 232, "#EAF4FF", INK, 14, 7, F["tiny"]) + 18


# L0 external touchpoints
layer_band("L0", "业务触点层", "用户和业务系统进入 AI 中台的入口，不直接依赖底层实现模块。", 330, 310, BLUE)
l0 = [
    ("患者端 / 小程序", ["问诊、报告解读、随访、建档"]),
    ("医生工作站", ["舌诊、面诊、病历生成、医嘱建议"]),
    ("院内运营后台", ["能力配置、合同、套餐、统计报表"]),
    ("医院系统集成", ["HIS / EMR / LIS / PACS 调用"]),
    ("IoMT / 语音设备", ["语音陪诊、床旁设备、可穿戴数据"]),
]
x = 520
for title, lines in l0:
    card(title, lines, x, 420, 850, 150, BLUE, "外部调用")
    x += 900


# L1 application entry
layer_band("L1", "应用入口层", "保持当前业务入口，向下只调用领域 API，不跨层访问具体实现。", 690, 330, CYAN)
l1 = [
    ("emoon-openplatform", ["开放接口、SSE 会话、院内集成入口", "当前已依赖 emoon-mcp-api"]),
    ("emoon-admin", ["平台运营后台、租户配置、能力包管理", "三期补齐计费、统计、审计运营页"]),
    ("emoon-app / 小程序端", ["患者问诊、报告、随访、AI 卡片交互", "只认业务 API 和 SSE 事件"]),
    ("emoon-extend / worker", ["文件处理、异步任务、外部通道适配", "后续承接 OCR/ASR/回调类任务"]),
]
x = 540
for title, lines in l1:
    card(title, lines, x, 780, 1040, 170, CYAN, "入口")
    x += 1140


# L2 current maven baseline and split
layer_band("L2", "当前 Maven 基座与目标 API 拆分", "不做微服务先行；先把 API 契约从 emoon-mcp-api 中按职责拆清楚。", 1070, 610, PURPLE)
card("当前 emoon-infra / emoon-modules-api", [
    "emoon-chat-api：聊天/消息领域契约",
    "emoon-system-api：租户、用户、字典、权限基础能力",
    "emoon-knowledge-api：知识库与文档能力契约",
    "emoon-mcp-api：当前混放 Agent、Card、MCP、三方 SDK 依赖",
], 520, 1180, 1250, 360, PURPLE, "现状")

card("emoon-mcp-api（现状待拆）", [
    "包含 AgentEngine / AgentRequest / AgentResponse",
    "包含 AiConversation / AiCardInstance 等领域对象",
    "依赖偏重：web、sse、websocket、langchain4j、pdfbox、tika 等",
    "问题：API 契约和实现依赖混在一起，调用边界不清晰",
], 1910, 1150, 1320, 420, RED, "不要先动代码")

target_cards = [
    ("ai-agent-api", ["统一 AI 调用契约", "会话、SSE、编排、Dify/DirectLLM 引擎抽象", "只放 DTO / SPI / 轻量接口"]),
    ("ai-card-api", ["AI 卡片契约", "卡片定义、实例、事件、状态、渲染数据", "面向前端和业务场景复用"]),
    ("ai-mcp-api", ["MCP 工具契约", "Tool Registry、Tool Call、权限、审计元数据", "仅表达工具协议，不承载 Agent 编排"]),
]
for i, (t, lines) in enumerate(target_cards):
    card(t, lines, 3420, 1125 + i * 165, 1500, 140, [BLUE, GREEN, ORANGE][i], "目标 API")

arrow(1775, 1360, 1900, 1360, RED, 10)
arrow(3235, 1255, 3405, 1195, BLUE, 8)
arrow(3235, 1360, 3405, 1360, GREEN, 8)
arrow(3235, 1465, 3405, 1525, ORANGE, 8)
label("拆分原则：API 包只定义契约；实现、持久化、三方 SDK 依赖下沉到对应业务模块", 3460, 1600, RED, F["small"])


# L3 target implementation modules
layer_band("L3", "目标领域实现模块", "仍保持 Maven 模块化单体优先，模块内高内聚，模块间通过 API 契约调用。", 1740, 620, GREEN)
impl = [
    ("ai-agent", BLUE, ["实现 ai-agent-api", "AgentEngineFactory、DifyEngine、DirectLLMEngine", "SSE 事件、幂等、降级、Outbox"]),
    ("ai-card", GREEN, ["实现 ai-card-api", "卡片定义/实例/版本/事件流", "卡片只经 Agent 调 MCP，不直连工具"]),
    ("ai-mcp", ORANGE, ["实现 ai-mcp-api", "工具注册、权限、调用路由、审计", "院内工具适配 HIS/EMR/LIS/PACS"]),
    ("ai-meter", PURPLE, ["计量事件中心", "Token/次数/分钟/项目/年费统一归集", "8 状态机、重放、幂等去重"]),
    ("ai-billing", RED, ["合同、能力值包、预扣款、结算", "pricing snapshot、special adjustment", "合理不限量与跨账户划拨"]),
    ("ai-knowledge", CYAN, ["知识库、模板、病种规则、提示词资产", "对接现有 emoon-knowledge-api", "支撑报告、病历、质控场景"]),
]
x0, y0 = 520, 1850
for i, (t, c, lines) in enumerate(impl):
    x = x0 + (i % 3) * 1540
    y = y0 + (i // 3) * 235
    card(t, lines, x, y, 1420, 190, c, "实现模块")

for sx, sy, tx, ty, c in [
    (4170, 1265, 1230, 1825, BLUE),
    (4170, 1425, 2770, 1825, GREEN),
    (4170, 1590, 4310, 1825, ORANGE),
    (1250, 2090, 2770, 2090, PURPLE),
    (2770, 2090, 4310, 2090, RED),
]:
    arrow(sx, sy, tx, ty, c, 7)


# L4 future independent boundaries
layer_band("L4", "未来可独立部署边界", "三开发阶段不建议先拆微服务；当吞吐、隔离或交付责任成熟后，再按这些边界拆。", 2420, 360, ORANGE)
future = [
    ("MCP Tool Server", ["医院内网工具调用隔离", "工具权限、审计、限流独立"]),
    ("AudioStreamGateway", ["电话/语音陪诊流式接入", "ASR/TTS、按分钟计量"]),
    ("IoMTEventIngestion", ["设备事件接入与清洗", "告警、趋势、随访触发"]),
    ("OutboundFollowup", ["随访任务、回访、消息触达", "失败重试与运营统计"]),
    ("File/OCR/ASR Worker", ["报告、影像、病历附件异步处理", "长任务与主链路解耦"]),
]
x = 520
for title, lines in future:
    card(title, lines, x, 2510, 850, 165, ORANGE, "可拆服务")
    x += 900


# L5 external systems + data base
layer_band("L5", "外部系统与数据可靠性底座", "所有调用、计量、账务、审计必须形成闭环，避免医疗场景不可追溯。", 2840, 360, BLUE)
base = [
    ("AI 引擎", ["Dify Workflow / DirectLLM / Vision / Mock"]),
    ("院内系统", ["HIS / EMR / LIS / PACS / 预约挂号"]),
    ("数据存储", ["MySQL：业务状态", "Redis：锁/限流/短期缓存"]),
    ("可靠性", ["Outbox：异步一致性", "幂等键：API/Meter/Tool 分层"]),
    ("账务审计", ["Credit Ledger：能力值流水", "Audit/Trace：全链路追踪"]),
]
x = 520
for title, lines in base:
    card(title, lines, x, 2930, 850, 165, BLUE, "底座")
    x += 900


shadow_card(80, 3180, W - 160, 82, "#EAF4FF", "#A9C8EE", 22)
label("读图顺序：先看 L2 的 emoon-mcp-api 现状待拆，再看右侧三类目标 API，最后沿箭头进入 L3 实现模块和 L5 计量/账务/审计闭环。", W // 2, 3210, INK, F["small"], "ma")

OUT.parent.mkdir(parents=True, exist_ok=True)
img.save(OUT, "PNG", optimize=True)
print(OUT)