py-easy-wecom 2.0.1
一个简化企业微信API使用的Python工具包，提供Webhook消息发送和服务器API调用的便捷封装。
Requires Python: >=3.6

# py-easy-wecom

一个简化企业微信API使用的Python工具包，提供Webhook消息发送和服务器API调用的便捷封装。

## 功能特性

- **Webhook操作**：支持企业微信机器人消息发送和媒体文件上传
- **Server API操作**：支持企业微信应用消息发送、素材上传等服务器API
- **同步与异步**：同时支持同步和异步操作
- **响应处理**：统一的响应处理机制，自动验证响应结果
- **缓存管理**：支持diskcache和redis缓存access_token
- **类型提示**：完整的类型注解，提供良好的IDE支持

## 安装

### 使用pip安装

```bash
pip install py-easy-wecom
```

### 从源码安装

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

## 依赖

- httpx
- jsonschema
- diskcache
- redis
- jsonpath-ng

## 快速开始

### Webhook使用

```python
from py_easy_wecom.webhook import Webhook

# 初始化Webhook实例
webhook = Webhook(key="YOUR_WEBHOOK_KEY")

# 发送文本消息
response = webhook.send(
    json={
        "msgtype": "text",
        "text": {
            "content": "Hello, World!",
            "mentioned_list": [],
            "mentioned_mobile_list": []
        }
    }
)
print(f"发送结果: {response.validated}")

# 上传媒体文件
response = webhook.upload_media(
    file_type="file",
    files={"file": open("example.txt", "rb")}
)

if response.validated:
    # 发送文件消息
    media_id = response.value
    webhook.send(
        json={
            "msgtype": "file",
            "file": {
                "media_id": media_id
            }
        }
    )
```

### 异步Webhook使用

```python
import asyncio
from py_easy_wecom.webhook import Webhook

async def main():
    webhook = Webhook(key="YOUR_WEBHOOK_KEY")
    
    # 异步发送文本消息
    response = await webhook.async_send(
        json={
            "msgtype": "text",
            "text": {
                "content": "Hello, Async World!",
                "mentioned_list": [],
                "mentioned_mobile_list": []
            }
        }
    )
    print(f"发送结果: {response.validated}")

asyncio.run(main())
```

### Server使用

```python
from py_easy_wecom.server import Server
import diskcache

# 初始化缓存
cache = diskcache.Cache('./runtime/diskcache/default')

# 初始化Server实例
server = Server(
    corpid="YOUR_CORPID",
    corpsecret="YOUR_CORPSECRET",
    agentid="YOUR_AGENTID",
    cache_config={
        "instance": cache
    }
)

# 发送应用消息
with server.client() as client:
    # 刷新access_token并发送消息
    response = server.refresh(client=client).message__send(
        json={
            "touser": "USER_ID",
            "msgtype": "text",
            "agentid": server.agentid,
            "text": {
                "content": "Hello from Server!"
            }
        }
    )
    print(f"发送结果: {response.validated}")
```

### 异步Server使用

```python
import asyncio
from py_easy_wecom.server import Server
import diskcache

async def main():
    cache = diskcache.Cache('./runtime/diskcache/default')
    server = Server(
        corpid="YOUR_CORPID",
        corpsecret="YOUR_CORPSECRET",
        agentid="YOUR_AGENTID",
        cache_config={
            "instance": cache
        }
    )
    
    async with server.async_client() as client:
        # 异步刷新access_token并发送消息
        response = await server.refresh(client=client).async_message__send(
            json={
                "touser": "USER_ID",
                "msgtype": "text",
                "agentid": server.agentid,
                "text": {
                    "content": "Hello from Async Server!"
                }
            }
        )
        print(f"发送结果: {response.validated}")

asyncio.run(main())
```

## API文档

### Webhook类

#### 初始化参数
- `base_url`：企业微信API基础URL，默认"https://qyapi.weixin.qq.com"
- `key`：企业微信机器人Webhook密钥
- `mentioned_list`：默认@用户ID列表
- `mentioned_mobile_list`：默认@手机号列表
- `client_kwargs`：传递给httpx.Client的额外参数

#### 方法
- `client(**kwargs)`：创建同步客户端实例
- `async_client(**kwargs)`：创建异步客户端实例
- `send(client=None, **kwargs)`：同步发送消息
- `async_send(client=None, **kwargs)`：异步发送消息
- `upload_media(client=None, file_type="file", **kwargs)`：同步上传媒体文件
- `async_upload_media(client=None, file_type="file", **kwargs)`：异步上传媒体文件

### Server类

#### 初始化参数
- `base_url`：企业微信API基础URL，默认"https://qyapi.weixin.qq.com"
- `corpid`：企业ID
- `corpsecret`：应用凭证密钥
- `agentid`：应用ID
- `cache_config`：缓存配置
- `client_kwargs`：传递给httpx.Client的额外参数

#### 方法
- `client(**kwargs)`：创建同步客户端实例
- `async_client(**kwargs)`：创建异步客户端实例
- `refresh(client=None, gettoken_kwargs=None, get_api_domain_ip_kwargs=None)`：刷新access_token
- `message__send(client=None, **kwargs)`：同步发送应用消息
- `async_message__send(client=None, **kwargs)`：异步发送应用消息
- `media__upload(client=None, file_type=None, **kwargs)`：同步上传临时素材
- `async_media__upload(client=None, file_type=None, **kwargs)`：异步上传临时素材
- `media__uploadimg(client=None, **kwargs)`：同步上传图文消息内的图片
- `async_media__uploadimg(client=None, **kwargs)`：异步上传图文消息内的图片
- `gettoken(client=None, **kwargs)`：获取access_token
- `get_api_domain_ip(client=None, **kwargs)`：获取企业微信API域名IP段
- `getcallbackip(client=None, **kwargs)`：获取企业微信回调IP段

### 响应处理

所有API调用都会返回`RequestResponse`对象，包含以下属性：
- `validated`：响应是否有效
- `value`：从响应中提取的值（如access_token、media_id等）
- `json`：响应的JSON数据
- `response`：原始的httpx.Response对象

## 测试

运行测试用例：

```bash
python -m unittest test_webhook.py
python -m unittest test_server.py
```

## 项目结构

```
py-easy-wecom/
├── py_easy_wecom/
│   ├── __init__.py        # 模块初始化和响应处理
│   ├── webhook.py         # Webhook操作类
│   └── server.py          # Server API操作类
├── runtime/
│   └── diskcache/         # 缓存目录
├── test_webhook.py        # Webhook测试
├── test_server.py         # Server测试
├── setup.py               # 安装配置
├── requirements.txt       # 依赖声明
├── LICENSE                # 许可证
└── README.md              # 项目文档
```

## 许可证

MIT License

## 贡献

欢迎提交Issue和Pull Request！

## 联系方式

- 作者：guolei
- 邮箱：174000902@qq.com
- 项目地址：https://gitee.com/guolei19850528/py_easy_wecom