--- name: cms-cwork-workflow description: 管理工作协同中的员工查询、汇报处理、待办闭环和任务协作流程。触发词：cwork/CWork/工作协同/发送汇报/发汇报/汇报/申请/周报/待办/任务/催办/搜索员工/查收件箱。 skillcode: cms-cwork-workflow github: https://github.com/xgjk/cwork-skills/tree/main/cwork-skills/cms-cwork-workflow dependencies: - cms-auth-skills version: 1.0.5 tools_provided: - name: cwork_client category: exec risk_level: medium permission: exec description: CWork API共享客户端，封装HTTP请求、认证和所有业务API方法 status: active - name: cwork-search-emp category: exec risk_level: low permission: read description: 搜索员工信息（支持模糊查询） status: active - name: cwork-send-report category: exec risk_level: medium permission: write description: 发送工作协同汇报（支持附件） status: active - name: cwork-query-report category: exec risk_level: low permission: read description: 查询汇报（收件箱/发件箱/详情/历史） status: active - name: cwork-create-task category: exec risk_level: medium permission: write description: 创建工作计划/任务 status: active - name: cwork-review-report category: exec risk_level: medium permission: write description: 审阅汇报（回复/标记已读） status: active - name: cwork-query-tasks category: exec risk_level: low permission: read description: 查询任务（我的/创建的/团队/详情） status: active - name: cwork-nudge-report category: exec risk_level: medium permission: write description: 催办闭环（识别未闭环/生成催办/发送） status: active - name: cwork-todo category: exec risk_level: medium permission: write description: 待办管理（查询/完成决策/建议/反馈） status: active - name: cwork-templates category: exec risk_level: low permission: read description: 查询汇报模板列表 status: active - name: cwork-report-issue category: exec risk_level: low permission: read description: 自动上报问题到 GitHub Issues（需环境变量 GITHUB_TOKEN） status: active --- # cms-cwork-workflow ## 概述本 Skill 将 CWork（工作协同平台）的完整 API 能力封装为 **9 个意图级编排脚本**，每个脚本独立可执行，Agent 通过 `exec python3 scripts/.py` 调用，JSON 输出到 stdout、错误到 stderr。 **设计原则**： - **Agent-First**：脚本负责 API 编排，Agent 负责 LLM 推理和用户交互 - **幂等安全**：所有写操作支持 `--dry-run` / `--preview-only` - **零外部依赖**：纯 Python 3.9+，仅需标准库 - **强制封装**：所有 API 调用必须通过脚本，**禁止直接 HTTP/curl 调用**（脚本已内置 URL 编码、参数验证、错误转换、重试机制） ## 快速开始 ### 发送汇报（标准 3 步流程） ``` 1. 搜索接收人，确认 empId python3 scripts/cwork-search-emp.py --name "张三" 2. 预览草稿（--preview-only 仅保存草稿，不发送） python3 scripts/cwork-send-report.py \ --title "周报" --content-html "

内容

" \ --receivers "张三" --preview-only 3. 确认发送 python3 scripts/cwork-send-report.py \ --title "周报" --content-html "

内容

" \ --receivers "张三" ``` ### 其他常用命令 ```bash # 查看未读汇报 python3 scripts/cwork-query-report.py --mode unread # 查看待办列表 python3 scripts/cwork-todo.py list --page-size 20 # 查询我的任务 python3 scripts/cwork-query-tasks.py --mode my ``` --- ## 9 个编排命令 ### 0. 搜索员工 — `cwork-search-emp.py` ✨ 新增 **意图**：根据姓名/关键词搜索员工 ID 和详细信息 **使用场景**： 1. ✅ **发送汇报前确认接收人** - 确保姓名和 empId 准确 2. ✅ **处理待办时确认发件人** - 查看发件人部门/职位 3. ✅ **创建任务时确认责任人** - 避免姓名错误（重名/错别字） 4. ✅ **催办时确认责任人信息** - 获取完整的员工信息 ```bash # 基础搜索（模糊匹配） python3 scripts/cwork-search-emp.py --name "张" # 精确搜索 python3 scripts/cwork-search-emp.py --name "成伟" # 详细模式（包含 personId、dingUserId 等） python3 scripts/cwork-search-emp.py --name "刘丽华" --verbose # 更多结果 python3 scripts/cwork-search-emp.py --name "刘" --max-results 10 # 原始 API 响应（调试用） python3 scripts/cwork-search-emp.py --name "张" --output-raw ``` | 参数 | 必填 | 说明 | |------|------|------| | `--name` / `-n` | ✅ | 员工姓名或关键词（支持模糊匹配） | | `--max-results` / `-m` | ❌ | 每个类别最多返回数量（默认 5） | | `--verbose` / `-v` | ❌ | 包含额外信息（personId、dingUserId、corpId） | | `--output-raw` | ❌ | 输出原始 API 响应（调试用） | **输出格式**： ```json { "success": true, "searchKey": "成伟", "inside": [ { "empId": "1514822118611259394", "name": "成伟", "title": "首席架构师", "mainDept": "技术部", "status": "在职" } ], "outside": [ { "empId": "1897870576398327809", "name": "成伟", "title": "", "mainDept": "其他", "status": "在职", "company": "德镁医药" } ], "totalInside": 1, "totalOutside": 1 } ``` **注意事项**： - ✅ **URL 编码已自动处理**（支持中文参数） - ✅ **模糊匹配**：搜索"刘"会返回所有姓刘的员工 - ✅ **内外部区分**：`inside`（玄关健康员工）+ `outside`（外部联系人/其他公司） - ⚠️ **重名问题**：可能返回多个同名员工，需要根据部门/职位区分 - 💡 **推荐用法**：发送汇报前先搜索确认 empId --- ### 1. 发送汇报 — `cwork-send-report.py` **意图**：先**全量**保存/更新草稿（5.23，更新前会拉 5.25 详情合并，避免覆盖丢字段）→ 输出接口返回的**完整**草稿（`draftDetail`）供用户过目 → 仅在用户明确同意后加 `--confirm-send` 调用 **5.27**（`draftBox/submit/{汇报id}`）发出。 **汇报 id 与 `draftId` 字段（避免歧义）** | 概念 | 含义 | 出现位置 | |------|------|----------| | **汇报 id** | 草稿对应的汇报记录主键 | 5.23 返回 `data.id`、5.25 路径与 `draftDetail.id`、5.27 路径 `{id}` | | **草稿箱记录 id** | 草稿箱列表里一行的主键，**仅用于 5.26 删除** | 5.24 列表项的 `id`（勿与汇报 id 混用） | **删除草稿（`cwork_client`）**：`delete_draft` 的参数必须是 **5.24 列表项的 `id`**。若只有汇报 id（与列表里的 `businessId` 相同），须调用 **`delete_draft_by_report_id(汇报id)`**；误把汇报 id 传给 `delete_draft` 时，接口可能仍返回 `true` 但列表中草稿未删（见开放 API 5.26 与 5.24 参数说明）。脚本 stdout 里同时有根字段 **`reportId`** 与 **`draftId`**：二者**不是**两种 id，而是**同一汇报 id 的重复输出**——`draftId` **并非**开放平台文档里的字段名，而是本脚本为衔接历史参数 `--draft-id` 而保留的 JSON 键名，容易让人误以为是「草稿箱 id」。**以 `reportId` / `draftDetail.id` 为准即可**；后续步骤一律传该汇报 id（`--draft-id <汇报id>` 中的值也是它）。 ```bash # 第一步：保存草稿并输出完整预览（默认不会发出） python3 scripts/cwork-send-report.py \ --title "周报标题" \ --content-html "

汇报内容

" \ --receivers "张三,李四" \ --grade "一般" # 第二步：用户确认 draftDetail 全文后，仅发出（无需再传标题正文） python3 scripts/cwork-send-report.py --draft-id "<汇报id>" --confirm-send # 一步保存并发出（仍须显式 --confirm-send，表示已确认预览） python3 scripts/cwork-send-report.py \ --title "周报标题" \ --content-html "

汇报内容

" \ --receivers "张三" \ --confirm-send ``` | 参数 | 必填 | 说明 | |------|------|------| | `--title` / `-t` | 保存草稿时 ✅ | 汇报标题（与 `--draft-id --confirm-send` 单独发出时勿传） | | `--content-html` / `-c` | 保存草稿时 ✅ | 正文 HTML（同上） | | `--receivers` / `-r` | ❌ | 接收人姓名；更新时若省略则沿用草稿详情中的接收人。**若本次传了姓名**且草稿已有 `reportLevelList`（且未使用 `--report-level-json`），脚本会把解析后的 empId **写回**对应节点的 `levelUserList`，与开放 API「接收人以 `reportLevelList` 为准」一致，避免仅 `summary` 显示新人而 `draftDetail` 仍为旧人 | | `--cc` | ❌ | 抄送；更新时若省略则沿用草稿中的抄送 | | `--grade` | ❌ | 优先级：`一般`（默认）/ `紧急` | | `--type-id` | ❌ | 汇报类型 ID（默认 9999） | | `--file-paths` | ❌ | 本地附件；**未传且为更新**时沿用草稿已有附件 | | `--file-names` | ❌ | 附件显示名称 | | `--plan-id` | ❌ | 关联任务 ID | | `--report-level-json` | ❌ | JSON 文件路径，`reportLevelList` 数组，覆盖流程节点 | | `--preview-only` | ❌ | 仅保存+预览；**即使带 `--confirm-send` 也不会发出** | | `--draft-id` | ❌ | **值为汇报 id**（参数名历史沿用）：更新草稿或配合 `--confirm-send` 仅执行 5.27 | | `--confirm-send` | ❌ | **必须**在用户确认完整 `draftDetail` 后再加，才会调用 5.27 | **流程步骤**： 1. **Resolve** — 按姓名搜索员工；本轮回填的姓名参与合并，未填则沿用 5.25 详情中的接收人/抄送 2. **Validate** — 姓名未找到或多匹配时报错终止 3. **Upload** — 若传了 `--file-paths` 则上传并作为附件；否则更新时保留原附件列表 4. **Detail（更新时）** — 若有 `--draft-id`，先 `get_draft_detail` 再与本次参数合并 5. **Draft（5.23）** — 全量 `saveOrUpdate`，返回汇报 id 6. **Preview** — 再次 `get_draft_detail`，stdout 含完整 `draftDetail`（含全文 `contentHtml`）。`summary` 含 `contentPlainText`（全文去标签）、`contentPreview`（极长纯文本时最多约 4000 字截断）、`contentPlainLength`；去标签后不足 50 字时有 `previewWarnings`。`confirmPrompt` 以纯文本展示正文（≤2000 字全文，更长截断），完整 HTML 始终以 `draftDetail.contentHtml` 为准 7. **Submit（5.27）** — 仅当 `--confirm-send` 且非 `--preview-only` 时 `submit`；**不要**再用 5.1 无 id 提交，以免产生重复汇报与孤儿草稿 --- ### 2. 查询汇报 — `cwork-query-report.py` **意图**：收件箱 / 发件箱 / 未读 / 汇报详情 / 节点详情 / **历史上下文检索** ✨ 新增 ```bash # 收件箱（默认） python3 scripts/cwork-query-report.py --mode inbox --page-size 20 # 未读汇报 python3 scripts/cwork-query-report.py --mode unread --page-size 20 # 发件箱 python3 scripts/cwork-query-report.py --mode outbox # 单条汇报详情（含回复链） python3 scripts/cwork-query-report.py --mode detail --report-id # 节点详情（含审批/建议/反馈状态与处理意见）✨ v3.1.0 新增 python3 scripts/cwork-query-report.py --mode node-detail --report-id # 历史上下文检索（审批决策支持）✨ v3.1.1 新增 # 查询发件人历史汇报 python3 scripts/cwork-query-report.py --mode sender-history \ --sender-emp-id \ --days 90 # 关键字搜索汇报（客户端过滤） python3 scripts/cwork-query-report.py --mode keyword-search \ --keyword "公章" \ --days 90 ``` | 参数 | 说明 | |------|------| | `--mode` | `inbox` / `outbox` / `unread` / `detail` / `node-detail` / `sender-history` / `keyword-search` / `pending` / `my-sent` | | `--page-size` | 分页大小（默认 20） | | `--page-index` | 页码（默认 1） | | `--report-id` | 汇报 ID（detail / node-detail 必填） | | `--sender-emp-id` | 发件人员工 ID（sender-history 必填） | | `--keyword` | 搜索关键词（keyword-search 必填） | | `--days` | 回溯天数（sender-history / keyword-search，默认 90） | | `--report-type` | 汇报类型：1-工作交流 / 2-工作指引 / 3-文件签批 / 4-AI汇报 / 5-工作汇报 | | `--status` | 已读状态：0=未读 / 1=已读 | | `--start-date` / `--end-date` | 时间范围（YYYY-MM-DD） | **输出格式**（sender-history）： ```json { "success": true, "data": { "senderEmpId": "1514822194347806721", "totalReports": 15, "recentReports": [ { "id": "2039993163862765570", "main": "互联网公司公章借出申请", "createTime": "2026-04-03 17:11:48", "reportRecordType": 5 } ] } } ``` **输出格式**（keyword-search）： ```json { "success": true, "data": { "keyword": "公章", "total": 5, "reports": [ { "id": "2039993163862765570", "main": "互联网公司公章借出申请", "content": "用于办理工商变更...", "createTime": "2026-04-03 17:11:48", "sendEmpName": "刘丽华" } ] } } ``` **输出格式**（node-detail）： ```json { "success": true, "data": { "id": 汇报ID, "main": "汇报标题", "content": "汇报正文", "writeEmpName": "汇报人", "createTime": "发起时间", "nodeList": [ { "nodeName": "建议人", "type": "建议", "status": "已完成", "level": 1, "userList": [ { "empId": 员工ID, "name": "张三", "status": "已处理", "operate": "建议", "content": "建议增加异常处理", "finishTime": "2026-04-03 11:00:00" } ] } ] } } ``` --- ### 3. 创建任务 — `cwork-create-task.py` **意图**：解析人员姓名 → 创建工作计划/任务 ```bash python3 scripts/cwork-create-task.py \ --task-main "完成XXX功能" \ --content "详细描述" \ --assignee "张三" \ --deadline 2026-05-01 ``` | 参数 | 必填 | 说明 | |------|------|------| | `--task-main` | ✅ | 任务标题 | | `--content` | ✅ | 任务描述 | | `--target` | ❌ | 预期目标（默认 = content） | | `--assignee` | ❌ | 责任人姓名（自动解析 empId） | | `--report-to` | ❌ | 汇报人姓名（不传则自动取 `--assignee` 的值；API 要求必填） | | `--assistant` | ❌ | 协办人姓名（逗号分隔多人） | | `--supervisor` | ❌ | 监督人姓名 | | `--copy` | ❌ | 抄送人姓名（逗号分隔多人） | | `--observer` | ❌ | 观察员姓名（逗号分隔多人） | | `--deadline` | ❌ | 截止时间（YYYY-MM-DD 或 Unix ms，默认 7 天后） | | `--push-now` | ❌ | 是否立即推送（true/false，默认 true）。开放 API 文档未说明 `pushNow=0` 时的额外字段；若服务端报「待办发送时间未设置」等错误，需向接口提供方确认是否另有未文档化参数或是否暂不支持延迟推送 | | `--dry-run` | ❌ | 仅验证+解析，不创建 | **流程步骤**： 1. 解析所有人员姓名 → empId 2. 校验必填项（task-main、content） 3. 汇总所有未匹配姓名 → 报错 4. `--dry-run` 时输出解析结果，不调用创建 API 5. 调用 `createPlan` API 创建任务 --- ### 4. 审阅汇报 — `cwork-review-report.py` **意图**：回复汇报 / 标记已读 / 查询待审汇报 ```bash # 标记已读 python3 scripts/cwork-review-report.py --mode mark-read --report-id # 回复（默认 markdown，可发内部链接；需纯 HTML 段落可加 --content-type html） python3 scripts/cwork-review-report.py --mode reply \ --report-id --reply "回复内容" # 查询待审汇报列表 python3 scripts/cwork-review-report.py --mode pending --page-size 20 ``` | 参数 | 说明 | |------|------| | `--mode` | `reply` / `mark-read` / `pending` | | `--report-id` | 汇报记录 ID（reply / mark-read 必填） | | `--reply` | 回复正文（reply 必填；默认按 markdown 原样提交，支持内部链接语法） | | `--content-type` | `markdown`（默认）或 `html`；`html` 时将正文包成 `

…

` | | `--at` | 回复中 @的人姓名（自动解析 empId） | | `--page-index` | 页码（pending 模式，默认 1） | | `--page-size` | 每页大小（pending 模式，默认 20） | | `--report-type` | 汇报类型筛选 1-5（pending 模式可选） | | `--dry-run` | 仅预览，不调用 API | --- ### 5. 查询任务 — `cwork-query-tasks.py` **意图**：我的任务 / 我创建的 / 团队任务 / 任务详情（含汇报链）/ 识别逾期和未闭环 ```bash # 分配给我的任务 python3 scripts/cwork-query-tasks.py --mode my --status 1 # 我创建的任务 python3 scripts/cwork-query-tasks.py --mode created # 下属任务 python3 scripts/cwork-query-tasks.py --mode manager --subordinate-ids "id1,id2" # 任务详情（含汇报链路） python3 scripts/cwork-query-tasks.py --mode detail --task-id # 识别逾期任务 python3 scripts/cwork-query-tasks.py --mode blocked --days-threshold 7 ``` | 参数 | 说明 | |------|------| | `--mode` | `my` / `created` / `team` / `assigned` / `detail` / `chain` / `blocked` / `unclosed` / `manager` / `nudge` | | `--task-id` | 任务/计划 ID（detail / chain 必填） | | `--subordinate-ids` | 下属 empId 列表（逗号分隔，manager 模式必填） | | `--assignee` | 责任人姓名（my / assigned 模式可选，自动解析 empId） | | `--status` | 任务状态：0=已关闭 / 1=进行中 / 2=未启动 | | `--report-status` | 汇报状态：0=关闭 / 1=待汇报 / 2=已汇报 / 3=逾期 | | `--key-word` | 关键词搜索 | | `--days-threshold` | 逾期天数阈值（blocked 模式，默认 7） | | `--page-index` | 页码（默认 1） | | `--page-size` | 每页大小（默认 20） | | `--dry-run` | 仅预览，不调用 API | --- ### 6. 催办闭环 — `cwork-nudge-report.py` **意图**：列出逾期未闭环任务 / 向责任人发送催办通知 ```bash # 列出逾期未闭环任务（超过阈值天数） python3 scripts/cwork-nudge-report.py --mode list --days-threshold 7 # 向责任人发送催办（通过 empId） python3 scripts/cwork-nudge-report.py --mode nudge \ --emp-id \ --task-main "完成XXX功能" \ --deadline 2026-05-01 \ --content "请尽快处理" # 通过姓名自动解析 empId 并催办 python3 scripts/cwork-nudge-report.py --mode nudge \ --assignee "张三" \ --task-main "完成XXX功能" \ --remind-style normal ``` | 参数 | 说明 | |------|------| | `--mode` | `list`=列出未闭环 / `nudge`=发送催办 | | `--days-threshold` | 逾期天数阈值（list 模式，默认 7） | | `--page-index` | 页码（list 模式，默认 1） | | `--page-size` | 每页大小（list 模式，默认 50） | | `--emp-id` | 催办对象 empId（nudge 必填，与 `--assignee` 二选一） | | `--assignee` | 责任人姓名（nudge 模式，自动解析 empId） | | `--task-main` | 任务名称（nudge 必填） | | `--deadline` | 截止日期（YYYY-MM-DD 或 Unix ms） | | `--content` | 催办内容描述（脚本自动构建 HTML 正文） | | `--target` | 目标描述 | | `--remind-style` | 催办风格：`polite`（默认，含礼貌用语）/ `normal`（简洁） | | `--dry-run` | 仅预览，不调用 API | --- ### 7. 待办管理 — `cwork-todo.py` **意图**：查询待办列表 / 完成待办（支持决策/建议/反馈三种类型） **支持的三种待办类型**： | 类型 | 英文标识 | 必填参数 | 说明 | |------|---------|---------|------| | **决策** | `decide` | `--operate agree/disagree` | 决策人必须明确同意或不同意 | | **建议** | `suggest` | `--content` | 建议人提供意见或建议 | | **反馈** | `feedback` | `--content` | 反馈人回复评论或补充信息 | ```bash # 查询待办列表（5.15 分页结构为 PageInfo，条目见 items，字段含 todoId、reportId、main、todoType 等） python3 scripts/cwork-todo.py list --page-size 20 --status pending # 完成决策待办（必须指定 operate） python3 scripts/cwork-todo.py complete \ --todo-id \ --content "同意该方案，建议增加异常处理" \ --operate agree # 完成建议待办 python3 scripts/cwork-todo.py complete \ --todo-id \ --content "从技术角度看，建议采用微服务架构" # 完成反馈待办 python3 scripts/cwork-todo.py complete \ --todo-id \ --content "已补充相关数据，详见附件" # 查看汇报详情（含节点与处理意见） python3 scripts/cwork-query-report.py --mode node-detail --report-id ``` | 参数 | 说明 | |------|------| | `action` | `list` / `complete` | | `--page-index` | 页码（默认 1） | | `--page-size` | 每页数量（默认 20） | | `--status` | 状态筛选 | | `--todo-id` | 待办 ID（complete 必填） | | `--content` | 完成说明（所有类型必填） | | `--operate` | 决策操作：`agree`（同意）/ `disagree`（不同意）（仅决策类待办需要，不传则不发送此字段） | | `--dry-run` | 仅预览（complete 可用） | **输出格式**（complete）： ```json { "success": true, "action": "complete", "todoId": "12345", "result": {} } ``` --- ### 8. 模板管理 — `cwork-templates.py` **意图**：查询汇报模板列表 ```bash # 查询模板列表 python3 scripts/cwork-templates.py list --limit 50 # 带时间范围 python3 scripts/cwork-templates.py list --begin-time 1710000000000 --end-time 1712000000000 ``` | 参数 | 说明 | |------|------| | `action` | `list` | | `--limit` | 返回数量限制（默认 50） | | `--begin-time` | 开始时间戳（毫秒） | | `--end-time` | 结束时间戳（毫秒） | | `--output-raw` | 输出原始 API 响应 | **输出字段**： - `id` — 模板 ID - `name` — 模板名称 - `type` — 类型 ID - `typeName` — 类型名称 - `grade` — 优先级 --- ## 辅助工具 ### 问题自动上报 — `cwork-report-issue.py` **意图**：当脚本报错或 API 异常时，自动将问题提交为 GitHub Issue，便于追踪和修复。 **前置条件**：设置环境变量 `GITHUB_TOKEN`（详见 `references/maintenance.md`）。 ```bash # 上报一个脚本报错 python3 scripts/cwork-report-issue.py \ --title "bug: cwork-send-report.py 发送失败" \ --script cwork-send-report.py \ --error '{"success": false, "error": "API Error (200003): 流程节点类型不正确"}' \ --body "发送汇报时传入 reportLevelList，type 字段使用了中文导致报错" # 先预览，不实际提交 python3 scripts/cwork-report-issue.py \ --title "bug: ..." \ --script cwork-query-report.py \ --error "..." \ --dry-run ``` | 参数 | 必填 | 说明 | |------|------|------| | `--title` / `-T` | ✅ | Issue 标题 | | `--script` / `-s` | ❌ | 出错的脚本名称 | | `--error` / `-e` | ❌ | 错误信息（脚本 stderr 的 JSON 输出） | | `--body` / `-b` | ❌ | 问题描述（复现步骤等） | | `--extra` | ❌ | 附加信息（环境、版本等） | | `--labels` | ❌ | 额外标签（逗号分隔，默认已含 `bug` 和 `cms-cwork-workflow`） | | `--dry-run` | ❌ | 预览将提交的内容，不实际创建 | | `--token` | ❌ | GitHub Token（仅调试用，生产环境请用环境变量 `GITHUB_TOKEN`） | **输出格式**： ```json { "success": true, "issue_number": 42, "issue_url": "https://github.com/xgjk/cwork-skills/issues/42", "title": "bug: cwork-send-report.py 发送失败" } ``` **Agent 调用建议**： - 脚本返回 `"success": false` 且错误类型为 API 异常（非参数错误）时，询问用户是否上报 - 上报前使用 `--dry-run` 预览内容，确认无敏感信息（如 appKey、empId 等）后再提交 - `--error` 传入脚本 stderr 的原始 JSON 输出即可，脚本会自动格式化 --- ## `reportLevelList` 字段格式 `cwork-send-report.py` 可通过 **`--report-level-json`** 指向 UTF-8 JSON 文件（根节点为数组），内容对应 API 字段 `reportLevelList`，用于指定建议人/决策人/传阅等节点；不传时新建草稿可为空，**更新**时默认从 5.25 详情中的 `reportLevelList` 原样转换后写回，避免全量更新被清空。每个节点结构如下： ```python report_level_list = [ { "level": 1, # 节点序号（从1开始） "nodeName": "建议人", # 节点显示名称 "type": "suggest", # suggest=建议 | decide=决策 | read=传阅 "levelUserList": [ {"empId": 1512393035869810694}, # empId 必须是整数（非字符串） ], } ] ``` > ⚠️ `type` 只接受英文小写 `suggest` / `decide` / `read`，不接受中文。`levelUserList` 是必填字段，不可为 `null` 或空列表。 --- ## Agent 调用模式 ### 模式 A：简单查询（单次 exec） ``` 用户：「帮我看看今天有没有未读汇报」 Agent → exec: python3 scripts/cwork-query-report.py --mode unread --page-size 10 Agent ← JSON → 摘要呈现给用户 ``` ### 模式 B：多步编排（Agent 协调多次 exec） ``` 用户：「给张三发一份周报，内容是XXX」 Agent → exec: python3 scripts/cwork-send-report.py \ --title "周报" --content-html "..." --receivers "张三" Agent ← JSON（含完整 draftDetail、confirmPrompt；默认不会发出） Agent → 向用户展示 **draftDetail 全文**（尤其 contentHtml、附件、reportLevelList）用户：「确认」 Agent → exec: python3 scripts/cwork-send-report.py \ --draft-id "<上一步的 reportId（与 draftId 同值）>" --confirm-send Agent ← JSON（success、已通过 5.27 发出） Agent → 告知发送成功 ``` ### 模式 C：催办闭环（3步分离） ``` Agent → exec: python3 scripts/cwork-nudge-report.py identify --days-threshold 7 Agent ← JSON（未闭环列表） Agent → （LLM 推理）筛选需要催办的事项 Agent → exec: python3 scripts/cwork-nudge-report.py reminder \ --item-id --recipient "张三" --days-unresolved 14 --style polite Agent ← JSON（催办文案） Agent → （可选 LLM 优化文案） Agent → exec: python3 scripts/cwork-nudge-report.py nudge \ --report-id --content-html "..." ``` --- ## 错误处理所有脚本遵循统一错误约定： - **成功**：JSON 到 stdout，含 `"success": true` - **失败**：JSON 到 stderr，含 `"success": false` 和 `"error"` 字段，exit code ≠ 0 - **Agent 应同时检查 stdout 和 stderr** 遇到 API 异常（如 `API Error (2xxxxx)`）时，可调用 `cwork-report-issue.py` 上报问题： ```bash python3 scripts/cwork-report-issue.py \ --title "bug: <出错脚本> <简短描述>" \ --script "<出错脚本>.py" \ --error '' \ --body "<复现步骤>" ``` > ⚠️ 上报前确认 `--error` 和 `--body` 中不含 appKey、empId 等敏感信息。 ### 通用参数所有脚本均支持以下通用参数： | 参数 | 说明 | |------|------| | `--params-file ` | 从 UTF-8 JSON 文件读取参数，key 与命令行参数名一致（连字符格式）。用于解决 Windows PowerShell 中文编码问题。 | **用法示例**： ```json { "title": "周报标题", "content-html": "

汇报内容

", "receivers": "张三" } ``` ```bash python3 scripts/cwork-send-report.py --params-file params.json ``` > 文件参数与命令行参数可混用，命令行参数优先级更高。文件必须为 UTF-8 编码（带或不带 BOM 均支持）。 --- ## 目录结构 ``` cms-cwork-workflow/ ├── SKILL.md ← 本文件（意图级接口文档） ├── scripts/ │ ├── cwork_client.py ← 共享 API 客户端（HTTP 封装 + 所有 API 方法） │ ├── cwork-search-emp.py ← 0. 搜索员工 ✨ v3.2.0 新增 │ ├── cwork-send-report.py ← 1. 发送汇报 │ ├── cwork-query-report.py ← 2. 查询汇报 │ ├── cwork-create-task.py ← 3. 创建任务 │ ├── cwork-review-report.py ← 4. 审阅汇报 │ ├── cwork-query-tasks.py ← 5. 查询任务 │ ├── cwork-nudge-report.py ← 6. 催办闭环 │ ├── cwork-todo.py ← 7. 待办管理 │ └── cwork-templates.py ← 8. 模板管理 ├── design/ │ └── DESIGN.md ← 架构设计文档 └── references/ └── maintenance.md ← 维护操作说明 ``` --- ## 参考资料 > 正常调用只需按本文档使用 CLI 脚本，不必直接查阅 API 文档。 > 遇到 API 错误码或需要扩展脚本时，可查阅以下官方文档： - **工作协同 Open API 接口文档**：[工作协同API说明.md](https://github.com/xgjk/dev-guide/blob/main/02.%E4%BA%A7%E5%93%81%E4%B8%9A%E5%8A%A1AI%E6%96%87%E6%A1%A3/%E5%B7%A5%E4%BD%9C%E5%8D%8F%E5%90%8C/%E5%B7%A5%E4%BD%9C%E5%8D%8F%E5%90%8CAPI%E8%AF%B4%E6%98%8E.md) FILE:design/DESIGN.md # cms-cwork-workflow — Agent-First 重构设计文档 ## 设计概述将 cms-cwork-workflow 从 TypeScript 源码模式重构为 Agent-First 架构，通过 Python 编排脚本提供可组合、可复用的能力层。 ## 核心设计理念 ### Agent-First 架构 - **设计原则**: Agent 负责业务逻辑和 LLM 推理，脚本负责 API 编排和数据处理 - **协作模式**: Agent → JSON（工具输出）→ Agent（进一步处理） - **边界清晰**: 脚本只做标准化的 API 调用，不做业务决策 ### 编排脚本规范 - **JSON 输出**: 所有脚本输出结构化 JSON 到 stdout - **错误处理**: stderr 输出调试信息，成功/failure 统一格式 - **参数校验**: 强制参数验证，可选参数支持默认值 - **干跑模式**: `--dry-run` / `--preview-only` 支持 ## 文件结构设计 ``` cms-cwork-workflow/ ├── SKILL.md ← 产品级接口文档 ├── scripts/ │ ├── cwork_client.py ← 共享 API 客户端（HTTP 封装 + 所有 API 方法） │ ├── cwork-report-issue.py ← 辅助：问题自动上报到 GitHub │ ├── cwork-search-emp.py ← 搜索员工 │ ├── cwork-send-report.py ← 发送汇报 │ ├── cwork-query-report.py ← 查询汇报 │ ├── cwork-create-task.py ← 创建任务 │ ├── cwork-review-report.py ← 审阅汇报 │ ├── cwork-query-tasks.py ← 查询任务 │ ├── cwork-nudge-report.py ← 催办闭环 │ ├── cwork-todo.py ← 待办管理 │ └── cwork-templates.py ← 模板管理 ├── design/ │ └── DESIGN.md ← 本文件 └── references/ └── maintenance.md ← 维护说明 ``` ## 编排脚本架构 ### 客户端层 (`cwork_client.py`) - 封装 HTTP 请求逻辑（含 307 重定向处理） - 统一错误处理和重试 - 参数编码和响应解析 - 所有业务 API 方法（`submit_report`、`save_draft` 等） - 共享工具函数（`resolve_names_to_empids`、`parse_deadline`、`make_client` 等） ### 编排脚本层 (`.py`) - argparse 命令行参数处理 - 业务逻辑编排 - 结构化 JSON 输出 ## cwork_client.py API 方法索引 | API 端点 | 方法 | |----------|------| | `/open-api/cwork-user/searchEmpByName` | `search_emp_by_name()` | | `/open-api/work-report/report/record/inbox` | `get_inbox_list()` | | `/open-api/work-report/report/record/outbox` | `get_outbox_list()` | | `/open-api/work-report/report/info` | `get_report_info()` | | `/open-api/work-report/report/record/submit` | `submit_report()` | | `/open-api/work-report/report/record/reply` | `reply_report()` | | `/open-api/work-report/reportInfoOpenQuery/unreadList` | `get_unread_list()` | | `/open-api/work-report/open-platform/report/readReport` | `mark_report_read()` | | `/open-api/work-report/report/plan/searchPage` | `search_task_page()` | | `/open-api/work-report/report/plan/getSimplePlanAndReportInfo` | `get_simple_plan_and_report_info()` | | `/open-api/work-report/open-platform/report/plan/create` | `create_plan()` | | `/open-api/work-report/draftBox/saveOrUpdate` | `save_draft()` | | `/open-api/work-report/draftBox/listByPage` | `list_drafts()` | | `/open-api/work-report/draftBox/detail/{id}` | `get_draft_detail()` | | `/open-api/work-report/draftBox/delete/{id}` | `delete_draft()`（路径 id = 5.24 列表项 `id`）；`delete_draft_by_report_id()`（按 `businessId` 查列表再删） | | `/open-api/work-report/draftBox/submit/{id}` | `submit_draft()`（5.27，路径 id 为汇报 id） | | `/open-api/cwork-file/uploadWholeFile` | `upload_file()` | | `/open-api/work-report/template/listTemplates` | `list_templates()` | | `/open-api/work-report/reportInfoOpenQuery/todoList` | `get_todo_list()` | | `/open-api/work-report/open-platform/todo/completeTodo` | `complete_todo()` | | `/open-api/work-report/report/getReportNodeDetail` | `get_report_node_detail()` | | — | `get_sender_history()` | | — | `search_reports_by_keyword()` | ## API 覆盖设计 | 功能域 | 脚本 | 主要 API | 输出格式 | |--------|------|----------|----------| | 汇报查询 | `cwork-query-report.py` | `get_inbox_list()`, `get_outbox_list()`, `get_unread_list()` | `{"success": true, "items": [...]}` | | 任务查询 | `cwork-query-tasks.py` | `search_task_page()`, `get_simple_plan_and_report_info()` | `{"success": true, "tasks": [...]}` | | 审阅回复 | `cwork-review-report.py` | `reply_report()`, `mark_report_read()` | `{"success": true, "result": {...}}` | | 催办闭环 | `cwork-nudge-report.py` | `submit_report(type=12)`, `complete_todo()` | `{"success": true, "action": "nudge"}` | | 创建任务 | `cwork-create-task.py` | `create_plan()`, `search_emp_by_name()` | `{"success": true, "planId": "..."}` | | 发送汇报 | `cwork-send-report.py` | `save_draft()`, `get_draft_detail()`, `submit_draft()`（5.27）, `upload_file()` | `{"success": true, "reportId": "..."}` | | 待办管理 | `cwork-todo.py` | `get_todo_list()`, `complete_todo()` | `{"success": true, "todos": [...]}` | | 模板管理 | `cwork-templates.py` | `list_templates()` | `{"success": true, "templates": [...]}` | ## 错误处理设计 ### 成功响应 ```json { "success": true, "action": "list", "total": 20, "items": [] } ``` ### 错误响应 ```json { "success": false, "error": "缺少必填参数: --report-id" } ``` ### 交互模式 ```bash # 错误信息输出到 stderr $ python3 cwork-send-report.py --to 张三 --content "测试" --dry-run {"success": false, "error": "缺少必填参数: --task-id"} # 干跑模式输出到 stdout $ python3 cwork-send-report.py --to 张三 --content "测试" --dry-run --preview-only {"success": true, "preview": {...}, "validation": {...}} ``` ## 与 Agent 协作模式 ### Agent → Script ```python Agent: "帮我查询张三的待办列表" Agent → exec: python3 cwork-todo.py list --status pending --page-size 10 Agent ← stdout: {"success": true, "items": [...]} ``` ### Script → Agent (可选) ```python # Agent 可进一步处理脚本输出 Agent: "我发现你有3个逾期待办，需要催办吗？" Agent → exec: python3 cwork-nudge-report.py identify --days-threshold 7 ``` ## 扩展性设计 ### 新功能添加流程 1. 在 `cwork_client.py` 添加新的 API 方法 2. 在对应脚本中调用新 API 3. 更新 `SKILL.md` 文档 4. 按需更新 `DESIGN.md` 和 `SKILL.md` 记录设计变更 ### 版本管理 - 主版本号：重大架构变更 - 次版本号：新功能增加 - 修订版本号：Bug 修复 --- ## 实现状态 (v3.1.0) ### ✅ 已完成 - [x] 8个编排脚本完整实现 - [x] 共享客户端封装（含 23 个 API 方法） - [x] JSON 输出规范 - [x] 参数校验和错误处理 - [x] Agent-First 架构设计 - [x] 文档体系完善 - [x] **决策/建议/反馈待办完整支持**（v3.1.0 新增） - [x] **汇报节点详情查询**（`get_report_node_detail()`） ### 🆕 v3.1.0 新增功能（2026-04-03） | 功能 | 说明 | API 接口 | |------|------|----------| | **决策待办** | 支持同意/不同意操作 | `complete_todo(operate="agree/disagree")` | | **建议待办** | 支持建议内容提交 | `complete_todo(content="...")` | | **反馈待办** | 支持反馈回复 | `complete_todo(content="...")` | | **节点详情** | 查询汇报的审批节点与处理意见 | `get_report_node_detail()` | ### 🔄 进行中 - [ ] 单元测试覆盖 - [ ] 性能优化 - [ ] 更多汇报类型支持 ### 📋 待优化 - [ ] 错误码标准化 - [ ] 重试机制完善 - [ ] 缓存策略优化 --- *最后更新: 2026-04-03* FILE:references/maintenance.md # 维护信息 ## 基本信息 - 版本：见 `_meta.json` - ClawHub slug：`cms-cwork-workflow` ## GitHub 地址 - 仓库：https://github.com/xgjk/cwork-skills - Skill 目录：`cwork-skills/cms-cwork-workflow/` ## 官方 API 文档 - [工作协同 Open API 接口文档](https://github.com/xgjk/dev-guide/blob/main/02.%E4%BA%A7%E5%93%81%E4%B8%9A%E5%8A%A1AI%E6%96%87%E6%A1%A3/%E5%B7%A5%E4%BD%9C%E5%8D%8F%E5%90%8C/%E5%B7%A5%E4%BD%9C%E5%8D%8F%E5%90%8CAPI%E8%AF%B4%E6%98%8E.md) > 脚本已封装所有 API 调用，正常使用无需查阅。排查 API 错误码或新增 API 支持时参考。 ## 如何提 Issue ### 方式一：自动上报（推荐）通过 `cwork-report-issue.py` 脚本直接创建 Issue，需提前配置 `GITHUB_TOKEN`： `cwork-report-issue.py` 已内置共享 token，所有用户无需任何配置，直接调用即可。 > 维护者如需更换 token，修改 `scripts/cwork-report-issue.py` 中的 `_BUILTIN_TOKEN` 常量。 **调用示例**： ```bash python3 scripts/cwork-report-issue.py \ --title "bug: cwork-send-report.py 发送失败" \ --script cwork-send-report.py \ --error '{"success": false, "error": "API Error (200003)"}' \ --body "复现步骤：..." ``` ### 方式二：手动提交 1. 访问 https://github.com/xgjk/cwork-skills/issues/new 2. 选择 Label：`cms-cwork-workflow` 3. 填写问题描述 + 复现步骤 ## 如何更新 **工厂内部开发：** 1. 修改 Skill 内容 2. 更新 `_meta.json` 版本号 3. 执行 `clawhub publish` **ClawHub 用户：** ```bash clawhub update cms-cwork-workflow ``` --- *最后更新：2026-04-05* FILE:scripts/cwork-create-task.py #!/usr/bin/env python3 """ CWork Create Task - Agent-First Usage: python3 scripts/cwork-create-task.py --task-main "name" --content "desc" --assignee "person" --deadline 2026-04-10 """ import sys import os import json import argparse sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from cwork_client import make_client, CWorkError, parse_deadline, resolve_names_to_empids, apply_params_file_pre_parse _DEFAULT_MS = int(__import__("datetime").datetime.now().timestamp() * 1000) + 7 * 86400000 def parse_args(argv=None): p = argparse.ArgumentParser(description="CWork create task (Agent-First)") p.add_argument("--task-main", required=True, help="Task name") p.add_argument("--deadline", help="Deadline YYYY-MM-DD or ms timestamp (default 7d)") p.add_argument("--content", required=True, help="Task description") p.add_argument("--target", help="Target description") p.add_argument("--assignee", help="Owner name") p.add_argument("--assistant", help="Assistant names (comma-separated)") p.add_argument("--supervisor", help="Supervisor name") p.add_argument("--copy", help="CC names (comma-separated)") p.add_argument("--observer", help="Observer names (comma-separated)") p.add_argument("--report-to", help="Report-to name") p.add_argument("--push-now", type=lambda x: x.lower() == "true", default=True) p.add_argument("--dry-run", action="store_true") p.add_argument("--params-file", dest="params_file", default=None, help="UTF-8 JSON 文件路径，从文件读取参数") return p.parse_args(argv) def _comma(val): if not val: return None return [v.strip() for v in val.split(",") if v.strip()] def _die(msg): print(json.dumps({"success": False, "error": msg}, ensure_ascii=False), file=sys.stderr) sys.exit(1) def main(): apply_params_file_pre_parse() args = parse_args() try: client = make_client() except CWorkError as e: _die(str(e)) def _resolve(name_or_list): names = _comma(name_or_list) if isinstance(name_or_list, str) else name_or_list if not names: return None return resolve_names_to_empids(client, names) dl = parse_deadline(args.deadline) if args.deadline else _DEFAULT_MS report_to = args.report_to or args.assignee if not report_to: _die("--report-to 或 --assignee 至少需要提供一个（API 要求 reportEmpIdList 必填）") info = dict(assignee=args.assignee, assistant=args.assistant, supervisor=args.supervisor, copy=args.copy, observer=args.observer, reportTo=report_to, deadlineMs=dl) if args.dry_run: print(json.dumps({"success": True, "dryRun": True, "task": dict( main=args.task_main, content=args.content, target=args.target or args.content, deadline=args.deadline, deadlineMs=dl, pushNow=args.push_now, ), "resolved": info}, ensure_ascii=False, indent=2)) return try: pid = client.create_plan( main=args.task_main, needful=args.content, target=args.target or args.content, end_time=dl, owner_emp_id_list=_resolve(args.assignee), assist_emp_id_list=_resolve(args.assistant), supervisor_emp_id_list=_resolve(args.supervisor), copy_emp_id_list=_resolve(args.copy), observer_emp_id_list=_resolve(args.observer), report_emp_id_list=_resolve(report_to), push_now=args.push_now, ) print(json.dumps({"success": True, "planId": pid, "task": dict(main=args.task_main, deadline=args.deadline, deadlineMs=dl), "resolved": info}, ensure_ascii=False, indent=2)) except CWorkError as e: _die(str(e)) if __name__ == "__main__": main() FILE:scripts/cwork-nudge-report.py #!/usr/bin/env python3 """ CWork Nudge Report — 催办通知脚本 Modes: list — 列出未闭环事项清单（用于催收） nudge — 发送催办通知 Usage: python cwork-nudge-report.py --mode list [--days-threshold 7] python cwork-nudge-report.py --mode nudge --emp-id --task-main "任务名" --deadline 2026-04-10 --content "催办内容" Output: JSON to stdout, error JSON to stderr + exit 1 """ import sys import os import json import argparse from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from cwork_client import (CWorkClient, make_client, CWorkError, output_json, output_error, parse_deadline, resolve_names_to_empids, apply_params_file_pre_parse) REPORT_TYPE_ID = 12 # 催收汇报 def parse_args(argv=None): parser = argparse.ArgumentParser(description="CWork 催办通知") parser.add_argument("--mode", required=True, choices=["list", "nudge"], help="操作模式: list=列出未闭环, nudge=发送催办") parser.add_argument("--emp-id", help="催办对象 empId（nudge模式必填）") parser.add_argument("--task-main", help="任务名称（nudge模式必填）") parser.add_argument("--deadline", help="截止日期 YYYY-MM-DD 或毫秒时间戳") parser.add_argument("--content", help="催办内容描述") parser.add_argument("--target", help="目标描述") parser.add_argument("--remind-style", choices=["polite", "normal"], default="polite", help="催办风格（默认polite）") parser.add_argument("--days-threshold", type=int, default=7, help="未闭环天数阈值（默认7天）") parser.add_argument("--assignee", help="责任人姓名（用于解析empId）") parser.add_argument("--page-index", type=int, default=1, help="页码（默认1）") parser.add_argument("--page-size", type=int, default=50, help="每页大小（默认50）") parser.add_argument("--interactive", action="store_true", help="交互模式") parser.add_argument("--dry-run", action="store_true", help="干跑模式") parser.add_argument("--params-file", dest="params_file", default=None, help="UTF-8 JSON 文件路径，从文件读取参数") return parser.parse_args(argv) def build_nudge_content(task_main: str, deadline: str | None, content: str | None, style: str) -> str: """Build HTML nudge content based on style.""" if style == "polite": body = f"""

您好，您有任务需要关注：

📌 任务名称：{task_main}

""" if deadline: body += f"

⏰ 截止日期：{deadline}

" if content: body += f"

📝 详情：{content}

" body += "

请及时处理，如有疑问请联系我。谢谢！

" else: body = f"""

【催办】任务：{task_main}

""" if deadline: body += f"

截止日期：{deadline}

" if content: body += f"

详情：{content}

" body += "

请尽快处理。

" return body def main(): apply_params_file_pre_parse() args = parse_args() if args.dry_run: preview = { "mode": args.mode, "empId": args.emp_id, "taskMain": args.task_main, "deadline": args.deadline, "content": args.content, "target": args.target, "remindStyle": args.remind_style, "daysThreshold": args.days_threshold, "assignee": args.assignee, } print("=== DRY RUN PREVIEW ===", file=sys.stderr) print(json.dumps(preview, ensure_ascii=False, indent=2), file=sys.stderr) output_json({"success": True, "message": "Dry run — no actual API call made"}) return if args.interactive: from cwork_client import interactive_confirm desc = f"催办任务 (mode={args.mode}, empId={args.emp_id})" if not interactive_confirm(f"nudge_{args.mode}", desc): output_json({"success": True, "message": "Skipped by user"}) return try: client = make_client() except CWorkError as e: output_error(str(e)) try: if args.mode == "list": threshold_ms = args.days_threshold * 24 * 60 * 60 * 1000 now_ms = int(datetime.now().timestamp() * 1000) result = client.search_task_page( page_size=args.page_size, page_index=args.page_index, report_status=1, ) items = result if isinstance(result, list) else result.get("list", []) unclosed = [ item for item in items if item.get("endTime") and (now_ms - item.get("endTime", 0)) > threshold_ms ] output_json({ "success": True, "data": unclosed, "total": len(unclosed), "daysThreshold": args.days_threshold, "message": f"Found {len(unclosed)} unclosed items older than {args.days_threshold} days" }) elif args.mode == "nudge": if not args.emp_id and not args.assignee: output_error("--emp-id or --assignee is required for nudge mode") if not args.task_main: output_error("--task-main is required for nudge mode") emp_id = args.emp_id if args.assignee and not args.emp_id: emp_ids = resolve_names_to_empids(client, [args.assignee]) emp_id = emp_ids[0] deadline_str = args.deadline if args.deadline else None nudge_content = build_nudge_content( args.task_main, deadline_str, args.content, args.remind_style ) result = client.submit_report( main=f"【催办】{args.task_main}", content_html=nudge_content, type_id=REPORT_TYPE_ID, accept_emp_id_list=[emp_id], ) output_json({ "success": True, "reportId": result.get("id"), "empId": emp_id, "message": f"Nudge sent to empId={emp_id} for task: {args.task_main}" }) except CWorkError as e: output_error(str(e)) except Exception as e: output_error(f"Unexpected error: {e}") if __name__ == "__main__": main() FILE:scripts/cwork-query-report.py #!/usr/bin/env python3 """ CWork Query Reports - Agent-First Modes: inbox / outbox / unread / detail / node-detail / sender-history / keyword-search / pending / my-sent Usage: python3 scripts/cwork-query-report.py --mode inbox [--page-size 20] python3 scripts/cwork-query-report.py --mode detail --report-id python3 scripts/cwork-query-report.py --mode node-detail --report-id python3 scripts/cwork-query-report.py --mode sender-history --sender-emp-id python3 scripts/cwork-query-report.py --mode keyword-search --keyword "公章" python3 scripts/cwork-query-report.py --mode pending python3 scripts/cwork-query-report.py --mode unread """ import sys import os import json import argparse sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from cwork_client import CWorkClient, make_client, CWorkError, apply_params_file_pre_parse def parse_args(argv=None): parser = argparse.ArgumentParser(description="CWork query reports (Agent-First)") parser.add_argument("--mode", required=True, choices=["inbox", "outbox", "unread", "detail", "node-detail", "sender-history", "keyword-search", "pending", "my-sent"]) parser.add_argument("--page-index", type=int, default=1) parser.add_argument("--page-size", type=int, default=20) parser.add_argument("--report-id", help="Report ID (required for detail/node-detail)") parser.add_argument("--sender-emp-id", help="Sender employee ID (required for sender-history)") parser.add_argument("--keyword", help="Search keyword (required for keyword-search)") parser.add_argument("--days", type=int, default=90, help="Days to look back (default 90)") parser.add_argument("--report-type", type=int, choices=[1, 2, 3, 4, 5]) parser.add_argument("--status", type=int, help="Read status: 0=unread 1=read") parser.add_argument("--keyword-filter", help="Legacy: Keyword filter") parser.add_argument("--start-date", help="Start date YYYY-MM-DD") parser.add_argument("--end-date", help="End date YYYY-MM-DD") parser.add_argument("--params-file", dest="params_file", default=None, help="UTF-8 JSON 文件路径，从文件读取参数") return parser.parse_args(argv) def _parse_date(value, end_of_day: bool = False): """Convert YYYY-MM-DD to millisecond timestamp (always interpreted as UTC+8). Binds the date to UTC+8 explicitly so the result is identical regardless of the system timezone where the script is executed. end_of_day=True: returns 23:59:59.999 CST of that day so --end-date covers the full Beijing calendar day (e.g. 2026-04-07 → 2026-04-07T23:59:59.999+08:00). """ if value is None: return None from datetime import datetime, timedelta, timezone _CST = timezone(timedelta(hours=8)) try: dt = datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=_CST) if end_of_day: dt = dt + timedelta(days=1) - timedelta(milliseconds=1) return int(dt.timestamp() * 1000) except ValueError: return None def _die(msg): print(json.dumps({"success": False, "error": msg}, ensure_ascii=False), file=sys.stderr) sys.exit(1) def main(): try: sys.stdout.reconfigure(encoding="utf-8") except AttributeError: pass apply_params_file_pre_parse() args = parse_args() try: client = make_client() except CWorkError as e: _die(str(e)) try: if args.mode == "detail": if not args.report_id: _die("--report-id is required for detail mode") data = client.get_report_info(args.report_id) print(json.dumps({"success": True, "data": data}, ensure_ascii=False, indent=2)) return if args.mode == "node-detail": if not args.report_id: _die("--report-id is required for node-detail mode") data = client.get_report_node_detail(args.report_id) print(json.dumps({"success": True, "data": data}, ensure_ascii=False, indent=2)) return if args.mode == "sender-history": if not args.sender_emp_id: _die("--sender-emp-id is required for sender-history mode") data = client.get_sender_history( sender_emp_id=args.sender_emp_id, days=args.days, max_count=args.page_size ) print(json.dumps({"success": True, "data": data}, ensure_ascii=False, indent=2)) return if args.mode == "keyword-search": if not args.keyword: _die("--keyword is required for keyword-search mode") data = client.search_reports_by_keyword( keyword=args.keyword, days=args.days, max_count=args.page_size ) print(json.dumps({"success": True, "data": data}, ensure_ascii=False, indent=2)) return if args.mode == "unread": data = client.get_unread_list(args.page_index, args.page_size) print(json.dumps({"success": True, "data": data}, ensure_ascii=False, indent=2)) return read_status = args.status if args.mode == "pending": read_status = 0 if args.mode in ("inbox", "pending"): data = client.get_inbox_list( page_size=args.page_size, page_index=args.page_index, report_record_type=args.report_type, read_status=read_status, begin_time=_parse_date(args.start_date), end_time=_parse_date(args.end_date, end_of_day=True)) else: data = client.get_outbox_list( page_size=args.page_size, page_index=args.page_index, report_record_type=args.report_type, begin_time=_parse_date(args.start_date), end_time=_parse_date(args.end_date, end_of_day=True)) print(json.dumps({"success": True, "data": data}, ensure_ascii=False, indent=2)) except CWorkError as e: _die(str(e)) if __name__ == "__main__": main() FILE:scripts/cwork-query-tasks.py #!/usr/bin/env python3 """ CWork Query Tasks — 任务查询脚本 Modes: my — 查询我负责的任务（进行中 status=1） created — 查询我创建的任务（未启动 status=0） team — 查询团队任务（进行中 status=1） assigned — 查询分配给我的任务 detail — 查看任务详情（需 --task-id） chain — 查看任务→汇报链路（需 --task-id） blocked — 识别卡点/逾期任务 unclosed — 识别未闭环事项 manager — 管理者仪表盘（需 --subordinate-ids） Usage: python cwork-query-tasks.py --mode my [--page-index 1] [--page-size 20] python cwork-query-tasks.py --mode detail --task-id python cwork-query-tasks.py --mode blocked [--days-threshold 7] python cwork-query-tasks.py --mode manager --subordinate-ids emp001,emp002 Output: JSON to stdout, error JSON to stderr + exit 1 """ import sys import os import json import argparse from datetime import datetime, timedelta sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from cwork_client import CWorkClient, make_client, CWorkError, output_json, output_error, resolve_names_to_empids, apply_params_file_pre_parse def parse_args(argv=None): parser = argparse.ArgumentParser(description="CWork 任务查询") parser.add_argument("--mode", required=True, choices=["my", "created", "team", "assigned", "detail", "chain", "blocked", "unclosed", "manager", "nudge"], help="查询模式") parser.add_argument("--task-id", help="任务ID（detail/chain模式必填）") parser.add_argument("--page-index", type=int, default=1, help="页码（默认1）") parser.add_argument("--page-size", type=int, default=20, help="每页大小（默认20）") parser.add_argument("--status", type=int, choices=[0, 1, 2], help="任务状态: 0=已关闭, 1=进行中, 2=未启动") parser.add_argument("--task-status", type=int, choices=[0, 1, 2, 3], help="汇报状态: 0=关闭, 1=待汇报, 2=已汇报, 3=逾期") parser.add_argument("--report-status", type=int, choices=[0, 1, 2, 3], help="汇报状态（同task-status）: 0=关闭, 1=待汇报, 2=已汇报, 3=逾期") parser.add_argument("--key-word", help="关键词搜索") parser.add_argument("--assignee", help="责任人姓名") parser.add_argument("--subordinate-ids", help="下属empId列表（逗号分隔，manager模式用）") parser.add_argument("--days-threshold", type=int, default=7, help="未闭环天数阈值（默认7）") parser.add_argument("--interactive", action="store_true", help="交互模式") parser.add_argument("--dry-run", action="store_true", help="干跑模式") parser.add_argument("--params-file", dest="params_file", default=None, help="UTF-8 JSON 文件路径，从文件读取参数") return parser.parse_args(argv) def main(): apply_params_file_pre_parse() args = parse_args() if args.dry_run: preview = { "mode": args.mode, "params": { "pageIndex": args.page_index, "pageSize": args.page_size, "status": args.status, "taskStatus": args.task_status, "reportStatus": args.report_status, "keyWord": args.key_word, "assignee": args.assignee, "subordinateIds": args.subordinate_ids, "daysThreshold": args.days_threshold, } } print("=== DRY RUN PREVIEW ===", file=sys.stderr) print(json.dumps(preview, ensure_ascii=False, indent=2), file=sys.stderr) output_json({"success": True, "message": "Dry run — no actual API call made"}) return if args.interactive: from cwork_client import interactive_confirm desc = f"查询任务 (mode={args.mode}, page={args.page_index})" if not interactive_confirm(f"query_task_{args.mode}", desc): output_json({"success": True, "message": "Skipped by user"}) return try: client = make_client() except CWorkError as e: output_error(str(e)) try: if args.mode in ("detail", "chain"): if not args.task_id: output_error("--task-id is required for detail/chain mode") result = client.get_simple_plan_and_report_info(args.task_id) output_json({"success": True, "data": result}) elif args.mode == "blocked": threshold_ms = args.days_threshold * 24 * 60 * 60 * 1000 now_ms = int(datetime.now().timestamp() * 1000) result = client.search_task_page( page_size=args.page_size, page_index=args.page_index, report_status=3, key_word=args.key_word, ) items = result if isinstance(result, list) else result.get("list", []) blocked = [ item for item in items if item.get("endTime") and (now_ms - item.get("endTime", 0)) > threshold_ms ] output_json({"success": True, "data": blocked, "total": len(blocked)}) elif args.mode == "unclosed": result = client.search_task_page( page_size=args.page_size, page_index=args.page_index, report_status=1, key_word=args.key_word, ) output_json({"success": True, "data": result}) elif args.mode == "manager": if not args.subordinate_ids: output_error("--subordinate-ids is required for manager mode") emp_ids = [e.strip() for e in args.subordinate_ids.split(",") if e.strip()] result = client.search_task_page( page_size=args.page_size, page_index=args.page_index, emp_id_list=emp_ids, status=args.status, key_word=args.key_word, ) output_json({"success": True, "data": result}) elif args.mode in ("my", "assigned"): status = args.status if args.status is not None else 1 if args.assignee: emp_ids = resolve_names_to_empids(client, [args.assignee]) else: emp_ids = None result = client.search_task_page( page_size=args.page_size, page_index=args.page_index, status=status, emp_id_list=emp_ids, key_word=args.key_word, ) output_json({"success": True, "data": result}) elif args.mode == "created": status = args.status if args.status is not None else 0 result = client.search_task_page( page_size=args.page_size, page_index=args.page_index, status=status, key_word=args.key_word, ) output_json({"success": True, "data": result}) elif args.mode == "team": status = args.status if args.status is not None else 1 result = client.search_task_page( page_size=args.page_size, page_index=args.page_index, status=status, key_word=args.key_word, ) output_json({"success": True, "data": result}) elif args.mode == "nudge": result = client.search_task_page( page_size=args.page_size, page_index=args.page_index, report_status=3, key_word=args.key_word, ) output_json({"success": True, "data": result, "message": "Use --emp-id with cwork-nudge-report.py to send nudge"}) except CWorkError as e: output_error(str(e)) except Exception as e: output_error(f"Unexpected error: {e}") if __name__ == "__main__": main() FILE:scripts/cwork-report-issue.py #!/usr/bin/env python3 """cwork-report-issue.py — 自动上报问题到 GitHub Issues。使用场景： Agent 遇到脚本报错或 API 异常时，调用本脚本将问题自动提交为 GitHub Issue。认证方式（优先级从高到低）： 1. 环境变量 GITHUB_TOKEN 2. --token 参数（仅调试用，不要在 CI/生产环境使用）用法示例： python3 scripts/cwork-report-issue.py \\ --title "bug: cwork-send-report.py 发送失败" \\ --script cwork-send-report.py \\ --error '{"success": false, "error": "API Error (200003)"}' \\ --body "复现步骤：python3 scripts/cwork-send-report.py --title 测试" """ from __future__ import annotations import sys import json import os import argparse import urllib.request import urllib.error from pathlib import Path sys.stdout.reconfigure(encoding="utf-8", errors="replace") sys.stderr.reconfigure(encoding="utf-8", errors="replace") GITHUB_REPO = "xgjk/cwork-skills" GITHUB_API_URL = f"https://api.github.com/repos/{GITHUB_REPO}/issues" DEFAULT_LABELS = ["bug", "cms-cwork-workflow"] # Fine-grained token，仅限 xgjk/cwork-skills 仓库的 Issues: Read and Write 权限。 # 无法读写代码、推送提交或访问其他仓库。所有安装此 skill 的用户共享此 token。 # 如需使用自己的 token，设置环境变量 GITHUB_TOKEN 即可覆盖。 _BUILTIN_TOKEN = "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" def output_json(data: dict, *, to_stderr: bool = False) -> None: out = sys.stderr if to_stderr else sys.stdout print(json.dumps(data, ensure_ascii=False, indent=2), file=out) def create_github_issue(token: str, title: str, body: str, labels: list[str]) -> dict: payload = json.dumps({ "title": title, "body": body, "labels": labels, }).encode("utf-8") req = urllib.request.Request( GITHUB_API_URL, data=payload, headers={ "Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3+json", "Content-Type": "application/json", "X-GitHub-Api-Version": "2022-11-28", }, method="POST", ) try: with urllib.request.urlopen(req) as resp: result = json.loads(resp.read().decode("utf-8")) return { "success": True, "issue_number": result["number"], "issue_url": result["html_url"], "title": result["title"], } except urllib.error.HTTPError as e: error_body = e.read().decode("utf-8", errors="replace") try: detail = json.loads(error_body).get("message", error_body) except Exception: detail = error_body return { "success": False, "error": f"GitHub API HTTP {e.code}: {detail}", } except urllib.error.URLError as e: return { "success": False, "error": f"网络错误: {e.reason}", } def build_issue_body( script: str | None, error: str | None, body: str | None, extra: str | None, ) -> str: parts: list[str] = [] if script: parts.append(f"## 出错脚本\n\n`{script}`") if error: # 尝试格式化 JSON 错误输出 try: parsed = json.loads(error) formatted = json.dumps(parsed, ensure_ascii=False, indent=2) except Exception: formatted = error parts.append(f"## 错误信息\n\n```json\n{formatted}\n```") if body: parts.append(f"## 问题描述\n\n{body}") if extra: parts.append(f"## 附加信息\n\n{extra}") parts.append("## Skill\n\n`cms-cwork-workflow`") return "\n\n---\n\n".join(parts) def apply_params_file_pre_parse() -> None: """从 --params-file 读取参数并注入 sys.argv（与其他脚本保持一致）。""" if "--params-file" not in sys.argv: return idx = sys.argv.index("--params-file") if idx + 1 >= len(sys.argv): return path = sys.argv[idx + 1] try: raw = Path(path).read_bytes() # 处理 UTF-8 BOM if raw.startswith(b"\xef\xbb\xbf"): raw = raw[3:] params = json.loads(raw.decode("utf-8")) injected: list[str] = [] for key, value in params.items(): arg_key = f"--{key}" if not key.startswith("--") else key injected.extend([arg_key, str(value)]) # 插入到 --params-file 之前 sys.argv = sys.argv[:idx] + injected + sys.argv[idx + 2:] except Exception as exc: print( json.dumps({"success": False, "error": f"读取 --params-file 失败: {exc}"}, ensure_ascii=False), file=sys.stderr, ) sys.exit(1) def main() -> None: sys.path.insert(0, str(Path(__file__).parent)) apply_params_file_pre_parse() parser = argparse.ArgumentParser( description="自动上报 cms-cwork-workflow 问题到 GitHub Issues", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument("--title", "-T", required=True, help="Issue 标题（必填）") parser.add_argument("--script", "-s", help="出错的脚本名称，如 cwork-send-report.py") parser.add_argument("--error", "-e", help="错误信息（脚本 stderr 的 JSON 输出）") parser.add_argument("--body", "-b", help="问题描述（补充说明、复现步骤等）") parser.add_argument("--extra", help="附加信息（环境、版本等）") parser.add_argument( "--labels", default="", help="额外标签（逗号分隔，默认已含 bug 和 cms-cwork-workflow）", ) parser.add_argument( "--token", help="GitHub Token（优先从环境变量 GITHUB_TOKEN 读取，此参数仅供调试）", ) parser.add_argument( "--dry-run", action="store_true", help="预览将要提交的 Issue 内容，不实际创建", ) parser.add_argument("--params-file", help="从 UTF-8 JSON 文件读取参数") args = parser.parse_args() # 解析 token：环境变量 > --token 参数 > 内置共享 token token = os.environ.get("GITHUB_TOKEN") or args.token or _BUILTIN_TOKEN if not token and not args.dry_run: output_json( { "success": False, "error": "缺少 GitHub Token，且内置 token 不可用。", "hint": "设置环境变量 GITHUB_TOKEN 或使用 --token 参数。", }, to_stderr=True, ) sys.exit(1) # 构建标签 labels = list(DEFAULT_LABELS) if args.labels: labels.extend(lbl.strip() for lbl in args.labels.split(",") if lbl.strip()) # 构建正文 issue_body = build_issue_body( script=args.script, error=args.error, body=args.body, extra=args.extra, ) # --dry-run：只展示，不提交 if args.dry_run: output_json( { "success": True, "dry_run": True, "would_create": { "repo": GITHUB_REPO, "title": args.title, "labels": labels, "body": issue_body, }, } ) return result = create_github_issue(token, args.title, issue_body, labels) if result["success"]: output_json(result) else: output_json(result, to_stderr=True) sys.exit(1) if __name__ == "__main__": main() FILE:scripts/cwork-review-report.py #!/usr/bin/env python3 """ CWork Review Report — 审阅/回复汇报脚本 Modes: reply — 回复/点评汇报 mark-read — 标记已读 pending — 查询待审汇报 Usage: python cwork-review-report.py --mode reply --report-id --reply "内容" python cwork-review-report.py --mode reply --report-id --reply "内容" --content-type html python cwork-review-report.py --mode reply --report-id --reply "内容" --at "张三" python cwork-review-report.py --mode mark-read --report-id python cwork-review-report.py --mode pending [--page-size 20] Output: JSON to stdout, error JSON to stderr + exit 1 """ import sys import os import json import argparse sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from cwork_client import (CWorkClient, make_client, CWorkError, output_json, output_error, resolve_names_to_empids, apply_params_file_pre_parse) def parse_args(argv=None): parser = argparse.ArgumentParser(description="CWork 审阅汇报") parser.add_argument("--mode", required=True, choices=["reply", "mark-read", "pending"], help="操作模式") parser.add_argument("--report-id", help="汇报ID（reply/mark-read模式必填）") parser.add_argument("--reply", help="回复内容（reply模式必填）") parser.add_argument( "--content-type", choices=["html", "markdown"], default="markdown", help="回复内容类型：markdown 支持 [@标题](reportId=…&linkType=report) 等内部链接（默认）；html 时包裹为

…

", ) parser.add_argument("--at", help="被@人的姓名（reply模式可选）") parser.add_argument("--page-index", type=int, default=1, help="页码（默认1）") parser.add_argument("--page-size", type=int, default=20, help="每页大小（默认20）") parser.add_argument("--report-type", type=int, choices=[1, 2, 3, 4, 5], help="汇报类型筛选") parser.add_argument("--interactive", action="store_true", help="交互模式") parser.add_argument("--dry-run", action="store_true", help="干跑模式") parser.add_argument("--params-file", dest="params_file", default=None, help="UTF-8 JSON 文件路径，从文件读取参数") return parser.parse_args(argv) def main(): apply_params_file_pre_parse() args = parse_args() if args.dry_run: preview = { "mode": args.mode, "reportId": args.report_id, "reply": args.reply, "contentType": args.content_type, "at": args.at, } print("=== DRY RUN PREVIEW ===", file=sys.stderr) print(json.dumps(preview, ensure_ascii=False, indent=2), file=sys.stderr) output_json({"success": True, "message": "Dry run — no actual API call made"}) return if args.interactive: from cwork_client import interactive_confirm desc = f"{args.mode} 汇报 (report_id={args.report_id})" if not interactive_confirm(f"review_{args.mode}", desc): output_json({"success": True, "message": "Skipped by user"}) return try: client = make_client() except CWorkError as e: output_error(str(e)) try: if args.mode == "reply": if not args.report_id: output_error("--report-id is required for reply mode") if not args.reply: output_error("--reply is required for reply mode") at_emp_ids = None if args.at: at_emp_ids = resolve_names_to_empids(client, [args.at]) if args.content_type == "html": content_body = f"

{args.reply}

" else: content_body = args.reply reply_id = client.reply_report( report_record_id=args.report_id, content_html=content_body, content_type=args.content_type, add_emp_id_list=at_emp_ids, send_msg=True, ) output_json({"success": True, "replyId": reply_id, "message": "Reply submitted successfully"}) elif args.mode == "mark-read": if not args.report_id: output_error("--report-id is required for mark-read mode") client.mark_report_read(args.report_id) output_json({"success": True, "message": f"Report {args.report_id} marked as read"}) elif args.mode == "pending": result = client.get_inbox_list( page_size=args.page_size, page_index=args.page_index, report_record_type=args.report_type, read_status=0, ) output_json({"success": True, "data": result}) except CWorkError as e: output_error(str(e)) except Exception as e: output_error(f"Unexpected error: {e}") if __name__ == "__main__": main() FILE:scripts/cwork-search-emp.py #!/usr/bin/env python3 """ CWork 员工搜索工具用途：根据姓名搜索员工 ID 和详细信息场景：发送汇报前确认接收人、处理待办时确认发件人、创建任务时确认责任人用法： python3 cwork-search-emp.py --name "张" python3 cwork-search-emp.py --name "成伟" --verbose python3 cwork-search-emp.py --name "刘" --max-results 10 """ import argparse import json import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from cwork_client import CWorkClient, CWorkError, make_client, apply_params_file_pre_parse def search_employees(client: CWorkClient, search_key: str, max_results: int = 5, verbose: bool = False) -> dict: """ Search employees by name keyword. Args: client: CWork API client search_key: Search keyword (supports fuzzy matching) max_results: Maximum results to return per category (inside/outside) verbose: Include additional details Returns: { "success": true, "searchKey": "张", "inside": [...], "outside": [...], "totalInside": 10, "totalOutside": 2 } """ try: result = client.search_emp_by_name(search_key) # Parse inside employees inside_list = [] inside_data = result.get("inside", {}) if inside_data: company = inside_data.get("companyVO", {}) emp_list = inside_data.get("empList", []) for emp in emp_list[:max_results]: emp_info = { "empId": emp.get("id"), "name": emp.get("name"), "title": emp.get("title", ""), "mainDept": emp.get("mainDept", ""), "status": "在职" if emp.get("status") == 1 else "离职" } if verbose: emp_info.update({ "personId": emp.get("personId"), "dingUserId": emp.get("dingUserId"), "corpId": emp.get("corpId"), "company": company.get("name", "") }) inside_list.append(emp_info) # Parse outside contacts outside_list = [] outside_data = result.get("outside", []) if outside_data: for item in outside_data: company = item.get("companyVO", {}) emp_list = item.get("empList", []) for emp in emp_list[:max_results]: emp_info = { "empId": emp.get("id"), "name": emp.get("name"), "title": emp.get("title", ""), "mainDept": emp.get("mainDept", ""), "status": "在职" if emp.get("status") == 1 else "离职", "company": company.get("name", "") } outside_list.append(emp_info) return { "success": True, "searchKey": search_key, "inside": inside_list, "outside": outside_list, "totalInside": len(result.get("inside", {}).get("empList", [])) if result.get("inside") else 0, "totalOutside": sum(len(item.get("empList", [])) for item in result.get("outside", [])) if result.get("outside") else 0 } except CWorkError as e: return { "success": False, "error": str(e), "searchKey": search_key } def main(): parser = argparse.ArgumentParser( description="Search CWork employees by name", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Basic search python3 cwork-search-emp.py --name "张" # Verbose mode (includes personId, dingUserId, etc.) python3 cwork-search-emp.py --name "成伟" --verbose # More results python3 cwork-search-emp.py --name "刘" --max-results 10 """ ) parser.add_argument( "--name", "-n", required=True, help="Employee name or keyword to search (supports fuzzy matching)" ) parser.add_argument( "--max-results", "-m", type=int, default=5, help="Maximum results to return per category (default: 5)" ) parser.add_argument( "--verbose", "-v", action="store_true", help="Include additional details (personId, dingUserId, etc.)" ) parser.add_argument( "--output-raw", action="store_true", help="Output raw API response" ) parser.add_argument( "--params-file", help="从 UTF-8 JSON 文件读取参数（用于 Windows 下传递中文内容）" ) apply_params_file_pre_parse() args = parser.parse_args() try: client = make_client() # Output raw response if requested if args.output_raw: result = client.search_emp_by_name(args.name) print(json.dumps(result, ensure_ascii=False, indent=2)) return # Normal search result = search_employees( client, args.name, max_results=args.max_results, verbose=args.verbose ) # Output JSON to stdout print(json.dumps(result, ensure_ascii=False, indent=2)) # Exit with error code if failed if not result.get("success"): sys.exit(1) except CWorkError as e: error_output = { "success": False, "error": str(e), "searchKey": args.name } print(json.dumps(error_output, ensure_ascii=False, indent=2), file=sys.stderr) sys.exit(1) except Exception as e: error_output = { "success": False, "error": f"Unexpected error: {e}", "searchKey": args.name } print(json.dumps(error_output, ensure_ascii=False, indent=2), file=sys.stderr) sys.exit(1) if __name__ == "__main__": main() FILE:scripts/cwork-send-report.py #!/usr/bin/env python3 """ cwork-send-report.py 发送汇报：解析接收人 → 全量保存/更新草稿（5.23）→ 输出完整草稿详情（5.25）供用户确认 → 仅在 --confirm-send 后调用 5.27 将草稿转为正式汇报。更新草稿前会先拉取详情，与本次参数合并后整包提交，避免全量覆盖导致字段丢失。 Usage: # 1) 仅存草稿 + 输出完整预览（默认，不发送） python3 scripts/cwork-send-report.py --title "..." --content-html "

...

" --receivers "张三" # 2) 用户确认后，仅凭汇报 id 发出（5.27） python3 scripts/cwork-send-report.py --draft-id <汇报id> --confirm-send # 3) 一步：保存并发出（需显式确认） python3 scripts/cwork-send-report.py --title "..." --content-html "..." --receivers "张三" --confirm-send Environment: CWORK_APP_KEY (required) CWORK_BASE_URL (optional) """ from __future__ import annotations import sys import json import copy import argparse from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from cwork_client import ( make_client, CWorkError, apply_params_file_pre_parse, flatten_emp_search_bucket, ) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def parse_args(): p = argparse.ArgumentParser(description="Send a CWork report (draft-first, 5.27 to publish)") p.add_argument("--title", "-t", default=None, help="汇报标题（发送-only 模式可不填）") p.add_argument("--content-html", "-c", default=None, help="正文 HTML（发送-only 模式可不填）") p.add_argument( "--receivers", "-r", default="", help="Comma-separated receiver names (will be resolved to empId)", ) p.add_argument( "--cc", dest="cc_names", default="", help="Comma-separated CC recipient names", ) p.add_argument( "--grade", "-g", default="一般", choices=["一般", "紧急"], help="Report urgency", ) p.add_argument( "--type-id", dest="type_id", type=int, default=9999, help="Report type ID (default 9999)", ) p.add_argument( "--file-paths", nargs="*", dest="file_paths", default=[], help="Local file paths to attach (up to 10)", ) p.add_argument( "--file-names", nargs="*", dest="file_names", default=[], help="File names for attachments (same order as --file-paths)", ) p.add_argument( "--plan-id", dest="plan_id", default=None, help="Linked plan/task ID", ) p.add_argument( "--preview-only", dest="preview_only", action="store_true", help="仅保存草稿并输出完整预览（与默认不发送行为一致，便于显式调用）", ) p.add_argument( "--draft-id", dest="draft_id", default=None, help="汇报 id：更新已有草稿；与 --confirm-send 单独使用时仅执行 5.27 发出", ) p.add_argument( "--confirm-send", dest="confirm_send", action="store_true", help="用户已预览完整草稿并同意发出后，再指定此参数才会调用 5.27", ) p.add_argument( "--report-level-json", dest="report_level_json", default=None, help="UTF-8 JSON 文件路径，内容为 reportLevelList 数组（覆盖详情中的流程节点）", ) p.add_argument( "--params-file", dest="params_file", default=None, help="UTF-8 JSON 文件路径，从文件读取参数（用于 Windows 下传递中文内容）", ) return p.parse_args() # --------------------------------------------------------------------------- # Resolve / validate names # --------------------------------------------------------------------------- def resolve_receivers(client, names: list[str]) -> dict: results = {} for name in names: if not name.strip(): continue try: data = client.search_emp_by_name(name.strip()) except CWorkError as e: results[name] = {"status": "error", "message": str(e)} continue inside = flatten_emp_search_bucket(data.get("inside")) if inside: candidates = inside else: candidates = flatten_emp_search_bucket(data.get("outside")) if len(candidates) == 0: results[name] = {"status": "not_found"} elif len(candidates) == 1: emp = candidates[0] results[name] = { "status": "found", "empId": emp["id"], "name": emp["name"], "title": emp.get("title", ""), "dept": emp.get("mainDept", ""), } else: results[name] = { "status": "multiple", "employees": [ {"empId": e["id"], "name": e["name"], "title": e.get("title", ""), "dept": e.get("mainDept", "")} for e in candidates ], } return results def validate_receivers(resolved: dict) -> tuple[list[dict], list[dict]]: confirmed = [] errors = [] for name, info in resolved.items(): if info["status"] == "not_found": errors.append({"name": name, "reason": "not_found"}) elif info["status"] == "multiple": errors.append({ "name": name, "reason": "multiple_matches", "candidates": [ {"empId": e["empId"], "name": e["name"], "title": e.get("title", ""), "dept": e.get("dept", "")} for e in info["employees"] ], }) elif info["status"] == "found": confirmed.append({"empId": info["empId"], "name": info["name"]}) return confirmed, errors # --------------------------------------------------------------------------- # Draft detail → saveOrUpdate 全量字段 # --------------------------------------------------------------------------- def _emp_ids_from_detail(emp_list: list | None) -> list[str]: if not emp_list: return [] out: list[str] = [] for e in emp_list: if isinstance(e, dict) and e.get("id") is not None: out.append(str(e["id"])) return out def _file_vos_from_detail(file_list: list | None) -> list[dict]: if not file_list: return [] out: list[dict] = [] for f in file_list: if not isinstance(f, dict): continue fid = f.get("fileId") if not fid: continue out.append({ "fileId": str(fid), "name": f.get("name") or "", "type": f.get("type") or "file", }) return out def _report_level_param_from_detail(nodes: list | None) -> list[dict] | None: if not nodes: return None result: list[dict] = [] for node in nodes: if not isinstance(node, dict): continue emp_list = node.get("empList") or [] level_users = [] for e in emp_list: if not isinstance(e, dict): continue eid = e.get("empId", e.get("id")) if eid is not None: level_users.append({"empId": eid}) entry = { "type": node.get("type"), "level": node.get("level"), "nodeCode": node.get("nodeCode"), "nodeName": node.get("nodeName"), "levelUserList": level_users, "groupIdList": node.get("groupIdList"), "requirement": node.get("requirement"), } result.append({k: v for k, v in entry.items() if v is not None}) return result or None def _receiver_node_index_for_cli_sync(nodes: list[dict]) -> int: """选择应将 ``--receivers`` 写入 ``levelUserList`` 的节点（与 5.1/5.23 一致：接收人以 reportLevelList 为准）。""" for i, n in enumerate(nodes): if isinstance(n, dict) and isinstance(n.get("type"), str) and n["type"].lower() == "read": return i for i, n in enumerate(nodes): if isinstance(n, dict) and n.get("levelUserList"): return i return 0 def _apply_cli_receivers_to_report_level_list( report_level_list: list[dict], accept_emp_ids: list[str], ) -> list[dict]: """用本次 CLI 解析出的接收人覆盖目标节点 ``levelUserList``，避免仅更新 acceptEmpIdList 时详情仍显示旧人。""" if not report_level_list or not accept_emp_ids: return report_level_list out = copy.deepcopy(report_level_list) idx = _receiver_node_index_for_cli_sync(out) if idx < 0 or idx >= len(out): return report_level_list level_users: list[dict] = [] for eid in accept_emp_ids: try: level_users.append({"empId": int(str(eid))}) except (TypeError, ValueError): level_users.append({"empId": eid}) node = dict(out[idx]) node["levelUserList"] = level_users node.pop("groupIdList", None) out[idx] = {k: v for k, v in node.items() if v is not None} return out def load_report_level_json(path: str) -> list[dict]: raw = Path(path).read_text(encoding="utf-8-sig") data = json.loads(raw) if not isinstance(data, list): raise ValueError("report-level-json 根节点须为 JSON 数组") return data # --------------------------------------------------------------------------- # Upload files # --------------------------------------------------------------------------- def upload_files(client, file_paths: list[str], file_names: list[str]) -> list[dict]: if not file_paths: return [] file_vos = [] for i, path in enumerate(file_paths): fname = file_names[i] if i < len(file_names) else Path(path).name try: result = client.upload_file(path) file_id = result.get("fileId", "") file_vos.append({ "fileId": str(file_id) if file_id is not None else "", "name": fname, "type": "file", }) except CWorkError as e: print(json.dumps({ "step": "upload", "file": fname, "error": str(e), }, ensure_ascii=False), file=sys.stderr) return file_vos # --------------------------------------------------------------------------- # Merge + save draft (5.23 全量) # --------------------------------------------------------------------------- def build_save_draft_kwargs( args, *, detail: dict | None, accept_emp_ids: list[str], cc_emp_ids: list[str], file_vos: list[dict], receiver_names_nonempty: bool, cc_names_nonempty: bool, new_uploads: bool, ) -> dict: """构造 save_draft 的完整参数，避免更新时省略字段导致服务端清空。""" if args.report_level_json: report_level_list = load_report_level_json(args.report_level_json) elif detail is not None: rll = detail.get("reportLevelList") if rll: converted = _report_level_param_from_detail(rll) report_level_list = converted if converted is not None else [] else: report_level_list = [] else: report_level_list = None if detail: if receiver_names_nonempty: final_accept = accept_emp_ids else: final_accept = _emp_ids_from_detail(detail.get("acceptEmployeeList")) if cc_names_nonempty: final_cc = cc_emp_ids else: final_cc = _emp_ids_from_detail(detail.get("copyEmployeeList")) if new_uploads: final_files = file_vos else: final_files = _file_vos_from_detail(detail.get("fileList")) privacy_level = detail.get("privacyLevel") or "非涉密" template_raw = detail.get("templateId") template_id = str(template_raw) if template_raw is not None else None plan_raw = args.plan_id if args.plan_id is not None else detail.get("planId") plan_id = str(plan_raw) if plan_raw is not None else None # CLI 约定正文为 HTML，与 --content-html 一致，避免沿用详情里 markdown 与 HTML 混用 content_type = "html" else: final_accept = accept_emp_ids final_cc = cc_emp_ids final_files = file_vos privacy_level = "非涉密" template_id = None plan_id = str(args.plan_id) if args.plan_id is not None else None content_type = "html" # 5.1/5.23：接收人以 reportLevelList 为准；acceptEmpIdList 仅在 reportLevelList 为空时兜底。 # 更新草稿且用户显式传 --receivers 时，必须把新人写入 reportLevelList，否则详情仍显示旧 empList。 if ( receiver_names_nonempty and not args.report_level_json and isinstance(report_level_list, list) and len(report_level_list) > 0 ): report_level_list = _apply_cli_receivers_to_report_level_list( report_level_list, final_accept ) return { "main": args.title, "content_html": args.content_html, "content_type": content_type, "type_id": args.type_id, "grade": args.grade, "privacy_level": privacy_level, "plan_id": plan_id, "template_id": template_id, "accept_emp_id_list": final_accept, "copy_emp_id_list": final_cc, "report_level_list": report_level_list, "file_vo_list": final_files, "draft_id": args.draft_id, } def save_draft_full(client, kwargs: dict) -> str | None: draft_id = kwargs.pop("draft_id", None) try: result = client.save_draft(draft_id=draft_id, **kwargs) rid = result.get("id") return str(rid) if rid is not None else None except CWorkError as e: print(json.dumps({"step": "save_draft", "error": str(e)}, ensure_ascii=False), file=sys.stderr) return None # --------------------------------------------------------------------------- # Preview output（完整草稿，来自 5.25） # --------------------------------------------------------------------------- def build_preview_shell(args, confirmed: list[dict], cc_confirmed: list[dict], file_vos: list[dict], *, from_api_detail: dict) -> dict: import re html = from_api_detail.get("contentHtml") or "" plain = re.sub(r"<[^>]+>", "", html) plain_stripped = plain.strip() plain_len = len(plain_stripped) # summary.contentPreview：极短正文不截断，避免占位符被「摘要」误伤；仅超长纯文本才截断 preview_cap = 4000 if len(plain) <= preview_cap: content_preview = plain else: content_preview = plain[:preview_cap] + "…" # confirmPrompt 内嵌正文：用纯文本预览，避免重复塞入整段 HTML；短正文全文展示便于审核 prompt_body_cap = 2000 prompt_plain = plain if len(plain) <= prompt_body_cap else plain[:prompt_body_cap] + "…" preview_warnings: list[str] = [] if plain_len < 50: preview_warnings.append( f"正文去标签后仅 {plain_len} 个字符，可能为过短或占位内容，发送前请与用户确认" ) accept_names = [e["name"] for e in confirmed] cc_names = [e["name"] for e in cc_confirmed] summary: dict = { "title": from_api_detail.get("main"), "grade": from_api_detail.get("grade"), "typeId": from_api_detail.get("typeId"), "planId": from_api_detail.get("planId"), "contentType": from_api_detail.get("contentType"), "receiversResolved": accept_names, "ccResolved": cc_names, "attachmentsThisRun": [{"name": f["name"]} for f in file_vos], "contentPlainText": plain, "contentPreview": content_preview, "contentPlainLength": plain_len, } if preview_warnings: summary["previewWarnings"] = preview_warnings warn_prefix = ("⚠ " + preview_warnings[0] + "\n\n") if preview_warnings else "" confirm_prompt = ( warn_prefix + "【请用户确认以下完整草稿后再发送】\n" f"标题：{from_api_detail.get('main')}\n" f"优先级：{from_api_detail.get('grade')}\n" "正文（纯文本预览；完整富文本见 draftDetail.contentHtml）：\n" f"{prompt_plain}\n" f"接收人（解析结果）：{', '.join(accept_names) or '（沿用草稿详情）'}\n" f"抄送：{', '.join(cc_names) or '（沿用草稿详情）'}\n" f"附件：{json.dumps(from_api_detail.get('fileList') or [], ensure_ascii=False)}\n" f"流程节点 reportLevelList：{json.dumps(from_api_detail.get('reportLevelList') or [], ensure_ascii=False)}\n" "\n用户同意后，执行：--draft-id <汇报id> --confirm-send" ) return { "reportId": from_api_detail.get("id"), "note": ( "以下为接口 5.25 返回的完整草稿数据（draftDetail），请向用户展示全文后再发送。" "确认无误后使用 --draft-id <汇报id> --confirm-send。" ), "draftDetail": from_api_detail, "summary": summary, "confirmPrompt": confirm_prompt, } # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): try: sys.stdout.reconfigure(encoding="utf-8") except AttributeError: pass apply_params_file_pre_parse() args = parse_args() send_only = bool(args.confirm_send and args.draft_id and not args.preview_only) if send_only: if args.title is not None or args.content_html is not None: print(json.dumps({ "success": False, "error": "发送-only 模式请勿再传 --title/--content-html；仅需 --draft-id 与 --confirm-send", }, ensure_ascii=False), file=sys.stderr) sys.exit(1) else: if not args.title or args.content_html is None: print(json.dumps({ "success": False, "error": "保存草稿需要 --title 与 --content-html（或使用 --draft-id + --confirm-send 仅发出）", }, ensure_ascii=False), file=sys.stderr) sys.exit(1) client = make_client() if send_only: try: ok = client.submit_draft(args.draft_id) print(json.dumps({ "success": bool(ok), "reportId": args.draft_id, "submitted": bool(ok), "message": "已通过 5.27 将草稿转为正式汇报" if ok else "5.27 返回未成功", }, ensure_ascii=False, indent=2)) sys.exit(0 if ok else 1) except CWorkError as e: print(json.dumps({ "success": False, "error": str(e), "reportId": args.draft_id, }, ensure_ascii=False), file=sys.stderr) sys.exit(1) receiver_names = [n.strip() for n in args.receivers.split(",") if n.strip()] cc_names = [n.strip() for n in args.cc_names.split(",") if n.strip()] receiver_nonempty = bool(receiver_names) cc_nonempty = bool(cc_names) resolved = resolve_receivers(client, receiver_names) confirmed, errors = validate_receivers(resolved) cc_resolved = resolve_receivers(client, cc_names) cc_confirmed, cc_errors = validate_receivers(cc_resolved) all_errors = [] if errors: all_errors.append({"field": "receivers", "errors": errors}) if cc_errors: all_errors.append({"field": "cc", "errors": cc_errors}) if all_errors: print(json.dumps({ "success": False, "step": "validate_names", "message": "部分姓名未能唯一匹配，请确认后重试", "details": all_errors, }, ensure_ascii=False, indent=2)) sys.exit(1) accept_emp_ids = [str(e["empId"]) for e in confirmed] cc_emp_ids = [str(e["empId"]) for e in cc_confirmed] detail: dict | None = None if args.draft_id: try: detail = client.get_draft_detail(args.draft_id) except CWorkError as e: print(json.dumps({ "success": False, "step": "get_draft_detail", "error": str(e), }, ensure_ascii=False), file=sys.stderr) sys.exit(1) file_vos = upload_files(client, args.file_paths, args.file_names) new_uploads = bool(args.file_paths) kwargs = build_save_draft_kwargs( args, detail=detail, accept_emp_ids=accept_emp_ids, cc_emp_ids=cc_emp_ids, file_vos=file_vos, receiver_names_nonempty=receiver_nonempty, cc_names_nonempty=cc_nonempty, new_uploads=new_uploads, ) saved_report_id = save_draft_full(client, kwargs) if not saved_report_id: print(json.dumps({ "success": False, "step": "save_draft", "message": "草稿保存失败", }, ensure_ascii=False, indent=2)) sys.exit(1) try: fresh_detail = client.get_draft_detail(saved_report_id) except CWorkError as e: print(json.dumps({ "success": False, "step": "get_draft_detail", "error": str(e), "reportId": saved_report_id, }, ensure_ascii=False), file=sys.stderr) sys.exit(1) preview = build_preview_shell( args, confirmed, cc_confirmed, file_vos, from_api_detail=fresh_detail, ) preview["success"] = True # draftId：历史 JSON 键名，值为汇报 id，与 reportId / draftDetail.id 相同（非草稿箱记录 id） preview["draftId"] = saved_report_id if not args.confirm_send or args.preview_only: preview["nextStep"] = ( "已向用户展示完整草稿（draftDetail）并确认无误后，再执行：" f"--draft-id {saved_report_id} --confirm-send" ) if args.preview_only and args.confirm_send: preview["noteOnFlags"] = "已指定 --preview-only，不会发出；忽略 --confirm-send" print(json.dumps(preview, ensure_ascii=False, indent=2)) return try: ok = client.submit_draft(saved_report_id) except CWorkError as e: print(json.dumps({ "success": False, "step": "submit_draft_5_27", "error": str(e), "draftId": saved_report_id, }, ensure_ascii=False), file=sys.stderr) sys.exit(1) if not ok: print(json.dumps({ "success": False, "step": "submit_draft_5_27", "draftId": saved_report_id, "message": "5.27 返回未成功", }, ensure_ascii=False, indent=2)) sys.exit(1) print(json.dumps({ "success": True, "reportId": saved_report_id, "submitted": True, "receivers": accept_emp_ids, "cc": cc_emp_ids, "attachmentsThisRun": len(file_vos), "message": "已通过 5.27 将草稿转为正式汇报", }, ensure_ascii=False, indent=2)) if __name__ == "__main__": main() FILE:scripts/cwork-templates.py #!/usr/bin/env python3 """ cwork-templates.py — 模板管理功能： 1. 查询汇报模板列表用法： python3 cwork-templates.py list --limit 50 """ import argparse import json import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from cwork_client import make_client, apply_params_file_pre_parse def list_templates(args): """查询模板列表""" client = make_client() result = client.list_templates( limit=args.limit, begin_time=getattr(args, "begin_time", None), end_time=getattr(args, "end_time", None), ) if args.output_raw: print(json.dumps(result, ensure_ascii=False, indent=2)) return # 结构化输出（API 返回 recentOperateTemplates 或 rows 或直接是列表） if isinstance(result, list): templates = result else: templates = ( result.get("recentOperateTemplates") or result.get("rows") or [] ) output = { "success": True, "action": "list", "total": len(templates), "items": [ { "id": t.get("id") or t.get("templateId"), "name": t.get("name") or t.get("templateName") or t.get("main"), "type": t.get("type"), "typeName": t.get("typeName"), "grade": t.get("grade"), } for t in templates ] } print(json.dumps(output, ensure_ascii=False, indent=2)) def main(): parser = argparse.ArgumentParser( description="CWork 模板管理", formatter_class=argparse.RawDescriptionHelpFormatter ) subparsers = parser.add_subparsers(dest="action", help="操作类型") # list 子命令 list_parser = subparsers.add_parser("list", help="查询模板列表") list_parser.add_argument("--limit", type=int, default=50, help="返回数量限制") list_parser.add_argument("--begin-time", type=int, help="开始时间戳") list_parser.add_argument("--end-time", type=int, help="结束时间戳") list_parser.add_argument("--output-raw", action="store_true", help="输出原始响应") parser.add_argument("--params-file", dest="params_file", default=None, help="UTF-8 JSON 文件路径，从文件读取参数") apply_params_file_pre_parse() args = parser.parse_args() if not args.action: parser.print_help() sys.exit(1) if args.action == "list": list_templates(args) if __name__ == "__main__": main() FILE:scripts/cwork-todo.py #!/usr/bin/env python3 """ cwork-todo.py — 待办管理功能： 1. 查询待办列表 2. 完成待办用法： python3 cwork-todo.py list --page-size 20 --status pending python3 cwork-todo.py complete --todo-id --content "已完成" """ import argparse import json import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from cwork_client import make_client, apply_params_file_pre_parse def list_todos(args): """查询待办列表""" client = make_client() result = client.get_todo_list( page_index=args.page_index, page_size=args.page_size ) if args.output_raw: print(json.dumps(result, ensure_ascii=False, indent=2)) return # 5.15 返回 PageInfo：列表在 ``list``（见开放 API 6.3），非 ``rows`` rows = result.get("list") or result.get("rows") or [] total = result.get("total", len(rows)) output = { "success": True, "action": "list", "total": total, "items": [ { "todoId": item.get("todoId"), "reportId": item.get("reportId"), "id": item.get("todoId"), "title": item.get("main") or item.get("title"), "type": item.get("todoType") or item.get("type"), "status": item.get("status"), "createTime": item.get("createTime"), "creator": item.get("writeEmpName") or item.get("creatorName"), } for item in rows ], } print(json.dumps(output, ensure_ascii=False, indent=2)) def complete_todo(args): """完成待办""" client = make_client() if not args.todo_id: print(json.dumps({ "success": False, "error": "缺少必填参数: --todo-id" }, ensure_ascii=False), file=sys.stderr) sys.exit(1) if not args.content: print(json.dumps({ "success": False, "error": "缺少必填参数: --content" }, ensure_ascii=False), file=sys.stderr) sys.exit(1) if args.dry_run: print(json.dumps({ "success": True, "action": "complete", "dryRun": True, "todoId": args.todo_id, "content": args.content }, ensure_ascii=False, indent=2)) return result = client.complete_todo( todo_id=args.todo_id, content=args.content, operate=args.operate ) output = { "success": True, "action": "complete", "todoId": args.todo_id, "result": result } print(json.dumps(output, ensure_ascii=False, indent=2)) def main(): parser = argparse.ArgumentParser( description="CWork 待办管理", formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument("--params-file", dest="params_file", default=None, help="UTF-8 JSON 文件路径，从文件读取参数") subparsers = parser.add_subparsers(dest="action", help="操作类型") # list 子命令 list_parser = subparsers.add_parser("list", help="查询待办列表") list_parser.add_argument("--page-index", type=int, default=1, help="页码") list_parser.add_argument("--page-size", type=int, default=20, help="每页数量") list_parser.add_argument("--status", type=str, help="状态筛选") list_parser.add_argument("--output-raw", action="store_true", help="输出原始响应") # complete 子命令 complete_parser = subparsers.add_parser("complete", help="完成待办") complete_parser.add_argument("--todo-id", type=str, required=True, help="待办 ID") complete_parser.add_argument("--content", type=str, required=True, help="完成说明") complete_parser.add_argument("--operate", type=str, default=None, choices=["agree", "disagree"], help="决策操作: agree=同意, disagree=不同意（仅决策类待办需要）") complete_parser.add_argument("--dry-run", action="store_true", help="仅预览") apply_params_file_pre_parse() args = parser.parse_args() if not args.action: parser.print_help() sys.exit(1) if args.action == "list": list_todos(args) elif args.action == "complete": complete_todo(args) if __name__ == "__main__": main() FILE:scripts/cwork_client.py #!/usr/bin/env python3 """ CWork API Client — 共享 API 封装由所有编排脚本 import 使用环境变量: CWORK_BASE_URL (default: https://sg-al-cwork-web.mediportal.com.cn) CWORK_APP_KEY (required) """ from __future__ import annotations import os import json import sys import urllib.request import urllib.parse import urllib.error import argparse from datetime import datetime # 在模块加载时强制 stdout/stderr 使用 UTF-8，避免在 LANG=en_US 等环境下 # argparse --help 或任何 print() 输出中文时崩溃 try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") sys.stderr.reconfigure(encoding="utf-8", errors="replace") except AttributeError: pass # --------------------------------------------------------------------------- # HTTP redirect handler — preserves method on 307/308 # --------------------------------------------------------------------------- class _MethodPreservingRedirectHandler(urllib.request.HTTPRedirectHandler): """urllib 默认不跟随 307 的 POST 请求；此 handler 保留原始 method 和 body。""" def redirect_request(self, req, fp, code, msg, headers, newurl): new_headers = {k: v for k, v in req.header_items() if k.lower() not in ("host", "content-length")} return urllib.request.Request( newurl, data=req.data, headers=new_headers, method=req.method, origin_req_host=req.origin_req_host, unverifiable=True, ) def http_error_307(self, req, fp, code, msg, headers): return self.http_error_302(req, fp, code, msg, headers) def http_error_308(self, req, fp, code, msg, headers): return self.http_error_302(req, fp, code, msg, headers) # --------------------------------------------------------------------------- # Exceptions # --------------------------------------------------------------------------- class CWorkError(Exception): """Raised on API errors (non-1 resultCode or HTTP non-OK).""" pass # --------------------------------------------------------------------------- # Client # --------------------------------------------------------------------------- class CWorkClient: BASE_URL = os.environ.get( "CWORK_BASE_URL", "https://sg-al-cwork-web.mediportal.com.cn" ) def __init__(self, app_key: str | None = None): self.app_key = app_key or os.environ.get("CWORK_APP_KEY") if not self.app_key: raise CWorkError("CWORK_APP_KEY environment variable is required") # ---- HTTP helpers ------------------------------------------------------- def _headers(self, json_body: bool = False) -> dict: h = {"appKey": self.app_key} if json_body: h["Content-Type"] = "application/json" return h def _get(self, path: str, params: dict | None = None) -> dict: url = f"{self.BASE_URL}{path}" if params: q = urllib.parse.urlencode({k: v for k, v in params.items() if v is not None}) url = f"{url}?{q}&appKey={self.app_key}" else: url = f"{url}?appKey={self.app_key}" req = urllib.request.Request(url, headers=self._headers(), method="GET") return self._request(req) def _post(self, path: str, body: dict | None = None) -> dict: url = f"{self.BASE_URL}{path}" data = json.dumps(body).encode() if body else None req = urllib.request.Request( url, data=data, headers=self._headers(json_body=body is not None), method="POST", ) return self._request(req) def _request(self, req: urllib.request.Request) -> dict: try: opener = urllib.request.build_opener(_MethodPreservingRedirectHandler) with opener.open(req, timeout=30) as resp: raw = resp.read() charset = resp.headers.get_content_charset() or "utf-8" result = json.loads(raw.decode(charset)) except urllib.error.HTTPError as e: raise CWorkError(f"HTTP {e.code}: {e.reason}") except urllib.error.URLError as e: raise CWorkError(f"URL error: {e.reason}") if result.get("resultCode", -1) != 1: raise CWorkError( f"API Error ({result.get('resultCode')}): {result.get('resultMsg', 'unknown')}" ) return result.get("data", {}) # ------------------------------------------------------------------------- # Employee # ------------------------------------------------------------------------- def search_emp_by_name(self, search_key: str) -> dict: """Returns {inside: {empList:[]}, outside: {empList:[]}}""" return self._get("/open-api/cwork-user/searchEmpByName", {"searchKey": search_key}) # ------------------------------------------------------------------------- # Reports — inbox / outbox / detail # ------------------------------------------------------------------------- def get_inbox_list( self, page_size: int, page_index: int = 1, *, report_record_type: int | None = None, emp_id_list: list[str] | None = None, begin_time: int | None = None, end_time: int | None = None, read_status: int | None = None, order_column: str | None = None, grade: str | None = None, template_id: int | None = None, ) -> dict: """PaginatedResult""" params = { "pageSize": page_size, "pageIndex": page_index, "reportRecordType": report_record_type, "empIdList": emp_id_list, "beginTime": begin_time, "endTime": end_time, "readStatus": read_status, "orderColumn": order_column, "grade": grade, "templateId": template_id, } return self._post("/open-api/work-report/report/record/inbox", params) def get_outbox_list( self, page_size: int, page_index: int = 1, *, report_record_type: int | None = None, emp_id_list: list[str] | None = None, begin_time: int | None = None, end_time: int | None = None, grade: str | None = None, template_id: int | None = None, ) -> dict: params = { "pageSize": page_size, "pageIndex": page_index, "reportRecordType": report_record_type, "empIdList": emp_id_list, "beginTime": begin_time, "endTime": end_time, "grade": grade, "templateId": template_id, } return self._post("/open-api/work-report/report/record/outbox", params) def get_report_info(self, report_id: str) -> dict: """ReportDetail""" return self._get("/open-api/work-report/report/info", {"reportId": report_id}) def get_report_node_detail(self, report_id: str | int) -> dict: """ 获取汇报详情（含节点与处理意见） API: 5.33 /work-report/report/getReportNodeDetail Returns: { "id": 汇报ID, "main": 标题, "content": 正文, "writeEmpId": 汇报人ID, "writeEmpName": 汇报人姓名, "createTime": 发起时间, "nodeList": [ { "nodeName": "建议人", "type": "建议/决策/传阅", "status": "未开始/已完成/进行中/已取消", "level": 1, "userList": [ { "empId": 员工ID, "name": 姓名, "status": "待处理/已处理", "operate": "同意/不同意/建议", "content": "处理意见", "finishTime": "完成时间" } ] } ] } """ return self._get("/open-api/work-report/report/getReportNodeDetail", {"reportId": report_id}) def get_unread_list(self, page_index: int, page_size: int) -> dict: return self._post("/open-api/work-report/reportInfoOpenQuery/unreadList", { "pageIndex": page_index, "pageSize": page_size, }) def is_report_read(self, report_id: str | int, employee_id: str | int) -> bool: return self._get( "/open-api/work-report/reportInfoOpenQuery/isReportRead", {"reportId": str(report_id), "employeeId": str(employee_id)}, ) def mark_report_read(self, report_id: str | int) -> None: self._get( "/open-api/work-report/open-platform/report/readReport", {"reportId": str(report_id)}, ) # ------------------------------------------------------------------------- # Report — reply / submit / remind # ------------------------------------------------------------------------- def submit_report( self, main: str, content_html: str, *, report_id: str | None = None, content_type: str = "html", type_id: int = 9999, grade: str = "一般", privacy_level: str = "非涉密", plan_id: str | None = None, template_id: int | None = None, accept_emp_id_list: list[str] | None = None, copy_emp_id_list: list[str] | None = None, report_level_list: list[dict] | None = None, file_vo_list: list[dict] | None = None, ) -> dict: """Submit a report. Pass report_id to promote an existing draft.""" payload = { "appKey": self.app_key, "main": main, "contentHtml": content_html, "contentType": content_type, "typeId": type_id, "grade": grade, "privacyLevel": privacy_level, "planId": plan_id, "templateId": template_id, "acceptEmpIdList": accept_emp_id_list, "copyEmpIdList": copy_emp_id_list, "reportLevelList": report_level_list, "fileVOList": file_vo_list, } if report_id is not None: payload["id"] = report_id return self._post("/open-api/work-report/report/record/submit", payload) def reply_report( self, report_record_id: str, content_html: str, *, content_type: str = "html", add_emp_id_list: list[str] | None = None, send_msg: bool = True, ) -> int: """Returns reply ID""" return self._post("/open-api/work-report/report/record/reply", { "appKey": self.app_key, "reportRecordId": report_record_id, "contentHtml": content_html, "contentType": content_type, "addEmpIdList": add_emp_id_list, "sendMsg": send_msg, }) # ------------------------------------------------------------------------- # Draft box # ------------------------------------------------------------------------- def save_draft( self, main: str, content_html: str, *, draft_id: str | None = None, id: str | None = None, # noqa: A002 — 兼容误用 save_draft(id=汇报id) 的调用方 content_type: str = "markdown", type_id: int = 9999, grade: str = "一般", privacy_level: str = "非涉密", plan_id: str | None = None, template_id: str | None = None, accept_emp_id_list: list[str] | None = None, copy_emp_id_list: list[str] | None = None, report_level_list: list[dict] | None = None, file_vo_list: list[dict] | None = None, ) -> dict: effective_draft_id = draft_id if draft_id is not None else id if draft_id is not None and id is not None and str(draft_id) != str(id): raise ValueError("save_draft: pass only one of draft_id and id") payload = { "appKey": self.app_key, "main": main, "contentHtml": content_html, "contentType": content_type, "typeId": type_id, "grade": grade, "privacyLevel": privacy_level, "planId": plan_id, "templateId": template_id, "acceptEmpIdList": accept_emp_id_list, "copyEmpIdList": copy_emp_id_list, "reportLevelList": report_level_list, "fileVOList": file_vo_list, } if effective_draft_id is not None: payload["id"] = effective_draft_id return self._post("/open-api/work-report/draftBox/saveOrUpdate", payload) def list_drafts(self, page_index: int, page_size: int) -> dict: """5.24 草稿箱分页。``data.list[]`` 中 ``id`` 为草稿箱记录 id（用于 5.26 删除）；``businessId`` 为汇报 id（``bizType=report`` 时）。""" return self._post("/open-api/work-report/draftBox/listByPage", { "pageIndex": page_index, "pageSize": page_size, }) def get_draft_detail(self, report_record_id: str) -> dict: return self._get(f"/open-api/work-report/draftBox/detail/{report_record_id}") def submit_draft(self, report_id: str) -> bool: """API 5.27: 将草稿转为正式汇报发出。路径参数 id 为汇报 id（与 saveOrUpdate 返回的 id 一致）。""" rid = urllib.parse.quote(str(report_id), safe="") return bool(self._post(f"/open-api/work-report/draftBox/submit/{rid}")) def delete_draft(self, draft_id: str) -> bool: """5.26 删除草稿。路径 ``id`` 必须是 **5.24 列表项的 ``id``**（草稿箱记录主键）。 **不是** ``businessId``，也**不是** 汇报 id（与 5.25/5.27、``--draft-id`` 所用相同的那份汇报 id 不同）。误传汇报 id 时，部分环境仍可能返回 ``data: true``，但草稿仍在列表中。若只有汇报 id，请用 ``delete_draft_by_report_id``。 """ did = urllib.parse.quote(str(draft_id), safe="") return bool(self._post(f"/open-api/work-report/draftBox/delete/{did}")) def delete_draft_by_report_id( self, report_id: str | int, *, page_size: int = 50, max_pages: int = 20, ) -> bool: """按汇报 id 删除草稿：先 5.24 查找 ``bizType=report`` 且 ``businessId`` 匹配的行，再 5.26。未在列表中找到对应行时返回 ``False``（不调用删除接口）。 """ rid = str(report_id) for page in range(1, max_pages + 1): data = self.list_drafts(page_index=page, page_size=page_size) items = data.get("list") or [] for row in items: if not isinstance(row, dict): continue if row.get("bizType") != "report": continue if str(row.get("businessId")) != rid: continue box_id = row.get("id") if box_id is None: continue return self.delete_draft(str(box_id)) if len(items) < page_size: break return False # ------------------------------------------------------------------------- # Tasks # ------------------------------------------------------------------------- def search_task_page( self, page_size: int, page_index: int = 1, *, key_word: str | None = None, status: int | None = None, report_status: int | None = None, emp_id_list: list[str] | None = None, grades: list[str] | None = None, label_list: list[str] | None = None, is_read: int | None = None, ) -> dict: params = { "pageSize": page_size, "pageIndex": page_index, "keyWord": key_word, "status": status, "reportStatus": report_status, "empIdList": emp_id_list, "grades": grades, "labelList": label_list, "isRead": is_read, } return self._post("/open-api/work-report/report/plan/searchPage", params) def get_simple_plan_and_report_info(self, plan_id: str) -> dict: return self._get( "/open-api/work-report/report/plan/getSimplePlanAndReportInfo", {"planId": plan_id}, ) def create_plan( self, main: str, needful: str, target: str, end_time: int, type_id: int = 9999, *, report_emp_id_list: list[str] | None = None, owner_emp_id_list: list[str] | None = None, assist_emp_id_list: list[str] | None = None, supervisor_emp_id_list: list[str] | None = None, copy_emp_id_list: list[str] | None = None, observer_emp_id_list: list[str] | None = None, push_now: bool = True, ) -> str: """Returns plan ID string.""" result = self._post("/open-api/work-report/open-platform/report/plan/create", { "appKey": self.app_key, "main": main, "needful": needful, "target": target, "typeId": type_id, "reportEmpIdList": report_emp_id_list, "ownerEmpIdList": owner_emp_id_list, "assistEmpIdList": assist_emp_id_list, "supervisorEmpIdList": supervisor_emp_id_list, "copyEmpIdList": copy_emp_id_list, "observerEmpIdList": observer_emp_id_list, "pushNow": 1 if push_now else 0, "endTime": end_time, }) return str(result) # ------------------------------------------------------------------------- # Todo / feedback # ------------------------------------------------------------------------- def list_created_feedbacks(self, emp_id: str | None = None) -> list: """API 5.12: GET, optional empId filter.""" params = {} if emp_id is not None: params["empId"] = emp_id return self._get( "/open-api/work-report/todoTask/listCreatedFeedbacks", params ) def get_todo_list(self, page_index: int, page_size: int, *, status: str | None = None) -> dict: """5.15 待办列表。成功时 ``data`` 为 PageInfo（``total`` + ``list``，见开放 API 6.3 / 6.18）。""" params = {"pageIndex": page_index, "pageSize": page_size} if status: params["status"] = status return self._post("/open-api/work-report/reportInfoOpenQuery/todoList", params) def complete_todo( self, todo_id: str, content: str, *, content_type: str = "html", operate: str | None = None, ) -> bool: payload: dict = { "appKey": self.app_key, "todoId": todo_id, "content": content, "contentType": content_type, } if operate is not None: payload["operate"] = operate return self._post("/open-api/work-report/open-platform/todo/completeTodo", payload) # ------------------------------------------------------------------------- # File upload # ------------------------------------------------------------------------- def upload_file(self, file_path: str) -> dict: import mimetypes boundary = "----FormBoundary7MA4YWxkTrZu0gW" filename = os.path.basename(file_path) mime_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream" with open(file_path, "rb") as f: file_data = f.read() body_parts = [ f"--{boundary}\r\n".encode(), f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n'.encode(), f"Content-Type: {mime_type}\r\n\r\n".encode(), file_data, f"\r\n--{boundary}--\r\n".encode(), ] body = b"".join(body_parts) req = urllib.request.Request( f"{self.BASE_URL}/open-api/cwork-file/uploadWholeFile", data=body, headers={ "appKey": self.app_key, "Content-Type": f"multipart/form-data; boundary={boundary}", }, method="POST", ) result = self._request(req) if isinstance(result, str): return {"fileId": result} return result # ------------------------------------------------------------------------- # Templates # ------------------------------------------------------------------------- def list_templates( self, begin_time: int | None = None, end_time: int | None = None, limit: int | None = None ) -> dict: return self._post("/open-api/work-report/template/listTemplates", { "appKey": self.app_key, "beginTime": begin_time, "endTime": end_time, "limit": limit, }) # ------------------------------------------------------------------------- # History context retrieval (for approval decision support) # ------------------------------------------------------------------------- def get_sender_history( self, sender_emp_id: str | int, days: int = 90, max_count: int = 20 ) -> dict: """ Get sender's historical reports (for approval context) Args: sender_emp_id: Sender employee ID days: Days to look back (default 90) max_count: Max results (default 20) Returns: { "senderEmpId": "xxx", "totalReports": 15, "recentReports": [...] } """ import time end_time = int(time.time() * 1000) begin_time = end_time - (days * 24 * 60 * 60 * 1000) # Query inbox for sender's reports inbox = self.get_inbox_list( page_size=max_count, emp_id_list=[str(sender_emp_id)], begin_time=begin_time, end_time=end_time ) return { "senderEmpId": str(sender_emp_id), "totalReports": inbox.get("total", 0), "recentReports": inbox.get("list", [])[:max_count] } def search_reports_by_keyword( self, keyword: str, days: int = 90, max_count: int = 100 ) -> dict: """ Search reports by keyword (client-side filtering) Args: keyword: Search keyword days: Days to look back (default 90) max_count: Max results to fetch (default 100) Returns: { "keyword": "公章", "total": 5, "reports": [...] } """ import time end_time = int(time.time() * 1000) begin_time = end_time - (days * 24 * 60 * 60 * 1000) # Fetch inbox inbox = self.get_inbox_list( page_size=max_count, begin_time=begin_time, end_time=end_time ) # Client-side filtering all_reports = inbox.get("list", []) matched = [ r for r in all_reports if keyword in r.get("main", "") or keyword in r.get("content", "") ] return { "keyword": keyword, "total": len(matched), "reports": matched[:20] # Return top 20 } # --------------------------------------------------------------------------- # Convenience factory # --------------------------------------------------------------------------- def make_client() -> CWorkClient: app_key = os.environ.get("CWORK_APP_KEY") if not app_key: raise CWorkError( "CWORK_APP_KEY environment variable is not set. " "Run: export CWORK_APP_KEY='your-key'" ) return CWorkClient(app_key) # --------------------------------------------------------------------------- # CLI argument helpers # --------------------------------------------------------------------------- def flatten_emp_search_bucket(raw) -> list: """Normalize ``inside`` / ``outside`` from searchEmpByName to a flat emp dict list. API may return a dict with ``empList``, a list of group dicts (each with ``empList``), or a flat list of employee records. """ if raw is None: return [] if isinstance(raw, dict): emp_list = raw.get("empList") return list(emp_list) if isinstance(emp_list, list) else [] if isinstance(raw, list): out: list = [] for item in raw: if not isinstance(item, dict): continue nested = item.get("empList") if isinstance(nested, list) and nested: out.extend(nested) elif item.get("id") is not None or item.get("empId") is not None or item.get("empid") is not None: out.append(item) return out return [] def parse_deadline(value: str | None) -> int | None: """Parse deadline string to milliseconds timestamp.""" if value is None: return None try: return int(value) except ValueError: pass try: dt = datetime.strptime(value, "%Y-%m-%d") return int(dt.timestamp() * 1000) except ValueError: pass raise argparse.ArgumentTypeError(f"Invalid deadline format: {value}. Use YYYY-MM-DD or milliseconds.") def resolve_names_to_empids(client: CWorkClient, names: list[str]) -> list[str]: """Resolve a list of names to empIds via search API. Priority: internal employees (inside) > external contacts (outside). If inside has any matches, outside is ignored entirely. Raises CWorkError if any name is not found or matches more than one employee within the same category (inside or outside). """ empids = [] for name in names: result = client.search_emp_by_name(name) inside_list = flatten_emp_search_bucket(result.get("inside")) # Only fall back to outside when inside has no match at all if inside_list: all_emps = inside_list else: all_emps = flatten_emp_search_bucket(result.get("outside")) if not all_emps: raise CWorkError( f'未找到姓名为"{name}"的员工，请确认姓名或直接提供员工 ID' ) if len(all_emps) > 1: candidates = [ { "empId": e.get("id") or e.get("empId") or e.get("empid"), "name": e.get("name", ""), "title": e.get("title", ""), "dept": e.get("mainDept", ""), } for e in all_emps ] raise CWorkError( f'"{name}" 匹配到多名员工，请指定唯一员工后重试：' + json.dumps(candidates, ensure_ascii=False) ) empids.append( all_emps[0].get("id") or all_emps[0].get("empId") or all_emps[0].get("empid") ) return empids def apply_params_file(args) -> None: """[Deprecated: use apply_params_file_pre_parse() instead] Post-parse merge — cannot satisfy required argparse args from file. """ apply_params_file_pre_parse() def apply_params_file_pre_parse() -> None: """Pre-scan sys.argv for --params-file, load JSON, and inject missing flags back into sys.argv BEFORE argparse.parse_args() is called. This allows required arguments (e.g. --mode) to be provided from a file, which is the primary workaround for Windows PowerShell encoding issues when passing Chinese content on the command line. Call this at the very start of main(), before parse_args(): def main(): apply_params_file_pre_parse() args = parse_args() ... File format example (params.json, UTF-8): { "mode": "inbox", "content": "本周工作进展", "receivers": "张三,李四" } CLI args always take precedence over file values. """ # Step 1: find --params-file in sys.argv without a full parse params_file = None argv = sys.argv[1:] for i, arg in enumerate(argv): if arg == "--params-file" and i + 1 < len(argv): params_file = argv[i + 1] break if arg.startswith("--params-file="): params_file = arg.split("=", 1)[1] break if not params_file: return # Step 2: load the JSON file try: # utf-8-sig strips the UTF-8 BOM that PowerShell Out-File adds by default with open(params_file, "r", encoding="utf-8-sig") as f: file_params = json.load(f) except (OSError, json.JSONDecodeError) as exc: print( json.dumps({"success": False, "error": f"--params-file: {exc}"}, ensure_ascii=False), file=sys.stderr, ) sys.exit(1) # Step 3: build set of flags already present in sys.argv (CLI wins) existing_flags: set[str] = set() for arg in sys.argv[1:]: if arg.startswith("--"): existing_flags.add(arg.split("=")[0]) # Step 4: append missing flags to sys.argv extra: list[str] = [] for key, value in file_params.items(): flag = f"--{key}" if flag in existing_flags: continue if isinstance(value, bool): if value: extra.append(flag) elif isinstance(value, list): for v in value: extra.extend([flag, str(v)]) else: extra.extend([flag, str(value)]) if extra: sys.argv.extend(extra) def _write_utf8(text: str, stream=None) -> None: """Write text as UTF-8 to stream's binary buffer, falling back to print.""" target = stream or sys.stdout try: target.buffer.write((text + "\n").encode("utf-8")) target.buffer.flush() except AttributeError: print(text, file=target) def output_json(data: dict) -> None: """Output JSON to stdout (UTF-8, regardless of terminal codepage).""" _write_utf8(json.dumps(data, ensure_ascii=False, indent=2)) def output_error(message: str) -> None: """Output error JSON to stderr (UTF-8) and exit.""" _write_utf8( json.dumps({"success": False, "message": message}, ensure_ascii=False, indent=2), sys.stderr, ) sys.exit(1) def interactive_confirm(step_name: str, description: str) -> bool: """Print step description and wait for 'confirm' from stdin.""" print(f"\n[STEP] {step_name}: {description}", file=sys.stderr) print("Type 'confirm' to proceed (or press Enter to skip): ", file=sys.stderr, end="") sys.stderr.flush() try: user_input = input().strip().lower() return user_input == "confirm" except (EOFError, KeyboardInterrupt): return False FILE:version.json { "skillcode": "cms-cwork-workflow", "version": "1.0.5" }

Cwork Skill Update

Comments (0)