---
name: wecom
description: 企业微信集成。通过自然语言发送消息、管理群机器人、操作审批流程、管理通讯录。当用户提到企业微信、微信工作、群机器人、企业号、wecom相关任务时自动激活。
---

# 企业微信集成技能

## 功能概述

### 消息推送
- **应用消息**: 向指定用户/部门发送文本、图片、文件等消息
- **群机器人**: 通过 Webhook 向群聊发送消息
- **模板消息**: 发送结构化的卡片消息

### 通讯录管理
- **部门管理**: 查询、创建、更新部门
- **成员管理**: 查询、创建、更新成员信息

### 审批流程
- **发起审批**: 通过 API 发起审批申请
- **审批状态**: 查询审批单状态

---

## 环境配置

### 已配置的企业微信应用

| 配置项 | 值 |
|--------|-----|
| 企业ID (CorpID) | `ww8ab927306fa235d2` |
| 应用ID (AgentId) | `1000003` |
| 可信域名 | `wecom.pipexerp.com` |

### 环境变量

```bash
# ~/.zshrc 已配置
export WECOM_CORP_ID="ww8ab927306fa235d2"
export WECOM_AGENT_ID="1000003"
export WECOM_SECRET="Dts8BmENzjxCRK1Qn4qmkO6mU81FLVEkhI2LitcBcjI"
```

---

## API 基础

### 获取 Access Token

```python
import os
import requests

CORP_ID = os.environ.get("WECOM_CORP_ID")
CORP_SECRET = os.environ.get("WECOM_SECRET")
AGENT_ID = os.environ.get("WECOM_AGENT_ID")

def get_access_token():
    """获取企业微信 access_token（有效期 7200 秒）"""
    url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
    params = {
        "corpid": CORP_ID,
        "corpsecret": CORP_SECRET
    }
    resp = requests.get(url, params=params)
    result = resp.json()

    if result.get("errcode") == 0:
        return result["access_token"]
    else:
        raise Exception(f"获取 token 失败: {result}")
```

---

## 消息推送

### 发送文本消息

```python
def send_text(user_id: str, content: str):
    """发送文本消息给指定用户"""
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"

    data = {
        "touser": user_id,  # 用 "@all" 发送给所有人
        "msgtype": "text",
        "agentid": int(AGENT_ID),
        "text": {"content": content}
    }

    resp = requests.post(url, json=data)
    return resp.json()

# 使用示例
send_text("@all", "这是一条测试消息")
```

### 发送 Markdown 消息

```python
def send_markdown(user_id: str, content: str):
    """发送 Markdown 格式消息"""
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"

    data = {
        "touser": user_id,
        "msgtype": "markdown",
        "agentid": int(AGENT_ID),
        "markdown": {"content": content}
    }

    resp = requests.post(url, json=data)
    return resp.json()
```

### 发送卡片消息

```python
def send_card(user_id: str, title: str, description: str, url: str):
    """发送文本卡片消息"""
    token = get_access_token()
    api_url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"

    data = {
        "touser": user_id,
        "msgtype": "textcard",
        "agentid": int(AGENT_ID),
        "textcard": {
            "title": title,
            "description": description,
            "url": url,
            "btntxt": "详情"
        }
    }

    resp = requests.post(api_url, json=data)
    return resp.json()
```

---

## 群机器人 Webhook

```python
def send_to_group(webhook_url: str, content: str, mentioned_list: list = None):
    """通过 Webhook 发送群消息"""
    data = {
        "msgtype": "text",
        "text": {
            "content": content,
            "mentioned_list": mentioned_list or []
        }
    }

    resp = requests.post(webhook_url, json=data)
    return resp.json()

# 使用示例
WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxx"
send_to_group(WEBHOOK_URL, "自动化任务完成通知", ["@all"])
```

---

## 通讯录管理

### 获取部门列表

```python
def get_departments():
    """获取部门列表"""
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/department/list?access_token={token}"
    resp = requests.get(url)
    return resp.json()
```

### 获取部门成员

```python
def get_department_users(department_id: int):
    """获取部门成员列表"""
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/user/simplelist"
    params = {"access_token": token, "department_id": department_id}
    resp = requests.get(url, params=params)
    return resp.json()
```

---

## 云文档操作

### 样式规范

企业微信云文档支持两种样式方案，根据文档类型选择：

| 文档类型 | 推荐方案 | 说明 |
|----------|----------|------|
| 合同、协议 | 带标题样式 | 使用 `create_contract_doc()` |
| 报告、记录 | 简单文本 | 使用 `create_simple_doc()` |
| 会议纪要 | 简单文本 | 使用 `create_simple_doc()` |

#### 简单文本样式符号规范

```
文档标题                        ← 由文档名称体现
━━━━━━━━━━━━━━━━━━━━━━━━━━━━    ← 分隔线
【章节标题】                    ← 用【】标记一级章节
内容正文...

【子章节】                      ← 同样用【】
1.1 条款内容
1.2 条款内容
```

#### 带标题样式（合同类）

- heading_level=1: 文档主标题（如"物流服务合同"）
- heading_level=2: 章节标题（如"第一条"、"甲方"）
- heading_level=0: 正文内容

---

### 创建文档

```python
def create_doc(doc_name: str, doc_type: int = 3):
    """
    创建企业微信文档

    Args:
        doc_name: 文档名称
        doc_type: 文档类型 (3=文档, 4=表格)

    Returns:
        dict: {"docid": "xxx", "url": "https://doc.weixin.qq.com/..."}
    """
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/create_doc?access_token={token}"

    resp = requests.post(url, json={
        "doc_type": doc_type,
        "doc_name": doc_name
    })
    return resp.json()

# 使用示例
result = create_doc("项目文档")
print(f"文档链接: {result['url']}")
```

### 获取文档内容

```python
def get_doc_content(docid: str):
    """获取文档内容"""
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/document/get?access_token={token}"

    resp = requests.post(url, json={"docid": docid})
    return resp.json()
```

### 编辑文档内容

```python
def update_doc(docid: str, text: str, index: int = 0):
    """
    更新文档内容

    Args:
        docid: 文档ID
        text: 要插入的文本内容
        index: 插入位置 (0=开头)
    """
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/document/batch_update?access_token={token}"

    resp = requests.post(url, json={
        "docid": docid,
        "requests": [
            {
                "insert_text": {
                    "text": text,
                    "location": {"index": index}
                }
            }
        ]
    })
    return resp.json()

# 使用示例
update_doc("DOCID", "# 标题\n\n这是正文内容")
```

### 获取文档列表

```python
def list_docs(limit: int = 20):
    """获取文档列表"""
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/doc_list?access_token={token}"

    resp = requests.post(url, json={"limit": limit})
    return resp.json()
```

### 上传图片到文档

```python
import base64

def upload_doc_image(docid: str, image_path: str):
    """
    上传图片到文档（获取图片 URL）

    Args:
        docid: 文档ID
        image_path: 本地图片路径

    Returns:
        dict: {"url": "https://wdcdn.qpic.cn/...", "width": 1280, "height": 800}
    """
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/image_upload?access_token={token}"

    with open(image_path, "rb") as f:
        img_base64 = base64.b64encode(f.read()).decode()

    resp = requests.post(url, json={
        "docid": docid,
        "base64_content": img_base64
    })
    return resp.json()

# 使用示例
result = upload_doc_image("DOCID", "/tmp/screenshot.png")
print(f"图片 URL: {result['url']}")
```

### 插入图片到文档

```python
def insert_image_to_doc(docid: str, image_url: str, index: int = 1):
    """
    插入图片到文档

    Args:
        docid: 文档ID
        image_url: 图片 URL（从 upload_doc_image 获取）
        index: 插入位置

    注意：使用 insert_paragraph + elements + image 格式
    """
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/document/batch_update?access_token={token}"

    resp = requests.post(url, json={
        "docid": docid,
        "requests": [
            {
                "insert_paragraph": {
                    "elements": [
                        {"image": {"url": image_url}}
                    ],
                    "location": {"index": index}
                }
            }
        ]
    })
    return resp.json()

# 使用示例
insert_image_to_doc("DOCID", "https://wdcdn.qpic.cn/xxx", index=1)
```

### 获取文档末尾位置

```python
def get_doc_end_index(docid: str) -> int:
    """获取文档末尾索引（用于追加内容）"""
    doc = get_doc_content(docid)

    if doc.get("errcode") != 0:
        return 1

    body = doc.get("document", {}).get("body", {})
    blocks = body.get("blocks", [])

    end_index = 1
    for block in blocks:
        if "paragraph" in block:
            for elem in block["paragraph"].get("elements", []):
                if "text_run" in elem:
                    end_index = max(end_index, elem["text_run"].get("end_index", 1))

    return end_index
```

### 完整示例：截图并插入文档

```python
import base64
from playwright.sync_api import sync_playwright

def screenshot_and_insert(url: str, docid: str, title: str = "网页截图"):
    """
    截取网页并插入到文档

    Args:
        url: 要截取的网页 URL
        docid: 目标文档 ID
        title: 截图标题
    """
    token = get_access_token()

    # 1. 截取网页
    print(f"截取网页: {url}")
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page(viewport={"width": 1280, "height": 800})
        page.goto(url, wait_until="networkidle")
        page.screenshot(path="/tmp/screenshot.png")
        browser.close()

    # 2. 上传图片
    print("上传图片...")
    with open("/tmp/screenshot.png", "rb") as f:
        img_base64 = base64.b64encode(f.read()).decode()

    upload_resp = requests.post(
        f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/image_upload?access_token={token}",
        json={"docid": docid, "base64_content": img_base64}
    ).json()

    img_url = upload_resp.get("url")

    # 3. 获取文档末尾位置
    end_index = get_doc_end_index(docid)

    # 4. 插入标题
    requests.post(
        f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/document/batch_update?access_token={token}",
        json={
            "docid": docid,
            "requests": [
                {"insert_text": {"text": f"\n\n{title}\n\n", "location": {"index": end_index}}}
            ]
        }
    )

    # 5. 插入图片
    new_end = get_doc_end_index(docid)
    insert_resp = requests.post(
        f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/document/batch_update?access_token={token}",
        json={
            "docid": docid,
            "requests": [
                {
                    "insert_paragraph": {
                        "elements": [{"image": {"url": img_url}}],
                        "location": {"index": new_end}
                    }
                }
            ]
        }
    ).json()

    if insert_resp.get("errcode") == 0:
        print("✅ 截图已插入文档!")
    else:
        print(f"❌ 插入失败: {insert_resp}")

# 使用示例
screenshot_and_insert("https://www.baidu.com", "YOUR_DOCID", "百度首页截图")
```

---

## 通用函数封装

### 简单文本文档

适用于报告、记录、会议纪要等。使用符号标记章节。

```python
def create_simple_doc(doc_name: str, content: str) -> dict:
    """
    创建简单文本文档

    Args:
        doc_name: 文档名称
        content: 文档内容（使用【】标记章节）

    Returns:
        dict: {"docid": "xxx", "url": "https://..."}

    Example:
        content = '''项目周报

━━━━━━━━━━━━━━━━━━━━━━━━━━━━

【本周完成】
1. 完成用户模块开发
2. 修复登录bug

【下周计划】
1. 开始订单模块
2. 编写测试用例
'''
        result = create_simple_doc("2026年第5周周报", content)
    """
    token = get_access_token()

    # 1. 创建文档
    create_url = f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/create_doc?access_token={token}"
    create_resp = requests.post(create_url, json={
        "doc_type": 3,
        "doc_name": doc_name
    }).json()

    if create_resp.get("errcode") != 0:
        return create_resp

    docid = create_resp["docid"]
    doc_url = create_resp["url"]

    # 2. 写入内容
    update_url = f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/document/batch_update?access_token={token}"
    update_resp = requests.post(update_url, json={
        "docid": docid,
        "requests": [{
            "insert_text": {
                "text": content,
                "location": {"index": 0}
            }
        }]
    }).json()

    return {
        "errcode": update_resp.get("errcode", 0),
        "docid": docid,
        "url": doc_url
    }
```

### 合同类文档（带标题样式）

适用于合同、协议等需要标题层级的正式文档。

```python
import time

def create_contract_doc(doc_name: str, sections: list) -> dict:
    """
    创建带标题样式的合同文档

    Args:
        doc_name: 文档名称
        sections: 内容列表，格式 [(文本, 标题级别), ...]
                  标题级别: 1=主标题, 2=章节标题, 0=正文

    Returns:
        dict: {"docid": "xxx", "url": "https://..."}

    Example:
        sections = [
            ("物流服务合同", 1),
            ("合同编号: WL-2026-0201", 0),
            ("", 0),
            ("甲方（托运方）", 2),
            ("公司名称: xxx公司", 0),
            ("", 0),
            ("第一条 服务内容", 2),
            ("1.1 服务类型：货物运输", 0),
            ("1.2 货物类型：xxx", 0),
        ]
        result = create_contract_doc("物流服务合同-甲方与乙方", sections)
    """
    token = get_access_token()

    # 1. 创建文档
    create_url = f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/create_doc?access_token={token}"
    create_resp = requests.post(create_url, json={
        "doc_type": 3,
        "doc_name": doc_name
    }).json()

    if create_resp.get("errcode") != 0:
        return create_resp

    docid = create_resp["docid"]
    doc_url = create_resp["url"]

    # 2. 倒序内容（因为每次在位置0插入）
    reversed_sections = list(reversed(sections))

    # 3. 分批插入（每批最多25个，留余量）
    update_url = f"https://qyapi.weixin.qq.com/cgi-bin/wedoc/document/batch_update?access_token={token}"
    batch_size = 25

    for i in range(0, len(reversed_sections), batch_size):
        batch = reversed_sections[i:i+batch_size]
        requests_list = []

        for text, level in batch:
            para = {"elements": [{"text_run": {"content": text}}]}
            if level > 0:
                para["paragraph_style"] = {"heading_level": level}

            requests_list.append({
                "insert_paragraph": {
                    "location": {"index": 0},
                    "paragraph": para
                }
            })

        resp = requests.post(update_url, json={
            "docid": docid,
            "requests": requests_list
        }).json()

        if resp.get("errcode") != 0:
            return {
                "errcode": resp.get("errcode"),
                "errmsg": resp.get("errmsg"),
                "docid": docid,
                "url": doc_url
            }

        time.sleep(0.3)  # 避免频率限制

    return {
        "errcode": 0,
        "docid": docid,
        "url": doc_url
    }


def build_contract_sections(
    title: str,
    party_a: dict,
    party_b: dict,
    clauses: list,
    signature: bool = True
) -> list:
    """
    构建合同内容结构

    Args:
        title: 合同标题
        party_a: 甲方信息 {"name": "", "code": "", "address": "", "phone": "", "legal_rep": ""}
        party_b: 乙方信息，格式同上
        clauses: 条款列表 [{"title": "第一条 xxx", "items": ["1.1 xxx", "1.2 xxx"]}, ...]
        signature: 是否包含签章栏

    Returns:
        list: 可直接传给 create_contract_doc 的 sections
    """
    sections = []

    # 标题
    sections.append((title, 1))
    sections.append(("", 0))

    # 甲方
    sections.append(("甲方", 2))
    if party_a.get("name"):
        sections.append((f"公司名称: {party_a['name']}", 0))
    if party_a.get("code"):
        sections.append((f"统一社会信用代码: {party_a['code']}", 0))
    if party_a.get("address"):
        sections.append((f"地址: {party_a['address']}", 0))
    if party_a.get("phone"):
        sections.append((f"联系电话: {party_a['phone']}", 0))
    if party_a.get("legal_rep"):
        sections.append((f"法定代表人: {party_a['legal_rep']}", 0))
    sections.append(("", 0))

    # 乙方
    sections.append(("乙方", 2))
    if party_b.get("name"):
        sections.append((f"公司名称: {party_b['name']}", 0))
    if party_b.get("code"):
        sections.append((f"统一社会信用代码: {party_b['code']}", 0))
    if party_b.get("address"):
        sections.append((f"地址: {party_b['address']}", 0))
    if party_b.get("phone"):
        sections.append((f"联系电话: {party_b['phone']}", 0))
    if party_b.get("legal_rep"):
        sections.append((f"法定代表人: {party_b['legal_rep']}", 0))
    sections.append(("", 0))

    # 条款
    for clause in clauses:
        sections.append((clause["title"], 2))
        for item in clause.get("items", []):
            sections.append((item, 0))
        sections.append(("", 0))

    # 签章
    if signature:
        sections.append(("签章", 2))
        sections.append(("", 0))
        sections.append((f"甲方（盖章）: {party_a.get('name', '')}", 0))
        sections.append(("法定代表人/授权代表：________________", 0))
        sections.append(("日期：________________", 0))
        sections.append(("", 0))
        sections.append((f"乙方（盖章）: {party_b.get('name', '')}", 0))
        sections.append(("法定代表人/授权代表：________________", 0))
        sections.append(("日期：________________", 0))

    return sections
```

### 使用示例

#### 简化示例

```python
# 简化示例：使用占位符
party_a = {"name": "甲方公司名称", "code": "甲方税号"}
party_b = {"name": "乙方公司名称", "code": "乙方税号"}
clauses = [
    {"title": "第一条 服务内容", "items": ["1.1 xxx", "1.2 xxx"]},
    {"title": "第二条 服务期限", "items": ["2.1 xxx"]},
]

sections = build_contract_sections("合同标题", party_a, party_b, clauses)
result = create_contract_doc("合同文档名称", sections)
```

#### 完整示例

```python
# 完整示例：创建物流合同
party_a = {
    "name": "重庆妗晨工贸有限公司",
    "code": "91500104MA7EJTPA6D",
    "address": "重庆市大渡口区跳磴镇海康路106号1-1",
    "phone": "15213397998"
}

party_b = {
    "name": "北京名风新能源科技有限公司",
    "code": "91110106092440790K",
    "legal_rep": "魏小健"
}

clauses = [
    {
        "title": "第一条 服务内容",
        "items": [
            "1.1 服务类型：货物运输配送服务",
            "1.2 货物类型：今麦郎系列产品",
        ]
    },
    {
        "title": "第二条 服务期限",
        "items": [
            "2.1 合同有效期：自 2026年2月1日 至 2027年1月31日",
        ]
    },
    {
        "title": "第三条 运费标准",
        "items": [
            "3.1 运费标准：今麦郎产品 2.50 元/件",
            "3.2 结算周期：月结",
        ]
    }
]

# 构建并创建文档
sections = build_contract_sections("物流服务合同", party_a, party_b, clauses)
result = create_contract_doc("物流服务合同-妗晨与名风", sections)
print(f"文档链接: {result['url']}")
```

---

## 常见错误码

| 错误码 | 说明 | 解决方案 |
|--------|------|----------|
| 40001 | access_token 无效 | 重新获取 token |
| 40058 | 请求参数错误 | 检查 API 参数格式 |
| 48002 | API 未授权 | 在应用设置中开启对应 API 权限 |
| 60011 | 用户不存在 | 检查 userid 是否正确 |
| 60020 | IP 不在白名单 | 在应用设置中添加可信 IP |
| 81013 | 缺少通讯录权限 | 在应用权限中开启通讯录读取权限 |
| 44001 | 文档不存在 | 检查 docid 是否正确 |
| 2050065 | 插入位置无效 | 使用 get_doc_end_index 获取正确位置 |
| 2400001 | 请求参数错误 | 检查 insert_paragraph 格式 |
| 93017 | JSON 格式错误 | 图片上传需要使用 base64_content |

---

## 相关资源

| 资源 | 链接 |
|------|------|
| API 文档 | https://developer.work.weixin.qq.com/document/ |
| 调试工具 | https://developer.work.weixin.qq.com/devtool/interface |
| 管理后台 | https://work.weixin.qq.com/wework_admin/ |