Metadata-Version: 2.4
Name: py-wecom-toolkit
Version: 1.0.11
Summary: 企业微信 SDK，提供企业微信 API 的 Python 封装，支持同步和异步调用。
Author-email: Guolei <174000902@qq.com>
Maintainer-email: Guolei <174000902@qq.com>
License: MIT License
        
        Copyright (c) 2026 郭磊
        
        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.git
Project-URL: Documentation, https://gitee.com/guolei19850528/py_wecom_toolkit
Keywords: wecom,python,client,api,企业微信,异步调用,server,消息推送,文件上传,webhook,机器人
Classifier: License :: OSI Approved :: MIT License
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
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: Operating System :: OS Independent
Classifier: Topic :: Communications
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pydantic>=2.0
Requires-Dist: diskcache>=5.6.3
Requires-Dist: redis>=4.6.0
Requires-Dist: py-httpx-toolkit>=1.0.1
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: setuptools>=61.0; extra == "dev"
Requires-Dist: twine>=4.0; extra == "dev"
Requires-Dist: build>=1.0; extra == "dev"
Dynamic: license-file

# py-wecom-toolkit

企业微信 SDK 工具包，提供企业微信 Webhook 和 Server API 的 Python 封装，支持同步和异步调用方式。

### 项目信息

- **项目地址**: https://gitee.com/guolei19850528/py_wecom_toolkit
- **作者**: Guo Lei
- **邮箱**: 174000902@qq.com
- **许可证**: MIT License

## 功能特性

- **Webhook 模块**: 企业微信机器人消息发送和媒体上传
- **Server 模块**: 企业微信服务端 API，包含 access_token 管理、消息发送、素材管理
- **支持同步/异步**: 所有 API 均提供同步和异步两种调用方式
- **缓存支持**: access_token 支持 diskcache 和 Redis 缓存
- **类型安全**: 使用 Pydantic 进行数据验证和类型提示

## 项目结构

```
src/py_wecom_toolkit/
├── __init__.py              # 主模块入口，包含版本信息
├── server/                  # Server API 模块
│   ├── __init__.py          # Server 基础类（access_token管理、IP查询）
│   ├── models/              # 响应模型
│   │   ├── __init__.py
│   │   └── response.py      # Base、Success 响应模型
│   ├── materials/           # 素材管理
│   │   ├── __init__.py
│   │   └── media.py         # Media 类（临时素材上传、永久图片上传）
│   └── messages/            # 消息发送
│       ├── __init__.py      # Message 类（消息发送）
│       └── model.py         # 消息类型模型（Text、Markdown、Image 等）
└── webhook/                 # Webhook 模块
    ├── __init__.py          # Webhook 客户端类
    ├── models/              # 消息和响应模型
    │   ├── __init__.py
    │   ├── msgtype.py       # 消息类型模型
    │   └── response.py      # 响应模型
    └── util.py              # 工具函数（图片转 Base64 和 MD5）
```

## 安装

```bash
pip install py-wecom-toolkit
```
## UV 安装
```bash
uv add py-wecom-toolkit
```

## 快速开始

### Webhook 使用示例

```python
from py_wecom_toolkit.webhook import Webhook
from py_wecom_toolkit.webhook.models import msgtype

# 初始化 Webhook
webhook = Webhook(key="your_webhook_key")

# 发送文本消息
webhook.send(msgtype.Text(text={"content": "Hello, World!"}))

# 发送 Markdown 消息
webhook.send(msgtype.Markdown(markdown={"content": "# 标题\n\n**粗体文本**"}))

# 发送图片消息（需先转换为 Base64 和 MD5）
from py_wecom_toolkit.webhook.util import image_to_base64_and_md5
base64_str, md5_str = image_to_base64_and_md5("image.jpg")
webhook.send(msgtype.Image(image={"base64": base64_str, "md5": md5_str}))

# 异步发送消息
import asyncio
asyncio.run(webhook.async_send(msgtype.Text(text={"content": "Async Message"})))

# 上传媒体文件
response = webhook.upload_media(ft="file", files={"media": open("test.pdf", "rb")})
media_id = response.json().get("media_id")

# 发送文件消息
webhook.send(msgtype.File(file={"media_id": media_id}))
```

### Server 使用示例

```python
from py_wecom_toolkit.server import Server
from py_wecom_toolkit.server.messages import Message
from py_wecom_toolkit.server.messages.model import Text
from py_wecom_toolkit.server.materials.media import Media

# 初始化 Server（基础类）
server = Server(
    corpid="your_corpid",
    corpsecret="your_corpsecret",
    agentid="your_agentid"
)

# 刷新 access_token（带缓存）
server.refresh_access_token()

# 使用 Message 类发送消息
message = Message(
    corpid="your_corpid",
    corpsecret="your_corpsecret",
    agentid="your_agentid"
)
message.refresh_access_token()
response = message.send(Text(text={"content": "Hello from Server API"}))

# 使用 Media 类上传素材
media = Media(
    corpid="your_corpid",
    corpsecret="your_corpsecret",
    agentid="your_agentid"
)
media.refresh_access_token()

# 上传临时素材
response = media.upload(ft="image", files={"media": open("test.jpg", "rb")})
media_id = response.json().get("media_id")

# 上传永久图片
response = media.uploadimg(files={"media": open("test.jpg", "rb")})
image_url = response.json().get("url")

# 异步调用示例
import asyncio
async def main():
    await message.async_refresh_access_token()
    await message.async_send(Text(text={"content": "Async Server Message"}))
asyncio.run(main())
```

## 官方文档

| 模块 | 官方文档链接 |
|------|-------------|
| Webhook API | [企业微信机器人配置说明](https://developer.work.weixin.qq.com/document/path/91770) |
| Server API | [企业微信服务端API文档](https://developer.work.weixin.qq.com/document/path/90664) |
| 获取 access_token | [获取access_token](https://developer.work.weixin.qq.com/document/path/91039) |
| 发送消息 | [发送应用消息](https://developer.work.weixin.qq.com/document/path/90236) |
| 素材管理 | [素材管理接口](https://developer.work.weixin.qq.com/document/path/90253) |
| IP 白名单 | [获取API域名IP](https://developer.work.weixin.qq.com/document/path/90964) |

## 核心模块说明

### Webhook 模块

**Webhook 类** (`webhook/__init__.py`)

| 方法 | 说明 |
|------|------|
| `send(msgtype_inst)` | 发送消息 |
| `upload_media(ft, files)` | 上传媒体文件 |
| `async_send(msgtype_inst)` | 异步发送消息 |
| `async_upload_media(ft, files)` | 异步上传媒体文件 |

**支持的消息类型**:
- Text: 文本消息（支持 @ 指定用户）
- Markdown: Markdown 消息
- MarkdownV2: MarkdownV2 消息
- Image: 图片消息
- News: 图文消息
- File: 文件消息
- Voice: 语音消息
- TemplateCard: 模板卡片消息

### Server 模块

**Server 类** (`server/__init__.py`)

| 方法 | 说明 |
|------|------|
| `gettoken()` | 获取 access_token |
| `refresh_access_token()` | 刷新 access_token（带缓存） |
| `get_api_domain_ip()` | 获取 API 域名 IP 列表 |
| `getcallbackip()` | 获取回调 IP 列表 |
| `request_with_access_token()` | 发送带 access_token 的请求 |

**Message 类** (`server/messages/__init__.py`)

| 方法 | 说明 |
|------|------|
| `send(msgtype_inst)` | 发送消息到企业微信 |
| `async_send(msgtype_inst)` | 异步发送消息 |

**Media 类** (`server/materials/media.py`)

| 方法 | 说明 |
|------|------|
| `upload(ft, files)` | 上传临时素材（image/video/voice/file） |
| `uploadimg(files)` | 上传永久图片 |
| `async_upload(ft, files)` | 异步上传临时素材 |
| `async_uploadimg(files)` | 异步上传永久图片 |

### 工具函数

**Webhook 工具** (`webhook/util.py`)

| 函数 | 说明 |
|------|------|
| `image_to_base64_and_md5(image_path)` | 将图片转换为 Base64 编码和 MD5 值 |

### 响应模型

**Server 响应** (`server/models/response.py`)

| 模型/函数 | 说明 |
|-----------|------|
| `Base` | 基础响应模型（errcode, errmsg） |
| `build_base_instance(response)` | 从响应构建 Base 实例 |
| `is_success()` | 判断响应是否成功 |

**Webhook 响应** (`webhook/models/response.py`)

| 模型/函数 | 说明 |
|-----------|------|
| `Base` | 基础响应模型（errcode, errmsg） |
| `build_base_instance(response)` | 从响应构建 Base 实例 |
| `is_success()` | 判断响应是否成功 |

**响应模型使用示例**：

```python
from py_wecom_toolkit.server.models.response import build_base_instance, Base

# 发送请求并解析响应
response = server.gettoken()
base_instance = build_base_instance(response)

# 检查响应是否成功
if base_instance.is_success():
    print("请求成功")
    print(f"access_token: {base_instance.access_token}")
else:
    print(f"请求失败: {base_instance.errcode} - {base_instance.errmsg}")

# 直接从字典构建
response_dict = {"errcode": 0, "errmsg": "ok", "access_token": "xxx"}
base_instance = Base(**response_dict)
if base_instance.is_success():
    print("响应成功")
```

## 缓存配置

Server 模块支持 access_token 缓存，减少 API 调用次数。

```python
# 使用 diskcache
import diskcache
cache = diskcache.Cache("./cache")
server = Server(
    corpid="xxx",
    corpsecret="xxx",
    agentid="xxx",
    cache_config={"instance": cache}
)

# 使用 Redis
import redis
r = redis.Redis(host="localhost", port=6379)
server = Server(
    corpid="xxx",
    corpsecret="xxx",
    agentid="xxx",
    cache_config={"instance": r}
)
```

## 客户端配置

支持自定义 HTTP 客户端配置：

```python
webhook = Webhook(
    key="your_key",
    client_kwargs={
        "timeout": 30,
        "verify": True,
        "proxies": {"http": "http://proxy:8080"}
    }
)

server = Server(
    corpid="xxx",
    corpsecret="xxx",
    client_kwargs={
        "timeout": 60,
        "verify": False
    }
)
```

## 依赖

- httpx >= 0.27.0
- pydantic >= 2.0
- py-httpx-toolkit >= 1.0.0
- diskcache >= 5.0（可选，用于缓存）
- redis >= 5.0（可选，用于缓存）

## 贡献

欢迎提交 Issue 和 Pull Request！

## 许可证

MIT License
