276 lines
13 KiB
Python
276 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""Hub-only DingTalk notifications. Python 3.9+, standard library, no shell eval."""
|
||
|
||
import argparse
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import shutil
|
||
import stat
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
PREFIX = "DINGTALK_PROGRESS_"
|
||
KEYS = {
|
||
"ENABLED", "DRY_RUN", "MODE", "CLIENT_ID", "TARGET", "ROBOT_CODE",
|
||
"CLIENT_SECRET", "PROJECT", "TITLE_PREFIX", "DWS_PATH", "PROFILE",
|
||
}
|
||
EVENTS = {
|
||
"start": "开始推进", "milestone": "阶段里程碑", "blocked": "遇到阻塞",
|
||
"complete": "本轮完成", "failed": "执行失败",
|
||
}
|
||
# Only an explicit DNS resolution failure is safe to replay automatically.
|
||
DNS_ERROR = re.compile(
|
||
r"lookup [\w.-]+(?: on [^:\s]+(?::\d+)?)?: [^\n]*"
|
||
r"(?:i/o timeout|no such host|server misbehaving|temporary failure in name resolution)",
|
||
re.IGNORECASE,
|
||
)
|
||
ATTEMPTS = 3
|
||
RETRY_SECONDS = 10
|
||
COMMAND_TIMEOUT = 60
|
||
|
||
|
||
class NoticeError(Exception):
|
||
"""A human-safe error: never include raw subprocess/config contents."""
|
||
|
||
|
||
def emit(status, message, **extra):
|
||
print(json.dumps({"status": status, "message": message, **extra}, ensure_ascii=False))
|
||
|
||
|
||
def read_config(root):
|
||
path = root / ".dingtalk" / "config.env"
|
||
if not path.exists():
|
||
return {}
|
||
if os.name != "nt" and stat.S_IMODE(path.stat().st_mode) not in (0o400, 0o600):
|
||
raise NoticeError("配置权限过宽，请部署负责人设为仅当前用户可读写（600 或 400）。")
|
||
result = {}
|
||
for line in path.read_text(encoding="utf-8-sig").splitlines():
|
||
if not line.strip() or line.lstrip().startswith("#"):
|
||
continue
|
||
key, sep, value = line.partition("=")
|
||
key = key.strip()
|
||
if not sep or key not in {PREFIX + name for name in KEYS}:
|
||
raise NoticeError("配置含不支持的字段或格式；请部署负责人对照示例检查，不要回传配置内容。")
|
||
if key in result:
|
||
raise NoticeError("配置存在重复字段，请部署负责人检查。")
|
||
result[key] = value.strip()
|
||
for name in ("ENABLED", "DRY_RUN"):
|
||
if result.get(PREFIX + name, "0") not in ("0", "1"):
|
||
raise NoticeError("通知开关只能填写 0 或 1。")
|
||
return {key[len(PREFIX):]: value for key, value in result.items()}
|
||
|
||
|
||
def parse_args(argv):
|
||
parser = argparse.ArgumentParser(description="Hub 钉钉辅助通知；不派发 Etunel 任务。")
|
||
commands = parser.add_subparsers(dest="command", required=True)
|
||
progress = commands.add_parser("progress", help="项目事件通知")
|
||
progress.add_argument("event", choices=EVENTS)
|
||
progress.add_argument("summary", nargs="*", help="简短中文 Markdown 摘要")
|
||
progress.add_argument("--file", help="UTF-8 正文文件，与摘要二选一")
|
||
message = commands.add_parser("message", help="自定义中文摘要或负责人提醒")
|
||
message.add_argument("--title", required=True)
|
||
body = message.add_mutually_exclusive_group(required=True)
|
||
body.add_argument("--text")
|
||
body.add_argument("--file", help="UTF-8 正文文件；不是上传附件")
|
||
message.add_argument("--at", action="append", default=[], help="完整姓名，逗号分隔，可重复")
|
||
message.add_argument("--at-id", action="append", default=[], help="兼容已核实的 姓名=userId；不得猜编号")
|
||
message.add_argument("--group", help="仅用于用户明确指定的其他群；默认使用部署配置")
|
||
for command in (progress, message):
|
||
command.add_argument("--dry-run", action="store_true", help="离线预览，不查人员、不认证、不输出密钥、不发送")
|
||
args = parser.parse_args(argv)
|
||
if args.command == "progress" and bool(args.summary) == bool(args.file):
|
||
parser.error("请提供摘要或 --file，且只能选择一种。")
|
||
return args
|
||
|
||
|
||
def body_text(args, root=ROOT):
|
||
if args.file:
|
||
body_path = Path(args.file)
|
||
config_path = root / ".dingtalk" / "config.env"
|
||
if config_path.exists() and body_path.samefile(config_path):
|
||
raise NoticeError("不能将通知配置文件作为正文发送或预览。")
|
||
value = body_path.read_text(encoding="utf-8-sig")
|
||
else:
|
||
value = " ".join(args.summary) if args.command == "progress" else args.text
|
||
if not value.strip():
|
||
raise NoticeError("通知正文不能为空。")
|
||
if len(value.encode("utf-8")) > 12000:
|
||
raise NoticeError("通知过长，请改为简短摘要；正文文件也不是日志或成果附件通道。")
|
||
return value
|
||
|
||
|
||
def render(args, config, body):
|
||
if args.command == "message":
|
||
title = args.title.strip()
|
||
text = body
|
||
else:
|
||
project = config.get("PROJECT") or ROOT.name
|
||
label = EVENTS[args.event]
|
||
title = f"{config.get('TITLE_PREFIX') or '[项目进度]'} {project}｜{label}"
|
||
timestamp = dt.datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %z")
|
||
text = f"## {title}\n\n**项目：** {project} \n**状态：** {label} \n**更新时间：** {timestamp}\n\n---\n\n{body}"
|
||
if not title or "\n" in title or "\r" in title or len(title) > 200:
|
||
raise NoticeError("请使用不超过 200 字的单行中文标题。")
|
||
return title, text
|
||
|
||
|
||
def clean_env():
|
||
return {key: value for key, value in os.environ.items()
|
||
if key not in ("DING_SEC", "DWS_CLIENT_ID", "DWS_CLIENT_SECRET")
|
||
and not key.startswith(PREFIX)}
|
||
|
||
|
||
def secret_for(config):
|
||
# One source on every platform; never fall back to inherited credentials.
|
||
secret = config.get("CLIENT_SECRET", "")
|
||
if not secret:
|
||
raise NoticeError("未配置机器人凭据，请部署负责人在本地 config.env 填写 DINGTALK_PROGRESS_CLIENT_SECRET；不要在会话中提供密钥。")
|
||
return secret
|
||
|
||
|
||
def dws_path(config):
|
||
supplied = config.get("DWS_PATH")
|
||
path = shutil.which(supplied or "dws")
|
||
if not path:
|
||
raise NoticeError("未找到 dws，请部署负责人安装原生可执行文件并加入 Hub 的 PATH，或配置 DWS_PATH。")
|
||
if os.name == "nt" and Path(path).suffix.lower() != ".exe":
|
||
raise NoticeError("Windows 请使用原生 dws.exe（可配置 DWS_PATH），不通过 cmd/bat 转发 Markdown。")
|
||
return path
|
||
|
||
|
||
def json_object(output):
|
||
try:
|
||
value = json.loads(output)
|
||
return value if isinstance(value, dict) else {}
|
||
except (ValueError, TypeError):
|
||
return {}
|
||
|
||
|
||
def accepted_key(data):
|
||
value = data.get("processQueryKey")
|
||
return value.strip() if isinstance(value, str) and value.strip() else None
|
||
|
||
|
||
def invoke(command, env, sending=False, run=subprocess.run, sleep=time.sleep, stdin_data=None):
|
||
for attempt in range(1, ATTEMPTS + 1):
|
||
try:
|
||
result = run(command, env=env, input=stdin_data, capture_output=True, text=True,
|
||
encoding="utf-8", errors="replace", timeout=COMMAND_TIMEOUT)
|
||
except subprocess.TimeoutExpired:
|
||
raise NoticeError("钉钉调用超时，结果未知；本次不自动重发，请负责人核对。") from None
|
||
except OSError:
|
||
raise NoticeError("钉钉命令无法启动，请部署负责人检查安装和进程权限。") from None
|
||
data = json_object(result.stdout)
|
||
if sending and accepted_key(data):
|
||
return data
|
||
if result.returncode == 0:
|
||
if sending:
|
||
raise NoticeError("未收到有效发送回执，结果未知；本次不自动重发，请负责人核对。")
|
||
return data
|
||
if DNS_ERROR.search(result.stdout + "\n" + result.stderr):
|
||
if attempt < ATTEMPTS:
|
||
print(f"钉钉域名解析失败，10 秒后进行第 {attempt + 1}/3 次尝试。", file=sys.stderr)
|
||
sleep(RETRY_SECONDS)
|
||
continue
|
||
raise NoticeError("钉钉域名解析连续失败 3 次，本次通知已停止；主任务继续。")
|
||
raise NoticeError("钉钉调用未成功，未自动重发；请部署负责人检查认证、权限或网络，主任务继续。")
|
||
raise AssertionError("unreachable")
|
||
|
||
|
||
def mentions(args, config, cli, run=subprocess.run, sleep=time.sleep):
|
||
people = {}
|
||
for item in args.at_id:
|
||
name, sep, user_id = item.partition("=")
|
||
if not sep or not name.strip() or not user_id.strip() or re.search(r"\s", user_id):
|
||
raise NoticeError("已核实人员映射的格式应为 姓名=userId；不要自行编造编号。")
|
||
people[user_id.strip()] = name.strip()
|
||
names = list(dict.fromkeys(name.strip() for group in args.at
|
||
for name in group.replace("，", ",").split(",") if name.strip()))
|
||
if args.at and not names:
|
||
raise NoticeError("提醒对象不能为空。")
|
||
if names and not config.get("PROFILE"):
|
||
raise NoticeError("按姓名提醒需要部署负责人配置同组织的 PROFILE；不要临时切换或猜测账号。")
|
||
for name in names:
|
||
data = invoke([cli, "aisearch", "person", "--query", name, "--dimension", "name",
|
||
"--profile", config["PROFILE"], "--format", "json"],
|
||
clean_env(), run=run, sleep=sleep)
|
||
candidates = {}
|
||
rows = data.get("result", [])
|
||
if not isinstance(rows, list):
|
||
rows = []
|
||
for row in rows:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
meta = row.get("meta") if isinstance(row.get("meta"), dict) else {}
|
||
actual_name = meta.get("name") or row.get("title") or row.get("author")
|
||
user_id = row.get("userId")
|
||
if actual_name == name and isinstance(user_id, str) and user_id.strip():
|
||
candidates[user_id] = name
|
||
if len(candidates) != 1:
|
||
raise NoticeError("提醒对象未唯一匹配，本条未发送；请负责人确认完整姓名或已有的有效人员映射。")
|
||
people.update(candidates)
|
||
return people
|
||
|
||
|
||
def main(argv=None, root=ROOT, run=subprocess.run, sleep=time.sleep):
|
||
args = parse_args(argv)
|
||
try:
|
||
config = read_config(root)
|
||
preview = args.dry_run or config.get("DRY_RUN") == "1"
|
||
if not preview and config.get("ENABLED") != "1":
|
||
emit("skipped", "钉钉通知未配置或未启用，已跳过。")
|
||
return 0
|
||
title, text = render(args, config, body_text(args, root))
|
||
if preview:
|
||
emit("preview", "仅离线预览，未验证群配置或提醒对象；未查人员、认证或发送，不输出配置中的密钥。",
|
||
title=title, text=text,
|
||
mentions_requested=args.command == "message" and bool(args.at or args.at_id))
|
||
return 0
|
||
if config.get("MODE", "app-bot") != "app-bot":
|
||
raise NoticeError("当前封装仅支持 app-bot，请部署负责人检查配置。")
|
||
if any(not config.get(key) for key in ("CLIENT_ID", "TARGET", "ROBOT_CODE")):
|
||
raise NoticeError("机器人配置不完整，请部署负责人检查，不要在会话中粘贴配置。")
|
||
cli = dws_path(config)
|
||
secret = secret_for(config)
|
||
people = mentions(args, config, cli, run, sleep) if args.command == "message" else {}
|
||
if people:
|
||
text += "\n\n" + " ".join("@" + user_id for user_id in people)
|
||
message = {"title": title, "text": text}
|
||
if people:
|
||
message["atUserIds"] = list(people)
|
||
payload = {
|
||
"msgParam": json.dumps(message, ensure_ascii=False), "msgKey": "sampleMarkdown",
|
||
"openConversationId": (args.group if args.command == "message" else None) or config["TARGET"],
|
||
"robotCode": config["ROBOT_CODE"],
|
||
}
|
||
# DWS accepts --data -: avoid shell quoting, argv length limits and
|
||
# exposing message/target contents in the process command line.
|
||
command = [cli, "api", "POST", "/v1.0/robot/groupMessages/send",
|
||
"--data", "-", "--yes", "--format", "json"]
|
||
if config.get("PROFILE"):
|
||
command += ["--profile", config["PROFILE"]]
|
||
env = clean_env()
|
||
env.update(DWS_CLIENT_ID=config["CLIENT_ID"], DWS_CLIENT_SECRET=secret)
|
||
data = invoke(command, env, sending=True, run=run, sleep=sleep,
|
||
stdin_data=json.dumps(payload, ensure_ascii=False))
|
||
emit("accepted", "钉钉服务端已接受通知，不代表群成员已读。", processQueryKey=accepted_key(data))
|
||
return 0
|
||
except NoticeError as error:
|
||
emit("error", str(error))
|
||
return 1
|
||
except (OSError, UnicodeError, subprocess.SubprocessError, ValueError):
|
||
emit("error", "通知配置、正文文件或本机调用异常；请部署负责人检查，本次停止且不影响主任务。")
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
for stream in (sys.stdout, sys.stderr):
|
||
if hasattr(stream, "reconfigure"):
|
||
stream.reconfigure(encoding="utf-8")
|
||
sys.exit(main())
|