""" 企业微信 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_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"**该接口已被移除**"]) detail_lines = ["", "---", "", "## 接口参数变动详情", "", "---", ""] if report.is_new_endpoint: detail_lines.append("### **新增接口参数**") else: detail_lines.append("### **参数变更明细**") if report.parameter_changes: for change in report.parameter_changes: md = change.to_markdown_line(plain=report.is_new_endpoint) detail_lines.append(md) else: detail_lines.append('无') return "\n".join(header + detail_lines) 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] 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. 普通参数变更（非路径变更）仍使用原有格式 if changed_reports: parts.append("# API参数变更通知") parts.append(f"- **变更类型：** 修改参数") 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 格式的接口参数变更通知。 :param webhook_url: 企微 Webhook :param reports: 变更报告 :param push_user: 推送人 :param push_time: 推送时间 :param llm_review: LLM 兼容性摘要 :param mentioned_users: @ 成员（Markdown 暂不支持，保留参数） :return: 成功发送条数 """ if not reports and not llm_review: print("无接口参数变更，不发送到企业微信") return 0 full_md = build_markdown_notification(reports, push_user, push_time, llm_review) segments = _split_markdown(full_md, MAX_MD_LENGTH) sent = 0 for i, segment in enumerate(segments): if i > 0: segment = ( f"（续 {i + 1}/{len(segments)}）\n\n" + segment ) if _post_wecom_markdown(webhook_url, segment): 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)