Metadata-Version: 2.4
Name: py-wecom-toolkit
Version: 3.0.10
Summary: 企业微信 Python 扩展包 - 支持 Webhook 与 Server 端 API
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: Documentation, https://gitee.com/guolei19850528/py_wecom_toolkit#readme
Keywords: wecom,wechat,work,webhook,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
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: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: python-dotenv>=1.0.0; extra == "dev"
Dynamic: license-file

# py-wecom-toolkit

企业微信 Python 扩展包 - 支持 Webhook 与 Server 端 API。

- **作者**: 郭磊
- **邮箱**: 174000902@qq.com
- **项目主页**: https://gitee.com/guolei19850528/py_wecom_toolkit
- **PyPI**: https://pypi.org/project/py-wecom-toolkit/

## 安装

### 基础安装（推荐使用 uv）

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

# 或使用 pip 安装
pip install py-wecom-toolkit

# 或从本地源码安装（开发模式）
uv pip install -e .
```

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

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

# 或从本地源码安装（开发模式，带缓存依赖）
uv pip install -e ".[cache]"
```

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

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

# 或从本地源码安装（开发模式）
uv pip install -e ".[dev]"
```

## 快速开始

### Webhook 消息发送（同步）

```python
from py_wecom_toolkit.webhook.client import WeComWebhookClient

# 创建同步客户端，支持上下文管理器自动管理连接池
with WeComWebhookClient(key="你的webhook_key") as client:

    # 发送文本消息
    resp = client.send_text("Hello, World!")
    print(f"发送结果: {resp.errmsg}")

    # 发送文本消息并 @某人
    resp = client.send_text(
        "你好, 请查看",
        mentioned_list=["zhangsan"],
        mentioned_mobile_list=["13800001111"],
    )
    print(f"发送结果: {resp.errmsg}")

    # 发送 Markdown 消息
    resp = client.send_markdown(
        '实时新增用户反馈<font color="warning">132例</font>'
    )
    print(f"发送结果: {resp.errmsg}")

    # 发送图文消息
    resp = client.send_news([{
        "title": "中秋节礼品领取",
        "description": "今年中秋节公司有豪礼相送",
        "url": "https://work.weixin.qq.com",
        "picurl": "https://example.com/pic.png",
    }])
    print(f"发送结果: {resp.errmsg}")

    # 发送图片
    resp = client.send_image("/path/to/image.png")
    print(f"发送结果: {resp.errmsg}")

    # 上传文件并发送
    media_id = client.upload_media("/path/to/file.pdf")
    resp = client.send_file(media_id)
    print(f"发送结果: {resp.errmsg}")
```

### Webhook 消息发送（异步）

```python
import asyncio
from py_wecom_toolkit.webhook.client import AsyncWeComWebhookClient

async def main():
    async with AsyncWeComWebhookClient(key="你的webhook_key") as client:

        # 发送文本消息（方法名与同步客户端完全一致）
        resp = await client.send_text("Hello, World!")
        print(f"发送结果: {resp.errmsg}")

        # 发送 Markdown 消息
        resp = await client.send_markdown(
            '实时新增用户反馈<font color="warning">132例</font>'
        )
        print(f"发送结果: {resp.errmsg}")

        # 上传文件并发送
        media_id = await client.upload_media("/path/to/file.pdf")
        resp = await client.send_file(media_id)
        print(f"发送结果: {resp.errmsg}")

asyncio.run(main())
```

### Server 端 API（同步）

```python
from py_wecom_toolkit.server.client import WeComServerClient
from py_wecom_toolkit.server.cache import WeComServerDiskCache

# 关键配置请通过环境变量或配置文件读取，不要硬编码在代码中
client = WeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
    cache=WeComServerDiskCache(directory=".runtime.cache.diskcache"),  # 默认使用 diskcache 缓存 access_token
)

# 获取 access_token（自动缓存）
token = client.get_access_token()
print(f"access_token: {token}")

# 获取企业微信接口 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}")

# 发送文本应用消息
resp = client.send_text(
    content="Hello, WeCom!",
    touser="Guolei",
)
print(f"发送结果: {resp.errmsg}")

# 上传素材并发送文件消息
media = client.upload_media("/path/to/README.md", media_type="file")
resp = client.send_file(media.media_id, touser="Guolei")
print(f"发送文件结果: {resp.errmsg}")

client.close()
```

### Server 端 API（异步）

```python
import asyncio
from py_wecom_toolkit.server.client import AsyncWeComServerClient
from py_wecom_toolkit.server.cache import WeComServerDiskCache

async def main():
    async with AsyncWeComServerClient(
        corpid="你的CorpID",
        corpsecret="你的CorpSecret",
        agentid="你的AgentID",
        cache=WeComServerDiskCache(directory=".runtime.cache.diskcache"),
    ) as client:

        # 获取 access_token（自动缓存，方法名与同步客户端一致）
        token = await client.get_access_token()
        print(f"access_token: {token}")

        # 获取接口 IP 段
        iplist = await client.get_iplist()
        print(f"接口 IP 段: {iplist.ip_list}")

        # 发送文本应用消息
        resp = await client.send_text(
            content="Hello, WeCom Async!",
            touser="Guolei",
        )
        print(f"发送结果: {resp.errmsg}")

        # 上传素材并发送文件消息
        media = await client.upload_media("/path/to/README.md", media_type="file")
        resp = await client.send_file(media.media_id, touser="Guolei")
        print(f"发送文件结果: {resp.errmsg}")

asyncio.run(main())
```

## API 参考

### Webhook 客户端

#### `WeComWebhookClient`（同步）

| 方法 | 说明 |
|------|------|
| `send(message)` | 发送任意类型的消息 |
| `send_text(content, ...)` | 发送文本消息 |
| `send_markdown(content)` | 发送 Markdown 消息 |
| `send_markdown_v2(content)` | 发送 Markdown V2 消息 |
| `send_image(path)` | 发送图片消息 |
| `send_news(articles)` | 发送图文消息 |
| `send_file(media_id)` | 发送文件消息 |
| `send_voice(media_id)` | 发送语音消息 |
| `send_template_card(card)` | 发送模版卡片消息 |
| `upload_media(path, type)` | 上传媒体文件，返回 media_id |

#### `AsyncWeComWebhookClient`（异步）

异步方法名与同步客户端完全一致（不带 `async_` 前缀），支持 `async with` 上下文管理器。

| 方法 | 说明 |
|------|------|
| `send(message)` | 发送任意类型的消息 |
| `send_text(content, ...)` | 发送文本消息 |
| `send_markdown(content)` | 发送 Markdown 消息 |
| `send_markdown_v2(content)` | 发送 Markdown V2 消息 |
| `send_image(path)` | 发送图片消息 |
| `send_news(articles)` | 发送图文消息 |
| `send_file(media_id)` | 发送文件消息 |
| `send_voice(media_id)` | 发送语音消息 |
| `send_template_card(card)` | 发送模版卡片消息 |
| `upload_media(path, type)` | 上传媒体文件，返回 media_id |

### Server 端客户端

#### `WeComServerClient`（同步）

| 方法 | 说明 |
|------|------|
| `get_access_token()` | 获取 access_token（自动缓存） |
| `clear_access_token_cache()` | 清除 access_token 缓存 |
| `get_iplist()` | 获取企业微信接口 IP 段 |
| `get_callback_ip()` | 获取企业微信回调 IP 段 |
| `send_text(content, touser, ...)` | 发送文本应用消息 |
| `send_news(articles, ...)` | 发送图文应用消息 |
| `send_file(media_id, touser, ...)` | 发送文件应用消息 |
| `send_voice(media_id, touser, ...)` | 发送语音应用消息 |
| `send_image(media_id, touser, ...)` | 发送图片应用消息 |
| `send_message(request, ...)` | 发送任意类型应用消息 |
| `upload_media(path, type)` | 上传临时素材，返回 media_id |
| `download_media(media_id)` | 下载临时素材，返回字节内容 |

#### `AsyncWeComServerClient`（异步）

异步方法名与同步客户端完全一致（不带 `async_` 前缀），支持 `async with` 上下文管理器。

| 方法 | 说明 |
|------|------|
| `get_access_token()` | 获取 access_token（自动缓存） |
| `clear_access_token_cache()` | 清除 access_token 缓存 |
| `get_iplist()` | 获取企业微信接口 IP 段 |
| `get_callback_ip()` | 获取企业微信回调 IP 段 |
| `send_text(content, touser, ...)` | 发送文本应用消息 |
| `send_news(articles, ...)` | 发送图文应用消息 |
| `send_file(media_id, touser, ...)` | 发送文件应用消息 |
| `send_voice(media_id, touser, ...)` | 发送语音应用消息 |
| `send_image(media_id, touser, ...)` | 发送图片应用消息 |
| `send_message(request, ...)` | 发送任意类型应用消息 |
| `upload_media(path, type)` | 上传临时素材，返回 media_id |
| `download_media(media_id)` | 下载临时素材，返回字节内容 |

### 缓存后端

| 类 | 说明 |
|----|------|
| `WeComServerMemoryCache` | 进程内内存缓存（默认，无需额外依赖） |
| `WeComServerDiskCache` | 基于 diskcache 的持久化缓存（需安装 `[cache]` 依赖） |
| `WeComServerRedisCache` | 基于 redis 的分布式缓存（需安装 `[cache]` 依赖） |

### 异常

| 类 | 说明 |
|----|------|
| `WeComWebhookError` | Webhook 基础异常 |
| `WeComWebhookAPIError` | Webhook API 返回错误 |
| `WeComWebhookRateLimitError` | 触发频率限制（errcode=45009） |
| `WeComWebhookValidationError` | 参数验证错误 |
| `WeComWebhookNetworkError` | 网络请求错误 |
| `WeComServerError` | Server 基础异常 |
| `WeComServerAPIError` | Server API 返回错误 |
| `WeComServerNetworkError` | 网络请求错误 |
| `WeComServerConfigError` | 配置错误 |

## 测试

测试用例位于 `tests/` 目录，基于 pytest + pytest-asyncio，统一从 `.env` 读取配置。

```bash
# 安装测试依赖
uv pip install -e ".[dev]"

# 运行全部测试
pytest tests/ -q

# 仅运行 Webhook 测试
pytest tests/webhook/ -q

# 仅运行 Server 端测试
pytest tests/server/ -q
```

未配置相关环境变量时，测试会自动跳过。

## 项目结构

```
py-wecom-toolkit/
├── pyproject.toml              # 项目构建配置
├── README.md                   # 项目说明文档
├── LICENSE                     # MIT 许可证
├── 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 示例
    └── server/                 # Server 端示例
```

## 消息类型参考

参考 [企业微信文档](https://developer.work.weixin.qq.com/document/path/91770) 获取 Webhook 消息格式说明。

参考 [企业微信服务端 API](https://developer.work.weixin.qq.com/document/path/90664) 获取 Server 端 API 说明。

## License

MIT
