py-wecom 0.1.1

企业微信 SDK，提供企业微信 API 的 Python 封装，支持同步和异步调用。

py_wecom

企业微信 SDK，提供企业微信 API 的 Python 封装，支持同步和异步调用。

功能特性

• Server API: 企业微信服务端 API，支持消息发送、素材上传等
• Webhook API: 企业微信机器人 Webhook 接口，支持多种消息类型
• 同步/异步支持: 所有接口均提供同步和异步版本
• 类型安全: 使用 Pydantic 进行数据验证，提供完整的类型提示
• 缓存支持: 支持 Redis 和 diskcache 缓存 access_token

安装

pip install py_wecom

快速开始

Server API

from py_wecom.server import Base, Sender, Uploader, Text, TextContent

# 初始化客户端
server = Base(
    corpid="your_corp_id",
    corpsecret="your_corp_secret",
    agentid="your_agent_id"
)

# 刷新 access_token
server.refresh_access_token()

# 发送文本消息
sender = Sender()
sender.agentid = server.agentid
sender.access_token = server.access_token

content = Text(
    touser="@all",
    text=TextContent(content="Hello from py_wecom!")
)
response = sender.send_text(content=content)
print(response.json())

Webhook API

from py_wecom.webhook import Webhook
from py_wecom.webhook.messages import Text, TextContent
from py_wecom.webhook.utils import convert_to_send

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

# 发送文本消息
response = webhook.send_text(
    text=Text(
        text=TextContent(
            content="Hello from py_wecom!",
            mentioned_list=["@all"]
        )
    )
)

# 转换响应为模型对象
send = convert_to_send(response)
print(f"发送结果: errcode={send.errcode}, errmsg={send.errmsg}")

支持的消息类型

Server API 支持的消息类型

消息类型	说明
Text	文本消息
Image	图片消息
Voice	语音消息
Video	视频消息
File	文件消息
TextCard	文本卡片消息
News	图文消息
MpNews	图文消息(mpnews)
Markdown	Markdown消息
MiniprogramNotice	小程序通知
TemplateCard	模板卡片消息

Webhook API 支持的消息类型

消息类型	说明
Text	文本消息
Markdown	Markdown消息
MarkdownV2	MarkdownV2消息（支持更多样式）
Image	图片消息
News	图文消息
Voice	语音消息
File	文件消息
TemplateCard	模板卡片消息

使用示例

发送图片消息 (Server API)

from py_wecom.server import Base, Sender, Image, ImageContent

server = Base(corpid="xxx", corpsecret="xxx", agentid="xxx")
server.refresh_access_token()

sender = Sender()
sender.agentid = server.agentid
sender.access_token = server.access_token

# 先上传图片获取 media_id
uploader = Uploader()
uploader.access_token = server.access_token

# 上传图片
with open("image.jpg", "rb") as f:
    upload_response = uploader.upload(
        file_type="image",
        files={"media": f}
    )
media_id = upload_response.json()["media_id"]

# 发送图片消息
content = Image(
    touser="user_id",
    image=ImageContent(media_id=media_id)
)
response = sender.send_image(content=content)

发送文件消息 (Webhook)

from py_wecom.webhook import Webhook
from py_wecom.webhook.messages import File, FileContent
from py_wecom.webhook.utils import convert_to_send, convert_to_upload_media

webhook = Webhook(key="your_webhook_key")

# 先上传文件获取 media_id
upload_response = webhook.upload_media(
    files={
        "file": ("test.txt", open("test.txt", "rb"))
    }
)

# 转换上传响应为模型对象
upload_media = convert_to_upload_media(upload_response)
print(f"上传结果: media_id={upload_media.media_id}")

# 发送文件消息
send_response = webhook.send_file(
    file=File(file=FileContent(media_id=upload_media.media_id))
)

# 转换发送响应为模型对象
send = convert_to_send(send_response)
print(f"发送结果: errcode={send.errcode}, errmsg={send.errmsg}")

使用缓存 (Server API)

import redis
from py_wecom.server import Base

# 使用 Redis 缓存 access_token
redis_client = redis.Redis(host="localhost", port=6379, db=0)

server = Base(
    corpid="xxx",
    corpsecret="xxx",
    agentid="xxx",
    cache_config={
        "instance": redis_client,
        "key": "pywecom_access_token",
        "expire": 7100
    }
)

# 刷新 access_token（会自动使用缓存）
server.refresh_access_token()

异步支持

所有方法均提供异步版本，方法名以 async_ 开头：

Server API 异步示例

import asyncio
from py_wecom.server import Base, Sender, Text, TextContent

async def main():
    server = Base(corpid="xxx", corpsecret="xxx", agentid="xxx")
    await server.async_refresh_access_token()
    
    sender = Sender()
    sender.agentid = server.agentid
    sender.access_token = server.access_token
    
    content = Text(
        touser="@all",
        text=TextContent(content="Hello!")
    )
    response = await sender.async_send_text(content=content)
    print(response.json())

asyncio.run(main())

Webhook API 异步示例

import asyncio
from py_wecom.webhook import Webhook
from py_wecom.webhook.messages import Text, TextContent, File, FileContent
from py_wecom.webhook.utils import convert_to_send, convert_to_upload_media

async def main():
    webhook = Webhook(key="your_webhook_key")
    
    # 异步发送文本消息
    response = await webhook.async_send_text(
        text=Text(text=TextContent(content="async hello world"))
    )
    send = convert_to_send(response)
    print(f"发送结果: errcode={send.errcode}")
    
    # 异步上传并发送文件
    upload_response = await webhook.async_upload_media(
        files={"file": ("test.txt", open("test.txt", "rb"))}
    )
    upload_media = convert_to_upload_media(upload_response)
    
    send_response = await webhook.async_send_file(
        file=File(file=FileContent(media_id=upload_media.media_id))
    )
    send = convert_to_send(send_response)
    print(f"文件发送结果: errcode={send.errcode}")

asyncio.run(main())

项目结构

py_wecom/
├── src/
│   └── py_wecom/
│       ├── __init__.py
│       ├── server/           # Server API
│       │   ├── __init__.py   # Base 类，access_token 管理
│       │   ├── materials.py  # 素材上传
│       │   ├── messages.py   # 消息类型和发送
│       │   ├── responses.py  # 响应模型
│       │   └── utils.py      # 工具函数
│       └── webhook/          # Webhook API
│           ├── __init__.py   # Webhook 客户端
│           ├── messages.py   # 消息类型
│           ├── responses.py  # 响应模型
│           └── utils.py      # 工具函数
└── README.md

参考文档

• 企业微信开发者文档
• 企业微信机器人文档

主页

https://gitee.com/guolei19850528/py_wecom

许可证

MIT License