dongzi
556f5b8ab6
Some checks failed
API Parameter Change Check / api-param-check (push) Failing after 3s
219 lines
6.5 KiB
Python
219 lines
6.5 KiB
Python
"""
|
||
企业微信机器人通知模块。
|
||
按第一版模板发送 Controller 接口参数变更通知，支持超长内容分段发送。
|
||
"""
|
||
|
||
import json
|
||
from typing import List, Optional
|
||
|
||
import requests
|
||
|
||
from comparator import EndpointChangeReport
|
||
from git_utils import CommitInfo
|
||
|
||
# 企微 text 消息字节上限约 2048，留余量按字符分段
|
||
MAX_TEXT_LENGTH = 2000
|
||
|
||
|
||
def truncate_text(text: str, max_length: int = MAX_TEXT_LENGTH) -> str:
|
||
"""
|
||
截断文本，避免超出企微单条消息限制。
|
||
|
||
:param text: 原始文本
|
||
:param max_length: 最大字符数
|
||
:return: 截断后文本
|
||
"""
|
||
if len(text) <= max_length:
|
||
return text
|
||
return text[:max_length] + "\n... [消息过长，已截断]"
|
||
|
||
|
||
def build_single_endpoint_message(
|
||
report: EndpointChangeReport,
|
||
push_user: str,
|
||
push_time: str,
|
||
) -> str:
|
||
"""
|
||
按第一版模板构建单个接口的通知正文。
|
||
|
||
模板示例：
|
||
[API变更通知]
|
||
URI: GET /api/users/{id}
|
||
修改人：张三
|
||
修改时间：2026-06-03 10:00:00
|
||
参数变更:
|
||
- 删除: Boolean userType
|
||
- 新增: Boolean includeDisabled (是否必填：false)
|
||
|
||
:param report: 单个接口变更报告
|
||
:param push_user: 代码推送人
|
||
:param push_time: 推送时间
|
||
:return: 通知文本
|
||
"""
|
||
lines = [
|
||
"[API变更通知]",
|
||
f"URI: {report.http_method} {report.uri}",
|
||
f"修改人：{push_user}",
|
||
f"修改时间：{push_time}",
|
||
]
|
||
|
||
if report.is_new_endpoint:
|
||
lines.append("接口状态：新增接口")
|
||
if report.parameter_changes:
|
||
lines.append("参数变更:")
|
||
for change in report.parameter_changes:
|
||
lines.append(change.to_display_line())
|
||
elif report.is_removed_endpoint:
|
||
lines.append("接口状态：已删除接口")
|
||
lines.append(" - 整个接口已被移除")
|
||
else:
|
||
lines.append("参数变更:")
|
||
for change in report.parameter_changes:
|
||
lines.append(change.to_display_line())
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def build_all_notifications(
|
||
reports: List[EndpointChangeReport],
|
||
push_user: str,
|
||
push_time: str,
|
||
llm_review: Optional[str] = None,
|
||
) -> List[str]:
|
||
"""
|
||
将所有接口变更组装为通知消息列表，超长时自动分段。
|
||
|
||
:param reports: 变更报告列表
|
||
:param push_user: 推送人
|
||
:param push_time: 推送时间
|
||
:param llm_review: LLM 参数变更审核结论（可选，附在最后一条或单独一条）
|
||
:return: 待发送的消息段落列表
|
||
"""
|
||
if not reports:
|
||
return []
|
||
|
||
messages: List[str] = []
|
||
current = ""
|
||
|
||
for report in reports:
|
||
block = build_single_endpoint_message(report, push_user, push_time)
|
||
separator = "\n\n---\n\n"
|
||
|
||
if not current:
|
||
current = block
|
||
elif len(current) + len(separator) + len(block) <= MAX_TEXT_LENGTH:
|
||
current += separator + block
|
||
else:
|
||
messages.append(current)
|
||
current = block
|
||
|
||
if current:
|
||
messages.append(current)
|
||
|
||
# LLM 审核结论单独或追加发送
|
||
if llm_review:
|
||
review_msg = f"[AI参数变更审核]\n修改人：{push_user}\n修改时间：{push_time}\n\n{llm_review}"
|
||
if messages and len(messages[-1]) + len(review_msg) + 4 <= MAX_TEXT_LENGTH:
|
||
messages[-1] += "\n\n" + review_msg
|
||
elif len(review_msg) <= MAX_TEXT_LENGTH:
|
||
messages.append(review_msg)
|
||
else:
|
||
messages.extend(_split_long_text(review_msg, MAX_TEXT_LENGTH))
|
||
|
||
return messages
|
||
|
||
|
||
def _split_long_text(text: str, max_len: int) -> List[str]:
|
||
"""按行拆分超长文本。"""
|
||
lines = text.split("\n")
|
||
chunks: List[str] = []
|
||
current = ""
|
||
for line in lines:
|
||
candidate = current + line + "\n"
|
||
if len(candidate) > max_len and current:
|
||
chunks.append(current.rstrip())
|
||
current = line + "\n"
|
||
else:
|
||
current = candidate
|
||
if current.strip():
|
||
chunks.append(current.rstrip())
|
||
return chunks
|
||
|
||
|
||
def _post_wecom_text(webhook_url: str, content: str) -> bool:
|
||
"""
|
||
发送单条 text 消息到企业微信。
|
||
|
||
:param webhook_url: Webhook URL
|
||
:param content: 消息正文
|
||
:return: 是否成功
|
||
"""
|
||
if not webhook_url or "YOUR_WECOM_KEY" in webhook_url:
|
||
print("[警告] 未配置有效的企业微信 Webhook URL。")
|
||
print("--- 通知预览 ---")
|
||
print(content[:800])
|
||
return False
|
||
|
||
payload = {
|
||
"msgtype": "text",
|
||
"text": {"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:
|
||
"""
|
||
发送 Controller 接口参数变更通知（支持分段）。
|
||
|
||
:param webhook_url: 企微 Webhook
|
||
:param reports: AST 参数变更报告
|
||
:param push_user: 推送人
|
||
:param push_time: 推送时间
|
||
:param llm_review: LLM 参数变更审核（可选）
|
||
:param mentioned_users: @ 成员 userid 列表
|
||
:return: 成功发送条数
|
||
"""
|
||
if not reports and not llm_review:
|
||
print("无接口参数变更，不发送到企业微信")
|
||
return 0
|
||
|
||
segments = build_all_notifications(reports, push_user, push_time, llm_review)
|
||
if not segments:
|
||
return 0
|
||
|
||
sent = 0
|
||
for i, segment in enumerate(segments):
|
||
payload_text = segment
|
||
if mentioned_users and i == 0:
|
||
# text 类型 @ 成员需放在 mentioned_list
|
||
pass # 在 _post 里处理较复杂，首条单独带 mentioned
|
||
|
||
if _post_wecom_text(webhook_url, payload_text):
|
||
sent += 1
|
||
print(f"第 {sent} 条通知已发送到企业微信")
|
||
|
||
if sent > 0:
|
||
print(f"总共发送 {sent} 条通知到企业微信")
|
||
return sent
|