我们在企业内部经常会建立一个新的群组进行一个项目的消息收发，但是钉钉除使用管理后台操作外没有办法批量导入名单进行快速建群，一个一个拉人非常痛苦(30人以上群)手发麻。
基于Flask开发了钉钉建群小助手，需要用到requests这个库进行钉钉接口的调用。
基本原理，钉钉后台创建一个企业内部应用（H5快捷方式）获取App密钥以及打开此APP的权限(在管理后台操作)，获取企业凭证后通过用户手机号获取用户id，最后请求建群接口批量将成员拉入建群。
static中的mb.xlsx表格文件存放姓名手机号，用于匹配人名查找到该用户手机号方便获取钉钉用户id 。
templates中存放html页面
leadin.html(不会前端，写的非常烂)
钉钉建群助手
* {
margin: 0;
padding: 0;
}
a {
text-decoration: none;
color: cornflowerblue;
}
.shang {
width: 100%;
height: 150px;
position: absolute;
top: 0;
box-shadow: 0 2px 10px #888888;
}
.zhong {
overflow: auto;
position: absolute;
top: 150px;
bottom: 0;
width: 100%;
}
.tl {
margin-top: 20px;
margin-bottom: 35px;
text-align: center;
font-size: 30px;
font-weight: bold;
}
input {
outline: none;
border: 1px solid #aaa;
width: 140px;
height: 26px;
border-radius: 10px;
margin-left: 10px;
margin-right: 10px;
}
.sub {
width: 80px;
line-height: 26px;
cursor: pointer;
background-color: indianred;
color: white;
font-weight: bold;
transition: 650ms;
}
.sub:hover {
background-color: cornflowerblue;
}
.tlmsg {
text-align: center;
color: indianred;
font-weight: bold;
font-size: 20px;
margin-top: 20px;
}
.up {
text-align: center;
margin-top: 20px;
}
.msg {
font-weight: bold;
color: #aaa;
}
.lb {
margin: 10px auto;
color: white;
width: 200px;
height: 30px;
line-height: 30px;
font-size: 18px;
font-weight: bold;
background-color: green;
display: block;
border-radius: 10px;
transition: 650ms;
}
.lb:hover {
box-shadow: 0 2px 10px #888888;
}
.list {
width: 300px;
margin: 10px auto 0;
border-collapse: collapse;
}
.list td {
text-align: center;
line-height: 50px;
border: 1px solid cornflowerblue;
}
自动建群
批量导入
app.py
from flask import Flask, render_template, request
from data_api import *
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def index():
if request.method == 'GET':
return render_template('leadin.html', **{'msg': '轻松创建'})
else:
name = request.form.get('group')
excel = request.files.get('excel')
excel.save('static/data.xlsx')
data = getData()
print(auto_create_group(name, data))
return f'请在钉钉查看'
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8866, threaded=True)
data_api.py
import json
import requests
from pf_excel import ExcelWork
AppKey = ''
AppSecret = ''
def load_mb():
"""
获取表格数据
:return: list: 手机号列表
"""
excel = ExcelWork('static/mb.xlsx')
data = excel.getRowAll()
excel.close()
mb_dict = {}
for temp in data:
mb_dict[temp[0]] = temp[1]
return mb_dict
MB_DICT = load_mb()
def get_access_token():
"""
获取企业凭证
:return: str
"""
data = requests.get(f'https://oapi.dingtalk.com/gettoken?appkey={AppKey}&appsecret={AppSecret}')
data = json.loads(data.text)
return data['access_token']
def get_userid(mb, token):
"""
根据手机号获取userid
:param mb: str: 手机号
:param token: str: 企业凭证
:return: str: userid
"""
url = f'https://oapi.dingtalk.com/topapi/v2/user/getbymobile?access_token={token}'
cont = {"mobile": mb}
data = requests.post(url, cont)
data = json.loads(data.text)
return data.get('result').get('userid')
def create_group(name, admin_userid, userid_list, token):
"""
建群
:param name: str: 群名
:param admin_userid: str: 管理员id
:param userid_list: list: 群组成员id列表
:param token: str: 企业凭证
:return: dict
"""
url = f'https://oapi.dingtalk.com/chat/create?access_token={token}'
cont = {
"name": name,
"owner": admin_userid,
"useridlist": userid_list
}
data = requests.post(url, json.dumps(cont))
data = json.loads(data.text)
return data
def getData():
"""
获取表格数据
:return: list: 手机号列表
"""
excel = ExcelWork('static/data.xlsx')
data = excel.getColumn(1)
excel.close()
return data[1:]
def auto_create_group(name, data):
"""
批量建群
:param name: str: 群名
:param data: list: 成员姓名
:return:
"""
tk = get_access_token()
userid_list = []
for user in data[1:]:
userid_list.append(get_userid(MB_DICT.get(user), tk))
return create_group(name, get_userid(MB_DICT.get(data[0]), tk), userid_list, tk)
pf_excel.py(新的版本，博客中有旧版本)根据openpyxl封装的操作excel数据的简单接口
"""
基于openpyxl的excel读写模块
提供更简单的读写方式
"""
from openpyxl import load_workbook, Workbook
class ExcelWork:
def __init__(self, filePath, readonly=False, dataonly=True):
"""
:param filePath: str: 文件地址
:param readOnly: bool: 读取模式(默认：False)
:param dataOnly: bool: 数据模式(默认：True)
"""
self.filePath = filePath
self.readonly = readonly
try:
# 加载excel
self.excel = load_workbook(self.filePath, read_only=self.readOnly, data_only=dataOnly)
except FileNotFoundError:
# 写入模式
self.readonly = False
# 创建excel
self.excel = Workbook(self.filePath)
# 创建sheet
self.createSheet('Sheet1')
# 保存excel
self.excel.save(self.filePath)
# 加载excel
self.excel = load_workbook(self.filePath, read_only=self.readOnly, data_only=dataOnly)
# sheet设置为第一个工作表
self.sheet = self.excel.active
def close(self):
"""
保存并退出
:return:
"""
if self.readOnly:
self.excel.close()
else:
self.excel.save(self.filePath)
def createSheet(self, sheetName):
"""
创建工作表
:param sheetName: str: 工作表名
:return:
"""
self.excel.create_sheet(sheetName)
def getSheetTitle(self):
"""
获取当前工作表名称
:return: str: 工作表名称
"""
return self.sheet.title
def getSheetTitleAll(self):
"""
获取excel所有工作表的名称
:return: list: [工作表名称,]
"""
return self.excel.sheetnames
def delSheet(self):
"""
删除当前工作表
:return:
"""
self.excel.remove(self.sheet)
def selectSheet(self, sheetName):
"""
选择工作表, 如果没有将创建
:param sheetName: str: 工作表名
:return:
"""
if sheetName in self.getSheetTitles():
# 选择工作表
self.sheet = self.excel[sheetName]
else:
# 创建工作表
self.createSheet(sheetName)
# 选择工作表
self.sheet = self.excel[sheetName]
def setCell(self, r, c, var):
"""
修改指定行， 列的单元格内容
:param r: int: 行数
:param c: int: 列数
:param var: str: 修改内容
:return:
"""
self.sheet.cell(row=r, column=c, value=var)
def getCell(self, r, c):
"""
获取指定行， 列的单元格内容
:param r: int: 行数
:param c: int: 列数
:return: str: 单元格内容
"""
return self.sheet.cell(row=r, column=c).value
def getRow(self, r):
"""
获取指定行所有数据
:param r: int: 行数
:return: list: [数据,]
"""
rowList = []
for cell in self.sheet[r]:
rowList.append(cell.value)
return rowList
def getColumn(self, c):
"""
获取指定列所有数据
:param c: int: 列数
:return: list: [数据,]
"""
columnList = []
for temp in range(1, self.sheet.max_row + 1):
columnList.append(self.getCell(temp, c))
return columnList
def getRowAll(self):
"""
以行分组获取所有数据
:return: list: [[数据,],]
"""
data = []
table = tuple(self.sheet.rows)
for temp in table:
tempList = []
for cell in temp:
tempList.append(cell.value)
data.append(tempList)
return data
def getColumnAll(self):
"""
以列分组获取所有数据
:return: list: [[数据,],]
"""
data = []
table = tuple(self.sheet.columns)
for temp in table:
tempList = []
for cell in temp:
tempList.append(cell.value)
data.append(tempList)
return data
