Metadata-Version: 2.4
Name: py-wecom-toolkit
Version: 3.0.14
Summary: 企业微信 Python 扩展包 / WeCom Python Extension - Webhook 群机器人与 Server 端 API 客户端库 / WeCom group robot and Server API client library
Author-email: 郭磊 <174000902@qq.com>
License: MIT License
        
        Copyright (c) 2024 py-wecom-toolkit
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Project-URL: Homepage, https://gitee.com/guolei19850528/py_wecom_toolkit
Project-URL: Repository, https://gitee.com/guolei19850528/py_wecom_toolkit
Project-URL: Package, https://pypi.org/project/py-wecom-toolkit/
Keywords: wecom,wechat,enterprise-wechat,work-wechat,企业微信,企业微信机器人,webhook,group-robot,server-api,pydantic,access-token,async,sync,enterprise
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Topic :: Communications :: Chat
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.24.0
Requires-Dist: pydantic>=2.0.0
Provides-Extra: cache
Requires-Dist: diskcache>=5.0.0; extra == "cache"
Requires-Dist: redis>=5.0.0; extra == "cache"
Provides-Extra: dev
Requires-Dist: ruff>=0.6.0; extra == "dev"
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0; extra == "dev"
Requires-Dist: python-dotenv>=1.0.0; extra == "dev"
Provides-Extra: docs
Requires-Dist: mkdocs>=1.0; extra == "docs"
Requires-Dist: mkdocs-material>=9.0; extra == "docs"
Dynamic: license-file

# py-wecom-toolkit

[![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)
[![PyPI](https://img.shields.io/badge/pypi-py--wecom--toolkit-blue.svg)](https://pypi.org/project/py-wecom-toolkit/)
[![Python](https://img.shields.io/badge/python-3.8%2B-blue.svg)](https://www.python.org/)
[![WeCom](https://img.shields.io/badge/WeCom-Enterprise%20WeChat-green.svg)](https://developer.work.weixin.qq.com/)

## 文档语言

- [中文 (README.md)](README.md)
- [English (README.en.md)](README.en.md)

## 介绍

**py-wecom-toolkit** 是一个企业微信 Python 扩展包，支持 **Webhook（群机器人）** 与 **Server 端 API**。提供同步/异步双模客户端、基于 Pydantic v2 的消息模型，以及自动缓存的 access_token 管理。

py-wecom-toolkit 让你极其简单地与企业微信 API 交互。无需手动拼接 URL，无需手动构建 JSON 负载，自动缓存 access_token，支持同步和异步客户端。

### 核心特性

- **双模客户端**：同时提供同步（`WeComWebhookClient` / `WeComServerClient`）与异步（`AsyncWeComWebhookClient` / `AsyncWeComServerClient`）客户端，方法名完全一致，支持 `with` / `async with` 上下文管理器。
- **Webhook 群机器人**：支持文本、Markdown、Markdown V2、图片、图文、文件、语音、模版卡片消息发送，以及媒体文件上传。
- **Server 端 API**：支持 access_token 自动获取与缓存、接口/回调 IP 段查询、应用消息发送（文本/Markdown/图片/语音/视频/文件/文本卡片/图文/mpnews/markdown_card）、素材上传/下载。
- **自动缓存 access_token**：支持内存（默认）、diskcache、Redis 三种缓存后端，自动处理 token 过期与刷新（提前 5 分钟视为过期）。
- **Pydantic v2 消息模型**：所有消息载荷基于 Pydantic v2 构建，字段自动校验，类型安全。
- **基于 httpx**：底层 HTTP 客户端使用 httpx，支持同步与异步请求。

### 参考文档

- [企业微信 Webhook 接口文档](https://developer.work.weixin.qq.com/document/path/91770)
- [企业微信 Server 端 API 文档](https://developer.work.weixin.qq.com/document/path/90664)
- [access_token 接口文档](https://developer.work.weixin.qq.com/document/path/90128)
- [应用消息发送接口文档](https://developer.work.weixin.qq.com/document/path/90236)
- [素材管理接口文档](https://developer.work.weixin.qq.com/document/path/91054)

## 软件架构

```
py-wecom-toolkit/
├── pyproject.toml              # 项目构建配置（PEP 621 / setuptools）
├── README.md                   # 项目说明文档（中文）
├── README.en.md                # 项目说明文档（英文）
├── LICENSE                     # MIT 许可证
├── uv.lock                     # 依赖锁定文件
├── src/
│   └── py_wecom_toolkit/       # 源代码包
│       ├── __init__.py         # 包入口，导出主要客户端类
│       ├── py.typed            # PEP 561 类型标记
│       ├── webhook/            # Webhook 消息发送模块（群机器人）
│       │   ├── __init__.py
│       │   ├── client.py       # WeComWebhookClient / AsyncWeComWebhookClient / WeComWebhookClientConfig
│       │   ├── models.py       # Pydantic v2 消息模型与异常定义
│       │   └── types.py        # 类型别名
│       └── server/             # Server 端 API 模块（企业微信服务端接口）
│           ├── __init__.py
│           ├── client.py       # WeComServerClient / AsyncWeComServerClient / WeComServerClientConfig
│           ├── token.py        # WeComServerAccessTokenManager / AsyncWeComServerAccessTokenManager
│           ├── cache.py        # WeComServerMemoryCache / WeComServerDiskCache / WeComServerRedisCache
│           ├── models.py       # Server API 响应模型与异常定义
│           ├── message.py      # 应用消息发送（WeComServerMessageSendMixin / AsyncWeComServerMessageSendMixin）
│           └── media.py        # 素材上传/下载（WeComServerMediaMixin / AsyncWeComServerMediaMixin）
├── tests/                      # 测试用例
│   ├── webhook/                # Webhook 测试（按消息类型组织）
│   └── server/                 # Server 端测试（按功能模块组织）
└── examples/                   # 演示代码
    ├── webhook/                # Webhook 示例（文本/Markdown/图片/图文/文件/语音/模版卡片，同步+异步）
    └── server/                 # Server 端示例（文本/token/素材/IP 段，同步+异步）
```

## 安装教程

推荐使用 [uv](https://github.com/astral-sh/uv) 进行安装：

### 1. 基础安装（推荐 uv）

```bash
uv pip install py-wecom-toolkit
```

### 2. 带缓存依赖安装（diskcache / redis）

```bash
uv pip install py-wecom-toolkit[cache]
```

### 3. 开发环境安装（包含测试、代码检查工具）

```bash
uv pip install py-wecom-toolkit[dev]
```

### 4. 从源码安装

```bash
git clone https://gitee.com/guolei19850528/py_wecom_toolkit.git
cd py_wecom_toolkit
uv pip install -e .
```

## 使用说明

### 一、Webhook 群机器人（Webhook）

Webhook 适用于企业微信群机器人场景，只需配置 Webhook Key 即可发送消息到群聊。

#### 1. 创建客户端

```python
from py_wecom_toolkit import WeComWebhookClient, AsyncWeComWebhookClient

# 同步客户端
client = WeComWebhookClient(key="你的webhook_key")

# 异步客户端
async_client = AsyncWeComWebhookClient(key="你的webhook_key")
```

推荐使用上下文管理器，自动管理 HTTP 连接池：

```python
with WeComWebhookClient(key="你的webhook_key") as client:
    # 使用 client 发送消息
    pass

async with AsyncWeComWebhookClient(key="你的webhook_key") as client:
    # 使用 client 发送消息
    pass
```

#### 2. 发送文本消息（同步 / 异步）

同步：

```python
from py_wecom_toolkit import WeComWebhookClient

with WeComWebhookClient(key="你的webhook_key") as client:
    resp = client.send_text("Hello, World!", mentioned_list=["@all"])
    print(f"发送结果: {resp.errmsg}")
```

异步：

```python
import asyncio
from py_wecom_toolkit import AsyncWeComWebhookClient

async def main():
    async with AsyncWeComWebhookClient(key="你的webhook_key") as client:
        resp = await client.send_text("Hello, World!", mentioned_list=["@all"])
        print(f"发送结果: {resp.errmsg}")

asyncio.run(main())
```

#### 3. 发送 Markdown 消息

```python
with WeComWebhookClient(key="你的webhook_key") as client:
    content = """### 标题示例

- 列表项一
- 列表项二
**加粗文本**"""
    resp = client.send_markdown(content)
```

Markdown V2：

```python
resp = client.send_markdown_v2("# 标题\n\n正文内容")
```

#### 4. 发送图片消息

```python
with WeComWebhookClient(key="你的webhook_key") as client:
    resp = client.send_image("path/to/image.png")
```

#### 5. 发送图文消息（最多 8 条）

```python
with WeComWebhookClient(key="你的webhook_key") as client:
    articles = [
        {
            "title": "标题 1",
            "description": "描述 1",
            "url": "https://example.com",
        },
        {
            "title": "标题 2",
            "description": "描述 2",
            "url": "https://example.com",
            "picurl": "https://example.com/image.png",
        },
    ]
    resp = client.send_news(articles)
```

#### 6. 发送文件消息（先上传获取 media_id）

```python
with WeComWebhookClient(key="你的webhook_key") as client:
    media_id = client.upload_media("path/to/file.pdf", media_type="file")
    resp = client.send_file(media_id)
```

#### 7. 发送语音消息（先上传获取 media_id）

```python
with WeComWebhookClient(key="你的webhook_key") as client:
    media_id = client.upload_media("path/to/voice.amr", media_type="voice")
    resp = client.send_voice(media_id)
```

#### 8. 发送模版卡片消息

文本通知卡片（text_notice）：

```python
with WeComWebhookClient(key="你的webhook_key") as client:
    card = {
        "card_type": "text_notice",
        "source": {"desc": "py_wecom_toolkit"},
        "main_title": {"title": "卡片标题", "desc": "标题辅助信息"},
        "emphasis_content": {"title": "关键数据", "desc": "数据描述"},
        "sub_title_text": "二级普通文本",
        "card_action": {"type": 1, "url": "https://example.com"},
    }
    resp = client.send_template_card(card)
```

图文展示卡片（news_notice）：

```python
card = {
    "card_type": "news_notice",
    "source": {"desc": "py_wecom_toolkit"},
    "main_title": {"title": "卡片标题"},
    "card_image": {"url": "https://example.com/image.png"},
    "image_text_area": {
        "type": 1,
        "url": "https://example.com",
        "title": "左图右文标题",
        "desc": "左图右文描述",
        "image_url": "https://example.com/image.png",
    },
    "card_action": {"type": 1, "url": "https://example.com"},
}
resp = client.send_template_card(card)
```

#### 9. 默认 @ 提醒（可选）

在初始化客户端时可设置默认的 `mentioned_list` 和 `mentioned_mobile_list`，每次发送文本消息时自动合并：

```python
client = WeComWebhookClient(
    key="你的webhook_key",
    mentioned_list=["@all"],
    mentioned_mobile_list=["13800000000"],
)
client.send_text("这条消息会自动 @ 所有人")
```

---

### 二、Server 端 API（企业微信服务端接口）

Server 端适用于企业微信应用开发，需要配置 CorpID、CorpSecret 和 AgentID。

#### 1. 创建客户端

```python
from py_wecom_toolkit import WeComServerClient, AsyncWeComServerClient
from py_wecom_toolkit.server.cache import WeComServerDiskCache

# 同步客户端（使用磁盘缓存）
client = WeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
    cache=WeComServerDiskCache(directory=".runtime.cache.diskcache"),
)

# 异步客户端（使用内存缓存，默认）
from py_wecom_toolkit.server.cache import WeComServerMemoryCache
async_client = AsyncWeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
    cache=WeComServerMemoryCache(),
)
```

#### 2. 获取 access_token（自动缓存）

```python
with WeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
) as client:
    token = client.get_access_token()
    print(f"access_token: {token}")

    # 二次获取将命中缓存
    token_again = client.get_access_token()
    print(f"缓存命中: {token == token_again}")

    # 清除缓存
    client.clear_access_token_cache()
```

异步：

```python
async with AsyncWeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
) as client:
    token = await client.get_access_token()
    print(f"access_token: {token}")
```

#### 3. 查询 IP 段

```python
with WeComServerClient(corpid="你的CorpID", corpsecret="你的CorpSecret") as client:
    # 接口 IP 段
    iplist = client.get_iplist()
    print(f"接口 IP: {iplist.ip_list}")

    # 回调 IP 段
    callback_ip = client.get_callback_ip()
    print(f"回调 IP: {callback_ip.ip_list}")
```

#### 4. 发送应用消息

发送文本消息：

```python
with WeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
) as client:
    resp = client.send_text(
        content="Hello, WeCom!",
        touser="@all",
    )
    print(f"发送结果: {resp.errmsg}")
```

异步：

```python
async with AsyncWeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
) as client:
    resp = await client.send_text(
        content="Hello, WeCom!",
        touser="@all",
    )
    print(f"发送结果: {resp.errmsg}")
```

发送 Markdown 消息：

```python
resp = client.send_markdown(
    content="### Markdown 消息\n\n**加粗文本**",
    touser="@all",
)
```

发送图片消息（需先上传素材获取 media_id）：

```python
# 上传素材
media_resp = client.upload_media("path/to/image.png", media_type="image")
media_id = media_resp.media_id

# 发送图片消息
resp = client.send_image(
    media_id=media_id,
    touser="@all",
)
```

发送文件消息：

```python
media_resp = client.upload_media("path/to/file.pdf", media_type="file")
resp = client.send_file(
    media_id=media_resp.media_id,
    touser="@all",
)
```

发送语音消息：

```python
media_resp = client.upload_media("path/to/voice.amr", media_type="voice")
resp = client.send_voice(
    media_id=media_resp.media_id,
    touser="@all",
)
```

发送视频消息：

```python
media_resp = client.upload_media("path/to/video.mp4", media_type="video")
resp = client.send_video(
    media_id=media_resp.media_id,
    title="视频标题",
    description="视频描述",
    touser="@all",
)
```

发送文本卡片消息：

```python
resp = client.send_textcard(
    title="卡片标题",
    description="卡片描述",
    url="https://example.com",
    btntxt="详情",
    touser="@all",
)
```

发送图文消息（news）：

```python
resp = client.send_news(
    articles=[
        {
            "title": "文章标题 1",
            "description": "文章描述 1",
            "url": "https://example.com/1",
            "picurl": "https://example.com/1.png",
        },
    ],
    touser="@all",
)
```

发送 mpnews 消息：

```python
resp = client.send_mpnews(
    articles=[
        {
            "title": "mpnews 标题",
            "thumb_media_id": "你的缩略图 media_id",
            "author": "作者",
            "content_source_url": "https://example.com",
            "content": "<h1>正文内容</h1>",
            "digest": "摘要",
        },
    ],
    touser="@all",
)
```

发送 markdown_card 消息：

```python
resp = client.send_markdown_card(
    title="卡片标题",
    content="### Markdown 内容\n\n**加粗**",
    touser="@all",
)
```

#### 5. 素材上传 / 下载

上传临时素材：

```python
resp = client.upload_media(
    file_path="path/to/image.png",
    media_type="image",  # image / voice / video / file
)
print(f"media_id: {resp.media_id}")
```

下载临时素材：

```python
data = client.download_media(media_id="你的media_id")
with open("downloaded_file", "wb") as f:
    f.write(data)
```

#### 6. 指定 access_token（跳过自动获取）

如果你已有 access_token，可直接传入，跳过自动获取：

```python
client = WeComServerClient(
    corpid="你的CorpID",
    access_token="你的access_token",  # 直接指定，不再自动获取
    agentid="你的AgentID",
)
resp = client.send_text(content="Hello", touser="@all")
```

---

### 三、异常处理

所有 API 调用失败时会抛出相应异常：

```python
from py_wecom_toolkit import (
    WeComWebhookAPIError,
    WeComWebhookRateLimitError,
    WeComWebhookValidationError,
    WeComWebhookNetworkError,
)
from py_wecom_toolkit.server.models import (
    WeComServerAPIError,
    WeComServerNetworkError,
    WeComServerConfigError,
)

try:
    resp = client.send_text("Hello")
except WeComWebhookRateLimitError as e:
    print(f"触发频率限制: {e}")
except WeComWebhookAPIError as e:
    print(f"API 错误: [{e.errcode}] {e.errmsg}")
except WeComWebhookNetworkError as e:
    print(f"网络错误: {e}")
```

---

### 四、贡献指南

1. Fork 本仓库
2. 新建 `Feat_xxx` 分支
3. 提交代码
4. 新建 Pull Request

### 五、许可证

本项目采用 [MIT 许可证](LICENSE)。
