352 lines
12 KiB
Python
352 lines
12 KiB
Python
"""
|
||
企业微信 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<font color=\"comment\">... 消息过长，已截断</font>"
|
||
|
||
|
||
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"- **全路径类名：** <font color=\"info\">**{file_path}**</font>"
|
||
|
||
header = [
|
||
f"- **变更类型：** <font color=\"warning\">**{change_type}**</font>",
|
||
f"- **URI：** {uri_line}",
|
||
class_line,
|
||
]
|
||
|
||
if report.is_removed_endpoint:
|
||
return "\n".join(header + ["", f"<font color=\"warning\">**该接口已被移除**</font>"])
|
||
|
||
detail_lines = ["", "---", "", "## 接口参数变动详情", "", "---", ""]
|
||
|
||
if report.is_new_endpoint:
|
||
detail_lines.append("### <font color=\"info\">**新增接口参数**</font>")
|
||
else:
|
||
detail_lines.append("### <font color=\"warning\">**参数变更明细**</font>")
|
||
|
||
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('<font color="comment">无</font>')
|
||
|
||
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:
|
||
path_md = build_path_change_markdown(
|
||
old_uri=report.uri,
|
||
new_uri=report.uri,
|
||
change_type="修改请求方式",
|
||
push_user=push_user,
|
||
push_time=push_time,
|
||
file_name=report.source_file or report.controller_class,
|
||
old_method=report.old_http_method,
|
||
new_method=report.http_method,
|
||
)
|
||
parts.append(path_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("### <font color=\"comment\">【兼容性提示】</font>")
|
||
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"<font color=\"comment\">（续 {i + 1}/{len(segments)}）</font>\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,
|
||
old_method: Optional[str] = None,
|
||
new_method: Optional[str] = None,
|
||
) -> str:
|
||
"""构建 API路径变更通知，完全匹配 model1.md 模板，并加强视觉区分。
|
||
|
||
支持的 change_type：
|
||
- 新增接口 / 删除接口 / 修改路径 / 修改请求方式
|
||
|
||
改进点：
|
||
- 标题使用【】风格
|
||
- 头部信息缩进 + 颜色高亮
|
||
- URI 详情使用列表（更直观）
|
||
- 「修改请求方式」额外展示方法变更
|
||
"""
|
||
# 变更类型高亮
|
||
type_highlight = f"<font color=\"warning\">**{change_type}**</font>"
|
||
|
||
# 全路径类名高亮
|
||
class_highlight = f"<font color=\"info\">**{file_name}**</font>"
|
||
|
||
# 根据变更类型优化 URI 展示
|
||
if change_type == "新增接口":
|
||
old_display = "`-`"
|
||
new_display = f"<font color=\"info\">**`{new_uri}`**</font> ← <font color=\"info\">**新增**</font>"
|
||
elif change_type == "删除接口":
|
||
old_display = f"<font color=\"warning\">**`{old_uri}`**</font> ← <font color=\"warning\">**已删除**</font>"
|
||
new_display = "`已删除`"
|
||
elif change_type == "修改请求方式":
|
||
# 方法变更场景：展示原方法 → 新方法
|
||
old_m = old_method or "?"
|
||
new_m = new_method or "?"
|
||
old_display = f"<font color=\"warning\">**`{old_uri}`</font> <font color=\"warning\">**[{old_m}]**</font>"
|
||
new_display = (
|
||
f"<font color=\"info\">**`{new_uri}`</font> "
|
||
f"<font color=\"warning\">**[{old_m} → {new_m}]**</font> ← <font color=\"info\">**请求方式变更**</font>"
|
||
)
|
||
else: # 修改路径
|
||
old_display = f"<font color=\"warning\">~~`{old_uri}`~~</font> ← <font color=\"warning\">**旧路径**</font>"
|
||
new_display = f"<font color=\"info\">**`{new_uri}`**</font> ← <font color=\"info\">**新路径**</font>"
|
||
|
||
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 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)
|