职贝云数AI新零售门户

标题: Python企业微信集成教程 [打印本页]

作者: 6KjHg7gd 时间: 2025-1-22 12:45
标题: Python企业微信集成教程
大家好！我是你们的Python教师。明天我们要学习如何完成Python与企业微信的集成。就像一位纯熟的通讯员，我们要学会如何发送音讯、管理通讯录、处理事情等技能。经过这些技巧，我们可以完成更智能的办公自动化。让我们末尾这次企业微信集成之旅吧！
一、基础配置模块

首先，我们来完成基础配置的功能：
import requests
import json
import logging
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
import time
import hashlib

class WeChatConfig:
def __init__(self,
               corp_id: str,
               corp_secret: str):
      self.corp_id = corp_id
      self.corp_secret = corp_secret
      self.access_token = None
      self.token_expires = datetime.now()
      self.setup_logging()

def setup_logging(self):
      """配置日志记录"""
      logging.basicConfig(
         level=logging.INFO,
         format='%(asctime)s - %(levelname)s - %(message)s'
      )

def get_access_token(self) -> str:
      """获取访问令牌"""
      try:
         # 检查令牌能否过期
         if (self.access_token and
            datetime.now() < self.token_expires):
            return self.access_token

         # 央求新的令牌
         url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
         params = {
            "corpid": self.corp_id,
            "corpsecret": self.corp_secret
         }

         response = requests.get(url, params=params)
         result = response.json()

         if result.get("errcode") == 0:
            self.access_token = result["access_token"]
            self.token_expires = (
                  datetime.now() +
                  timedelta(seconds=7000)  # 保险起见，比2小时少些
            )
            return self.access_token
         else:
            raise Exception(
                  f"获取令牌失败: {result}"
            )

      except Exception as e:
         logging.error(f"令牌获取失败: {str(e)}")
         raise

class WeChatAPI:
def __init__(self, config: WeChatConfig):
      self.config = config

def _request(self,
               method: str,
               url: str,
               **kwargs) -> Dict:
      """发送央求"""
      try:
         # 添加访问令牌
         params = kwargs.get('params', {})
         params['access_token'] = self.config.get_access_token()
         kwargs['params'] = params

         # 发送央求
         response = requests.request(
            method,
            url,
            **kwargs
         )
         result = response.json()

         # 检查呼应
         if result.get("errcode", 0) != 0:
            raise Exception(
                  f"央求失败: {result}"
            )

         return result

      except Exception as e:
         logging.error(f"央求失败: {str(e)}")
         raise

# 运用示例
config = WeChatConfig(
"your_corp_id",
"your_corp_secret"
)
api = WeChatAPI(config)

小贴士：记得定期刷新访问令牌，避免过期！

二、音讯发送模块

让我们完成音讯发送功能：
class MessageSender:
def __init__(self, api: WeChatAPI):
      self.api = api

def send_text(self,
               agentid: int,
               content: str,
               touser: Optional[str] = "@all",
               toparty: Optional[str] = None,
               totag: Optional[str] = None
               ) -> bool:
      """发送文本音讯"""
      try:
         url = "https://qyapi.weixin.qq.com/cgi-bin/message/send"

         data = {
            "msgtype": "text",
            "agentid": agentid,
            "text": {"content": content},
            "touser": touser
         }

         if toparty:
            data["toparty"] = toparty
         if totag:
            data["totag"] = totag

         result = self.api._request(
            "POST",
            url,
            json=data
         )

         return True

      except Exception as e:
         logging.error(f"音讯发送失败: {str(e)}")
         return False

def send_markdown(self,
                  agentid: int,
                  content: str,
                  touser: Optional[str] = "@all"
                  ) -> bool:
      """发送Markdown音讯"""
      try:
         url = "https://qyapi.weixin.qq.com/cgi-bin/message/send"

         data = {
            "msgtype": "markdown",
            "agentid": agentid,
            "markdown": {"content": content},
            "touser": touser
         }

         result = self.api._request(
            "POST",
            url,
            json=data
         )

         return True

      except Exception as e:
         logging.error(f"Markdown发送失败: {str(e)}")
         return False

# 运用示例
sender = MessageSender(api)

# 发送文本音讯
sender.send_text(
1000002,  # 运用ID
"你好，这是一条测试音讯",
"@all"
)

# 发送Markdown音讯
sender.send_markdown(
1000002,
"**重要告诉**\n> 请于明天下班前提交周报"
)
三、通讯录管理模块

如今我们完成通讯录管理功能：
class ContactManager:
def __init__(self, api: WeChatAPI):
      self.api = api

def get_department_list(self,
                        id: Optional[int] = None
                        ) -> List[Dict]:
      """获取部门列表"""
      try:
         url = "https://qyapi.weixin.qq.com/cgi-bin/department/list"

         params = {}
         if id is not None:
            params['id'] = id

         result = self.api._request(
            "GET",
            url,
            params=params
         )

         return result['department']

      except Exception as e:
         logging.error(f"获取部门列表失败: {str(e)}")
         return []

def get_user_list(self,
                  department_id: int,
                  fetch_child: bool = False
                  ) -> List[Dict]:
      """获取部门成员列表"""
      try:
         url = "https://qyapi.weixin.qq.com/cgi-bin/user/list"

         params = {
            'department_id': department_id,
            'fetch_child': 1 if fetch_child else 0
         }

         result = self.api._request(
            "GET",
            url,
            params=params
         )

         return result['userlist']

      except Exception as e:
         logging.error(f"获取成员列表失败: {str(e)}")
         return []

def create_user(self,
                  user_data: Dict[str, Any]
                  ) -> bool:
      """创建成员"""
      try:
         url = "https://qyapi.weixin.qq.com/cgi-bin/user/create"

         result = self.api._request(
            "POST",
            url,
            json=user_data
         )

         return True

      except Exception as e:
         logging.error(f"创建成员失败: {str(e)}")
         return False

# 运用示例
contact = ContactManager(api)

# 获取部门列表
departments = contact.get_department_list()

# 获取成员列表
users = contact.get_user_list(1)  # 获取部门ID为1的成员

# 创建新成员
new_user = {
"userid": "zhangsan",
"name": "张三",
"mobile": "13800000000",
"department": [1],
"email": "zhangsan@example.com"
}
contact.create_user(new_user)

留意事项：

保护敏感信息处理API限流记录操作日志定期同步数据

练习题

适用技巧

def generate_message_template(
template_type: str,
**kwargs
) -> str:
"""生成音讯模板"""
templates = {
      'notice': (
         "📢 告诉：{title}\n"
         "━━━━━━━━━━\n"
         "{content}\n"
         "发布工夫：{time}"
      ),
      'report': (
         "📊 {report_type}报告\n"
         "━━━━━━━━━━\n"
         "部门：{department}\n"
         "工夫：{time}\n"
         "概况：{content}"
      )
}
return templates[template_type].format(**kwargs)

def batch_send_message(
sender: MessageSender,
messages: List[Dict]
) -> Dict[str, int]:
"""批量发送音讯"""
results = {
      'success': 0,
      'failed': 0
}
for msg in messages:
      success = sender.send_text(**msg)
      if success:
         results['success'] += 1
      else:
         results['failed'] += 1
return results

def check_sync_status(
local_data: List[Dict],
remote_data: List[Dict],
key: str = 'userid'
) -> Dict[str, List]:
"""检查同步形态"""
local_ids = {item[key] for item in local_data}
remote_ids = {item[key] for item in remote_data}

return {
      'to_add': list(remote_ids - local_ids),
      'to_delete': list(local_ids - remote_ids),
      'to_check': list(local_ids & remote_ids)
}

小伙伴们，明天的Python学习之旅就到这里啦！记得动手敲代码，有成绩随时在评论区问我哦。祝大家学习愉快，Python学习节节高！

欢迎光临职贝云数AI新零售门户 (https://www.taojin168.com/cloud/)