py-easy-wecom 3.0.3

企业微信 Python SDK，提供简单易用的企业微信 API 封装，支持消息发送、媒体上传等功能。

py_easy_wecom

企业微信 Python SDK，提供简单易用的企业微信 API 封装，支持消息发送、媒体上传等功能。

功能特性

• Webhook 机器人：支持同步和异步发送消息，上传媒体文件
• 企业微信服务器：支持获取和刷新 access_token，管理缓存（支持 diskcache 和 redis）
• 消息管理：支持发送应用消息
• 媒体管理：支持上传临时素材和图文消息内的图片
• 同步/异步支持：同时支持同步和异步操作
• 灵活配置：支持通过参数自定义客户端行为
• 响应处理：提供 JSON 数据验证和提取功能
• 类型提示：完整的类型注解，提供良好的 IDE 支持

安装

使用 pip 安装

pip install py_easy_wecom

从源码安装

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

依赖

• httpx
• diskcache
• redis
• py_easy_httpx

快速开始

Webhook 机器人

同步发送消息

from py_easy_wecom.webhook import Webhook

# 初始化机器人
robot = Webhook(key="YOUR_WEBHOOK_KEY")

# 发送文本消息
response = robot.send(
    json={
        "msgtype": "text",
        "text": {
            "content": "Hello, World!",
            "mentioned_list": ["@all"],
            "mentioned_mobile_list": []
        }
    }
)
print(response)

异步发送消息

import asyncio
from py_easy_wecom.webhook import Webhook

async def main():
    # 初始化机器人
    robot = Webhook(key="YOUR_WEBHOOK_KEY")
    
    # 发送文本消息
    response = await robot.async_send(
        json={
            "msgtype": "text",
            "text": {
                "content": "Hello, World!",
                "mentioned_list": ["@all"],
                "mentioned_mobile_list": []
            }
        }
    )
    print(response)

asyncio.run(main())

企业微信服务器

初始化服务器

from py_easy_wecom.server import Server

# 初始化服务器
server = Server(
    corpid="YOUR_CORPID",
    corpsecret="YOUR_CORPSECRET",
    agentid="YOUR_AGENTID"
)

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

发送应用消息

from py_easy_wecom.server.message import Message

# 初始化消息客户端
message_client = Message(
    corpid="YOUR_CORPID",
    corpsecret="YOUR_CORPSECRET",
    agentid="YOUR_AGENTID"
)

# 刷新 access_token
message_client.refresh_access_token()

# 发送文本消息
response = message_client.send(
    json={
        "touser": "UserID1|UserID2",
        "msgtype": "text",
        "text": {
            "content": "Hello, World!"
        }
    }
)
print(response)

上传媒体文件

from py_easy_wecom.server.media import Media

# 初始化媒体客户端
media_client = Media(
    corpid="YOUR_CORPID",
    corpsecret="YOUR_CORPSECRET",
    agentid="YOUR_AGENTID"
)

# 刷新 access_token
media_client.refresh_access_token()

# 上传临时素材
with open("example.jpg", "rb") as f:
    response = media_client.upload(
        file_type="image",
        files={"media": f}
    )
print(response)  # 返回 media_id

API 文档

Webhook 类

初始化参数

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

方法

• 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：缓存配置，支持 diskcache 或 redis
• client_kwargs：传递给 httpx.Client 的额外参数

方法

• request(client=None, **kwargs)：同步发送请求
• async_request(client=None, **kwargs)：异步发送请求
• get_access_token(client=None, **kwargs)：获取 access_token
• async_get_access_token(client=None, **kwargs)：异步获取 access_token
• get_api_domain_ip(client=None, **kwargs)：获取企业微信 API 域名 IP 段
• async_get_api_domain_ip(client=None, **kwargs)：异步获取企业微信 API 域名 IP 段
• get_callback_ip(client=None, **kwargs)：获取企业微信回调 IP 段
• async_get_callback_ip(client=None, **kwargs)：异步获取企业微信回调 IP 段
• refresh_access_token(client=None, get_access_token_kwargs=None, get_api_domain_ip_kwargs=None)：刷新 access_token
• async_refresh_access_token(client=None, get_access_token_kwargs=None, get_api_domain_ip_kwargs=None)：异步刷新 access_token

Message 类

方法

• send(client=None, **kwargs)：同步发送应用消息
• async_send(client=None, **kwargs)：异步发送应用消息

Media 类

方法

• upload(client=None, file_type=None, **kwargs)：同步上传临时素材
• async_upload(client=None, file_type=None, **kwargs)：异步上传临时素材
• upload_img(client=None, **kwargs)：同步上传图文消息内的图片
• async_upload_img(client=None, **kwargs)：异步上传图文消息内的图片

高级用法

缓存配置

使用 diskcache

import diskcache
from py_easy_wecom.server import Server

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

# 初始化服务器
server = Server(
    corpid="YOUR_CORPID",
    corpsecret="YOUR_CORPSECRET",
    agentid="YOUR_AGENTID",
    cache_config={
        "instance": cache,
        "expire": 7100  # 缓存过期时间（秒）
    }
)

使用 redis

import redis
from py_easy_wecom.server import Server

# 创建 redis 实例
redis_client = redis.Redis(host="localhost", port=6379, db=0)

# 初始化服务器
server = Server(
    corpid="YOUR_CORPID",
    corpsecret="YOUR_CORPSECRET",
    agentid="YOUR_AGENTID",
    cache_config={
        "instance": redis_client,
        "expire": 7100  # 缓存过期时间（秒）
    }
)

项目结构

py_easy_wecom/
├── py_easy_wecom/              # 主包目录
│   ├── __init__.py            # 包初始化文件
│   ├── webhook/               # Webhook 模块
│   │   └── __init__.py        # Webhook 初始化文件
│   └── server/                # 服务器模块
│       ├── __init__.py        # 服务器初始化文件
│       ├── message.py         # 消息管理模块
│       └── media.py           # 媒体管理模块
├── README.md                  # 项目文档
├── setup.py                   # 安装配置
├── requirements.txt           # 依赖列表
├── LICENSE                    # 许可证文件
└── .gitignore                 # Git 忽略文件

测试

运行测试：

# 运行测试
python -m pytest

许可证

MIT License

贡献

欢迎提交 Issue 和 Pull Request！

联系方式

• 作者：guolei
• 邮箱：[e-mail]
• 项目地址：https://gitee.com/guolei19850528/py_easy_wecom

致谢

• httpx - 高性能的异步 HTTP 客户端库
• diskcache - 简单易用的本地缓存库
• redis - 高性能的键值存储
• py_easy_httpx - 简化 HTTP 请求的工具包