# -*- coding: utf-8 -*- from pathlib import Path import math from PIL import Image, ImageDraw, ImageFont BASE_DIR = Path(__file__).resolve().parent FONT_CANDIDATES = [ r"C:\Windows\Fonts\simhei.ttf", r"C:\Windows\Fonts\Dengb.ttf", r"C:\Windows\Fonts\NotoSansSC-VF.ttf", r"C:\Windows\Fonts\Deng.ttf", r"C:\Windows\Fonts\simsun.ttc", ] def make_canvas(width, height): img = Image.new("RGB", (width, height), "#F7F9FC") draw = ImageDraw.Draw(img) return img, draw def load_font(size, bold=False): candidates = [] if bold: candidates.extend([r"C:\Windows\Fonts\Dengb.ttf", r"C:\Windows\Fonts\simhei.ttf"]) candidates.extend(FONT_CANDIDATES) for path in candidates: try: return ImageFont.truetype(path, size=size) except Exception: pass return ImageFont.load_default() class Diagram: def __init__(self, width, height): self.width = width self.height = height self.img, self.draw = make_canvas(width, height) self.font_title = load_font(72, True) self.font_subtitle = load_font(34) self.font_section = load_font(40, True) self.font_box = load_font(33) self.font_small = load_font(29) self.font_note = load_font(31) self.font_edge = load_font(24) def text_size(self, text, font): bbox = self.draw.textbbox((0, 0), text, font=font) return bbox[2] - bbox[0], bbox[3] - bbox[1] def wrap_line(self, line, font, max_width): if not line: return [""] lines = [] current = "" for ch in line: test = current + ch if self.text_size(test, font)[0] <= max_width or not current: current = test else: lines.append(current) current = ch if current: lines.append(current) return lines def wrap_text(self, text, font, max_width): lines = [] for raw in text.split("\n"): lines.extend(self.wrap_line(raw, font, max_width)) return lines def text_center(self, text, cx, cy, max_width, font=None, fill="#172033", line_gap=10): font = font or self.font_box lines = self.wrap_text(text, font, max_width) heights = [self.text_size(line, font)[1] for line in lines] total_h = sum(heights) + line_gap * (len(lines) - 1) y = cy - total_h / 2 for line, h in zip(lines, heights): w, _ = self.text_size(line, font) self.draw.text((cx - w / 2, y), line, font=font, fill=fill) y += h + line_gap def text_left(self, text, x, y, max_width, font=None, fill="#344054", line_gap=10): font = font or self.font_note yy = y for line in self.wrap_text(text, font, max_width): self.draw.text((x, yy), line, font=font, fill=fill) yy += self.text_size(line, font)[1] + line_gap return yy def title(self, title, subtitle): tw, _ = self.text_size(title, self.font_title) self.draw.text(((self.width - tw) / 2, 50), title, font=self.font_title, fill="#111827") sw, _ = self.text_size(subtitle, self.font_subtitle) self.draw.text(((self.width - sw) / 2, 138), subtitle, font=self.font_subtitle, fill="#475467") def box(self, cx, cy, w, h, text, fill="#FFFFFF", outline="#2E5A8A", radius=28, width=5, font=None): font = font or self.font_box bounds = (cx - w / 2, cy - h / 2, cx + w / 2, cy + h / 2) self.draw.rounded_rectangle(bounds, radius=radius, fill=fill, outline=outline, width=width) self.text_center(text, cx, cy, w - 60, font=font) return bounds def panel(self, x, y, w, h, title, fill="#FFFFFF", outline="#C7D2E5"): self.draw.rounded_rectangle((x, y, x + w, y + h), radius=34, fill=fill, outline=outline, width=5) self.draw.text((x + 38, y + 26), title, font=self.font_section, fill="#172033") def diamond(self, cx, cy, w, h, text, fill="#FFF4CC", outline="#B27700"): points = [(cx, cy - h / 2), (cx + w / 2, cy), (cx, cy + h / 2), (cx - w / 2, cy)] self.draw.polygon(points, fill=fill, outline=outline) self.draw.line(points + [points[0]], fill=outline, width=5) self.text_center(text, cx, cy, w - 70, font=self.font_box) return points def arrow(self, points, color="#344054", width=6, label=None, label_pos=None): for p1, p2 in zip(points, points[1:]): self.draw.line((p1[0], p1[1], p2[0], p2[1]), fill=color, width=width) x1, y1 = points[-2] x2, y2 = points[-1] angle = math.atan2(y2 - y1, x2 - x1) size = 24 left = (x2 - size * math.cos(angle - math.pi / 7), y2 - size * math.sin(angle - math.pi / 7)) right = (x2 - size * math.cos(angle + math.pi / 7), y2 - size * math.sin(angle + math.pi / 7)) self.draw.polygon([(x2, y2), left, right], fill=color) if label: lx, ly = label_pos if label_pos else ((x1 + x2) / 2, (y1 + y2) / 2) tw, th = self.text_size(label, self.font_edge) self.draw.rounded_rectangle( (lx - tw / 2 - 12, ly - th / 2 - 7, lx + tw / 2 + 12, ly + th / 2 + 7), radius=10, fill="#F7F9FC", ) self.draw.text((lx - tw / 2, ly - th / 2), label, font=self.font_edge, fill=color) def note_box(self, x, y, w, h, text, fill="#FFF1F3", outline="#C01048"): self.draw.rounded_rectangle((x, y, x + w, y + h), radius=28, fill=fill, outline=outline, width=5) self.text_center(text, x + w / 2, y + h / 2, w - 90, font=self.font_note, fill="#7A271A") def footer(self, text): tw, _ = self.text_size(text, self.font_edge) self.draw.text(((self.width - tw) / 2, self.height - 52), text, font=self.font_edge, fill="#667085") def save(self, name): path = BASE_DIR / name self.img.save(path, format="PNG", dpi=(220, 220), optimize=True) print(f"{path} {self.width}x{self.height}") def draw_main(): d = Diagram(5400, 3400) d.title("MESCommonBase 响应输出与 SqlWebCall 主流程图", "主图：入口 type 分支如何进入二进制下载、JSON、IP 文本和 SqlWebCall") cx = d.width / 2 d.box(cx, 290, 1050, 125, "MESCommonBase.ashx\nProcessRequest", "#E8F1FF", "#2864B4") d.box(cx, 480, 1400, 140, "解析请求并取得 type\nJSON body / jsonobj / Request[\"param\"]", "#EAF7EF", "#227A45") d.diamond(cx, 690, 480, 170, "switch\n(type)") d.arrow([(cx, 352), (cx, 410)]) d.arrow([(cx, 550), (cx, 607)]) panel_x, panel_y, panel_w, panel_h = 190, 860, 5020, 1120 d.panel(panel_x, panel_y, panel_w, panel_h, "1. Type 首层分支") d.arrow([(cx, 775), (cx, panel_y + 5)]) lanes = [ (830, "2001/2002/2003/2004/16\n文件下载类分支\n生成或读取 bytes / fileName / extension", "#EAF7EF", "#227A45"), (2040, "15\n文件上传分支\n读取 Request.Files[0]\n调用 ExePROCEDURE_Type15", "#EEF2FF", "#4F46E5"), (3250, "4000\nIP 查询分支\n读取 ServerVariables / UserHostAddress", "#FFF7E6", "#B27700"), (4460, "default\n普通业务分支\nDataLink.SqlWebCall(type,jsonData,dataobj)", "#F3E8FF", "#7E22CE"), ] for x, text, fill, outline in lanes: d.box(x, 1110, 1080, 250, text, fill, outline, font=d.font_small) d.diamond(830, 1450, 430, 150, "bytes\n有效?", "#FFF4CC", "#B27700") d.box(830, 1760, 1080, 250, "二进制下载响应\napplication/octet-stream\nContent-Disposition: attachment\nBinaryWrite + Flush + End/Close", "#EAF7EF", "#227A45", font=d.font_small) d.box(1320, 1545, 480, 130, "无文件\n直接 return\n可能空响应", "#FFF1F3", "#C01048", font=d.font_small) d.arrow([(830, 1235), (830, 1375)]) d.arrow([(830, 1525), (830, 1635)], label="是", label_pos=(780, 1580)) d.arrow([(1045, 1450), (1165, 1450), (1165, 1545), (1080, 1545)], label="否", label_pos=(1145, 1408)) d.box(2040, 1760, 1080, 250, "JSON 文本响应\nResponse.Write(responseText)\n常见 result=1 / result=0", "#EEF2FF", "#4F46E5", font=d.font_small) d.arrow([(2040, 1235), (2040, 1635)]) d.box(3250, 1760, 1080, 250, "文本 IP 响应\nResponse.Write(userIP)\nContent-Type 仍可能是 application/json", "#FFF7E6", "#B27700", font=d.font_small) d.arrow([(3250, 1235), (3250, 1635)]) d.box(4460, 1450, 1080, 220, "SqlWebCall 返回字符串\n数组 JSON / code-message-data\n未支持 Type 可能为空字符串", "#F3E8FF", "#7E22CE", font=d.font_small) d.arrow([(4460, 1235), (4460, 1340)]) d.box(4460, 1760, 1080, 250, "JSON 文本响应\nHeaders.Remove(\"Server\")\nResponse.Write(responseText)", "#EEF2FF", "#4F46E5", font=d.font_small) d.arrow([(4460, 1560), (4460, 1635)]) d.panel(360, 2140, 2280, 660, "2. 响应出口规则") d.text_left( "文件下载：设置 application/octet-stream，写 Content-Disposition，BinaryWrite(bytes)。\n\n" "上传/default：使用 Response.Write 写 JSON 字符串，格式由下游决定。\n\n" "IP 查询：直接写 userIP，实际是文本。\n\n" "Type=16 bytes 为 null 时直接 return，调用方可能收到空响应。", 430, 2245, 2140, font=d.font_note, ) d.panel(2860, 2140, 2180, 660, "3. 外层异常兜底") d.text_left( "responseText 初始值为 NULL。\n\n" "ProcessRequest 外层 catch(Exception err) 不记录日志、不改 HTTP 状态码，只写当前 responseText。\n\n" "如果异常发生在下游调用前，通常返回 NULL；如果发生在赋值后，可能返回旧结果。", 2930, 2245, 2040, font=d.font_note, ) d.note_box( 680, 2940, 4040, 150, "主流程风险：响应格式混合（二进制/JSON/文本/NULL/空响应），Response.End 位于 try 内，default 分支执行能力取决于 SqlWebCall 内部 Type。", ) d.box(d.width - 520, 3180, 420, 92, "请求结束", "#E8F1FF", "#2864B4", radius=45, font=d.font_small) d.arrow([(d.width / 2, 3090), (d.width - 730, 3180)]) d.footer("源：MESCommonBase响应输出与SqlWebCall主流程图.mmd 图：MESCommonBase响应输出与SqlWebCall主流程图.png") d.save("MESCommonBase响应输出与SqlWebCall主流程图.png") def draw_secondary(): d = Diagram(5800, 4300) d.title("DataLink.SqlWebCall 默认分支次流程图", "次图：SqlWebCall 内部初始化、Type 二次分发、参数解析、数据库执行和返回格式") cx = d.width / 2 d.box(cx, 285, 1260, 120, "MESCommonBase default 分支\nDataLink.SqlWebCall(type,jsonData,dataobj)", "#F3E8FF", "#7E22CE") d.diamond(cx, 500, 520, 160, "initSystemIsOk?", "#FFF4CC", "#B27700") d.box(4150, 500, 1040, 140, "InitSystemReg\n读取 Web.config ConnectionString\ninitSystemIsOk = true", "#E8F1FF", "#2864B4", font=d.font_small) d.diamond(cx, 735, 500, 160, "switch\n(type)", "#FFF4CC", "#B27700") d.arrow([(cx, 345), (cx, 420)]) d.arrow([(cx + 260, 500), (3630, 500)], label="否", label_pos=(3430, 458)) d.arrow([(4150, 570), (4150, 735), (3150, 735)]) d.arrow([(cx, 580), (cx, 655)], label="是", label_pos=(2820, 625)) panel_y = 930 d.panel(150, panel_y, 5500, 1630, "1. Type 二次分发与处理") cols = [ (860, "登录/加密类\n8888 / 5001 / 5002", "#E8F1FF", "#2864B4"), (2180, "旧协议存储过程\n1 / 2 / 5", "#EEF2FF", "#4F46E5"), (3500, "新协议存储过程\n11 / 111 / 12 / 13 / 21", "#EAF7EF", "#227A45"), (4820, "SQL/命令执行类\n1001 / 1002 / 3 / 4 / 7 / 22 / 3001", "#FFF7E6", "#B27700"), ] for x, header, fill, outline in cols: d.box(x, 1140, 1180, 180, header, fill, outline, font=d.font_small) d.box(860, 1420, 1180, 290, "8888：读取 Name/name 并加密\n5001：Password MD5 后执行存储过程\n5002：解密密码、比对数据库密码、成功生成 token", "#E8F1FF", "#2864B4", font=d.font_small) d.box(860, 1805, 1180, 300, "返回格式\n加密字符串\n用户表 JSON + token\n失败 result=0,msg=...", "#FFFFFF", "#2864B4", font=d.font_small) d.arrow([(860, 1230), (860, 1275)]) d.arrow([(860, 1565), (860, 1655)]) d.box(2180, 1420, 1180, 290, "输入对象：jsonobj dataobj\nParam 拼接字符串\nType 1/2: @p=value=type\nType 5: @p&value&type|...", "#EEF2FF", "#4F46E5", font=d.font_small) d.box(2180, 1805, 1180, 300, "SQLCommon.GetCmdParam\n按 & / = / | 拆参数\n支持 output 参数\n随后 ExecuteStoredProcedure", "#FFFFFF", "#4F46E5", font=d.font_small) d.arrow([(2180, 1230), (2180, 1275)]) d.arrow([(2180, 1565), (2180, 1655)]) d.box(3500, 1420, 1180, 290, "GetString_JsonData\n读取 Type/Name/Param/token\nParam 是 JSON 数组字符串\n兼容大小写字段", "#EAF7EF", "#227A45", font=d.font_small) d.diamond(3500, 1785, 440, 150, "token\n非空?", "#FFF4CC", "#B27700") d.box(3030, 2045, 470, 150, "CheckToken\n失败返回\nresult=3", "#FFF1F3", "#C01048", font=d.font_small) d.box(3970, 2045, 650, 150, "Param 转 SqlParameter[]\n数组值转逗号字符串\noutput=1 设输出参数", "#EAF7EF", "#227A45", font=d.font_small) d.arrow([(3500, 1230), (3500, 1275)]) d.arrow([(3500, 1565), (3500, 1710)]) d.arrow([(3280, 1785), (3030, 1970)], label="是", label_pos=(3195, 1890)) d.arrow([(3720, 1785), (3970, 1970)], label="否/通过", label_pos=(3880, 1890)) d.box(4820, 1420, 1180, 290, "Name/name 可能是 SQL 文本或表名\n1001/1002：带参数 SQL\n3/4/22：直接 SQL\n7：DROP/CREATE 表并导入", "#FFF7E6", "#B27700", font=d.font_small) d.box(4820, 1805, 1180, 300, "SQLCommon.ExecuteDataTable / ExecuteDataset\nExecuteInsertMesWork / ExecuteSelectMesWork\n或 DbCallType1003_SqlCmd.SqlExec", "#FFFFFF", "#B27700", font=d.font_small) d.arrow([(4820, 1230), (4820, 1275)]) d.arrow([(4820, 1565), (4820, 1655)]) d.panel(300, 2740, 2440, 780, "2. 底层数据库执行链路") d.text_left( "存储过程：SQLCommon.ExecuteStoredProcedure\n- SqlCommand.CommandType = StoredProcedure\n- CommandTimeout = 0\n- SqlDataAdapter.Fill(DataSet/DataTable)\n\n" "SQL 查询：ExecuteSelectMesWork / ExecuteDataTable / ExecuteDataset\n- SqlDataAdapter(sql, conn)\n- Fill(dt/ds)\n\n" "SQL 增删改：ExecuteInsertMesWork / ExecuteNonQuery\n- SqlCommand.CommandText = 请求传入 SQL\n- ExecuteNonQuery / ExecuteScalar", 380, 2845, 2290, font=d.font_note, ) d.panel(3060, 2740, 2440, 780, "3. 返回格式与响应出口") d.text_left( "DataTable JSON：Type 1/3/5/11/1002 常见。\n\n" "result 标志：Type 2/4/12/1001 常见，成功 result=1，失败 result=0。\n\n" "分页：Type 111 返回 { rows, total }，total 来自 ItemCount 输出参数。\n\n" "新结构：Type 21/22 返回 { code, message, data }。\n\n" "未支持 Type：SqlWebCall 可能返回空字符串。", 3140, 2845, 2290, font=d.font_note, ) d.note_box( 560, 3700, 4680, 210, "次流程风险：token 多数是非空才校验；客户端可控制 Name/name；SQL 文本、存储过程名、表名缺少统一白名单；CommandTimeout=0 可能导致长时间阻塞。", ) d.box(d.width - 620, 4080, 560, 95, "返回 MESCommonBase\nResponse.Write", "#E8F1FF", "#2864B4", radius=45, font=d.font_small) d.arrow([(d.width / 2, 3910), (d.width - 900, 4080)]) d.footer("源：MESCommonBase响应输出与SqlWebCall次流程图.mmd 图：MESCommonBase响应输出与SqlWebCall次流程图.png") d.save("MESCommonBase响应输出与SqlWebCall次流程图.png") if __name__ == "__main__": draw_main() draw_secondary()