From 20c0de9b3829534685d1f931dc0b574c988eec6b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E9=93=AD=E8=BD=A9?= Date: Thu, 10 Sep 2026 17:14:23 +0800 Subject: [PATCH] =?UTF-8?q?release:=20v1.0.3=20=E7=BB=9F=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E4=BA=BA=20DWS=20=E9=80=9A=E7=9F=A5=E4=B8=8E=E7=AE=80=E5=8C=96?= =?UTF-8?q?=E5=88=9D=E5=A7=8B=E5=8C=96=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .dingtalk/README.md | 131 ++-- .dingtalk/config.env.example | 28 +- .gitattributes | 2 +- AGENTS.md | 7 +- README.md | 27 +- etunel-role-collaboration/SKILL.md | 2 +- .../references/dingtalk-progress-reporting.md | 14 +- .../references/dingtalk-script-usage.md | 81 ++- .../references/hub-workflow.md | 1 + .../中心角色精简职责与每轮提醒.md | 4 +- scripts/dingtalk-bot-message.ps1 | 11 - ...{dingtalk-bot-message => dingtalk-message} | 2 +- scripts/dingtalk-message.ps1 | 3 + scripts/dingtalk-progress.ps1 | 12 +- scripts/dingtalk-python.ps1 | 50 ++ scripts/dingtalk_notify.py | 426 ++++++------- scripts/test_dingtalk_notify.py | 562 +++++++++--------- 17 files changed, 670 insertions(+), 693 deletions(-) delete mode 100644 scripts/dingtalk-bot-message.ps1 rename scripts/{dingtalk-bot-message => dingtalk-message} (76%) create mode 100644 scripts/dingtalk-message.ps1 create mode 100644 scripts/dingtalk-python.ps1 diff --git a/.dingtalk/README.md b/.dingtalk/README.md index a2a6b0e..ca47d82 100644 --- a/.dingtalk/README.md +++ b/.dingtalk/README.md @@ -1,105 +1,66 @@ -# Hub 钉钉通知接入 +# 个人 DWS 项目通知 -这套封装只帮助 Hub 向项目群发送进度、异常摘要和负责人提醒，不是 Etunel 任务通道，也不负责审批、阶段门禁或项目留档。仅更新此仓库不会部署到实际项目；由部署负责人明确安装，成员 Agent 不调用。 +Hub 使用负责人确认的个人 DWS 账号发阶段汇报和 @ 提醒。只需要原生 DWS、Python 3.9+ 和一个项目配置文件；不使用机器人、应用密钥、Webhook 或 Stream 监听。 -## 文件与依赖 +**旧版升级：** 用新的个人账号配置替换旧配置，旧机器人配置不会自动转成个人发送授权。摘要入口已改名为 `dingtalk-message`／`dingtalk-message.ps1`，删除目标部署中的旧 `dingtalk-bot-message` 入口及 Stream 实验脚本。更新脚本和 Hub Skill／Hook 不等于已经迁移在途项目；按实际部署安排确认。 -- [dingtalk_notify.py](../scripts/dingtalk_notify.py)：共用 Python 实现，含配置、消息、人员解析和有限重试。 -- [dingtalk-progress](../scripts/dingtalk-progress)／[dingtalk-progress.ps1](../scripts/dingtalk-progress.ps1)：项目事件入口。 -- [dingtalk-bot-message](../scripts/dingtalk-bot-message)／[dingtalk-bot-message.ps1](../scripts/dingtalk-bot-message.ps1)：自定义 Markdown 与负责人提醒入口。 -- [离线测试](../scripts/test_dingtalk_notify.py)：假配置、假凭据及模拟命令，不发送真实消息。 +## 首次由 Hub 完成配置 -依赖 Python 3.9+，不需要第三方 Python 包、jq 或 Bash 才能在 Windows 运行。PowerShell 入口要求 7.3+；也可直接运行 Python。macOS/Linux 保留 Shell 入口。 +负责人不需要找编号、复制配置模板或填写密钥。Hub 在项目初始化时执行： -真实发送需要 [官方 DingTalk Workspace CLI](https://github.com/DingTalk-Real-AI/dingtalk-workspace-cli) 的原生可执行文件。按官方安装说明选择系统版本；Windows 使用 `dws.exe`，不要把 npm/Bun 的 `dws.cmd` 或 `dws.bat` 当原生程序。可以把原生文件加入 Hub 进程的 PATH，或在项目配置的 `DINGTALK_PROGRESS_DWS_PATH` 填它的绝对路径。本仓库不保存二进制，不自动安装、登录或更新全局 Skills。 +1. 检查本机 DWS；未登录时请负责人通过官方 DWS 登录流程授权，不索取密码或 Token。 +2. 用 `dws profile list --format json` 查看可用账号，以姓名和组织向负责人确认发送身份。 +3. 在选定的同一账号下，用 `dws chat +chat-search --query "负责人提供的群名称" --page-all --profile "查询返回的账号标识" --format json` 查群；结果不完整或有多个候选时不能自行选第一个。 +4. 一次问清：“是否由这个账号向这个项目群自动发送进度、阻塞和必要的 @ 提醒？”得到确认后，Hub 把工具实际返回的账号标识、群标识及项目名称传给初始化入口。下面为参数结构，不是需要人填写的编号表： -部署后检查 `python --version`（macOS/Linux 可用 `python3`）及原生 `dws version`。所用 DWS 需支持 `api --data -` 从标准输入读 JSON；正文、群参数不放命令行，密钥仅临时注入子进程环境。更换 DWS 版本后由部署负责人核对兼容性，离线预览不证明真实接口可用。 + ```text + python scripts/dingtalk_notify.py init --profile "实际查询结果" --group "实际查询结果" --project "项目名称" --yes + ``` -## 部署到 Hub 项目 +初始化检查账号存在且可登录，创建 `.dingtalk/config.env` 并启用通知；**不会发测试消息，也不会替人确认群选择**。群标识必须来自该账号的查询结果。初始化不创建钉钉群、不申请权限。已有配置默认不覆盖；迁移或变更账号、群时重新确认，再加 `--replace`，原配置被替换且不另存含密钥的备份。 -将共用 Python 文件和需要的平台入口一起放入目标项目的 `scripts/`，不要只复制一个入口。重试逻辑已合入共用实现，无需复制源项目的 `dingtalk-dns-retry.sh`。保持下列布局： +配置仅 4 项： + +| 字段 | 含义 | 谁填写 | +| --- | --- | --- | +| `DINGTALK_PROGRESS_ENABLED` | `1` 启用、`0` 暂停 | 初始化写入，负责人可随时要求暂停 | +| `DINGTALK_PROGRESS_PROFILE` | 固定个人账号 | Hub 使用 DWS 真实返回值 | +| `DINGTALK_PROGRESS_TARGET` | 固定项目群 | Hub 使用同账号查询并经确认的结果 | +| `DINGTALK_PROGRESS_PROJECT` | 项目名称 | Hub 使用已确认名称 | + +配置不包含凭据，但仍是本机信息，本仓库已排除出 Git；Hub 在其他目标项目初始化时也应把 `.dingtalk/config.env` 加入该项目的 Git 忽略规则。不要把它或登录态复制给其他同事；各部署分别确认账号和群。登录由 DWS 自己管理，后续切换 DWS 默认账号不会改变本项目的发送身份。 + +## Windows 与 macOS/Linux + +- Windows 使用 PowerShell 7.3+，优先运行 `./scripts/dingtalk-python.ps1 init ...`、`check ...`；事件用 `./scripts/dingtalk-progress.ps1 ...`，提醒用 `./scripts/dingtalk-message.ps1 ...`。 +- PowerShell 共用启动器选择独立 Python，避免商店版 Python 无法读取终端的 DWS 登录。必要时用 `DINGTALK_PYTHON` 指定独立解释器路径；原生 `dws.exe` 须在当前进程 PATH 中。 +- macOS/Linux 通常用 `python3 scripts/dingtalk_notify.py ...`；也可用 `sh scripts/dingtalk-progress ...`、`sh scripts/dingtalk-message ...`。初始化生成的配置在 POSIX 系统为仅当前用户读写；Windows 使用部署目录的访问权限。 +- 将共用 Python 文件和所需平台入口放在目标 Hub 项目的 `scripts/` 中。安装职责 Skill 不会自动部署这些脚本。 + +DWS 安装与登录以 [官方项目](https://github.com/DingTalk-Real-AI/dingtalk-workspace-cli) 和当前命令帮助为准；不要为通知分发机器人应用密钥。 + +## 日常调用 ```text -目标项目/ -├── scripts/ -│ ├── dingtalk_notify.py -│ ├── dingtalk-progress -│ ├── dingtalk-bot-message -│ ├── dingtalk-progress.ps1 -│ └── dingtalk-bot-message.ps1 -└── .dingtalk/ - └── config.env +python scripts/dingtalk_notify.py progress milestone "**已完成**：阶段成果已归档。下一步安排联调。" +python scripts/dingtalk_notify.py progress blocked "缺少接口说明，请补充后回自己的角色会话同步。" --at "完整姓名" +python scripts/dingtalk_notify.py message --title "资料补充提醒" --text "请补齐已约定资料，并回自己的角色会话确认。" --at "完整姓名" ``` -由部署负责人从 [配置示例](config.env.example) 新建实际配置，别覆盖已有有效配置： +通知保留 AI 标识并标注自动汇报；摘要用简短中文 Markdown。`--at` 按完整姓名查询唯一人员，再生成个人消息的原生提及标记。两个发送命令都支持 `--file` 正文和完全离线的 `--dry-run`。不支持临时换群、手填人员编号或退回机器人发送。 -- UTF-8，可带 BOM，支持 Windows CRLF 和 LF；值按字面解析，不执行 Shell、不展开变量，不加 Shell 引号。 -- 初始保持 `ENABLED=0`、`DRY_RUN=1`（实际字段均有 `DINGTALK_PROGRESS_` 前缀）。 -- 配置包括应用、机器人、目标群、项目名称、开关及机器人密钥。机器人必须属于配置的应用，已经加入目标群，并具备企业内机器人发送消息权限（`qyapi_robot_sendmsg`）；脚本不自动加群或申请权限。 -- macOS/Linux 对实际配置执行 `chmod 600 .dingtalk/config.env`；接受 400 或 600。Windows 不以 Unix 权限位作门槛，由部署负责人用 NTFS 权限限制其他用户读取或修改配置。 -- 实际配置必须被目标项目的 Git 忽略。不要把它、真实凭据或内部标识回传给 AI 会话或提交仓库。 +事件含义见 [通知规则](../etunel-role-collaboration/references/dingtalk-progress-reporting.md)，详细调用与结果见 [脚本用法](../etunel-role-collaboration/references/dingtalk-script-usage.md)。 -路径与当前工作目录无关：配置总从 Python 脚本所在项目根目录读取。消息的 `--file` 路径则相对当前工作目录解析。 +## 检查与失败处理 -## 凭据：所有系统统一读取配置 +- `python scripts/dingtalk_notify.py check --at "完整姓名"`：只检查固定账号登录和可选姓名，不发消息，不证明实际 @ 提醒成功。不是每条通知的必做步骤。 +- `accepted` 加有效 `openTaskId`：个人发送请求已接受，不等于已投递或已读。必要时在同账号下用 `dws chat +messages-query-send-status --open-task-id "回执中的实际值" --profile "配置中的账号" --format json` 只读查询；不把任务回执当消息编号，不循环重发。 +- 仅明确 DNS 解析失败最多尝试 3 次、间隔 10 秒；认证、权限或未知结果停止。登录失效请负责人恢复登录，不切账号、不退回机器人。通知失败不阻塞 Etunel 主任务。 -普通进度发送使用内部应用身份，不需要用户 OAuth 登录。Windows、macOS、Linux **只从本地 `.dingtalk/config.env` 的 `DINGTALK_PROGRESS_CLIENT_SECRET` 读取机器人密钥**，不再从 `DING_SEC`、其他环境变量或 macOS 钥匙串回退取值。 - -由部署负责人在本地编辑器中填写该字段，保持单行字面值，不加 Shell 引号；不要在会话、命令历史或仓库示例里填写真实值。密钥为空或字段缺失时，真实发送停止并提示负责人补配置；离线预览不要求填写密钥。 - -脚本仅在发送时把该值临时注入 DWS 子进程的 `DWS_CLIENT_SECRET`，同时注入配置中的 `DWS_CLIENT_ID`。这只是脚本内部对接 DWS 的方式，不需要负责人再配置环境变量。密钥不进入命令参数、消息正文、预览或错误输出；人员搜索进程也不会取得机器人密钥。 - -实际 `config.env` 现在包含明文密钥：macOS/Linux 设为 400 或 600；Windows 通过文件“属性 → 安全”限制为 Hub 运行账号及必要管理员可访问，不放在公共共享目录。目标项目必须忽略实际配置及其备份，只提交空值示例。不要把配置文件用作 `--file` 正文、成果或通知附件。 - -### 从旧配置迁移 - -1. 更新共用 Python 脚本及需要的平台入口。 -2. 由部署负责人在本地配置补填 `DINGTALK_PROGRESS_CLIENT_SECRET`；删除旧的 `DINGTALK_PROGRESS_KEYCHAIN_SERVICE` 字段。 -3. 检查文件权限与 Git 忽略，再做离线预览；正式发送仍需要启用开关、正确的应用权限和目标群。 -4. 旧环境变量与钥匙串不再被脚本读取；本次改动不会删除其中原有内容。 - -脚本每次调用重新读取本地配置，不再因为密钥配置依赖 Hub 继承某个终端的环境变量；Python、dws 路径及按姓名查询所需的 DWS 登录仍须对实际运行 Hub 的操作系统账号可用。 - -## 可选的按姓名提醒 - -`--at` 会调用 DWS 人员搜索，和应用机器人发送是不同的认证需求。只有使用按姓名提醒时，部署负责人需要： - -1. 在 DWS 中登录需要使用的组织账号； -2. 将已确认的稳定 profile 填入 `DINGTALK_PROGRESS_PROFILE`； -3. 确认该账号、机器人应用、目标群属于同一组织，且账号允许查询人员。 - -脚本在搜索和发送中传入同一个 profile，不自动选择或切换账号；应用归属与群归属仍需部署负责人核对，脚本不能仅凭 profile 字符串证明二者一致。查询只接受完整姓名的唯一匹配；查不到、重名、权限不足或返回结构不支持时整条提醒停止，不默认选择第一个，也不静默删掉 @ 后发送。 - -`--at-id` 和 `--group` 保留源脚本兼容能力，仅用于已有真实映射或用户明确指定的目标；不让 AI 猜编号或在群之间试发。普通进度通知不需要配置人员账号。 - -## 先做离线预览 - -在目标项目根目录选一种入口。以下命令只预览，不联网，不需要 dws，不查通讯录、不使用密钥认证。已有配置会在本地读取，但密钥不会输出： - -```powershell -python scripts/dingtalk_notify.py progress milestone '本阶段成果已确认并归档，下一步安排联调。' --dry-run -./scripts/dingtalk-bot-message.ps1 --title '需要补充接口说明' --text '请接口负责人补齐返回值说明，并回自己的角色会话同步结果。' --dry-run -``` - -macOS/Linux 等价命令： +维护时只做离线测试；真实发送必须另获明确授权： ```sh -sh scripts/dingtalk-progress milestone '本阶段成果已确认并归档，下一步安排联调。' --dry-run -sh scripts/dingtalk-bot-message --title '需要补充接口说明' --text '请接口负责人补齐返回值说明，并回自己的角色会话同步结果。' --dry-run -``` - -两种入口都遵守配置的 `DRY_RUN=1`，显式 `--dry-run` 不能被配置取消。即使没有配置，也能显式预览正文；此时不表示项目配置可用。正文支持中文多行 Markdown；复杂正文建议先保存 UTF-8 文件，用 `--file` 代替内联摘要或 `--text`，它不是附件上传。 - -确认部署与发送授权后，负责人再将 `ENABLED=1` 且 `DRY_RUN=0`。仓库维护测试不能用真实发送作冒烟测试。实际 Hub 的事件判断、参数示例和结果含义见 [通知规则](../etunel-role-collaboration/references/dingtalk-progress-reporting.md) 和 [脚本用法](../etunel-role-collaboration/references/dingtalk-script-usage.md)。 - -## 失败处理与测试 - -仅明确的 DNS 解析失败每隔 10 秒重试，总计最多 3 次；已获有效回执即停止。认证、权限、未知错误不重试；超时或缺少回执不盲目重发。单次 DWS 调用超时 60 秒；本次结束后由 Hub 说明，不在外层循环重试。只输出安全摘要和发送回执，不输出原始请求或错误日志。 - -本封装不建消息队列、计划任务、重发历史或通知状态库；避免重复事件仍由 Hub 利用已有项目留档判断，不重造 Etunel 软件机制。 - -```text python -X utf8 scripts/test_dingtalk_notify.py +bash -n scripts/dingtalk-progress +bash -n scripts/dingtalk-message ``` - -测试仅在临时目录使用假配置、假密钥和模拟 DWS；同时检查当前系统可用的 PowerShell/Shell 入口，不访问真实项目配置。覆盖配置密钥为唯一来源、密钥不回显、不通过命令参数传递以及预览不认证，不等于完成真实钉钉发送验收。 diff --git a/.dingtalk/config.env.example b/.dingtalk/config.env.example index f413a85..2712f6f 100644 --- a/.dingtalk/config.env.example +++ b/.dingtalk/config.env.example @@ -1,25 +1,7 @@ -# 由部署负责人复制为 .dingtalk/config.env；macOS/Linux 设置 chmod 600。 -# Windows 用当前 Hub 运行用户的文件权限，不套用 Unix 权限检查。 -# UTF-8（可带 BOM），支持 LF/CRLF；等号右侧是字面值，不加 Shell 引号，不执行变量。 - -# 两个入口共用开关；显式 --dry-run 也永远不发送。 +# Hub 查询账号和群、取得负责人确认后，用 init 生成真实 config.env。 +# 不需手工复制或填编号；不保存机器人密钥或个人登录凭据。 +# 等号右侧为字面值，不加 Shell 引号。ENABLED=0 暂停通知。 DINGTALK_PROGRESS_ENABLED=0 -DINGTALK_PROGRESS_DRY_RUN=1 - -# 企业内部应用机器人：发送本身不需要用户 OAuth 登录。 -DINGTALK_PROGRESS_MODE=app-bot -DINGTALK_PROGRESS_CLIENT_ID= -# 仅在本地 config.env 填真实 Client Secret；本示例必须保持空值，绝不提交真实配置。 -DINGTALK_PROGRESS_CLIENT_SECRET= -DINGTALK_PROGRESS_TARGET= -DINGTALK_PROGRESS_ROBOT_CODE= - -# 所有系统统一从 config.env 取密钥，不再读取 DING_SEC 或 macOS 钥匙串。 -# 可选：原生 dws 的绝对路径（Windows 填 dws.exe，不填 cmd/bat 包装器）。 -DINGTALK_PROGRESS_DWS_PATH= -# 可选：按姓名 @ 时必需。由部署负责人确认与应用、目标群同组织的已登录 profile。 -# 普通进度通知可留空；不是 Etunel 人员或会话编号。 DINGTALK_PROGRESS_PROFILE= - -DINGTALK_PROGRESS_PROJECT=你的项目名称 -DINGTALK_PROGRESS_TITLE_PREFIX=[项目进度] +DINGTALK_PROGRESS_TARGET= +DINGTALK_PROGRESS_PROJECT= diff --git a/.gitattributes b/.gitattributes index 6e2127f..4939071 100644 --- a/.gitattributes +++ b/.gitattributes @@ -13,4 +13,4 @@ *.sh text eol=lf *.ps1 text eol=lf scripts/dingtalk-progress text eol=lf -scripts/dingtalk-bot-message text eol=lf +scripts/dingtalk-message text eol=lf diff --git a/AGENTS.md b/AGENTS.md index f23c21e..dbf3a81 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -61,7 +61,7 @@ - 只有 Hub／主 Agent 可以调用通知；成员 Agent 只向 Hub 返回成果、状态或阻塞。 - 通知是项目群的单向辅助可见性与异常提醒，不是角色间通信、任务派发、审批、人类确认、项目门禁或项目事实源。 -- Agent 只用 `scripts/dingtalk_notify.py` 的 `progress`／`message`，或等价的 `dingtalk-progress`／`dingtalk-bot-message` Shell、PowerShell 入口。进度事件仍限定为 `start`、`milestone`、`blocked`、`complete`、`failed`；负责人提醒不新增状态。同一事项不重复通知，不得绕过封装直接调用 OpenAPI 或 `dws api`。可验证里程碑先完成项目留档，再发送简短中文 Markdown。例如： +- Agent 只用 `scripts/dingtalk_notify.py` 的 `progress`／`message`，或等价的 `dingtalk-progress`／`dingtalk-message` Shell、PowerShell 入口，统一以已确认的个人 DWS 账号发送。进度事件仍限定为 `start`、`milestone`、`blocked`、`complete`、`failed`；负责人提醒不新增状态。同一事项不重复通知，不得绕过封装直接调用 OpenAPI 或 `dws api`。可验证里程碑先完成项目留档，再发送简短中文 Markdown。例如： ```sh python scripts/dingtalk_notify.py progress milestone "**已完成** @@ -75,7 +75,8 @@ - 事件语义与消息格式以 [dingtalk-progress-reporting.md](etunel-role-collaboration/references/dingtalk-progress-reporting.md) 为准；安装与本机配置参考 [.dingtalk/README.md](.dingtalk/README.md)。 - 通知封装是辅助接入参考，不得据此反推或改变 Etunel 主流程；不创建通知队列或外层重试。仅明确 DNS 失败最多 3 次、间隔 10 秒，结果未知不盲目重发。 - 仓库维护和测试期间不得把真实发送当作冒烟测试。只有任务明确要求发送且当前环境已经授权时，才允许触发真实通知。 -- 不读取、输出、修改或提交 `.dingtalk/config.env`、Client Secret、`DING_SEC`、App Token 或其他凭据和内部标识。 +- Hub 首次查询个人账号及项目群，以姓名、组织和群名取得负责人确认后，用封装 `init` 创建 `.dingtalk/config.env`；只含通知开关、固定账号、群和项目名，不保存机器人或登录密钥。AI 可以在授权范围内维护该配置，但不提交或在群里展示内部标识；换账号或群须重新确认。 +- 不读取、输出或提交 DWS 登录凭据、Client Secret、`DING_SEC`、App Token。旧机器人配置须经确认重新初始化，不静默启用个人代发；保留 AI 标识，个人通知不代表负责人亲自批准。 ## 编辑约定 @@ -104,7 +105,7 @@ rtk python -X utf8 "$env:USERPROFILE\.codex\skills\.system\skill-creator\scripts\quick_validate.py" "etunel-role-collaboration" rtk python -X utf8 scripts/test_dingtalk_notify.py rtk "C:\Program Files\Git\bin\bash.exe" -n scripts/dingtalk-progress -rtk "C:\Program Files\Git\bin\bash.exe" -n scripts/dingtalk-bot-message +rtk "C:\Program Files\Git\bin\bash.exe" -n scripts/dingtalk-message rtk proxy git diff --check rtk proxy git status --short ``` diff --git a/README.md b/README.md index a5f95cf..32ed879 100644 --- a/README.md +++ b/README.md @@ -14,7 +14,7 @@ ## 版本与升级 -本版为 `v1.0.2`，流程设计基线仍为 **V1.0**。版本采用 `vX.Y.Z`：前两位 `X.Y` 跟随 HTML 主流程中明确标注且已确认的版本，第三位 `Z` 表示该流程版本下的仓库发布迭代。正式发布以 Git 标签和对应发布包为准，主分支可能包含尚未发布的修改。 +本版为 `v1.0.3`，流程设计基线仍为 **V1.0**。版本采用 `vX.Y.Z`：前两位 `X.Y` 跟随 HTML 主流程中明确标注且已确认的版本，第三位 `Z` 表示该流程版本下的仓库发布迭代。正式发布以 Git 标签和对应发布包为准，主分支可能包含尚未发布的修改。 | 变更 | 版本规则 | | --- | --- | @@ -26,7 +26,17 @@ 发布说明应写清：流程基线、改了什么、使用者需要更新哪些文件、在途项目是否需要重新确认。不要把“仓库已发布”写成“角色已更新”。更新插件缓存、Etunel 中独立配置的 Hook 或迁移在途项目，都需另外明确确认；已运行项目继续使用其登记基线。 -### v1.0.2 更新范围 +### v1.0.3 更新范围 + +阶段汇报和负责人 @ 提醒统一改为个人 DWS。Hub 首次查询账号和群，经负责人确认后用 `init` 自动创建四项无密钥配置；旧机器人配置不能直接沿用。摘要入口改名为 `dingtalk-message`／`dingtalk-message.ps1`，不再部署机器人或 Stream 实验。更新时同步脚本、通知 reference 和中心 Hook；其他角色边界及 V1.0 主流程不变。详见 [部署与迁移](.dingtalk/README.md)。 + +- `progress` 与 `message` 都支持按姓名 @，查询和发送固定同一账号；保留 AI 标识和中文自动汇报说明，不自动更换身份或通知群。 +- Windows 共用启动器使用独立 Python，修复商店版 Python 子进程无法识别终端 DWS 登录的部署问题。 +- 27 项离线回归通过，覆盖配置初始化、旧配置替换、个人发送回执、@ 参数、有限重试和 Windows 实际传参。 +- Windows 个人账号里程碑消息已通过新封装实测，投递状态为 `SUCCESS`；个人 @ 请求此前已成功提交，接收端提醒效果仍待本人确认。原生 macOS/Linux 运行尚未实测，Shell 语法已检查。 +- 配置迁移、目标项目脚本部署、Etunel 中心 Hook 和插件缓存更新须按部署安排确认，不自动迁移其他项目。通知仍不是任务通道、审批、门禁或事实源。 + +### v1.0.2 更新范围（历史，不是当前部署说明） - 明确前两位跟随 HTML 流程、其余发布迭代只升第三位的版本约定；主流程仍为 V1.0，八角色及阶段门禁不变。 - 钉钉通知改为共用 Python 实现，保留 macOS/Linux 入口，新增 Windows PowerShell 入口；支持项目事件、自定义中文 Markdown、文件正文和按姓名提醒负责人。 @@ -121,9 +131,10 @@ EP-Hub-Skill/ └── scripts/ # Hub 辅助通知，不是 Etunel 任务机制 ├── dingtalk_notify.py # Python 共用实现，含有限重试 ├── dingtalk-progress # macOS/Linux 进度入口 - ├── dingtalk-bot-message # macOS/Linux 摘要与负责人提醒 + ├── dingtalk-message # macOS/Linux 个人摘要与负责人提醒 ├── dingtalk-progress.ps1 # Windows PowerShell 进度入口 - ├── dingtalk-bot-message.ps1 # Windows PowerShell 摘要与提醒 + ├── dingtalk-message.ps1 # Windows PowerShell 个人摘要与提醒 + ├── dingtalk-python.ps1 # Windows 共用启动器，含初始化与检查 └── test_dingtalk_notify.py # 不联网的隔离测试 ``` @@ -185,11 +196,11 @@ python scripts/dingtalk_notify.py progress /dev/null 2>&1; then diff --git a/scripts/dingtalk-message.ps1 b/scripts/dingtalk-message.ps1 new file mode 100644 index 0000000..686da94 --- /dev/null +++ b/scripts/dingtalk-message.ps1 @@ -0,0 +1,3 @@ +# Personal DWS message entry; preserves Chinese, Markdown and exit status. +& "$PSScriptRoot/dingtalk-python.ps1" message @args +exit $LASTEXITCODE diff --git a/scripts/dingtalk-progress.ps1 b/scripts/dingtalk-progress.ps1 index 1d30041..c3e0238 100644 --- a/scripts/dingtalk-progress.ps1 +++ b/scripts/dingtalk-progress.ps1 @@ -1,11 +1,3 @@ -# Requires PowerShell 7.3+ for native argument passing (quotes and Markdown). -$ErrorActionPreference = 'Stop' -if ($PSVersionTable.PSVersion -lt [version]'7.3') { throw 'Please use PowerShell 7.3+ or the Python entry.' } -$PSNativeCommandArgumentPassing = 'Standard' -$env:PYTHONIOENCODING = 'utf-8' -if (Get-Command python -CommandType Application -ErrorAction SilentlyContinue) { - & python "$PSScriptRoot/dingtalk_notify.py" progress @args -} elseif (Get-Command py -CommandType Application -ErrorAction SilentlyContinue) { - & py -3 "$PSScriptRoot/dingtalk_notify.py" progress @args -} else { throw 'Python 3.9+ is required.' } +# PowerShell 7.3+; shared launcher preserves Chinese, Markdown and exit status. +& "$PSScriptRoot/dingtalk-python.ps1" progress @args exit $LASTEXITCODE diff --git a/scripts/dingtalk-python.ps1 b/scripts/dingtalk-python.ps1 new file mode 100644 index 0000000..32cf966 --- /dev/null +++ b/scripts/dingtalk-python.ps1 @@ -0,0 +1,50 @@ +# Shared Windows launcher. Also supports: ./scripts/dingtalk-python.ps1 check --at '姓名' +$ErrorActionPreference = 'Stop' +if ($PSVersionTable.PSVersion -lt [version]'7.3') { throw '请使用 PowerShell 7.3+ 或独立 Python 入口。' } +$PSNativeCommandArgumentPassing = 'Standard' + +function Resolve-NoticePython { + $candidates = @() + if ($env:DINGTALK_PYTHON) { + # An explicit choice must work; never silently substitute another runtime. + $candidates = @($env:DINGTALK_PYTHON) + } else { + if ($env:USERPROFILE) { + $bundled = Join-Path $env:USERPROFILE '.cache/codex-runtimes/codex-primary-runtime/dependencies/python/python.exe' + if (Test-Path -LiteralPath $bundled -PathType Leaf) { $candidates += $bundled } + } + $candidates += @(Get-Command python, python3 -All -CommandType Application -ErrorAction SilentlyContinue | + Select-Object -ExpandProperty Source) + # py is probed, not trusted: it can also resolve to Store Python. + $candidates += @(Get-Command py -All -CommandType Application -ErrorAction SilentlyContinue | + Select-Object -ExpandProperty Source) + } + foreach ($candidate in ($candidates | Select-Object -Unique)) { + if ($candidate -match '(?i)[\\/]WindowsApps[\\/]|PythonSoftwareFoundation\.Python\.') { continue } + $prefix = if ([IO.Path]::GetFileName($candidate) -eq 'py.exe') { @('-3') } else { @() } + try { + $probe = & $candidate @prefix -I -c 'import json,sys; print(json.dumps({"exe":sys.executable,"supported":sys.version_info >= (3,9)}))' 2>$null + if ($LASTEXITCODE -ne 0) { continue } + $info = ($probe -join "`n") | ConvertFrom-Json + if ($info.supported -and $info.exe -and + $info.exe -notmatch '(?i)[\\/]WindowsApps[\\/]|PythonSoftwareFoundation\.Python\.' -and + (Test-Path -LiteralPath $info.exe -PathType Leaf)) { return $info.exe } + } catch { continue } + } + throw '未找到独立 Python 3.9+。请安装 python.org 版本或将 DINGTALK_PYTHON 设为独立 python.exe 的路径；不使用商店版，也不自动安装。' +} + +$pythonExe = Resolve-NoticePython +$previousArgs = [Environment]::GetEnvironmentVariable('DINGTALK_PROGRESS_ARGV_B64', 'Process') +$previousEncoding = [Environment]::GetEnvironmentVariable('PYTHONIOENCODING', 'Process') +try { + $encodedArgs = ConvertTo-Json -InputObject @($args) -Compress + $env:DINGTALK_PROGRESS_ARGV_B64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($encodedArgs)) + $env:PYTHONIOENCODING = 'utf-8' + & $pythonExe "$PSScriptRoot/dingtalk_notify.py" + $exitCode = $LASTEXITCODE +} finally { + [Environment]::SetEnvironmentVariable('DINGTALK_PROGRESS_ARGV_B64', $previousArgs, 'Process') + [Environment]::SetEnvironmentVariable('PYTHONIOENCODING', $previousEncoding, 'Process') +} +exit $exitCode diff --git a/scripts/dingtalk_notify.py b/scripts/dingtalk_notify.py index d3cd37f..1bb4761 100644 --- a/scripts/dingtalk_notify.py +++ b/scripts/dingtalk_notify.py @@ -1,275 +1,277 @@ #!/usr/bin/env python3 -"""Hub-only DingTalk notifications. Python 3.9+, standard library, no shell eval.""" - +"""Hub notifications via a confirmed personal DWS account. Python 3.9+, stdlib.""" import argparse +import base64 import datetime as dt import json import os from pathlib import Path import re import shutil -import stat import subprocess import sys +import tempfile 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 +PREFIX = 'DINGTALK_PROGRESS_' +KEYS = {'ENABLED', 'PROFILE', 'TARGET', 'PROJECT'} +EVENTS = {'start': '开始推进', 'milestone': '阶段里程碑', 'blocked': '遇到阻塞', + 'complete': '本轮完成', 'failed': '执行失败'} +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.I) class NoticeError(Exception): - """A human-safe error: never include raw subprocess/config contents.""" + """Safe diagnostics: never echo raw DWS errors or configuration.""" def emit(status, message, **extra): - print(json.dumps({"status": status, "message": message, **extra}, ensure_ascii=False)) + print(json.dumps(dict(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): +def json_object(text): try: - value = json.loads(output) - return value if isinstance(value, dict) else {} + data = json.loads(text) + return data if isinstance(data, 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 read_config(root): + path = root / '.dingtalk' / 'config.env' + if not path.exists(): + return {} + 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 + k for k in KEYS} or key in result: + raise NoticeError('通知配置不是当前个人账号格式，请由 Hub 确认账号和群后重新 init；不沿用机器人配置。') + result[key] = value.strip() + config = {k[len(PREFIX):]: v for k, v in result.items()} + if config.get('ENABLED', '0') not in ('0', '1'): + raise NoticeError('通知开关只能是 0 或 1。') + return config -def invoke(command, env, sending=False, run=subprocess.run, sleep=time.sleep, stdin_data=None): - for attempt in range(1, ATTEMPTS + 1): +def clean_env(): + return {k: v for k, v in os.environ.items() + if k not in ('DING_SEC', 'DWS_CLIENT_ID', 'DWS_CLIENT_SECRET') and not k.startswith(PREFIX)} + + +def dws_path(): + executable = sys.executable.replace('\\', '/').lower() + if sys.platform == 'win32' and ('/windowsapps/' in executable or 'pythonsoftwarefoundation.python.' in executable): + raise NoticeError('请使用项目 PowerShell 入口或独立 Python 3.9+；商店版 Python 可能无法识别 DWS 登录。') + path = shutil.which('dws') + if not path or (os.name == 'nt' and Path(path).suffix.lower() != '.exe'): + raise NoticeError('未找到原生 dws，请安装并加入当前 Hub 运行环境的 PATH。') + return path + + +def accepted_task(data): + if (data.get('ok') is not True or data.get('identity') != 'user' + or data.get('tool') != 'send_personal_message' or data.get('partial') or data.get('failures')): + return None + result = data.get('result') + if not isinstance(result, dict) or result.get('success') is not True or result.get('errorCode') not in (None, 0, ''): + return None + payload = result.get('result') + task = payload.get('openTaskId') if isinstance(payload, dict) else None + return task.strip() if isinstance(task, str) and task.strip() else None + + +def invoke(command, run=subprocess.run, sleep=time.sleep, sending=False): + for attempt in range(3): try: - result = run(command, env=env, input=stdin_data, capture_output=True, text=True, - encoding="utf-8", errors="replace", timeout=COMMAND_TIMEOUT) + result = run(command, env=clean_env(), capture_output=True, text=True, + encoding='utf-8', errors='replace', timeout=60) except subprocess.TimeoutExpired: - raise NoticeError("钉钉调用超时，结果未知；本次不自动重发，请负责人核对。") from None + raise NoticeError('钉钉调用超时，结果未知，不自动重发；主任务继续。') from None except OSError: - raise NoticeError("钉钉命令无法启动，请部署负责人检查安装和进程权限。") from None + raise NoticeError('无法启动 DWS，请检查安装；本次停止。') from None data = json_object(result.stdout) - if sending and accepted_key(data): + if sending and accepted_task(data): return data if result.returncode == 0: if sending: - raise NoticeError("未收到有效发送回执，结果未知；本次不自动重发，请负责人核对。") + raise NoticeError('未取得有效的个人发送回执，结果未知，不自动重发。') + if not data or data.get('success') is False or data.get('ok') is False: + raise NoticeError('DWS 查询未返回有效结果，本次停止。') 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) + error = json_object(result.stderr).get('error', {}) + if isinstance(error, dict) and error.get('category') == 'auth': + raise NoticeError('个人 DWS 登录失效，请在 Hub 的运行账号下恢复登录；不换身份发送。') + if isinstance(error, dict) and error.get('retryable') is False: + raise NoticeError('DWS 拒绝本次调用，请检查权限或参数；不自动重试。') + if DNS_ERROR.search(result.stdout + '\n' + result.stderr): + if attempt < 2: + sleep(10) continue - raise NoticeError("钉钉域名解析连续失败 3 次，本次通知已停止；主任务继续。") - raise NoticeError("钉钉调用未成功，未自动重发；请部署负责人检查认证、权限或网络，主任务继续。") - raise AssertionError("unreachable") + raise NoticeError('域名解析连续失败 3 次，通知停止，主任务继续。') + raise NoticeError('钉钉调用失败或结果未知；请检查登录、权限或网络，不自动重发。') -def mentions(args, config, cli, run=subprocess.run, sleep=time.sleep): +def check_login(cli, profile, run, sleep): + if not profile or ':' not in profile or any(c.isspace() for c in profile): + raise NoticeError('缺少已确认的发送账号，请先由 Hub 初始化通知。') + data = invoke([cli, 'auth', 'status', '--profile', profile, '--format', 'json'], run, sleep) + if data.get('authenticated') is not True or data.get('token_valid') is not True: + raise NoticeError('个人 DWS 未登录或授权已失效，请负责人完成登录；本次未发送。') + + +def mentions(names, cli, profile, run, 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；不要临时切换或猜测账号。") + names = list(dict.fromkeys(n.strip() for value in names for n in value.replace('，', ',').split(','))) for name in names: - data = invoke([cli, "aisearch", "person", "--query", name, "--dimension", "name", - "--profile", config["PROFILE"], "--format", "json"], - clean_env(), run=run, sleep=sleep) + if not name: + raise NoticeError('提醒对象不能为空。') + data = invoke([cli, 'aisearch', 'person', '--query', name, '--dimension', 'name', + '--profile', profile, '--format', 'json'], run, sleep) candidates = {} - rows = data.get("result", []) - if not isinstance(rows, list): - rows = [] - for row in rows: + for row in data.get('result', []): 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 + meta = row.get('meta') if isinstance(row.get('meta'), dict) else {} + person = row.get('openDingTalkId') + if (meta.get('name') or row.get('title') or row.get('author')) == name and isinstance(person, str) and person.strip(): + candidates[person] = name if len(candidates) != 1: - raise NoticeError("提醒对象未唯一匹配，本条未发送；请负责人确认完整姓名或已有的有效人员映射。") + raise NoticeError('提醒对象未唯一匹配，请确认完整姓名；不猜编号、不删掉 @ 后发送。') people.update(candidates) return people -def main(argv=None, root=ROOT, run=subprocess.run, sleep=time.sleep): - args = parse_args(argv) +def initialize(args, root, run, sleep): + if not args.yes: + raise NoticeError('先向负责人确认发送账号、项目群及自动通知，再使用 init --yes。') + values = {'ENABLED': '1', 'PROFILE': args.profile.strip(), 'TARGET': args.group.strip(), 'PROJECT': args.project.strip()} + if any(not v or '\n' in v or '\r' in v or '\0' in v for v in values.values()): + raise NoticeError('初始化内容不能为空或含换行。') + path = root / '.dingtalk' / 'config.env' + if path.is_symlink() or path.parent.is_symlink(): + raise NoticeError('配置路径不能是符号链接。') + if path.exists() and not args.replace: + raise NoticeError('配置已存在；变更账号或群须重新确认后使用 --replace。') + cli = dws_path() + profiles = invoke([cli, 'profile', 'list', '--format', 'json'], run, sleep).get('profiles', []) + if len([p for p in profiles if isinstance(p, dict) and p.get('profile') == values['PROFILE']]) != 1: + raise NoticeError('初始化账号不在 DWS 实际账号列表中。') + check_login(cli, values['PROFILE'], run, sleep) + # Hub supplies a group ID from same-profile discovery after human confirmation. + path.parent.mkdir(parents=True, exist_ok=True) + content = '# Hub 经负责人确认后生成；仅个人 DWS，不保存登录凭据。\n' + ''.join( + f'{PREFIX}{k}={v}\n' for k, v in values.items()) + descriptor, temporary = tempfile.mkstemp(prefix='.notify-', suffix='.tmp', dir=path.parent) try: + with os.fdopen(descriptor, 'w', encoding='utf-8', newline='\n') as stream: + stream.write(content) + os.replace(temporary, path) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + emit('initialized', '已保存负责人确认的个人账号和项目群，通知已启用；未发送测试消息。') + + +def parse_args(argv): + parser = argparse.ArgumentParser(description='Hub 个人 DWS 通知：初始化、进度、提醒、只读检查。') + commands = parser.add_subparsers(dest='command', required=True) + init = commands.add_parser('init', help='负责人确认后创建配置，不发送消息') + init.add_argument('--profile', required=True) + init.add_argument('--group', required=True) + init.add_argument('--project', required=True) + init.add_argument('--yes', action='store_true') + init.add_argument('--replace', action='store_true') + progress = commands.add_parser('progress') + progress.add_argument('event', choices=EVENTS) + progress.add_argument('summary', nargs='*') + progress.add_argument('--file') + message = commands.add_parser('message') + message.add_argument('--title', required=True) + body = message.add_mutually_exclusive_group(required=True) + body.add_argument('--text') + body.add_argument('--file') + check = commands.add_parser('check', help='检查固定账号登录及可选姓名，不发送') + for command in (progress, message, check): + command.add_argument('--at', action='append', default=[], help='完整姓名；可重复或用逗号分隔') + for command in (progress, message): + command.add_argument('--dry-run', action='store_true', help='完全离线，只预览中文正文') + return parser.parse_args(argv) + + +def render(args, config, root): + if args.command == 'progress' and bool(args.summary) == bool(args.file): + raise NoticeError('进度摘要和 --file 必须且只能提供一项。') + if args.file: + path = Path(args.file).resolve() + config_path = root / '.dingtalk' / 'config.env' + if path == config_path.resolve() or (path.exists() and config_path.exists() and path.samefile(config_path)): + raise NoticeError('不能把配置文件当作消息正文。') + text = path.read_text(encoding='utf-8') + else: + text = ' '.join(args.summary) if args.command == 'progress' else args.text + if not text.strip() or len(text.encode('utf-8')) > 12000: + raise NoticeError('请提供不超过 12000 字节的简短中文摘要。') + title = EVENTS[args.event] if args.command == 'progress' else args.title.strip() + if not title or len(title) > 200 or '\n' in title or '\r' in title: + raise NoticeError('请使用不超过 200 字的单行标题。') + project = config.get('PROJECT') or root.name + timestamp = dt.datetime.now().astimezone().strftime('%Y-%m-%d %H:%M %z') + return f'## {project}｜{title}\n\n项目进度自动汇报 · {timestamp}\n\n{text}' + + +def main(argv=None, root=ROOT, run=subprocess.run, sleep=time.sleep): + try: + encoded = os.environ.pop(PREFIX + 'ARGV_B64', '') + if argv is None and encoded: + argv = json.loads(base64.b64decode(encoded, validate=True).decode('utf-8')) + if not isinstance(argv, list) or not all(isinstance(v, str) for v in argv): + raise NoticeError('PowerShell 参数无效。') + args = parse_args(argv) + if args.command == 'init': + initialize(args, root, run, sleep) + return 0 config = read_config(root) - preview = args.dry_run or config.get("DRY_RUN") == "1" - if not preview and config.get("ENABLED") != "1": - emit("skipped", "钉钉通知未配置或未启用，已跳过。") + if args.command != 'check' and args.dry_run: + emit('preview', '仅离线预览，账号、群与 @ 效果未验证。', text=render(args, config, root), mentions=args.at) 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)) + if args.command != 'check' and config.get('ENABLED') != '1': + emit('skipped', '通知尚未初始化或未启用；本次未发送，主任务继续。') + return 0 + if not config.get('PROFILE') or not config.get('TARGET'): + raise NoticeError('请由 Hub 先确认个人账号和项目群并执行 init。') + text = render(args, config, root) if args.command != 'check' else None + cli = dws_path() + profile = config['PROFILE'] + if args.command == 'check': + check_login(cli, profile, run, sleep) + people = mentions(args.at, cli, profile, run, sleep) + if args.command == 'check': + emit('checked', '登录及所请求的姓名查询通过；未发送，未验证实际提醒效果。', names=list(people.values())) 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} + text += '\n\n' + ' '.join(f'<@{person}>' for person in people) + command = [cli, 'chat', '+messages-send', '--as', 'user', '--group', config['TARGET'], + '--markdown', text, '--profile', profile, '--yes', '--format', 'json'] 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)) + command += ['--at-open-dingtalk-ids', ','.join(people)] + data = invoke(command, run, sleep, sending=True) + emit('accepted', '个人 DWS 已接受发送请求，不代表已投递、已读或已触发接收端提醒。', + openTaskId=accepted_task(data), mentions=list(people.values())) return 0 except NoticeError as error: - emit("error", str(error)) + emit('error', str(error)) return 1 - except (OSError, UnicodeError, subprocess.SubprocessError, ValueError): - emit("error", "通知配置、正文文件或本机调用异常；请部署负责人检查，本次停止且不影响主任务。") + except (OSError, ValueError, UnicodeError, TypeError, subprocess.SubprocessError): + emit('error', '配置、正文或调用结果异常；本次停止，不自动重发，主任务继续。') return 1 -if __name__ == "__main__": +if __name__ == '__main__': for stream in (sys.stdout, sys.stderr): - if hasattr(stream, "reconfigure"): - stream.reconfigure(encoding="utf-8") + if hasattr(stream, 'reconfigure'): + stream.reconfigure(encoding='utf-8') sys.exit(main()) diff --git a/scripts/test_dingtalk_notify.py b/scripts/test_dingtalk_notify.py index bb55852..83f2b3f 100644 --- a/scripts/test_dingtalk_notify.py +++ b/scripts/test_dingtalk_notify.py @@ -1,4 +1,5 @@ -"""Offline tests: fake configs, fake credentials, fake DWS. Never use live config.""" +"""Offline regression tests. All DWS calls use isolated fake responses.""" +import base64 import contextlib import io import json @@ -9,314 +10,305 @@ import subprocess import sys import tempfile import unittest -from unittest import mock +from unittest.mock import patch -import dingtalk_notify as notify +import dingtalk_notify as n -DNS = 'lookup api.dingtalk.com on 127.0.0.1:53: i/o timeout' -OK = '{"processQueryKey":"test-receipt"}' +PROFILE = 'fake-corp:fake-user' +GROUP = 'fake-group' +PERSON = 'fake-person' +OK = {'ok': True, 'identity': 'user', 'tool': 'send_personal_message', + 'result': {'success': True, 'errorCode': None, 'result': {'openTaskId': 'fake-task'}}} -def response(code=0, stdout=OK, stderr=''): - return subprocess.CompletedProcess(['fake-dws'], code, stdout, stderr) +def reply(data=None, code=0, error=''): + return subprocess.CompletedProcess([], code, json.dumps(data or {}, ensure_ascii=False), error) class NotifyTests(unittest.TestCase): def setUp(self): - self.temp = tempfile.TemporaryDirectory(prefix='通知测试 with spaces ') + self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) - (self.root / '.dingtalk').mkdir() - self.config = self.root / '.dingtalk/config.env' - self.configure() - self.environment = mock.patch.dict(os.environ, {'DING_SEC': 'ignored-env-secret', - 'DWS_CLIENT_SECRET': 'ignored-dws-secret'}, clear=True) - self.environment.start() - self.addCleanup(self.environment.stop) - self.locator = mock.patch.object(notify, 'dws_path', return_value='fake-dws') - self.locator.start() - self.addCleanup(self.locator.stop) + self.path = self.root / '.dingtalk' / 'config.env' + self.path.parent.mkdir() + self.config() + self.calls = [] + self.sleeps = [] + self.addCleanup(patch.stopall) + patch.object(n, 'dws_path', return_value='fake-dws').start() - def configure(self, **updates): - values = dict(ENABLED='1', DRY_RUN='0', MODE='app-bot', CLIENT_ID='fake-client', - TARGET='fake-group', ROBOT_CODE='fake-robot', PROJECT='离线测试', - PROFILE='fake-corp:fake-user', CLIENT_SECRET='fake-secret') - values.update(updates) - self.config.write_text('\n'.join(notify.PREFIX + k + '=' + v for k, v in values.items()), - encoding='utf-8') - self.config.chmod(0o600) + def config(self, **changes): + config = {'ENABLED': '1', 'PROFILE': PROFILE, 'TARGET': GROUP, 'PROJECT': '测试项目'} + config.update(changes) + self.path.write_text(''.join(n.PREFIX + k + '=' + v + '\n' for k, v in config.items()), encoding='utf-8') - def call(self, args, responses=None): - run = mock.Mock(side_effect=responses or [response()]) - sleep = mock.Mock() - output, errors = io.StringIO(), io.StringIO() - with contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors): - code = notify.main(args, root=self.root, run=run, sleep=sleep) - return code, json.loads(output.getvalue()), run, sleep, errors.getvalue() + def fake(self, command, **kwargs): + self.calls.append((command, kwargs)) + if command[1:3] == ['profile', 'list']: + return reply({'profiles': [{'profile': PROFILE, 'userName': '测试人'}]}) + if command[1:3] == ['auth', 'status']: + return reply({'authenticated': True, 'token_valid': True}) + if command[1:3] == ['aisearch', 'person']: + return reply({'result': [{'openDingTalkId': PERSON, 'meta': {'name': command[command.index('--query') + 1]}}]}) + self.assertEqual(command[1:3], ['chat', '+messages-send']) + return reply(OK) - def entries(self): - return [['progress', 'milestone', '**完成**\n\n- 内容'], - ['message', '--title', '负责人提醒', '--text', '**请确认**\n\n- 内容']] + def execute(self, args, run=None): + out = io.StringIO() + with contextlib.redirect_stdout(out): + code = n.main(args, root=self.root, run=run or self.fake, sleep=self.sleeps.append) + return code, json.loads(out.getvalue()) - def test_events(self): - for event, label in notify.EVENTS.items(): - with self.subTest(event=event): - code, data, run, _, _ = self.call(['progress', event, '中文摘要']) - self.assertEqual((code, data['status']), (0, 'accepted')) - payload = json.loads(run.call_args.kwargs['input']) - self.assertIn(label, json.loads(payload['msgParam'])['title']) - - def test_payload_and_credentials_not_on_argv(self): - text = '**结果**\n\n- "中文" 😀 `代码` $变量 & <内容> \\路径' - code, data, run, _, _ = self.call(['message', '--title', '摘要', '--text', text]) - self.assertEqual(code, 0) - command = run.call_args.args[0] - kwargs = run.call_args.kwargs - payload = json.loads(kwargs['input']) - self.assertEqual(json.loads(payload['msgParam'])['text'], text) - self.assertEqual(payload['openConversationId'], 'fake-group') - self.assertEqual(payload['msgKey'], 'sampleMarkdown') - self.assertEqual(command[command.index('--data') + 1], '-') - self.assertNotIn('fake-secret', str(command)) - self.assertNotIn('fake-group', str(command)) - self.assertEqual(kwargs['env']['DWS_CLIENT_SECRET'], 'fake-secret') - self.assertNotIn('DING_SEC', kwargs['env']) - self.assertNotIn('fake-secret', json.dumps(data)) - - def test_dns_retries_both_entries(self): - for args in self.entries(): - with self.subTest(args=args): - code, data, run, sleep, _ = self.call(args, [response(1, '', DNS)] * 3) - self.assertEqual((code, data['status'], run.call_count), (1, 'error', 3)) - self.assertEqual(sleep.call_args_list, [mock.call(10), mock.call(10)]) - - def test_transient_dns_success_stops(self): - for args in self.entries(): - code, data, run, sleep, _ = self.call(args, [response(1, '', DNS), response()]) - self.assertEqual((code, data['status'], run.call_count, sleep.call_count), (0, 'accepted', 2, 1)) - - def test_ambiguous_and_auth_errors_never_replayed_or_echoed(self): - for args in self.entries(): - for error in ['i/o timeout', 'EOF', 'HTTP 503', 'permission denied', 'fake-secret App Token']: - with self.subTest(error=error): - code, data, run, sleep, stderr = self.call(args, [response(1, '', error)]) - self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0)) - self.assertNotIn(error, json.dumps(data, ensure_ascii=False) + stderr) - - def test_receipt_wins_even_nonzero(self): - for args in self.entries(): - code, data, run, sleep, _ = self.call(args, [response(1, OK, DNS)]) - self.assertEqual((code, data['status'], run.call_count, sleep.call_count), (0, 'accepted', 1, 0)) - - def test_bad_receipts_do_not_replay(self): - for raw in ['{}', '{"processQueryKey":" "}', '{"processQueryKey":123}', 'not json', '[]']: - code, _, run, sleep, _ = self.call(self.entries()[0], [response(0, raw)]) - self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0)) - - def test_timeout_no_replay(self): - code, data, run, sleep, _ = self.call(self.entries()[0], [subprocess.TimeoutExpired('fake', 60)]) - self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0)) - self.assertIn('结果未知', data['message']) - self.assertEqual(run.call_args.kwargs['timeout'], 60) - - def test_startup_failure_no_replay(self): - code, _, run, sleep, _ = self.call(self.entries()[0], [OSError('private machine detail')]) - self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0)) - - def test_missing_credentials_no_commands(self): - self.configure(CLIENT_SECRET='') - code, data, run, _, _ = self.call(self.entries()[0]) - self.assertEqual((code, run.call_count), (1, 0)) - self.assertIn('凭据', data['message']) - - def test_skipped_no_tools(self): - self.configure(ENABLED='0') - for args in self.entries(): - code, data, run, _, _ = self.call(args) - self.assertEqual((code, data['status'], run.call_count), (0, 'skipped', 0)) - self.config.unlink() - self.assertEqual(self.call(self.entries()[0])[1]['status'], 'skipped') - - def test_preview_never_uses_or_prints_secret_or_resolves_people(self): - for configured in [False, True]: - self.configure(DRY_RUN='1' if configured else '0') - for args in self.entries(): - if args[0] == 'message': - args += ['--at', '完整姓名'] - if not configured: - args += ['--dry-run'] - with mock.patch.object(notify, 'secret_for', side_effect=AssertionError('secret accessed')): - code, data, run, _, _ = self.call(args) - self.assertEqual((code, data['status'], run.call_count), (0, 'preview', 0)) - self.assertNotIn('fake-group', str(data)) - self.assertNotIn('fake-secret', str(data)) - - def test_preview_without_secret(self): - self.configure(CLIENT_SECRET='') - code, data, run, _, _ = self.call(self.entries()[0] + ['--dry-run']) - self.assertEqual((code, data['status'], run.call_count), (0, 'preview', 0)) - - def test_config_cannot_be_used_as_body_even_in_preview(self): - for args in [['progress', 'milestone'], ['message', '--title', '摘要']]: - for flags in [[], ['--dry-run']]: - code, data, run, _, _ = self.call(args + ['--file', str(self.config)] + flags) - self.assertEqual((code, run.call_count), (1, 0)) - self.assertNotIn('fake-secret', str(data)) - - def test_preview_without_config(self): - self.config.unlink() - self.assertEqual(self.call(self.entries()[0] + ['--dry-run'])[1]['status'], 'preview') - - def test_utf8_bom_crlf_literal_config(self): - raw = self.config.read_text(encoding='utf-8') + '\nDINGTALK_PROGRESS_TITLE_PREFIX=$(not-executed)' - self.config.write_bytes(('\ufeff' + raw.replace('\n', '\r\n')).encode('utf-8')) - self.assertEqual(notify.read_config(self.root)['TITLE_PREFIX'], '$(not-executed)') - - def test_invalid_config_safe_error(self): - for raw in ['DING_SEC=fake-secret', 'invalid fake-secret', - 'DINGTALK_PROGRESS_ENABLED=1\nDINGTALK_PROGRESS_ENABLED=0', - 'DINGTALK_PROGRESS_ENABLED=yes']: - self.config.write_text(raw, encoding='utf-8') - code, data, run, _, _ = self.call(self.entries()[0]) - self.assertEqual((code, run.call_count), (1, 0)) - self.assertNotIn('fake-secret', str(data)) - - def test_body_files_both_entries(self): - body = self.root / '中文 summary.md' - body.write_bytes('\ufeff**结果**\r\n\r\n- 已确认 😀'.encode('utf-8')) - for args in [['progress', 'milestone'], ['message', '--title', '摘要']]: - code, _, run, _, _ = self.call(args + ['--file', str(body)]) + def test_each_event_personal_markdown(self): + for event, label in n.EVENTS.items(): + code, result = self.execute(['progress', event, '**进展**\n\n- 已归档']) self.assertEqual(code, 0) - text = json.loads(json.loads(run.call_args.kwargs['input'])['msgParam'])['text'] - self.assertIn('**结果**\n\n- 已确认 😀', text) + self.assertEqual(result['status'], 'accepted') + cmd = self.calls[-1][0] + self.assertEqual(cmd[cmd.index('--as') + 1], 'user') + self.assertEqual(cmd[cmd.index('--group') + 1], GROUP) + self.assertEqual(cmd[cmd.index('--profile') + 1], PROFILE) + body = cmd[cmd.index('--markdown') + 1] + self.assertIn(label, body) + self.assertIn('项目进度自动汇报', body) + self.assertIn('**进展**\n\n- 已归档', body) + self.assertNotIn('--ai-tag=false', cmd) + self.assertNotIn('--robot-code', cmd) - def test_empty_or_large_body_fails_before_tools(self): - for text in ['', ' ', '字' * 5000]: - code, _, run, _, _ = self.call(['message', '--title', '摘要', '--text', text]) - self.assertEqual((code, run.call_count), (1, 0)) - - def test_exact_name_and_dedup_same_profile(self): - found = json.dumps({'result': [{'meta': {'name': '完整姓名'}, 'userId': 'fake-person'}]}) - args = self.entries()[1] + ['--at', '完整姓名，完整姓名', '--at-id', '完整姓名=fake-person'] - code, _, run, _, _ = self.call(args, [response(0, found), response()]) - self.assertEqual((code, run.call_count), (0, 2)) - for call in run.call_args_list: - command = call.args[0] - self.assertEqual(command[command.index('--profile') + 1], 'fake-corp:fake-user') - self.assertNotIn('DWS_CLIENT_SECRET', run.call_args_list[0].kwargs['env']) - message = json.loads(json.loads(run.call_args.kwargs['input'])['msgParam']) - self.assertEqual(message['atUserIds'], ['fake-person']) - self.assertTrue(message['text'].endswith('@fake-person')) - - def test_names_zero_multiple_malformed_stop_before_send(self): - for rows in [[], [{'title': '完整姓名', 'userId': 'a'}, {'title': '完整姓名', 'userId': 'b'}], - [{'title': '别的姓名', 'userId': 'a'}], {}, [None]]: - code, _, run, sleep, _ = self.call(self.entries()[1] + ['--at', '完整姓名'], - [response(0, json.dumps({'result': rows}))]) - self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0)) - - def test_name_search_requires_explicit_profile(self): - self.configure(PROFILE='') - code, _, run, _, _ = self.call(self.entries()[1] + ['--at', '完整姓名']) - self.assertEqual((code, run.call_count), (1, 0)) - - def test_known_mapping_and_group_compatibility(self): - code, _, run, _, _ = self.call(self.entries()[1] + ['--at-id', '姓名=fake-id', '--group', 'fake-other']) + def test_message_mentions_exact_name_fixed_profile(self): + code, data = self.execute(['message', '--title', '提醒', '--text', '请补充资料', '--at', '测试人,测试人']) self.assertEqual(code, 0) - self.assertEqual(json.loads(run.call_args.kwargs['input'])['openConversationId'], 'fake-other') + self.assertEqual(len(self.calls), 2) + for cmd, _ in self.calls: + self.assertEqual(cmd[cmd.index('--profile') + 1], PROFILE) + cmd = self.calls[-1][0] + self.assertEqual(cmd[cmd.index('--at-open-dingtalk-ids') + 1], PERSON) + self.assertIn('<@fake-person>', cmd[cmd.index('--markdown') + 1]) + self.assertEqual(data['mentions'], ['测试人']) + self.assertEqual(data['openTaskId'], 'fake-task') - def test_config_secret_is_only_source_on_all_platforms(self): - for platform in ['darwin', 'win32', 'linux']: - with self.subTest(platform=platform), mock.patch.object(sys, 'platform', platform): - self.assertEqual(notify.secret_for({'CLIENT_SECRET': 'fake-secret'}), 'fake-secret') - with self.assertRaises(notify.NoticeError): - notify.secret_for({}) - - def test_secret_special_characters_are_literal(self): - secret = 'fake=secret#literal$not-expanded' - self.configure(CLIENT_SECRET=secret) - code, data, run, _, _ = self.call(self.entries()[0]) + def test_progress_can_include_mention_once(self): + code, _ = self.execute(['progress', 'blocked', '缺输入', '--at', '测试人']) self.assertEqual(code, 0) - self.assertEqual(run.call_args.kwargs['env']['DWS_CLIENT_SECRET'], secret) - self.assertNotIn(secret, str(data)) - self.assertNotIn(secret, str(run.call_args.args)) - self.assertNotIn(secret, run.call_args.kwargs['input']) + self.assertEqual(sum(c[0][1:3] == ['chat', '+messages-send'] for c in self.calls), 1) + def test_lookup_ambiguity_no_match_and_malformed_never_send(self): + for rows in [[], [{'openDingTalkId': 'a', 'meta': {'name': '甲'}}, + {'openDingTalkId': 'b', 'meta': {'name': '甲'}}], + [{'userId': 'wrong-id-kind', 'meta': {'name': '甲'}}], None]: + self.calls.clear() + def lookup(command, **kwargs): + self.calls.append(command) + return reply({'result': rows}) + code, data = self.execute(['message', '--title', '提醒', '--text', '正文', '--at', '甲'], lookup) + self.assertEqual(code, 1) + self.assertEqual(data['status'], 'error') + self.assertEqual(len(self.calls), 1) -class LauncherTests(unittest.TestCase): - def test_real_subprocess_stdin_without_live_dws(self): - with tempfile.TemporaryDirectory(prefix='通知 fake transport ') as folder: - root = Path(folder) - (root / '.dingtalk').mkdir() - config = root / '.dingtalk/config.env' - config.write_text('DINGTALK_PROGRESS_ENABLED=1\nDINGTALK_PROGRESS_DRY_RUN=0\n' - 'DINGTALK_PROGRESS_CLIENT_ID=fake-client\n' - 'DINGTALK_PROGRESS_CLIENT_SECRET=fake-secret\n' - 'DINGTALK_PROGRESS_TARGET=fake-group\n' - 'DINGTALK_PROGRESS_ROBOT_CODE=fake-robot\n', encoding='utf-8') - config.chmod(0o600) - fake = root / 'fake dws.py' - fake.write_text( - 'import json, os, sys\n' - 'assert sys.argv[1:] == ["api", "POST", "/v1.0/robot/groupMessages/send", ' - '"--data", "-", "--yes", "--format", "json"]\n' - 'payload = json.loads(sys.stdin.buffer.read().decode("utf-8"))\n' - 'assert payload["openConversationId"] == "fake-group"\n' - 'assert json.loads(payload["msgParam"])["text"] == "中文\\n\\n- \\\"引号\\\" 😀"\n' - 'assert os.environ["DWS_CLIENT_SECRET"] == "fake-secret"\n' - 'assert "DING_SEC" not in os.environ\n' - 'print(json.dumps({"processQueryKey": "fake-transport-receipt"}))\n', encoding='utf-8') + def test_check_is_read_only(self): + code, data = self.execute(['check', '--at', '测试人']) + self.assertEqual(code, 0) + self.assertEqual(data['status'], 'checked') + self.assertEqual(len(self.calls), 2) + self.assertNotIn('fake-person', json.dumps(data)) - def transport(command, **kwargs): - return subprocess.run([sys.executable, str(fake), *command[1:]], **kwargs) + def test_check_bad_login_stops(self): + code, _ = self.execute(['check'], lambda *a, **kw: reply({'authenticated': False, 'token_valid': False})) + self.assertEqual(code, 1) - out = io.StringIO() - with mock.patch.object(notify, 'dws_path', return_value='fake-dws'), \ - mock.patch.dict(os.environ, {'DING_SEC': 'ignored-env-secret'}), \ - contextlib.redirect_stdout(out): - code = notify.main(['message', '--title', '摘要', '--text', '中文\n\n- "引号" 😀'], - root=root, run=transport) - self.assertEqual(code, 0, out.getvalue()) - self.assertEqual(json.loads(out.getvalue())['status'], 'accepted') + def test_disabled_and_missing_config_no_tools(self): + self.config(ENABLED='0') + self.assertEqual(self.execute(['progress', 'start', '开始'])[1]['status'], 'skipped') + self.path.unlink() + self.assertEqual(self.execute(['progress', 'start', '开始'])[1]['status'], 'skipped') + self.assertFalse(self.calls) - def test_real_launchers_preview_in_isolated_copy(self): - """Exercise OS processes/quoting, but copy no real config and never run dws.""" - with tempfile.TemporaryDirectory(prefix='通知 launch with spaces ') as folder: - scripts = Path(folder) / 'scripts' - scripts.mkdir() - for name in ['dingtalk_notify.py', 'dingtalk-progress', 'dingtalk-bot-message', - 'dingtalk-progress.ps1', 'dingtalk-bot-message.ps1']: - shutil.copy2(Path(__file__).parent / name, scripts / name) - text = '**结果**\n\n- "中文" 😀 `代码` $变量 & 内容' - body = Path(folder) / '中文正文.md' - body.write_text(text, encoding='utf-8') - env = {k: v for k, v in os.environ.items() - if not k.startswith(('DING', 'DWS_'))} - # Run from outside the script project: --file is cwd-relative. - commands = [[sys.executable, str(scripts / 'dingtalk_notify.py'), 'progress', 'milestone', - '--file', body.name, '--dry-run']] - pwsh = shutil.which('pwsh') - if os.name == 'nt': - self.assertIsNotNone(pwsh, 'Windows launcher test needs PowerShell 7.3+') - if pwsh: - commands += [[pwsh, '-NoProfile', '-File', str(scripts / 'dingtalk-progress.ps1'), - 'milestone', text, '--dry-run'], - [pwsh, '-NoProfile', '-File', str(scripts / 'dingtalk-bot-message.ps1'), - '--title', '中文摘要', '--text', text, '--dry-run']] - bash = Path('C:/Program Files/Git/bin/bash.exe') if os.name == 'nt' else shutil.which('bash') - if bash and Path(bash).exists(): - for name, args in [('dingtalk-progress', ['milestone']), - ('dingtalk-bot-message', ['--title', '中文摘要'])]: - commands.append([str(bash), str(scripts / name), *args, - '--file', body.name, '--dry-run']) - for command in commands: - with self.subTest(entry=command[:3]): - result = subprocess.run(command, cwd=folder, env=env, capture_output=True, - text=True, encoding='utf-8', timeout=30) - self.assertEqual(result.returncode, 0, result.stderr) - data = json.loads(result.stdout) - self.assertEqual(data['status'], 'preview') - self.assertIn(text, data['text']) + def test_preview_without_config_or_dws(self): + self.path.unlink() + code, result = self.execute(['progress', 'blocked', '说明', '--at', '甲', '--dry-run']) + self.assertEqual(code, 0) + self.assertEqual(result['status'], 'preview') + self.assertFalse(self.calls) + + def test_legacy_and_invalid_config_fail_without_echo(self): + for suffix in ['CLIENT_SECRET=fake-secret', 'MODE=app-bot', 'ENABLED=1', 'UNKNOWN=x']: + self.config() + with self.path.open('a', encoding='utf-8') as f: + f.write(n.PREFIX + suffix + '\n') + code, data = self.execute(['progress', 'start', '开始']) + self.assertEqual(code, 1) + self.assertNotIn('fake-secret', json.dumps(data)) + self.assertFalse(self.calls) + + def test_config_bom_crlf_and_literal_values(self): + raw = self.path.read_text(encoding='utf-8').replace('测试项目', '中文 "引号" $()') + self.path.write_bytes(b'\xef\xbb\xbf' + raw.replace('\n', '\r\n').encode()) + code, data = self.execute(['progress', 'start', '开始', '--dry-run']) + self.assertEqual(code, 0) + self.assertIn('中文 "引号" $()', data['text']) + + def test_files_and_invalid_bodies(self): + body = self.root / '正文.md' + body.write_text('**进展**\n\n- 中文 "引号" $()', encoding='utf-8') + for args in [['progress', 'milestone'], ['message', '--title', '说明']]: + self.assertEqual(self.execute(args + ['--file', str(body)])[0], 0) + before = len(self.calls) + for args in [['progress', 'start', ''], ['progress', 'start', 'x' * 12001], + ['progress', 'start', '正文', '--file', str(body)], + ['message', '--title', '坏\n标题', '--text', '内容'], + ['progress', 'start', '--file', str(self.path), '--dry-run']]: + self.assertEqual(self.execute(args)[0], 1) + self.assertEqual(len(self.calls), before) + + def test_dns_three_attempts_only(self): + calls = [] + def failed(*a, **kw): + calls.append(a) + return reply(code=1, error='lookup api.dingtalk.com: no such host') + code, _ = self.execute(['progress', 'start', '开始'], failed) + self.assertEqual(code, 1) + self.assertEqual(len(calls), 3) + self.assertEqual(self.sleeps, [10, 10]) + + def test_dns_then_success(self): + responses = iter([reply(code=1, error='lookup api.dingtalk.com: no such host'), reply(OK)]) + self.assertEqual(self.execute(['progress', 'start', '开始'], lambda *a, **kw: next(responses))[0], 0) + self.assertEqual(self.sleeps, [10]) + + def test_failure_and_unknown_never_retry(self): + cases = [reply(code=1, error='fake-secret connection reset'), + reply(code=1, error=json.dumps({'error': {'category': 'auth', 'message': 'fake-secret'}})), + reply(code=1, error=json.dumps({'error': {'retryable': False, 'message': 'lookup api.dingtalk.com: no such host'}})), + reply({'success': True}), reply({'ok': False}), reply(OK | {'partial': True})] + for result in cases: + count = [] + def failed(*a, **kw): + count.append(1) + return result + code, data = self.execute(['progress', 'start', '开始'], failed) + self.assertEqual(code, 1) + self.assertEqual(len(count), 1) + self.assertNotIn('fake-secret', json.dumps(data)) + self.assertFalse(self.sleeps) + + def test_receipt_wins_nonzero_exit(self): + self.assertEqual(self.execute(['progress', 'start', '开始'], lambda *a, **kw: reply(OK, code=1))[0], 0) + + def test_wrong_receipts_rejected(self): + for bad in [OK | {'identity': 'bot'}, OK | {'tool': 'other'}, OK | {'failures': [1]}, + OK | {'result': {'success': False, 'result': {'openTaskId': 'x'}}}, + OK | {'result': {'success': True, 'result': {'openTaskId': ''}}}]: + self.assertIsNone(n.accepted_task(bad)) + + def test_timeout_and_launch_error_no_retry(self): + for error in [subprocess.TimeoutExpired('fake', 60), OSError('fake-secret')]: + calls = [] + def fail(*a, **kw): + calls.append(1) + raise error + code, _ = self.execute(['progress', 'start', '开始'], fail) + self.assertEqual(code, 1) + self.assertEqual(len(calls), 1) + + def test_no_app_credentials_in_child_environment(self): + with patch.dict(os.environ, {'DWS_CLIENT_SECRET': 'fake-secret', 'DWS_CLIENT_ID': 'fake-app', 'DING_SEC': 'fake-key'}): + self.execute(['progress', 'start', '开始']) + env = self.calls[-1][1]['env'] + self.assertNotIn('DWS_CLIENT_SECRET', env) + self.assertNotIn('DWS_CLIENT_ID', env) + self.assertNotIn('DING_SEC', env) + self.assertEqual(self.calls[-1][1]['timeout'], 60) + + def init_args(self): + return ['init', '--profile', PROFILE, '--group', GROUP, '--project', '项目甲'] + + def test_init_requires_confirmation_and_explicit_replace(self): + original = self.path.read_bytes() + self.assertEqual(self.execute(self.init_args())[0], 1) + self.assertEqual(self.execute(self.init_args() + ['--yes'])[0], 1) + self.assertEqual(self.path.read_bytes(), original) + self.assertFalse(self.calls) + + def test_init_creates_minimal_config_no_send(self): + self.path.unlink() + code, data = self.execute(self.init_args() + ['--yes']) + self.assertEqual(code, 0) + self.assertEqual(data['status'], 'initialized') + self.assertEqual(set(n.read_config(self.root)), n.KEYS) + self.assertNotIn(b'\r', self.path.read_bytes()) + self.assertEqual(len(self.calls), 2) + if os.name != 'nt': + self.assertEqual(self.path.stat().st_mode & 0o777, 0o600) + + def test_init_replaces_legacy_without_retaining_secret(self): + self.path.write_text('DINGTALK_PROGRESS_CLIENT_SECRET=fake-secret\n', encoding='utf-8') + code, data = self.execute(self.init_args() + ['--yes', '--replace']) + self.assertEqual(code, 0) + self.assertNotIn('fake-secret', self.path.read_text()) + self.assertNotIn('fake-secret', json.dumps(data)) + self.assertEqual(len(list(self.path.parent.iterdir())), 1) + + def test_init_bad_profile_and_line_injection_leave_config(self): + before = self.path.read_bytes() + for args in [self.init_args() + ['--profile', 'other:user'], + self.init_args() + ['--project', 'x\nBAD=1']]: + self.assertEqual(self.execute(args + ['--yes', '--replace'])[0], 1) + self.assertEqual(self.path.read_bytes(), before) + + def test_init_failed_auth_leaves_old_config(self): + before = self.path.read_bytes() + def fake(command, **kw): + if command[1] == 'auth': + return reply({'authenticated': False}) + return self.fake(command, **kw) + self.assertEqual(self.execute(self.init_args() + ['--yes', '--replace'], fake)[0], 1) + self.assertEqual(self.path.read_bytes(), before) + + def test_encoded_powershell_arguments(self): + args = ['message', '--title', '中文 "标题"', '--text', '**进展**\n\n- 引号 "$()"', '--dry-run'] + with patch.dict(os.environ, {n.PREFIX + 'ARGV_B64': base64.b64encode(json.dumps(args).encode()).decode()}): + code, data = self.execute(None) + self.assertEqual(code, 0) + self.assertIn('中文 "标题"', data['text']) + self.assertIn('**进展**\n\n- 引号 "$()"', data['text']) + + def test_bad_encoded_arguments(self): + for value in ['!!!', base64.b64encode(b'{}').decode()]: + with patch.dict(os.environ, {n.PREFIX + 'ARGV_B64': value}): + self.assertEqual(self.execute(None)[0], 1) + self.assertFalse(self.calls) + + def test_hardlinked_config_cannot_be_message_body(self): + other = self.root / 'not-a-summary.md' + os.link(self.path, other) + self.assertEqual(self.execute(['progress', 'start', '--file', str(other), '--dry-run'])[0], 1) + + def test_real_powershell_launcher_in_isolated_project(self): + pwsh = shutil.which('pwsh') + if not pwsh or 'windowsapps' in sys.executable.lower(): + self.skipTest('需要 PowerShell 和独立 Python，不能用商店版作启动器验收') + folder = self.root / 'scripts' + folder.mkdir() + source = Path(n.__file__).parent + for name in ['dingtalk_notify.py', 'dingtalk-python.ps1', 'dingtalk-progress.ps1', 'dingtalk-message.ps1']: + shutil.copyfile(source / name, folder / name) + env = n.clean_env() + env['DINGTALK_PYTHON'] = sys.executable + text = '**中文**\n\n- "引号" $() `原样`' + for script, args in [('dingtalk-progress.ps1', ['milestone', text]), + ('dingtalk-message.ps1', ['--title', '中文 "标题"', '--text', text])]: + result = subprocess.run([pwsh, '-NoProfile', '-File', str(folder / script), *args, '--dry-run'], + cwd=self.root, env=env, capture_output=True, encoding='utf-8', timeout=30) + self.assertEqual(result.returncode, 0, result.stderr) + data = json.loads(result.stdout) + self.assertEqual(data['status'], 'preview') + self.assertIn(text, data['text']) + result = subprocess.run([pwsh, '-NoProfile', '-File', str(folder / 'dingtalk-python.ps1'), 'init', + '--profile', PROFILE, '--group', GROUP, '--project', '甲'], + cwd=self.root, env=env, capture_output=True, encoding='utf-8', timeout=30) + self.assertEqual(result.returncode, 1) + self.assertIn('先向负责人确认', result.stdout) if __name__ == '__main__':