import json
import requests
class OperateKnowledgeBaseExcel():
robot_code = '' #机器人
agent_id = ''
app_key = '' #机器人密钥 获取方法见我的其他博客
app_secret = '' #机器人密钥
def __init__(self,union_id, workbook_id, worksheet_id=''):
self.union_id = union_id
self.workbook_id = workbook_id # 获取方式：... > 文档信息 > 表格ID
self.worksheet_id = worksheet_id # 获取方式：self.get_workbook_sheet_names() 拿到所有sheet名称和id
if not self.union_id:
raise Exception('请先设置union_id')
self.access_token = self.get_access_token()
def get_access_token(self):
""" 获取token： https://open.dingtalk.com/document/orgapp/obtain-orgapp-token """
headers = {'Content-Type': "application/x-www-form-urlencoded"}
url = f"https://oapi.dingtalk.com/gettoken/?appkey={self.app_key}&appsecret={self.app_secret}"
resp = requests.get(url, headers=headers)
# print(resp.json()['access_token'])
return resp.json()['access_token']
def get_workbook_sheet_names(self):
"""
获取工作表信息，包含工作表名称、工作表id
开发者文档：https://open.dingtalk.com/document/orgapp/obtain-all-worksheets
"""
url = f'https://oapi.dingtalk.com/v1.0/doc/workbooks/{self.workbook_id}/sheets?operatorId={self.union_id}'
headers = {
'Content-Type': "application/json",
'Host': 'api.dingtalk.com',
'x-acs-dingtalk-access-token': self.access_token
}
resp = requests.get(url, headers=headers)
# print(resp.json())
return resp.json()
def get_worksheet_describe(self):
"""
获取工作表的描述
开发者文档：https://open.dingtalk.com/document/orgapp/obtain-worksheet-properties
"""
url = f'https://oapi.dingtalk.com/v1.0/doc/workbooks/{self.workbook_id}/sheets/{self.worksheet_id}?operatorId={self.union_id}'
headers = {
'Content-Type': "application/json",
'Host': 'api.dingtalk.com',
'x-acs-dingtalk-access-token': self.access_token
}
resp = requests.get(url, headers=headers)
# print(resp.json())
return resp.json()
def get_all_cells_area(self):
""" 获取表格的全部区域，如：A1:D4 """
worksheet_describe = self.get_worksheet_describe()
last_row = worksheet_describe['lastNonEmptyRow']
last_column = worksheet_describe['lastNonEmptyColumn']
range_address = f'A1:{get_column_letter(last_column + 1)}{last_row + 1}'
return range_address
def read(self, range_address=False):
"""
读取工作表数据
开发者文档：https://open.dingtalk.com/document/orgapp/get-cell-properties
"""
if not range_address: # 如果没有指定读取区域，则读取整个工作表
range_address = self.get_all_cells_area()
url = f'https://oapi.dingtalk.com/v1.0/doc/workbooks/{self.workbook_id}/sheets/{self.worksheet_id}/ranges/{range_address}?select=values&operatorId={self.union_id}'
headers = {
'Content-Type': "application/json",
'Host': 'api.dingtalk.com',
'x-acs-dingtalk-access-token': self.access_token
}
resp = requests.get(url, headers=headers)
data = pd.DataFrame(columns=resp.json()['values'][0], data=resp.json()['values'][1:])
data.replace('', pd.NA, inplace=True)
return data
def update(self, range_address, values):
"""
更新工作表指定区域的数据
开发者文档：https://open.dingtalk.com/document/orgapp/get-cell-properties
:param range_address: 更新区域，如：A2:B3
:param values: 更新的数据，如：[['张三', 18], ['李四', 19]]
"""
url = f'https://oapi.dingtalk.com/v1.0/doc/workbooks/{self.workbook_id}/sheets/{self.worksheet_id}/ranges/{range_address}?operatorId={self.union_id}'
headers = {
'Content-Type': "application/json",
'Host': 'api.dingtalk.com',
'x-acs-dingtalk-access-token': self.access_token
}
data = {"values": values}
resp = requests.put(url, data=json.dumps(data), headers=headers)
print("更新工作表区域：", resp.json())
def delete(self, range_address=False):
"""
清空工作表指定区域的数据，仅清除数据，不清除格式
开发者文档：https://open.dingtalk.com/document/orgapp/get-cell-properties
"""
if not range_address: # 如果没有指定读取区域，则清空整个工作表
range_address = self.get_all_cells_area()
url = f'https://oapi.dingtalk.com/v1.0/doc/workbooks/{self.workbook_id}/sheets/{self.worksheet_id}/ranges/{range_address}/clearData?operatorId={self.union_id}'
headers = {
'Content-Type': "application/json",
'Host': 'api.dingtalk.com',
'x-acs-dingtalk-access-token': self.access_token
}
resp = requests.post(url, headers=headers)
print('清空工作表区域：', resp.json())
userid = '' #钉钉账号
workid = '' #文档id
def worksheetinsert(df,number=1,worksheetid=''):
if df.empty:
return
# 转换列索引为Excel的列名称
def col_num_to_letter(col_num):
string = ""
while col_num > 0:
col_num, remainder = divmod(col_num-1, 26)
string = chr(65 + remainder) + string
return string
# 获取行数和列数
num_rows, num_cols = df.shape
# 因为包括列名，所以行数需要加1
range = f"A{number}:{col_num_to_letter(num_cols)}{num_rows + number}"
print(range)
# 将数据转换为列表形式
values = df.values.tolist()
values = [df.columns.tolist()] + [[str(item) for item in row] for row in values]
# 构造字典对象
data = values
try:
temp = OperateKnowledgeBaseExcel(userid,workid,worksheetid).read()
OperateKnowledgeBaseExcel(userid,workid,worksheetid).delete()
except:
pass
OperateKnowledgeBaseExcel(userid,workid,worksheetid).update(range,data)
df =
excelsheet_json = OperateKnowledgeBaseExcel(userid,workid).get_workbook_sheet_names()
excelsheet = {item['name']: item['id'] for item in excelsheet_json['value']}
#指定表sheet名插入
worksheetinsert(df,number=1,worksheetid=excelsheet['XXX'])-
获取访问令牌：通过设置机器人的密钥和密钥获取访问令牌。
-
获取工作表信息：获取工作表的名称和ID。
-
获取工作表描述：获取工作表的描述信息。
-
获取全部单元格区域：获取工作表的全部区域。
-
读取数据：读取指定区域的数据，并将其转换为DataFrame格式返回。
-
更新数据：更新指定区域的数据。
-
清空数据：清空指定区域的数据，但保留格式。
-
插入数据到指定表格：将DataFrame格式的数据插入到指定的工作表中。
