python3调用企业微信api!开发一款属于自己的企业微信

python3调用企业微信api最后更新时间:2020/5/11
前段时间 , 我将企业微信官方提供的python接口代码的部分功能修改成了python3的 , 并且自己也使用并测试过部分功能;
因为并没有将公司的敏感信息抹去 , 所以代码一直没有更新到github;
等度过了springboot的学习期并完成开发任务就来重新整理一版本;
项目地址: RandolphCYG/husky_weworkapi
注意: 目前还没提交修改的测试代码 , 2020/5/11 本周将整理
0.初期代码这是自己写的发消息的测试代码(非接口) , 以测试一下调用企业微接口是不是需要很多东西 , 后来发现 , 完全没这个必要 。
import jsonimport sysimport urllibimport urllib.requestfrom urllib import request, parse CORPID= ''# CorpID 企业IDCORPSECRET= '' # Secret 应用密钥touser= ''# useridcontent = '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 = ';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 = "" + 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 = http://kandian.youth.cn/index/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同步:
python3调用企业微信api!开发一款属于自己的企业微信文章插图
确定一个部门的ID: 测试部门ID:2, 下面通过调API批量从AD域中将用户同步进企业微信
python3调用企业微信api!开发一款属于自己的企业微信文章插图
1.2【魔改企业微信api源码 --tt-darkmode-bgcolor: #131313;">文件位置 C:\Users\randolph\Desktop\wechatapi\api\examples\ADTest.py
1.2.1批量从AD域导入用户到企业微信from ldap3 import Server, Connection, ALLfrom api.src.CorpApi import *from api.src.AbstractApi import *from api.examples.TestConf import *import logging# AD域LDAP_IP = '192.168.255.222'# LDAP服务器IPLDAP_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.entriesfor 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) + '】 ' + "同步用户失败 , 用户未配置【手机号】与【邮箱】!")continueif 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:passelse: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=Trueif __name__ == '__main__':LDAP_CON = ldap_con_ssl(*INF)ad2wx(LDAP_CON, SEARCH_BASE)