Skip to content

KevinFan1/wecomkit

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

2 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

wecomkit 企业微信 SDK

English | 中文

wecomkit 是一个基于 asyncio + httpx 的企业微信异步 SDK。当前版本覆盖常用服务端 API 场景,包括通讯录、应用消息、素材、打卡、日程、审批、企业邮箱、人事助手、微盘和 WeDoc 的部分接口。

当前 SDK 不是企业微信全量 API 封装。接口状态以 src/wecomkit/services/ 中实际实现为准。

安装

从 PyPI 安装:

uv pip install wecomkit
#
pip install wecomkit

如果当前项目由 uv 管理,也可以直接添加依赖:

uv add wecomkit

需要 Redis token 缓存时安装可选依赖:

uv pip install "wecomkit[redis]"
#
uv add "wecomkit[redis]"

从源码开发安装:

git clone https://github.com/KevinFan1/wecomkit.git
cd wecomkit
uv pip install -e ".[dev]"

构建本地 wheel:

uv build

构建产物会输出到 dist/,可用于本地验证安装:

uv pip install ./dist/wecomkit-*.whl
#
pip install ./dist/wecomkit-*.whl

快速开始

import asyncio

from wecomkit import AsyncWeComClient, JSONFileTokenCache, WeComConfig


async def main() -> None:
    config = WeComConfig(
        corp_id="wwxxxxxx",
        corp_secret="your-secret",
    )

    async with AsyncWeComClient(
        config,
        token_cache=JSONFileTokenCache(".cache/wecom-token.json"),
    ) as client:
        token = await client.auth.get_access_token()
        departments = await client.departments.list()
        users = await client.users.simple_list_by_department(1, fetch_child=True)

        print(token)
        print(departments)
        print(users)


asyncio.run(main())

Examples

示例代码位于 examples

运行示例前设置环境变量:

export WECOM_CORP_ID="wwxxxxxx"
export WECOM_CORP_SECRET="your-secret"
export WECOM_AGENT_ID="1000001"
uv run python examples/quickstart.py

客户端结构

client.auth          # 授权
client.users         # 成员
client.departments   # 部门
client.tags          # 标签
client.agent         # 应用
client.menu          # 应用菜单
client.messages      # 应用消息
client.media         # 素材
client.mail          # 企业邮箱
client.robot         # 群机器人 webhook
client.hr            # 人事助手
client.checkin       # 打卡
client.approval      # 审批
client.calendar      # 日历
client.schedule      # 日程
client.wedrive       # 微盘
client.wedoc         # WeDoc 聚合服务

WeDoc 子服务:

client.wedoc.documents     # 文档
client.wedoc.smartsheets  # 智能表格
client.wedoc.permissions   # 权限
client.wedoc.forms         # 收集表通用调用
client.wedoc.materials     # 文档素材

已实现接口

授权

SDK 方法 企业微信接口
client.auth.get_access_token() /cgi-bin/gettoken
client.auth.get_user_info_by_code(code) /cgi-bin/auth/getuserinfo
client.auth.get_user_detail(user_ticket) /cgi-bin/auth/getuserdetail

通讯录

模块 已实现
成员 create, update, delete, batch_delete, get, list_by_department, simple_list_by_department, list_id, get_join_qrcode, invite
部门 create, update, delete, list, simple_list, get
标签 create, update, delete, get, add_users, delete_users, list

对应接口包括 /cgi-bin/user/*/cgi-bin/department/*/cgi-bin/tag/*/cgi-bin/invite/send

应用与消息

模块 已实现
应用 agent.get, agent.list, agent.set
菜单 menu.create, menu.get, menu.delete
应用消息 messages.send, send_text, send_markdown, send_textcard, send_news, send_mpnews, send_miniprogram_notice, send_template_card, send_image, send_file, send_video, send_voice
群机器人 robot.send, send_text, send_markdown, send_image, send_file, send_news, send_textcard, send_template

messages.send() 可直接发送任意已支持消息类型;常用消息类型也提供了便捷方法。

素材

SDK 方法 企业微信接口
media.upload(file_path, media_type) /cgi-bin/media/upload
media.upload_bytes(content, media_type, filename) /cgi-bin/media/upload
media.get(media_id) /cgi-bin/media/get
media.download(media_id, path) /cgi-bin/media/get
media.upload_image(file_path) /cgi-bin/media/uploadimg
media.upload_attachment(file_path) /cgi-bin/media/upload_attachment
media.upload_jssdk_image(file_path) /cgi-bin/media/upload_attachment

打卡

已实现:

  • get_checkin_data
  • get_checkin_schedules
  • get_checkin_option
  • add_checkin_user_face
  • get_checkin_daydata
  • get_checkin_monthdata
  • get_checkin_all_options
  • get_checkin_user_options

日历与日程

模块 已实现
日历 calendar.create, calendar.get, calendar.update, calendar.delete
日程 schedule.create, schedule.get, schedule.list_by_calendar, schedule.update, schedule.cancel, schedule.add_attendees, schedule.del_attendees

审批

已实现:

  • approval.get_template
  • approval.apply
  • approval.get_approve
  • approval.get_approve_list
  • approval.get_vacation_config
  • approval.get_user_vacation

企业邮箱

模块 已实现
邮件收发 mail.compose_send, send_mail, send_schedule_mail, send_meeting_mail, get_mail_list, get_mail_content, get_mail_unread_count
应用邮箱 mail.update_app_mailbox, get_app_mailbox
邮件群组 mail.create_mail_group, update_mail_group, delete_mail_group, get_mail_group, search_mail_group
公共邮箱 mail.create_public_mailbox, update_public_mailbox, delete_public_mailbox, get_public_mailbox, search_public_mailbox
客户端密码 mail.get_client_password_list, delete_client_password
高级账号 mail.allocate_mail_advanced_account, deallocate_mail_advanced_account, get_mail_advanced_account_list
用户邮箱属性 mail.toggle_mailbox_status, enable_mailbox, disable_mailbox, get_user_mail_attribute, update_user_mail_attribute

人事助手

已实现:

  • hr.get_fields
  • hr.get_staff_info
  • hr.update_staff_info
  • hr.text_attr
  • hr.uint32_attr

微盘

已实现:

  • wedrive.get_space_info
  • wedrive.create_space
  • wedrive.rename_space
  • wedrive.dismiss_space
  • wedrive.set_space
  • wedrive.share_space
  • wedrive.add_space_acl
  • wedrive.delete_space_acl
  • wedrive.create_folder
  • wedrive.upload_file
  • wedrive.download_file
  • wedrive.delete_file
  • wedrive.list_file
  • wedrive.get_file_meta
  • wedrive.rename_file
  • wedrive.move_file
  • wedrive.set_file
  • wedrive.set_file_security
  • wedrive.add_file_acl
  • wedrive.delete_file_acl
  • wedrive.get_share_link
  • wedrive.cancel_share

WeDoc

模块 已实现
文档 documents.create, rename, delete, get_base_info, share, get_content, edit_content
智能表格记录 smartsheets.add_records, delete_records, update_records, get_records
智能表格子表 smartsheets.add_sheet, delete_sheet, update_sheet, get_sheets
智能表格视图 smartsheets.add_view, delete_view, update_view, get_views
智能表格字段 smartsheets.add_fields, delete_fields, update_fields, get_fields
智能表格编组 smartsheets.add_group, delete_group, update_group, get_groups
权限 permissions.get_auth, set_auth, mod_member, mod_join_rule
收集表 forms.call(endpoint, payload), create_collect, modify_collect, get_info, get_answer, get_statistic
素材 materials.upload_image

未实现或仅部分覆盖

以下模块当前没有封装,或只覆盖了少量常用接口:

  • 客户联系、客户群、客户朋友圈
  • 微信客服
  • 会话内容存档
  • 企业支付、红包
  • 会议、会议室、直播
  • 家校沟通
  • 上下游、企业互联
  • 汇报、公费电话、紧急通知
  • 数据与智能专区

返回值与类型

服务方法默认返回企业微信原始 JSON 响应,即 dict[str, Any]wecomkit.types 提供 Pydantic v2 模型,调用方可以按需自行校验:

from wecomkit.types import BaseResp, UserInfo

resp = await client.users.get("zhangsan")
user = UserInfo.model_validate(resp)

ok = BaseResp.model_validate({"errcode": 0, "errmsg": "ok"})

当前服务方法没有统一的 response_model 参数。

智能表格和 WeDoc 权限也提供了宽松模型,未知字段会保留,便于兼容接口扩展:

from wecomkit.types import SmartSheetRecordListResponse

resp = await client.wedoc.smartsheets.get_records(docid, sheet_id)
records = SmartSheetRecordListResponse.model_validate(resp)

重试策略

默认不重试。需要对网络错误或企业微信限流错误做简单重试时,在 WeComConfig 中配置:

config = WeComConfig(
    corp_id="wwxx",
    corp_secret="secret",
    max_retries=2,
    retry_backoff=0.5,
)

等待时间按 retry_backoff * 2 ** attempt 递增。

Token 缓存

SDK 默认使用内存缓存 token,并在过期前 token_refresh_buffer 秒提前刷新。生产环境建议使用文件缓存或自定义缓存:

from wecomkit import AsyncWeComClient, JSONFileTokenCache, WeComConfig

client = AsyncWeComClient(
    WeComConfig(corp_id="wwxx", corp_secret="secret"),
    token_cache=JSONFileTokenCache(".cache/wecom-token.json"),
)

多进程或多实例部署建议使用 Redis。Redis 是可选依赖:

uv pip install "wecomkit[redis]"
#
uv add "wecomkit[redis]"
from redis.asyncio import Redis
from wecomkit import AsyncWeComClient, RedisTokenCache, WeComConfig

redis = Redis.from_url("redis://localhost:6379/0")

client = AsyncWeComClient(
    WeComConfig(corp_id="wwxx", corp_secret="secret"),
    token_cache=RedisTokenCache(redis, key="wecomkit:access_token"),
)

也可以直接通过 URL 创建:

from wecomkit import RedisTokenCache

token_cache = RedisTokenCache.from_url(
    "redis://localhost:6379/0",
    key="wecomkit:access_token",
)

自定义缓存:

from wecomkit.token_cache import TokenCache


class RedisTokenCache(TokenCache):
    async def get(self) -> str | None:
        ...

    async def set(self, token: str, expires_in: int) -> None:
        ...

异常处理

from wecomkit.exceptions import WeComAPIError, WeComNetworkError

try:
    user = await client.users.get("invalid_userid")
except WeComAPIError as e:
    print(e.errcode, e.errmsg)
except WeComNetworkError as e:
    print("network error", e)

开发

uv run pytest
uv run ruff check .
uv build

License

MIT

About

No description, website, or topics provided.

Resources

License

Stars

0 stars

Watchers

0 watching

Forks

Releases

No releases published

Packages

 
 
 

Contributors

Languages