Metadata-Version: 2.4
Name: py-wecom-toolkit
Version: 1.0.6
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: jsonpath-ng>=1.5.3
Requires-Dist: jsonschema>=4.21.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

企业微信 API 的 Python SDK，提供完整的企业微信接口封装，支持同步和异步调用方式。

## 项目主页

- **项目地址**: [https://gitee.com/guolei19850528/py_wecom_toolkit](https://gitee.com/guolei19850528/py_wecom_toolkit)


## 功能特性

### Webhook 模块 (`py_wecom_toolkit.webhook`)
- 企业微信 Webhook 消息发送
- 支持文本、Markdown、MarkdownV2、图片、图文、文件、语音、模板卡片等消息类型
- 支持同步和异步两种调用方式
- 媒体文件上传功能

### Server 模块 (`py_wecom_toolkit.server`)
- 企业微信 Server API 集成
- 获取 access_token（支持 diskcache 和 Redis 缓存）
- 获取 API 域名 IP 列表
- 获取回调 IP 列表
- 消息发送和素材管理
- 支持同步和异步两种调用方式

## 依赖包

```txt
httpx>=0.27.0
pydantic>=2.0
jsonpath-ng>=1.5.3
jsonschema>=4.21.0
diskcache>=5.6.3
redis>=4.6.0
py-httpx-toolkit>=1.0.1
```

## 安装方式

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

```bash

uv add py-wecom-toolkit
```


## 项目结构

```
src/py_wecom_toolkit/
├── __init__.py              # 主模块入口
├── server/                  # Server API 模块
│   ├── __init__.py          # Base 类（access_token管理、IP查询）
│   ├── materials.py         # 素材管理
│   ├── messages.py          # 消息模型
│   ├── responses.py         # 响应模型
│   └── utils.py             # 工具函数
└── webhook/                 # Webhook 模块
    ├── __init__.py          # Webhook 客户端类
    ├── messages.py          # 消息模型
    ├── responses.py         # 响应模型
    └── utils.py             # 工具函数
```

## 使用示例

### Webhook 使用示例

**同步发送文本消息**：

```python
from py_wecom_toolkit.webhook import Webhook
from py_wecom_toolkit.webhook.messages import Text, TextContent

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

# 发送文本消息
response = webhook.send(Text(text=TextContent(content="Hello World")))
print(response.json())
```

**异步发送消息**：

```python
import asyncio
from py_wecom_toolkit.webhook import Webhook
from py_wecom_toolkit.webhook.messages import Markdown, MarkdownContent

async def main():
    webhook = Webhook(key="your_webhook_key")
    response = await webhook.async_send(
        Markdown(markdown=MarkdownContent(content="**Hello** from async"))
    )
    print(response.json())

asyncio.run(main())
```

**上传媒体文件**：

```python
from py_wecom_toolkit.webhook import Webhook

webhook = Webhook(key="your_webhook_key")

with open("test.txt", "rb") as f:
    response = webhook.upload_media(files={"file": ("test.txt", f)})
print(response.json())  # {"errcode": 0, "media_id": "xxx", ...}
```

### Server 使用示例

**获取 access_token**：

```python
from py_wecom_toolkit.server import Base
from diskcache import Cache

# 创建缓存实例
cache = Cache("./cache")

# 初始化 Server 客户端
server = Base(
    corpid="your_corpid",
    corpsecret="your_corpsecret",
    cache_config={"instance": cache}
)

# 刷新 access_token
server.refresh_access_token()
print(server.access_token)
```

**异步获取 access_token**：

```python
import asyncio
import redis
from py_wecom_toolkit.server import Base

async def main():
    r = redis.Redis(host='localhost', port=6379, db=0)
    server = Base(
        corpid="your_corpid",
        corpsecret="your_corpsecret",
        cache_config={"instance": r}
    )
    await server.async_refresh_access_token()
    print(server.access_token)

asyncio.run(main())
```

**获取回调 IP**：

```python
from py_wecom_toolkit.server import Base

server = Base(corpid="your_corpid", corpsecret="your_corpsecret")
server.refresh_access_token()
response = server.getcallbackip()
print(response.json())
```

## API 说明

### Webhook 模块 API

#### Webhook 类

**初始化参数**：

| 参数 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| key | str | 是 | - | Webhook密钥 |
| base_url | str | 否 | `https://qyapi.weixin.qq.com/` | 企业微信API基础URL |
| mentioned_list | list/tuple | 否 | `[]` | 默认@用户ID列表 |
| mentioned_mobile_list | list/tuple | 否 | `[]` | 默认@用户手机号列表 |
| client_kwargs | dict | 否 | `{}` | HTTP客户端配置参数 |

**方法**：

| 方法名 | 说明 | 参数 | 返回值 |
|--------|------|------|--------|
| `send(message)` | 同步发送消息 | `message`: 消息对象或字典 | `httpx.Response` |
| `async_send(message)` | 异步发送消息 | `message`: 消息对象或字典 | `await httpx.Response` |
| `upload_media(file_type, **kwargs)` | 同步上传媒体文件 | `file_type`: "file"或"voice"，`files`: 文件参数 | `httpx.Response` |
| `async_upload_media(file_type, **kwargs)` | 异步上传媒体文件 | `file_type`: "file"或"voice"，`files`: 文件参数 | `await httpx.Response` |

#### Webhook 消息类型

| 消息类型 | 类名 | 说明 |
|----------|------|------|
| 文本 | `Text` | 支持@用户，最长2048字节 |
| Markdown | `Markdown` | 支持标准Markdown语法 |
| MarkdownV2 | `MarkdownV2` | 支持字体颜色等扩展样式 |
| 图片 | `Image` | 需要base64编码和md5值 |
| 图文 | `News` | 支持1-8篇文章 |
| 文件 | `File` | 需要先上传获取media_id |
| 语音 | `Voice` | 需要先上传获取media_id |
| 模板卡片 | `TemplateCard` | 支持多种卡片类型 |

**文本消息示例**：

```python
from py_wecom_toolkit.webhook.messages import Text, TextContent

message = Text(
    text=TextContent(
        content="Hello World",
        mentioned_list=["user1", "@all"],
        mentioned_mobile_list=["13800138000"]
    )
)
```

**图文消息示例**：

```python
from py_wecom_toolkit.webhook.messages import News, NewsContent, NewsArticle

message = News(
    news=NewsContent(
        articles=[
            NewsArticle(
                title="文章标题",
                description="文章描述",
                url="https://example.com",
                picurl="https://example.com/image.jpg"
            )
        ]
    )
)
```

### Server 模块 API

#### Base 类

**初始化参数**：

| 参数 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| corpid | str | 是 | - | 企业ID |
| corpsecret | str | 是 | - | 应用密钥 |
| agentid | str/int | 否 | - | 企业应用ID |
| base_url | str | 否 | `https://qyapi.weixin.qq.com/` | 企业微信API基础URL |
| cache_config | dict | 否 | `{}` | 缓存配置 |
| client_kwargs | dict | 否 | `{}` | HTTP客户端配置参数 |

**cache_config 参数**：

| 参数 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| instance | Cache/Redis | 否 | `None` | 缓存实例（diskcache.Cache或redis.Redis） |
| key | str | 否 | `pywecom_server_{corpid}_{agentid}` | 缓存键名 |
| expire | int | 否 | `7100` | 缓存过期时间（秒） |

**方法**：

| 方法名 | 说明 | 参数 | 返回值 |
|--------|------|------|--------|
| `gettoken()` | 同步获取access_token | - | `httpx.Response` |
| `async_gettoken()` | 异步获取access_token | - | `await httpx.Response` |
| `get_api_domain_ip()` | 同步获取API域名IP | - | `httpx.Response` |
| `async_get_api_domain_ip()` | 异步获取API域名IP | - | `await httpx.Response` |
| `getcallbackip()` | 同步获取回调IP | - | `httpx.Response` |
| `async_getcallbackip()` | 异步获取回调IP | - | `await httpx.Response` |
| `refresh_access_token()` | 同步刷新access_token | `cache_config` | `self` |
| `async_refresh_access_token()` | 异步刷新access_token | `cache_config` | `await self` |
| `request_with_access_token()` | 同步请求，自动添加access_token | `client`, `client_kwargs`, `**kwargs` | `httpx.Response` |
| `async_request_with_access_token()` | 异步请求，自动添加access_token | `client`, `client_kwargs`, `**kwargs` | `await httpx.Response` |

#### Sender 类

继承自 `Base` 类，新增消息发送方法：

| 方法名 | 说明 | 参数 | 返回值 |
|--------|------|------|--------|
| `send(message)` | 同步发送消息 | `message`: 消息对象或字典 | `httpx.Response` |
| `async_send(message)` | 异步发送消息 | `message`: 消息对象或字典 | `await httpx.Response` |

#### Server 消息类型

| 消息类型 | 类名 | 说明 |
|----------|------|------|
| 文本 | `Text` | 支持@用户 |
| 图片 | `Image` | 需要media_id |
| 语音 | `Voice` | 需要media_id |
| 视频 | `Video` | 需要media_id |
| 文件 | `File` | 需要media_id |
| 文本卡片 | `TextCard` | 带标题和链接的卡片 |
| 图文 | `News` | 支持1-8篇文章 |
| 图文(mpnews) | `MpNews` | 素材库文章 |
| Markdown | `Markdown` | Markdown格式消息 |
| 小程序通知 | `MiniprogramNotice` | 小程序模板消息 |
| 模板卡片 | `TemplateCard` | 支持多种卡片类型 |

**发送文本消息示例**：

```python
from py_wecom_toolkit.server.messages import Sender, Text, TextContent

sender = Sender(corpid="your_corpid", corpsecret="your_corpsecret", agentid=1000001)
sender.refresh_access_token()
response = sender.send(Text(text=TextContent(content="Hello")))
```

**发送图文消息示例**：

```python
from py_wecom_toolkit.server.messages import Sender, News, NewsArticlesContent, NewsArticleContent

sender = Sender(corpid="your_corpid", corpsecret="your_corpsecret", agentid=1000001)
sender.refresh_access_token()

message = News(
    news=NewsArticlesContent(
        articles=[
            NewsArticleContent(
                title="文章标题",
                description="描述",
                url="https://example.com",
                picurl="https://example.com/image.jpg"
            )
        ]
    )
)
response = sender.send(message)
```

### 响应模型

响应结果可通过 `response.json()` 获取，企业微信API响应结构：

```python
{
    "errcode": 0,          # 错误码，0表示成功
    "errmsg": "ok",        # 错误信息
    "access_token": "...", # access_token（仅gettoken接口）
    "media_id": "...",     # 媒体文件ID（仅上传接口）
    "ip_list": [...]       # IP列表（仅IP查询接口）
}
```

**错误码说明**：

| 错误码 | 说明 |
|--------|------|
| 0 | 成功 |
| 40001 | 获取access_token时AppSecret错误 |
| 40002 | 不合法的凭证类型 |
| 40003 | 不合法的UserID |
| 40014 | 不合法的access_token |
| 40033 | 不合法的字符 |
| 40056 | 不合法的agentid |
| 40060 | 不合法的media_id |
| 50000 | 企业不存在 |
| 60001 | 应用不存在 |
| 60002 | 应用已停用 |

## 工具函数说明

### Webhook 工具函数 (`py_wecom_toolkit.webhook.utils`)

| 函数名 | 说明 | 参数 | 返回值 |
|--------|------|------|--------|
| `json_find_first(expression, data)` | 使用JSONPath查找第一个匹配项 | `expression`: JSONPath表达式，`data`: JSON数据 | 匹配值或None |
| `json_is_valid(schema, data)` | JSON Schema校验 | `schema`: Schema字典，`data`: 待校验数据 | bool |
| `image_to_base64_and_md5(image_path)` | 图片转Base64和MD5 | `image_path`: 图片文件路径 | Tuple[str, str] |
| `build_success_instance(response)` | 转换为Success响应模型 | `response`: Response对象或dict | Success |
| `build_send_instance(response)` | 转换为Send响应模型 | `response`: Response对象或dict | Send |
| `build_upload_media_instance(response)` | 转换为UploadMedia响应模型 | `response`: Response对象或dict | UploadMedia |
| `success_validator(response, schema)` | 校验errcode是否为0 | `response`: Response对象或dict | bool |

**图片转换示例**：

```python
from py_wecom_toolkit.webhook.utils import image_to_base64_and_md5
from py_wecom_toolkit.webhook.messages import Image, ImageContent

base64_str, md5_str = image_to_base64_and_md5("path/to/image.jpg")
message = Image(image=ImageContent(base64=base64_str, md5=md5_str))
```

**响应校验示例**：

```python
from py_wecom_toolkit.webhook import Webhook
from py_wecom_toolkit.webhook.utils import build_send_instance, success_validator

webhook = Webhook(key="your_key")
response = webhook.send(message)

# 校验响应
if success_validator(response):
    result = build_send_instance(response)
    print(f"发送成功: {result.errmsg}")
else:
    print("发送失败")
```

### Server 工具函数 (`py_wecom_toolkit.server.utils`)

| 函数名 | 说明 | 参数 | 返回值 |
|--------|------|------|--------|
| `json_find_first(expression, data)` | 使用JSONPath查找第一个匹配项 | `expression`: JSONPath表达式，`data`: JSON数据 | 匹配值或None |
| `json_is_valid(schema, data)` | JSON Schema校验 | `schema`: Schema字典，`data`: 待校验数据 | bool |
| `build_success_instance(response)` | 转换为Success响应模型 | `response`: Response对象或dict | Success |
| `success_is_valid(response, schema)` | 校验errcode是否为0 | `response`: Response对象或dict | bool |

**响应校验示例**：

```python
from py_wecom_toolkit.server.messages import Sender, Text, TextContent
from py_wecom_toolkit.server.utils import build_success_instance, success_is_valid

sender = Sender(corpid="your_corpid", corpsecret="your_corpsecret", agentid=1000001)
sender.refresh_access_token()
response = sender.send(Text(text=TextContent(content="Hello")))

# 校验响应
if success_is_valid(response):
    result = build_success_instance(response)
    print(f"发送成功: {result.errmsg}")
else:
    print("发送失败")
```

**JSONPath查找示例**：

```python
from py_wecom_toolkit.server.utils import json_find_first

data = {
    "result": {
        "items": [
            {"id": 1, "name": "item1"},
            {"id": 2, "name": "item2"}
        ]
    }
}

# 使用JSONPath查找第一个item的name
name = json_find_first("$.result.items[0].name", data)
print(name)  # item1
```

## 参考文档

- [企业微信 Webhook API](https://developer.work.weixin.qq.com/document/path/91770)
- [企业微信 Server API](https://developer.work.weixin.qq.com/document/path/90664)

## 许可证

MIT License
