feat: add draw-prototype-flow

This commit is contained in:
2026-08-04 13:55:32 +08:00
parent 5ff8899b48
commit c113f68bf4
8 changed files with 975 additions and 0 deletions
+259
View File
@@ -0,0 +1,259 @@
#!/usr/bin/env python3
"""Generate an editable multi-page Draw.io prototype skeleton from JSON."""
from __future__ import annotations
import argparse
import json
import re
import sys
import xml.etree.ElementTree as ET
from pathlib import Path
FRAME_W = 1600
FRAME_H = 900
GAP_X = 120
GAP_Y = 260
STYLE = {
"label": "text;html=0;strokeColor=none;fillColor=none;fontColor=#10284B;fontSize=24;fontStyle=1;fontFamily=Microsoft YaHei;",
"bg": "rounded=0;whiteSpace=wrap;html=0;fillColor=#F4F7FB;strokeColor=#D9E2EF;",
"nav": "rounded=0;whiteSpace=wrap;html=0;fillColor=#061B3A;strokeColor=#061B3A;fontColor=#C8D6EA;fontSize=16;fontFamily=Microsoft YaHei;align=left;verticalAlign=top;spacingTop=32;spacingLeft=32;",
"title": "text;html=0;strokeColor=none;fillColor=none;fontColor=#10284B;fontSize=30;fontStyle=1;fontFamily=Microsoft YaHei;align=left;",
"card": "rounded=1;whiteSpace=wrap;html=0;fillColor=#FFFFFF;strokeColor=#D9E2EF;fontColor=#718198;fontSize=16;fontFamily=Microsoft YaHei;arcSize=8;",
"primary": "rounded=1;whiteSpace=wrap;html=0;fillColor=#1768E8;strokeColor=#1768E8;fontColor=#FFFFFF;fontSize=16;fontStyle=1;fontFamily=Microsoft YaHei;arcSize=10;",
"secondary": "rounded=1;whiteSpace=wrap;html=0;fillColor=#FFFFFF;strokeColor=#AFC1DA;fontColor=#27405F;fontSize=16;fontStyle=1;fontFamily=Microsoft YaHei;arcSize=10;",
"meta": "text;html=0;strokeColor=none;fillColor=none;fontColor=#718198;fontSize=13;fontFamily=Microsoft YaHei;align=left;",
"table_head": "rounded=0;whiteSpace=wrap;html=0;fillColor=#EAF0F8;strokeColor=#D9E2EF;fontColor=#263750;fontSize=14;fontStyle=1;fontFamily=Microsoft YaHei;align=left;spacingLeft=18;",
"table_row": "rounded=0;whiteSpace=wrap;html=0;fillColor=#FFFFFF;strokeColor=#D9E2EF;fontColor=#40516A;fontSize=14;fontFamily=Microsoft YaHei;align=left;spacingLeft=18;",
}
EDGE_COLORS = {
"primary": "#1768E8",
"success": "#18864B",
"warning": "#D97706",
"danger": "#D92D20",
"secondary": "#5B6ABF",
}
def args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("spec", type=Path, help="JSON scaffold specification")
parser.add_argument("output", type=Path, help="Target .drawio path")
parser.add_argument("--force", action="store_true", help="Overwrite an existing output file")
return parser.parse_args()
def safe_id(value: str) -> str:
value = re.sub(r"[^A-Za-z0-9_-]+", "-", value).strip("-").lower()
return value or "frame"
def geometry(parent: ET.Element, x: int, y: int, width: int, height: int) -> None:
ET.SubElement(parent, "mxGeometry", {
"x": str(x), "y": str(y), "width": str(width), "height": str(height), "as": "geometry"
})
def cell(root: ET.Element, cell_id: str, value: str, style: str, x: int, y: int,
width: int, height: int) -> ET.Element:
node = ET.SubElement(root, "mxCell", {
"id": cell_id, "value": value, "style": style, "vertex": "1", "parent": "1"
})
geometry(node, x, y, width, height)
return node
def edge(root: ET.Element, edge_id: str, value: str, source: str, target: str, kind: str) -> None:
color = EDGE_COLORS.get(kind, EDGE_COLORS["primary"])
node = ET.SubElement(root, "mxCell", {
"id": edge_id,
"value": value,
"style": (
"edgeStyle=orthogonalEdgeStyle;rounded=1;html=0;endArrow=classic;endFill=1;"
f"strokeWidth=2;strokeColor={color};fontColor=#40516A;"
),
"edge": "1",
"parent": "1",
"source": source,
"target": target,
})
ET.SubElement(node, "mxGeometry", {"relative": "1", "as": "geometry"})
def screen_frame(root: ET.Element, prefix: str, number: int, frame: dict, x: int, y: int,
product: str) -> None:
stable_id = frame["id"]
title = frame["title"]
state = frame.get("state", "")
role = frame.get("role", "")
suffix = f" · {state}" if state else ""
cell(root, f"{prefix}-label", f"{number:02d} · {stable_id} · {title}{suffix}", STYLE["label"], x, y, 1300, 40)
top = y + 55
cell(root, f"{prefix}-bg", "", STYLE["bg"], x, top, FRAME_W, FRAME_H)
cell(root, f"{prefix}-nav", f"{product}\n工作台\n核心任务\n数据概览\n系统设置", STYLE["nav"], x, top, 260, FRAME_H)
cell(root, f"{prefix}-title", title, STYLE["title"], x + 310, top + 50, 700, 50)
cell(root, f"{prefix}-card", "在此绘制页面内容、字段、表格和反馈状态", STYLE["card"], x + 310, top + 150, 1240, 560)
cell(root, f"{prefix}-secondary", "次要操作", STYLE["secondary"], x + 1060, top + 745, 200, 50)
cell(root, f"{prefix}-primary", "主要操作", STYLE["primary"], x + 1310, top + 745, 240, 50)
cell(root, f"{prefix}-meta", f"角色:{role or '待定义'}|状态:{state or '待定义'}|稳定 ID{stable_id}|下一步:待定义", STYLE["meta"], x + 310, top + 845, 1240, 30)
def flow_frame(root: ET.Element, prefix: str, number: int, frame: dict, x: int, y: int) -> None:
stable_id, title = frame["id"], frame["title"]
cell(root, f"{prefix}-label", f"{number:02d} · {stable_id} · {title}", STYLE["label"], x, y, 1300, 40)
top = y + 55
cell(root, f"{prefix}-bg", "", STYLE["bg"], x, top, FRAME_W, FRAME_H)
cell(root, f"{prefix}-title", title, STYLE["title"], x + 80, top + 55, 900, 50)
cell(root, f"{prefix}-card", "按角色或阶段添加泳道;使用带 source/target 的正交连接线表达动作与结果", STYLE["card"], x + 80, top + 150, 1440, 620)
cell(root, f"{prefix}-meta", f"稳定 ID{stable_id}|覆盖:入口、判断、跨角色交接、成功与失败终态", STYLE["meta"], x + 80, top + 835, 1440, 30)
def flow_graph(root: ET.Element, prefix: str, transitions: list[dict], frames: dict[str, dict],
x: int, y: int) -> None:
ordered_ids: list[str] = []
for transition in transitions:
for stable_id in (transition["from"], transition["to"]):
if stable_id not in ordered_ids:
ordered_ids.append(stable_id)
node_ids: dict[str, str] = {}
for index, stable_id in enumerate(ordered_ids):
col, row = index % 5, index // 5
node_id = f"{prefix}-node-{safe_id(stable_id)}"
node_ids[stable_id] = node_id
frame = frames[stable_id]
cell(
root,
node_id,
f"{stable_id}\n{frame['title']}",
"rounded=1;whiteSpace=wrap;html=0;fillColor=#FFFFFF;strokeColor=#1768E8;"
"fontColor=#10284B;fontSize=14;fontStyle=1;fontFamily=Microsoft YaHei;arcSize=10;",
x + 110 + col * 280,
y + 245 + row * 155,
220,
82,
)
for index, transition in enumerate(transitions):
edge(
root,
f"{prefix}-edge-{index}",
transition["label"],
node_ids[transition["from"]],
node_ids[transition["to"]],
transition.get("kind", "primary"),
)
def spec_frame(root: ET.Element, prefix: str, number: int, frame: dict, x: int, y: int) -> None:
stable_id, title = frame["id"], frame["title"]
cell(root, f"{prefix}-label", f"{number:02d} · {stable_id} · {title}", STYLE["label"], x, y, 1300, 40)
top = y + 55
cell(root, f"{prefix}-bg", "", STYLE["bg"], x, top, FRAME_W, FRAME_H)
cell(root, f"{prefix}-title", title, STYLE["title"], x + 80, top + 55, 900, 50)
cell(root, f"{prefix}-head", "稳定 ID 名称/条件 角色 动作/规则 结果", STYLE["table_head"], x + 80, top + 150, 1440, 58)
for row in range(4):
cell(root, f"{prefix}-row-{row}", "待填写 待填写 待填写 待填写 待填写", STYLE["table_row"], x + 80, top + 208 + row * 66, 1440, 66)
cell(root, f"{prefix}-meta", f"规格 ID{stable_id}|所有引用使用页面/状态/动作稳定 ID", STYLE["meta"], x + 80, top + 835, 1440, 30)
def load_spec(path: Path) -> dict:
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ValueError(f"Cannot read spec: {exc}") from exc
if not isinstance(data.get("pages"), list) or not data["pages"]:
raise ValueError("spec.pages must be a non-empty list")
columns = data.get("columns", 3)
if not isinstance(columns, int) or not 1 <= columns <= 5:
raise ValueError("columns must be an integer from 1 to 5")
seen: set[str] = set()
for page in data["pages"]:
if page.get("kind") not in {"screen", "flow", "spec"}:
raise ValueError(f"Unsupported page kind: {page.get('kind')!r}")
if not page.get("name") or not isinstance(page.get("frames"), list) or not page["frames"]:
raise ValueError("each page needs name and a non-empty frames list")
for frame in page["frames"]:
if not frame.get("id") or not frame.get("title"):
raise ValueError("each frame needs id and title")
if frame["id"] in seen:
raise ValueError(f"duplicate stable frame id: {frame['id']}")
seen.add(frame["id"])
transitions = data.get("transitions", [])
if not isinstance(transitions, list):
raise ValueError("transitions must be a list")
for transition in transitions:
if not all(transition.get(key) for key in ("from", "to", "label")):
raise ValueError("each transition needs from, to and label")
if transition["from"] not in seen or transition["to"] not in seen:
raise ValueError(f"transition references unknown frame: {transition}")
if transition.get("kind", "primary") not in EDGE_COLORS:
raise ValueError(f"unsupported transition kind: {transition.get('kind')!r}")
return data
def build(spec: dict) -> ET.ElementTree:
mxfile = ET.Element("mxfile", {
"host": "app.diagrams.net", "agent": "draw-prototype-flow", "version": "24.7.17",
"type": "device", "compressed": "false", "pages": str(len(spec["pages"]))
})
product = spec.get("title", "产品名称")
columns = spec.get("columns", 3)
frames_by_id = {
frame["id"]: frame
for page in spec["pages"]
for frame in page["frames"]
}
number = 0
for page_index, page in enumerate(spec["pages"]):
diagram = ET.SubElement(mxfile, "diagram", {
"id": f"page-{page_index}-{safe_id(page['name'])}", "name": page["name"]
})
model = ET.SubElement(diagram, "mxGraphModel", {
"dx": "1200", "dy": "800", "grid": "1", "gridSize": "10", "guides": "1",
"tooltips": "1", "connect": "1", "arrows": "1", "fold": "1", "page": "0",
"pageScale": "1", "math": "0", "shadow": "0"
})
root = ET.SubElement(model, "root")
ET.SubElement(root, "mxCell", {"id": "0"})
ET.SubElement(root, "mxCell", {"id": "1", "parent": "0"})
for frame_index, frame in enumerate(page["frames"]):
number += 1
col, row = frame_index % columns, frame_index // columns
x, y = col * (FRAME_W + GAP_X), row * (FRAME_H + GAP_Y) + 15
prefix = f"p{page_index}-f{frame_index}-{safe_id(frame['id'])}"
if page["kind"] == "screen":
screen_frame(root, prefix, number, frame, x, y, product)
elif page["kind"] == "flow":
flow_frame(root, prefix, number, frame, x, y)
else:
spec_frame(root, prefix, number, frame, x, y)
if page["kind"] == "flow" and spec.get("transitions"):
flow_graph(root, f"p{page_index}-auto-flow", spec["transitions"], frames_by_id, 0, 15)
ET.indent(mxfile, space=" ")
return ET.ElementTree(mxfile)
def main() -> int:
options = args()
if options.output.exists() and not options.force:
print(f"Refusing to overwrite existing file: {options.output}", file=sys.stderr)
return 2
try:
spec = load_spec(options.spec)
tree = build(spec)
options.output.parent.mkdir(parents=True, exist_ok=True)
tree.write(options.output, encoding="utf-8", xml_declaration=True)
except (ValueError, OSError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 1
frame_count = sum(len(page["frames"]) for page in spec["pages"])
print(f"Created {options.output}: {len(spec['pages'])} page(s), {frame_count} frame(s)")
return 0
if __name__ == "__main__":
sys.exit(main())
+236
View File
@@ -0,0 +1,236 @@
#!/usr/bin/env python3
"""Validate the structure of an uncompressed Draw.io product prototype."""
from __future__ import annotations
import argparse
import base64
import html
import json
import re
import sys
import urllib.parse
import xml.etree.ElementTree as ET
import zlib
from dataclasses import asdict, dataclass
from pathlib import Path
FRAME_RE = re.compile(r"^(?:\d{2,3}|P-[A-Z0-9-]+)\s*(?:[·.\-—-]|\s)")
NUMERIC_REF_RE = re.compile(r"(?:下一步|关联|进入|见|去向)[^|\n]{0,40}(?<!\d)\d{2,3}(?!\d)")
FLOW_PAGE_RE = re.compile(r"(?:总览|流程|动线|flow)", re.I)
@dataclass
class Issue:
level: str
code: str
message: str
page: str | None = None
cell: str | None = None
def label_text(value: str | None) -> str:
value = html.unescape(value or "")
value = re.sub(r"<br\s*/?>", " ", value, flags=re.I)
value = re.sub(r"<[^>]+>", "", value)
return " ".join(value.replace("\xa0", " ").split())
def is_frame_title(cell: ET.Element) -> bool:
"""Recognize large external frame headings, not numbered table content."""
if not FRAME_RE.match(label_text(cell.get("value"))):
return False
geometry = cell.find("mxGeometry")
if geometry is None:
return False
try:
width = float(geometry.get("width", "0"))
height = float(geometry.get("height", "0"))
except ValueError:
return False
style = cell.get("style", "")
size_match = re.search(r"(?:^|;)fontSize=(\d+(?:\.\d+)?)(?:;|$)", style)
font_size = float(size_match.group(1)) if size_match else 0
return width >= 500 and height <= 80 and font_size >= 20
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("file", type=Path, help="Uncompressed .drawio file")
parser.add_argument("--min-frames", type=int, default=0)
parser.add_argument("--min-edges", type=int, default=0)
parser.add_argument("--strict", action="store_true", help="Treat warnings as failures")
parser.add_argument("--json", action="store_true", dest="as_json")
return parser.parse_args()
def graph_model(diagram: ET.Element) -> tuple[ET.Element | None, bool]:
"""Return a graph model and whether the page used Draw.io compression."""
model = diagram.find("mxGraphModel")
if model is not None:
return model, False
payload = (diagram.text or "").strip()
if not payload:
return None, False
try:
decoded = base64.b64decode(payload)
xml_text = urllib.parse.unquote(zlib.decompress(decoded, -15).decode("utf-8"))
model = ET.fromstring(xml_text)
except (ValueError, UnicodeDecodeError, zlib.error, ET.ParseError):
return None, True
return (model if model.tag == "mxGraphModel" else None), True
def validate(path: Path, min_frames: int, min_edges: int) -> tuple[dict, list[Issue]]:
issues: list[Issue] = []
try:
root = ET.parse(path).getroot()
except (OSError, ET.ParseError) as exc:
return {"file": str(path), "pages": []}, [Issue("error", "xml", str(exc))]
if root.tag != "mxfile":
issues.append(Issue("error", "root", f"Expected mxfile root, found {root.tag!r}"))
diagrams = root.findall("diagram")
if not diagrams:
issues.append(Issue("error", "pages", "No diagram pages found"))
names = [diagram.get("name", "") for diagram in diagrams]
for name in sorted(set(names)):
if name and names.count(name) > 1:
issues.append(Issue("warning", "duplicate-page-name", f"Duplicate page name: {name}"))
page_reports = []
total_frames = 0
total_edges = 0
for index, diagram in enumerate(diagrams, start=1):
page_name = diagram.get("name") or f"page-{index}"
model, compressed = graph_model(diagram)
if model is None:
issues.append(Issue(
"error",
"compressed-page",
"Page has no readable mxGraphModel",
page_name,
))
page_reports.append({"name": page_name, "vertices": 0, "edges": 0, "frames": 0})
continue
if compressed:
issues.append(Issue(
"warning",
"compressed-page",
"Compressed page is readable, but agent-generated output should use compressed=false",
page_name,
))
graph_root = model.find("root")
if graph_root is None:
issues.append(Issue("error", "graph-root", "Missing graph root", page_name))
continue
cells = graph_root.findall("mxCell")
by_id: dict[str, ET.Element] = {}
for cell in cells:
cell_id = cell.get("id")
if not cell_id:
issues.append(Issue("error", "missing-id", "mxCell has no id", page_name))
continue
if cell_id in by_id:
issues.append(Issue("error", "duplicate-id", f"Duplicate cell id: {cell_id}", page_name, cell_id))
by_id[cell_id] = cell
vertices = [cell for cell in cells if cell.get("vertex") == "1"]
edges = [cell for cell in cells if cell.get("edge") == "1"]
frame_cells = [cell for cell in vertices if is_frame_title(cell)]
for cell in cells:
cell_id = cell.get("id")
parent = cell.get("parent")
if parent and parent not in by_id:
issues.append(Issue("error", "dangling-parent", f"Unknown parent {parent}", page_name, cell_id))
if (cell.get("vertex") == "1" or cell.get("edge") == "1") and cell.find("mxGeometry") is None:
issues.append(Issue("error", "geometry", "Vertex/edge has no mxGeometry", page_name, cell_id))
if cell.get("vertex") == "1" and NUMERIC_REF_RE.search(label_text(cell.get("value"))):
issues.append(Issue(
"warning",
"numeric-cross-reference",
"Cross-reference appears to use a display number; use a stable page/state ID",
page_name,
cell_id,
))
for edge in edges:
edge_id = edge.get("id")
source = edge.get("source")
target = edge.get("target")
if not source or not target:
issues.append(Issue("error", "unbound-edge", "Edge needs both source and target", page_name, edge_id))
continue
if source not in by_id:
issues.append(Issue("error", "dangling-source", f"Unknown source {source}", page_name, edge_id))
if target not in by_id:
issues.append(Issue("error", "dangling-target", f"Unknown target {target}", page_name, edge_id))
if not label_text(edge.get("value")):
issues.append(Issue("warning", "unlabelled-edge", "User-flow edge has no action/result label", page_name, edge_id))
if len(frame_cells) > 15:
issues.append(Issue(
"warning",
"crowded-page",
f"{len(frame_cells)} numbered frames on one page; consider splitting into multiple pages",
page_name,
))
if FLOW_PAGE_RE.search(page_name) and not edges:
issues.append(Issue("warning", "no-flow", "Flow/overview page has no linked user-flow edges", page_name))
total_frames += len(frame_cells)
total_edges += len(edges)
page_reports.append({
"name": page_name,
"cells": len(cells),
"vertices": len(vertices),
"edges": len(edges),
"frames": len(frame_cells),
})
if total_frames < min_frames:
issues.append(Issue("error", "min-frames", f"Found {total_frames} frames; expected at least {min_frames}"))
if total_edges < min_edges:
issues.append(Issue("error", "min-edges", f"Found {total_edges} edges; expected at least {min_edges}"))
return {
"file": str(path),
"page_count": len(diagrams),
"frame_count": total_frames,
"edge_count": total_edges,
"pages": page_reports,
}, issues
def main() -> int:
args = parse_args()
report, issues = validate(args.file, args.min_frames, args.min_edges)
errors = sum(issue.level == "error" for issue in issues)
warnings = sum(issue.level == "warning" for issue in issues)
if args.as_json:
print(json.dumps({**report, "issues": [asdict(issue) for issue in issues]}, ensure_ascii=False, indent=2))
else:
print(f"{report['file']}: {report.get('page_count', 0)} page(s), "
f"{report.get('frame_count', 0)} frame(s), {report.get('edge_count', 0)} edge(s)")
for page in report.get("pages", []):
print(f" {page['name']}: {page.get('frames', 0)} frames, "
f"{page.get('vertices', 0)} vertices, {page.get('edges', 0)} edges")
for issue in issues:
where = " / ".join(value for value in (issue.page, issue.cell) if value)
suffix = f" ({where})" if where else ""
print(f" {issue.level.upper()} [{issue.code}] {issue.message}{suffix}")
print(f"Result: {errors} error(s), {warnings} warning(s)")
return 1 if errors or (args.strict and warnings) else 0
if __name__ == "__main__":
sys.exit(main())