python3调用企业微信api!开发一款属于自己的企业微信
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, parseCORPID = '' # 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.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)
导入的所有用户例子:
导入的单个用户示例:
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.entriesfor 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查询出父IDp_id = m['id'] # 父IDrequest_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.entriesfor 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) + '】 ' + "同步用户失败,用户未配置【手机号】与【邮箱】!")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=Truedef ad_ou2wx(conn, search_base):"""每一层OU下员工的同步思路:结合上一步代码,将search_base根据每一次生成的子OU变化即可,并将Ou_id带入 """conn.search(search_base, '(objectclass=organizationalUnit)', attributes=['*'])entry = conn.entriesfor 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查询出父IDp_id = m['id'] # 父IDou_id = i+3request_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.distinguishedNamead2wx(conn, sear_base, ou_id)except Exception as e:logging.error(e)break
效果:
完整项目代码获取后台私信小编01
python3调用企业微信api!开发一款属于自己的企业微信相关推荐
- python调用百度翻译-python3调用百度翻译API实时翻译的实例代码
python3调用百度翻译API实现实时翻译 今天需要做一个翻译的工具,找到之前写过的有道翻译,已经不能用了,最后看到百度翻译还不错,不过官方版本是Python2,我需要Python3,就自己写了一个 ...
- insert时调用本身字段_python3调用企业微信api!开发一款属于自己的企业微信
python3调用企业微信api 最后更新时间:2020/5/11 前段时间,我将企业微信官方提供的python接口代码的部分功能修改成了python3的,并且自己也使用并测试过部分功能: 因为并没有 ...
- python企业微信特定用户_python3调用企业微信api!开发一款属于自己的企业微信...
python3调用企业微信api 最后更新时间:2020/5/11 前段时间,我将企业微信官方提供的python接口代码的部分功能修改成了python3的,并且自己也使用并测试过部分功能: 因为并没有 ...
- 基于企业微信api 开发 消息推送提醒 Python与Java
1.关于脚本 身为一个程序猿,不管你从事什么开发,学会使用一门脚本语言可以提高你的开发效率.我目前使用的是Python,因为它支持的库非常多,拿来即用.在这个快节奏的开发,是非常有效率的,所以我非常强 ...
- Python3调用百度翻译API进行英文翻译
一.API的概念 API(Application Programming Interface,应用程序编程接口)是一些预先定义的函数,目的是提供应用程序与开发人员基于某软件或硬件得以访问一组例程的能力 ...
- 微信企业应用java开发_移动办公-创建自定义企业微信应用
关键词:移动办公,O2OA微信办公,企业微信办公,微信办公,手机办公 O2OA平台拥有配套的原生开发的安卓和IOS移动APP,可以以自建应用的方式集成到企业微信,同步企业微信的企业通讯录作为本地组织人 ...
- 企业微信API开发前准备
1,术语介绍 参考文档:http://work.weixin.qq.com/api/doc#10013 企业微信管理后台: https://work.weixin.qq.com/wework_admi ...
- 前端: 开发一款有点意思的仿微信朋友圈应用
儿童节快乐 前言 今天要写的H5朋友圈也是基于笔者开发的cms搭建的,我将仿照微信朋友圈,带大家一起开发一个能发布动态(包括图片上传)的朋友圈应用.有关服务端部分笔者在本文中不会细讲,如果感兴趣的朋友 ...
- java调用高德地图API开发,高德在线地图开发——未完待续
这是目录 一.引入高德地图API 二.高德地图开发 1.定义一个div来存放地图 2.生成地图 3.添加一个跳跃的点 4.添加控件 5.有其他需要的请留言 一.引入高德地图API 高德地图官方示例:h ...
最新文章
- ubuntu18.04.4 中 下载 github 代码 并创建 python 虚拟环境virtualenv
- 支付宝扫一下就能体验的深度学习模型
- Py之keras-resnet:keras-resnet的简介、安装、使用方法之详细攻略
- 调试Spring源码 5.2.2.RELEASE
- linux-shell命令之cp(copy)【拷贝复制】
- Apache2 部署flask项目
- 存数据返回他的序列号id_使用excel进行数据分析
- mysql数据库相关基础知识02
- 以过来人身份给新手博主站长的几点小建议
- 北京航空航天大学、浙江大学等27支海内外高校队伍晋级ASC超算大赛总决赛
- 如何将陈述句变为疑问句
- puppeteer 初步实战 测试
- G1 Concurrent Refinement Thread 在干啥?
- SDHC介绍(收集网上资料-仅供学习之用)
- 基于linux2.6.30.4内核的DM9000网卡驱动编译成模块成功ping通
- 海量GIS文章,软件教程,GIs编程奉送!!(转)
- silverlight java通信_Silverlight使用JavaSocket连接jabber服务器
- 程序人生(一) 初生牛犊
- 微信小程序---- 获取手机号(微信最新更新)
- Windows电脑怎么重装系统?超详细教程!装机不再求人!
热门文章
- Zigbee W-Z-09 模块规格书
- 基于PHP+MySQL保险业务管理系统的设计与实现
- protues有linux客户端吗,串口实现linux系统的终端(用proteus)
- MYSQLg高级-----SQL注入的理解(初级篇)以及如何防止注入
- matlab汽车驱动力与行驶阻力,汽车理论课程设计:基于Matlab的汽车动力性的仿真...
- 1+X云计算平台运维与开发认证(初级)样卷D
- [附源码]Java计算机毕业设计SSM潮购购物网站
- 到了2020年,年薪80w的阿里P7专家,顶尖的技术人才只因做到了这几点
- 计算机一级操作题题库在线,全国计算机一级考试题库操作题
- 【Python学习】自定义对象转JSON