""" 企业微信 Markdown 通知模块。 支持加粗、颜色（info/comment/warning），新增接口与变更接口使用不同展示模板。 """ import json import re from typing import List, Optional import requests from comparator import EndpointChangeReport # 企微 Markdown 单条上限 4096 字符，留余量 MAX_MD_LENGTH = 3800 def truncate_text(text: str, max_length: int = MAX_MD_LENGTH) -> str: """截断超长消息。""" if len(text) <= max_length: return text return text[:max_length] + "\n\n... 消息过长，已截断" def _format_param_table(changes: List) -> List[str]: """按 model.md 生成参数变更 Markdown 表格。""" if not changes: return ['无'] lines = [ "", "| 字段 | 说明 | 变更 |", "|------|------|------|", ] for change in changes: lines.append(change.to_table_row()) return lines def _format_param_details_section(report: EndpointChangeReport) -> List[str]: """生成接口参数变动详情区块（对齐 model.md）。""" body_changes = [c for c in report.parameter_changes if c.source == "body"] regular_changes = [c for c in report.parameter_changes if c.source != "body"] lines = ["", "---", "", "## 接口参数变动详情", "", "---", ""] if body_changes: lines.append("### 类对象变更") lines.append("") lines.append("- **参数变更列表：**") lines.extend(_format_param_table(body_changes)) lines.append("") if regular_changes: lines.append("### 普通参数变更（非对象字段）") lines.append("") lines.append("- **参数变更列表：**") lines.extend(_format_param_table(regular_changes)) lines.append("") if not body_changes and not regular_changes: lines.append('无') return lines def _format_endpoint_block(report: EndpointChangeReport) -> str: """ 格式化单个接口块，按模板匹配格式输出。 全路径类名显示为 source_file（相对仓库根的完整 .java 路径）。 """ change_type = "新增接口" if report.is_new_endpoint else ("删除接口" if report.is_removed_endpoint else "修改参数") uri_line = f"**{report.http_method}** `{report.uri}`" file_path = report.source_file or report.controller_class class_line = f"- **全路径类名：** **{file_path}**" header = [ f"- **变更类型：** **{change_type}**", f"- **URI：** {uri_line}", class_line, ] if report.is_removed_endpoint: return "\n".join(header + ["", f"**该接口已被移除**"]) return "\n".join(header + _format_param_details_section(report)) def build_markdown_notification( reports: List[EndpointChangeReport], push_user: str, push_time: str, llm_summary: Optional[str] = None, ) -> str: """ 构建完整 Markdown 通知正文。 :param reports: AST 变更报告 :param push_user: 推送人 :param push_time: 推送时间 :param llm_summary: LLM 兼容性摘要（可选，简短） :return: Markdown 文本 """ parts: List[str] = [] # 所有 API 级变更（新增、修改路径、修改请求方式、删除、参数变更）统一走 model1.md 路径变更通知 method_changed_reports = [r for r in reports if r.is_method_changed] renamed_reports = [r for r in reports if r.is_renamed_endpoint] new_reports = [r for r in reports if r.is_new_endpoint] # 参数变更报告：只包含「URI/方法未变，仅参数变化」的报告 # 路径变更 + 参数变更、方法变更 + 参数变更 场景已在上层 comparator 中拆分为独立报告 changed_reports = [ r for r in reports if not r.is_new_endpoint and not r.is_removed_endpoint and not r.is_renamed_endpoint and not r.is_method_changed ] removed_reports = [r for r in reports if r.is_removed_endpoint] # 1. 新增接口 → 走 API路径变更通知 for report in new_reports: path_md = build_path_change_markdown( old_uri="-", new_uri=report.uri, change_type="新增接口", push_user=push_user, push_time=push_time, file_name=report.source_file or report.controller_class, ) parts.append(path_md) parts.append("") # 2. 修改请求方式 → 使用独立的新模板 【API请求方式变更通知】 for report in method_changed_reports: method_md = build_method_change_markdown( uri=report.uri, old_method=report.old_http_method or "?", new_method=report.http_method, push_user=push_user, push_time=push_time, file_name=report.source_file or report.controller_class, ) parts.append(method_md) parts.append("") # 3. 修改路径 → 走 API路径变更通知 for report in renamed_reports: path_md = build_path_change_markdown( old_uri=report.old_uri or "-", new_uri=report.uri, change_type="修改路径", push_user=push_user, push_time=push_time, file_name=report.source_file or report.controller_class, ) parts.append(path_md) parts.append("") # 4. 删除接口 → 走 API路径变更通知 for report in removed_reports: path_md = build_path_change_markdown( old_uri=report.uri, new_uri="已删除", change_type="删除接口", push_user=push_user, push_time=push_time, file_name=report.source_file or report.controller_class, ) parts.append(path_md) parts.append("") # 4. 普通参数变更（非路径变更）仍使用 model.md 格式 if changed_reports: parts.append("# 【API参数变更通知】") parts.append(f"- **修改人：** {push_user}") parts.append(f"- **修改时间：** {push_time}") parts.append("") for report in changed_reports: parts.append(_format_endpoint_block(report)) parts.append("") if llm_summary: cleaned = llm_summary.strip() # 去掉 LLM 可能输出的「排除框架注入」类说明 cleaned = re.sub( r"（排除Spring MVC框架自动注入的[^）]+）", "", cleaned, ) cleaned = re.sub( r"排除Spring MVC框架自动注入的[`\w/]+[`\w/、/]*[。\.]?", "", cleaned, ) if cleaned: parts.append("### 【兼容性提示】") parts.append(cleaned) return "\n".join(parts).strip() def _split_markdown(text: str, max_len: int) -> List[str]: """按 ### 标题块拆分超长 Markdown。""" if len(text) <= max_len: return [text] lines = text.split("\n") chunks: List[str] = [] current: List[str] = [] for line in lines: if line.startswith("### ") and current and len("\n".join(current)) > 200: chunks.append("\n".join(current)) current = [line] else: current.append(line) if len("\n".join(current)) >= max_len: chunks.append("\n".join(current)) current = [] if current: if chunks and len("\n".join(current)) < 200: chunks[-1] = chunks[-1] + "\n" + "\n".join(current) else: chunks.append("\n".join(current)) return chunks or [truncate_text(text)] def _post_wecom_markdown(webhook_url: str, content: str) -> bool: """发送企微 Markdown 消息。""" if not webhook_url or "YOUR_WECOM_KEY" in webhook_url: print("[警告] 未配置有效的企业微信 Webhook URL。") print("--- 通知预览 ---") print(content[:1000]) return False payload = { "msgtype": "markdown", "markdown": {"content": truncate_text(content)}, } try: resp = requests.post( webhook_url, headers={"Content-Type": "application/json"}, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), timeout=10, ) if resp.status_code == 200 and resp.json().get("errcode", 0) == 0: return True print(f"[错误] 企微返回异常: {resp.status_code} {resp.text}") return False except requests.RequestException as exc: print(f"[错误] 发送企微消息失败: {exc}") return False def send_parameter_change_notification( webhook_url: str, reports: List[EndpointChangeReport], push_user: str, push_time: str, llm_review: Optional[str] = None, mentioned_users: Optional[List[str]] = None, ) -> int: """ 发送 Markdown 格式的接口变更通知。 严格按变更类型拆分，各自独立构建和发送企微通知： - 方法变更 → 独立调用 build_method_change_markdown - 路径变更（新增/修改/删除） → 独立调用 build_path_change_markdown - 参数变更 → 独立调用 _format_endpoint_block 不同类型之间完全互不干扰，各自走独立分支。 """ if not reports and not llm_review: print("无接口参数变更，不发送到企业微信") return 0 # 按类型严格分组（互不重叠） method_changed_reports = [r for r in reports if r.is_method_changed] renamed_reports = [r for r in reports if r.is_renamed_endpoint] new_reports = [r for r in reports if r.is_new_endpoint] removed_reports = [r for r in reports if r.is_removed_endpoint] changed_reports = [ r for r in reports if not r.is_new_endpoint and not r.is_removed_endpoint and not r.is_renamed_endpoint and not r.is_method_changed ] sent = 0 # ========== 1. 请求方式变更通知（独立分支） ========== for report in method_changed_reports: md = build_method_change_markdown( uri=report.uri, old_method=report.old_http_method or "?", new_method=report.http_method, push_user=push_user, push_time=push_time, file_name=report.source_file or report.controller_class, ) if _post_wecom_markdown(webhook_url, md): sent += 1 print(f"第 {sent} 条通知已发送到企业微信（请求方式变更）") # ========== 2. 路径变更通知（新增/修改/删除） ========== # 新增接口 for report in new_reports: md = build_path_change_markdown( old_uri="-", new_uri=report.uri, change_type="新增接口", push_user=push_user, push_time=push_time, file_name=report.source_file or report.controller_class, ) if report.parameter_changes: param_section = "\n".join(_format_param_details_section(report)).strip() md = f"{md}\n\n{param_section}" if _post_wecom_markdown(webhook_url, md): sent += 1 print(f"第 {sent} 条通知已发送到企业微信（新增接口）") # 修改路径 for report in renamed_reports: md = build_path_change_markdown( old_uri=report.old_uri or "-", new_uri=report.uri, change_type="修改路径", push_user=push_user, push_time=push_time, file_name=report.source_file or report.controller_class, ) if _post_wecom_markdown(webhook_url, md): sent += 1 print(f"第 {sent} 条通知已发送到企业微信（修改路径）") # 删除接口 for report in removed_reports: md = build_path_change_markdown( old_uri=report.uri, new_uri="已删除", change_type="删除接口", push_user=push_user, push_time=push_time, file_name=report.source_file or report.controller_class, ) if _post_wecom_markdown(webhook_url, md): sent += 1 print(f"第 {sent} 条通知已发送到企业微信（删除接口）") # ========== 3. 参数变更通知（独立分支） ========== if changed_reports: # 构建参数变更通知（只包含参数变更报告，对齐 model.md） parts: List[str] = [] parts.append("# 【API参数变更通知】") parts.append(f"- **修改人：** {push_user if push_user.startswith('@') else '@' + push_user}") parts.append(f"- **修改时间：** {push_time}") parts.append("") for report in changed_reports: parts.append(_format_endpoint_block(report)) parts.append("") if llm_review: parts.append("---") parts.append("### 兼容性提示") parts.append(llm_review.strip()) md = "\n".join(parts).strip() if _post_wecom_markdown(webhook_url, md): sent += 1 print(f"第 {sent} 条通知已发送到企业微信（参数变更）") if sent > 0: print(f"总共发送 {sent} 条通知到企业微信") return sent def build_path_change_markdown( old_uri: str, new_uri: str, change_type: str, push_user: str, push_time: str, file_name: str, ) -> str: """构建 API路径变更通知，完全匹配 model1.md 模板，并加强视觉区分。 支持的 change_type： - 新增接口 / 删除接口 / 修改路径 / 修改请求方式 改进点： - 标题使用【】风格 - 头部信息缩进 + 颜色高亮 - URI 详情使用列表（更直观） - 「修改请求方式」额外展示方法变更 """ # 变更类型高亮 type_highlight = f"**{change_type}**" # 全路径类名高亮 class_highlight = f"**{file_name}**" # 根据变更类型优化 URI 展示 if change_type == "新增接口": old_display = "`-`" new_display = f"**`{new_uri}`** ← **新增**" elif change_type == "删除接口": old_display = f"**`{old_uri}`** ← **已删除**" new_display = "`已删除`" else: # 修改路径 old_display = f"~~`{old_uri}`~~ ← **旧路径**" new_display = f"**`{new_uri}`** ← **新路径**" parts = [ "# 【API路径变更通知】", "", f" 变更类型： {type_highlight}", f" 全路径类名： {class_highlight}", f" 修改人： {push_user}", f" 修改时间： {push_time}", "", "---------------------------------------", "", "#### 【URI变更详情】", f"- **原路径：** {old_display}", f"- **新路径：** {new_display}", "", ] return "\n".join(parts).strip() def build_method_change_markdown( uri: str, old_method: str, new_method: str, push_user: str, push_time: str, file_name: str, ) -> str: """构建【API请求方式变更通知】独立模板。 格式参考 model1.md，但专门针对 HTTP 方法变更场景设计， 突出「原请求方式 → 新请求方式」的对比。 """ type_highlight = '**修改请求方式**' class_highlight = f'**{file_name}**' uri_highlight = f'**`{uri}`**' old_m = f'**{old_method}**' new_m = f'**{new_method}**' parts = [ "# 【API请求方式变更通知】", "", f" 变更类型： {type_highlight}", f" 全路径类名： {class_highlight}", f" 修改人： {push_user}", f" 修改时间： {push_time}", "", "---------------------------------------", "", "#### 【请求方式变更详情】", f"- **URI：** {uri_highlight}", f"- **原请求方式：** {old_m}", f"- **新请求方式：** {new_m} ← **请求方式已变更**", "", ] return "\n".join(parts).strip() def send_path_change_notification( webhook_url: str, old_uri: str, new_uri: str, change_type: str, push_user: str, push_time: str, file_name: str, ) -> bool: """发送路径变更通知。""" md = build_path_change_markdown(old_uri, new_uri, change_type, push_user, push_time, file_name) return _post_wecom_markdown(webhook_url, md)