python3调用企业微信api

最后更新时间：2020/5/11

前段时间，我将企业微信官方提供的python接口代码的部分功能修改成了python3的，并且自己也使用并测试过部分功能；

因为并没有将公司的敏感信息抹去，所以代码一直没有更新到github；

等度过了springboot的学习期并完成开发任务就来重新整理一版本；

项目地址： RandolphCYG/husky_weworkapi

注意： 目前还没提交修改的测试代码，2020/5/11 本周将整理

0.初期代码

这是自己写的发消息的测试代码（非接口），以测试一下调用企业微接口是不是需要很多东西，后来发现，完全没这个必要。

import json
import sys
import urllib
import urllib.request
from urllib import request, parse
 
CORPID  = ''		# CorpID 企业ID
CORPSECRET  = ''	# Secret 应用密钥
touser  = ''		# userid
content = 'WECHAT2LDAP 测试'
headers = {
   "User-Agent":"Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36"}
 
 
def getToken(corpid, corpsecret):
    """功能获取access_token corpid:企业ID corpsecret:应用密钥 """
    url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=%s&corpsecret=%s' % (corpid, corpsecret)
    req = urllib.request.Request(url, headers=headers)
    results = json.loads(urllib.request.urlopen(req).read())
    print(results)
    return results['access_token']
 

def sendmsg(access_token, content):
    """功能:发送消息 """
    url = "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" + access_token
    """ touser 成员 @all 就是所有 toparty 部门ID @all 就是所有 msgtype 文本类型 agentid 企业应用ID content 内容 safe 是否保密 0是不保密 """
    values = {
   
               "touser"  : touser,
               "toparty" : '2',
               "msgtype" : "text",
               "agentid" : 1000005,		# 新建企业微信应用
               "text"    : {
   
                            "content" : content
                           },
     "safe"    :"0"
    }
    send_data = json.dumps(values).encode()
    send_request = urllib.request.Request(url, send_data, headers=headers)
    response = json.loads(urllib.request.urlopen(send_request).read())
    print(response)
    if response['errcode'] == 0:
        print('发送消息成功')
 
if __name__ == '__main__':
    access_token = getToken(CORPID, CORPSECRET)
    content = "发送消息测试"
    sendmsg(access_token, content)

1.python3企业微信接口代码魔改（例子）

1.1企业微信准备

通讯录同步 应用开启API同步：

确定一个部门的ID： 测试部门ID：2 ，下面通过调API批量从AD域中将用户同步进企业微信

1.2【魔改企业微信api源码&使用】准备

文件位置 C:\Users\randolph\Desktop\wechatapi\api\examples\ADTest.py

1.2.1批量从AD域导入用户到企业微信

from ldap3 import Server, Connection, ALL
from api.src.CorpApi import *
from api.src.AbstractApi import *
from api.examples.TestConf import *
import logging

# AD域
LDAP_IP = '192.168.255.222'  										# LDAP服务器IP
LDAP_ADMIN_USER = 'CN=Administrator,CN=Users,DC=bilibili,DC=com'  	# LDAP服务器管理员账户
LDAP_ADMIN_PWD = 'QQqq#123'  										# LDAP服务器管理员密码
INF = (LDAP_IP, LDAP_ADMIN_USER, LDAP_ADMIN_PWD)					# AD域信息汇总
SEARCH_BASE = "OU=B站,DC=bilibili,DC=com"							# 搜索请求的基础,在这个OU下开始搜索
# 企业微信CorpApi接口
api = CorpApi(TestConf['CORP_ID'], TestConf['CONTACT_SYNC_SECRET'])
# 日志设置
LOG_FORMAT = "%(asctime)s %(levelname)s %(filename)s %(lineno)d %(message)s"
logging.basicConfig(filename='localAD.log', level=logging.INFO, format=LOG_FORMAT)


def ad2wx(conn, search_base):
    """从AD域到企业微信导入用户(可邀请)，修改了请求包体，测试通过; 代码质量需要再次检查优化; """
    conn.search(search_base, '(objectclass=person)', attributes=['*'])		# 从AD域搜用户并遍历
    entry = conn.entries
    for i, user in enumerate(entry):
        # print(i+1, user)
        try:
            request_package = {
   
                "userid": str(user.sAMAccountName),  				# 由AD的必填字段登录名[sAMAccountName]到微信的[账号]字段
                "name": re.sub("[A-Za-z0-9]", "", str(user.cn)),  	# 由AD的必填字段cn例[小明2314] 截取数字之外的文字[小明]
                "enable": 1,										# 启用用户
                "to_invite": False,         						# 默认邀请,测试时候不邀请
                "department": [2],          						# 部门是[测试]，还没有改成同步部门【todo】
            }
            if hasattr(user, 'telephoneNumber') or hasattr(user, 'mail'):       # 手机号/邮箱必须有一项（邀请需要）
                if hasattr(user, 'telephoneNumber'):
                    request_package['mobile'] = str(user.telephoneNumber)
                if hasattr(user, 'mail'):
                    request_package['email'] = str(user.mail)
            else:
                logging.error('【' + str(user.sAMAccountName) + '】 ' + "同步用户失败，用户未配置【手机号】与【邮箱】！")
                continue
            if hasattr(user, 'title'):
                request_package['position'] = str(user.title)
                request_package['external_position'] = str(user.title)
            response = api.httpCall(CORP_API_TYPE['USER_CREATE'], request_package)
            if response['errcode'] == 0:
                pass
            else:
                print("@@@@创建错误，已经存在或其他错误" + str(user.displayName))
        except ApiException as e:
            logging.debug(e.errCode, e.errMsg)
        except Exception as e:
            logging.debug('【' + str(user.sAMAccountName) + '】 ' + str(e), exc_info=True)  # ,stack_info=True


if __name__ == '__main__':
    LDAP_CON = ldap_con_ssl(*INF)
    ad2wx(LDAP_CON, SEARCH_BASE)

导入的所有用户例子：

导入的单个用户示例：

1.2.2批量从AD域导入OU到企业微信

def ad_ou2wx(conn, search_base):
    """每一层OU下员工的同步思路：结合上一步代码，将search_base根据每一次生成的子OU变化即可 """
    # TODO: 检查优化此部分代码，带上对应OU下员工的同步导入
    conn.search(search_base, '(objectclass=organizationalUnit)', attributes=['distinguishedName'])
    entry = conn.entries
    for i, c in enumerate(entry):
        ou_list = [x.split('OU=')[1] for x in str(c.distinguishedName).split(',')[:-2][::-1]]
        ou_list.insert(0, '测试')
        print(i+3, ou_list)
        response = api.httpCall(CORP_API_TYPE['DEPARTMENT_LIST'])
        for m in response['department']:
            if m['name'] == ou_list[-2]:            # m是查询出部门列表信息，根据部门名字name查询出父ID
                p_id = m['id']                      # 父ID
                request_package = {
   
                   "name": ou_list[-1],             # 必须,中文名
                   "parentid": p_id,                # 必须,父id
                   "id": i+3,                       # 不必须，子id，为了方便下一层的创建，部门的ID为自然数
                }
                try:
                    response = api.httpCall(CORP_API_TYPE['DEPARTMENT_CREATE'], request_package)
                    if response['errcode'] == 0:
                        logging.info('创建【' + ou_list[-1] + '】成功')
                except Exception as e:
                    logging.error(e)
                break                               # 找到父子对应信息，就可以创建完子的部门，然后中断内层循环即可


if __name__ == '__main__':
    LDAP_CON = ldap_con_ssl(*INF)
    # ad2wx(LDAP_CON)
    ad_ou2wx(LDAP_CON, SEARCH_BASE)

企业微信批量导入成功举例：

日志举例：

附带批量删除部门department的代码【注意部门下不能有OU或人员，没写删这些的，测试时候经常需要批量删除空的depart】：

for n in range(3, 8)[::-1]:
        print(n)
        req_pack = {
   
            'id': str(n),
        }
        res = api.httpCall(CORP_API_TYPE['DEPARTMENT_DELETE'], req_pack)

1.2.3 【总结】批量从AD域导入OU以及OU下面的用户到企业微信

将上面的流程优化整合了一下，会在【测试】部门下面完整的 将AD域的多级OU及每一层OU下的用户导入企业微信 ！

待优化点：错误处理、search搜索参数（保证内存占用尽量低）

from ldap3 import Server, Connection, ALL, ASYNC, BASE, LEVEL, SUBTREE
from api.src.CorpApi import *
from api.src.AbstractApi import *
from api.examples.TestConf import *
import logging

# AD域
LDAP_IP = '192.168.255.222'  										# LDAP服务器IP
LDAP_ADMIN_USER = 'CN=Administrator,CN=Users,DC=bilibili,DC=com'  	# LDAP服务器管理员账户
LDAP_ADMIN_PWD = 'QQqq#123'  										# LDAP服务器管理员密码
INF = (LDAP_IP, LDAP_ADMIN_USER, LDAP_ADMIN_PWD)					# AD域信息汇总
SEARCH_BASE = "OU=B站,DC=bilibili,DC=com"							# 搜索请求的基础,在这个OU下开始搜索
# 企业微信CorpApi接口
api = CorpApi(TestConf['CORP_ID'], TestConf['CONTACT_SYNC_SECRET'])
# 日志设置
LOG_FORMAT = "%(asctime)s %(levelname)s %(filename)s %(lineno)d %(message)s"
logging.basicConfig(filename='localAD.log', level=logging.INFO, format=LOG_FORMAT)


def ad2wx(conn, search_base, ou_id):
    """从AD域到企业微信导入用户(可邀请)，修改了请求包体，测试通过; 代码质量需要再次检查优化; """
    conn.search(str(search_base), '(objectclass=person)', attributes=['*'], search_scope=LEVEL)		# 从AD域搜用户并遍历,LEVEL代表只搜索此OU下的用户
    entry = conn.entries
    for i, user in enumerate(entry):
        print(i+1, user.sAMAccountName)
        try:
            request_package = {
   
                "userid": str(user.sAMAccountName),  				# 由AD的必填字段登录名[sAMAccountName]到微信的[账号]字段
                "name": re.sub("[A-Za-z0-9]", "", str(user.cn)),  	# 由AD的必填字段cn例[小明2314] 截取数字之外的文字[小明]
                "enable": 1,										# 启用用户
                "to_invite": False,         						# 默认邀请,测试时候不邀请
                "department": [ou_id],          					# 部门是对应AD域中所在OU
            }
            if hasattr(user, 'telephoneNumber') or hasattr(user, 'mail'):       # 手机号/邮箱必须有一项（邀请需要）
                if hasattr(user, 'telephoneNumber'):
                    request_package['mobile'] = str(user.telephoneNumber)
                if hasattr(user, 'mail'):
                    request_package['email'] = str(user.mail)
            else:
                logging.error('【' + str(user.sAMAccountName) + '】 ' + "同步用户失败，用户未配置【手机号】与【邮箱】！")
                continue
            if hasattr(user, 'title'):
                request_package['position'] = str(user.title)
                request_package['external_position'] = str(user.title)
            response = api.httpCall(CORP_API_TYPE['USER_CREATE'], request_package)
            if response['errcode'] == 0:
                pass
            else:
                print("@@@@创建错误，已经存在或其他错误" + str(user.displayName))
        except ApiException as e:
            logging.debug(e.errCode, e.errMsg)
        except Exception as e:
            logging.debug('【' + str(user.sAMAccountName) + '】 ' + str(e), exc_info=True)  # ,stack_info=True


def ad_ou2wx(conn, search_base):
    """每一层OU下员工的同步思路：结合上一步代码，将search_base根据每一次生成的子OU变化即可，并将Ou_id带入 """
    conn.search(search_base, '(objectclass=organizationalUnit)', attributes=['*'])
    entry = conn.entries
    for i, c in enumerate(entry):
        # print(c.distinguishedName)
        ou_list = [x.split('OU=')[1] for x in str(c.distinguishedName).split(',')[:-2][::-1]]
        ou_list.insert(0, '测试')
        print(i+3, ou_list)
        response = api.httpCall(CORP_API_TYPE['DEPARTMENT_LIST'])
        for m in response['department']:
            if m['name'] == ou_list[-2]:            # m是查询出部门列表信息，根据部门名字name查询出父ID
                p_id = m['id']                      # 父ID
                ou_id = i+3
                request_package = {
   
                   "name": ou_list[-1],             # 必须,中文名
                   "parentid": p_id,                # 必须,父id
                   "id": ou_id,                       # 不必须，子id，为了方便下一层的创建，部门的ID为自然数
                }
                try:
                    response = api.httpCall(CORP_API_TYPE['DEPARTMENT_CREATE'], request_package)
                    if response['errcode'] == 0:
                        logging.info('创建【' + ou_list[-1] + '】成功')
                        # 这里可以将OU下面的用户查询并导入
                        sear_base = c.distinguishedName
                        ad2wx(conn, sear_base, ou_id)
                except Exception as e:
                    logging.error(e)
                break

效果：

完整项目代码获取后台私信小编01

python调用企业微信接口实现关联添加用户

# --*-- coding: utf-8 --*--

import json
import urllib2

coreID = 
secret = 
apisecret = 
agent_id = 

获取token：

def getToken(coreID, apisecret):
    url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=%s&corpsecret=%s' % (coreID, apisecret)
    req = urllib2.Request(url)
    result = urllib2.urlopen(req)
    access_token = json.loads(result.read())
    #print access_token['access_token']
    return access_token['access_token']

添加用户：

def AddWechartUser(userid, name, mobile, email, groupid):
    token = getToken(coreID, apisecret)
    url = 'https://qyapi.weixin.qq.com/cgi-bin/user/create?access_token=%s' % token
    values = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
        "email": email,
        "department": groupid,
    }
    add_user = json.dumps(values)
    request = urllib2.Request(url, add_user)
    response = json.loads(urllib2.urlopen(request).read())

    #print response

    if response['errcode'] == 0:
        return "添加成功"
    return response['errmsg']

发送告警：

def SendMsg(msg, touser, toparty):
    #这里以text为例，touser和toparty不可同时为空
    token = getToken(coreID, secret)
    msgtype = "text"
    url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=%s' % token
    values = {
        "touser": touser,
        "toparty": toparty,
        "msgtype": msgtype,
        "agentid": agent_id,
        "text": {
            "content": msg
        }
    }
    send = json.dumps(values)
    request = urllib2.Request(url, send)
    response = json.loads(urllib2.urlopen(request).read())

    if response['errcode'] != 0:
        return response['errmsg']

    return "发送成功"