685 lines
27 KiB
Python
685 lines
27 KiB
Python
"""
|
|
API 模块
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
from typing import Optional, List, Dict, Any
|
|
from config import Config
|
|
from decimal import Decimal
|
|
import time
|
|
import numpy as np
|
|
|
|
|
|
class NpEncoder(json.JSONEncoder):
|
|
def default(self, obj):
|
|
if isinstance(obj, np.integer):
|
|
return int(obj)
|
|
elif isinstance(obj, np.floating):
|
|
return float(obj)
|
|
elif isinstance(obj, np.ndarray):
|
|
return obj.tolist()
|
|
else:
|
|
return super(NpEncoder, self).default(obj)
|
|
|
|
|
|
def replace_decimals(obj):
|
|
if isinstance(obj, dict):
|
|
return {k: replace_decimals(v) for k, v in obj.items()}
|
|
elif isinstance(obj, list):
|
|
return [replace_decimals(item) for item in obj]
|
|
elif isinstance(obj, Decimal):
|
|
return float(obj) # 或者 str(obj)
|
|
return obj
|
|
|
|
|
|
class API:
|
|
def entry_data_get(self, data: dict, replace: bool = False) -> Dict: # 获取单条表单数据
|
|
"""
|
|
获取单条表单数据
|
|
:param replace: 是否替换字段,默认为关
|
|
:param data: 简道云插件发送过来的data，包含应用id、表单id、数据id等信息
|
|
:return:
|
|
"""
|
|
|
|
url = 'https://api.jiandaoyun.com/api/v5/app/entry/data/get'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 app_key
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
payload = json.dumps({
|
|
"app_id": data['api_key'], # 应用ID
|
|
"entry_id": data['entry_id'], # 表单ID
|
|
"data_id": data['data_id'] # 数据ID
|
|
})
|
|
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
data_get = res.json()
|
|
print(data_get)
|
|
|
|
if replace:
|
|
data_get = self.field_replacement(data, data_get) # 字段替换，由id替换为标签名
|
|
|
|
return data_get
|
|
|
|
def entry_data_list(self, data: dict, replace: bool = False, max_retries: int = 20) -> Dict: # 获取多条表单数据
|
|
"""
|
|
获取多条表单数据
|
|
:param max_retries: 最大重试次数
|
|
:param replace: 是否替换字段
|
|
:param data:
|
|
api_key: 应用id
|
|
entry_id: 表单id
|
|
:return:
|
|
"""
|
|
|
|
url = 'https://api.jiandaoyun.com/api/v5/app/entry/data/list'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 app_key
|
|
'Content-Type': 'application/json'
|
|
}
|
|
all_data_batches = [] # 用于存储每次请求返回的数据批次
|
|
last_data_id = None
|
|
exit_flag = False
|
|
while True:
|
|
payload = json.dumps({
|
|
"app_id": data['api_key'], # 应用ID
|
|
"entry_id": data['entry_id'], # 表单ID
|
|
"limit": 100,
|
|
"data_id": last_data_id
|
|
})
|
|
|
|
# res = requests.post(url=url, data=payload, headers=headers)
|
|
# data_get = res.json() # 此时返回的数据是一个字典
|
|
# if 'data' not in data_get or len(data_get['data']) == 0:
|
|
# break
|
|
#
|
|
# # 将当前批次的数据作为一个整体添加到 all_data_batches 中
|
|
# all_data_batches.extend(data_get['data'])
|
|
#
|
|
# last_data_id = data_get['data'][-1].get('_id')
|
|
# time.sleep(0.2)
|
|
# print(f"已获取 {len(all_data_batches)} 条数据")
|
|
|
|
retries = 0
|
|
while retries <= max_retries:
|
|
try:
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
res.raise_for_status() # 检查HTTP响应状态码，如果不等于200会抛出异常
|
|
data_get = res.json()
|
|
# print("返回结果:", data_get)
|
|
if data_get["data"]:
|
|
all_data_batches.extend(data_get['data'])
|
|
last_data_id = data_get['data'][-1].get('_id')
|
|
print(f"已获取 {len(all_data_batches)} 条数据")
|
|
break # 成功则跳出循环
|
|
else:
|
|
if 'data' not in data_get or len(data_get['data']) == 0:
|
|
exit_flag = True
|
|
break
|
|
print("请求失败, 将重新请求")
|
|
retries += 1
|
|
time.sleep(0.1) # 在重试之间稍作停顿
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"请求异常: {e}, 将重新请求")
|
|
retries += 1
|
|
time.sleep(0.1) # 在重试之间稍作停顿
|
|
if retries > max_retries:
|
|
print(f"超过最大重试次数({max_retries})，放弃此次请求")
|
|
all_data_batches.append(None) # 或者可以选择记录失败的payload以便后续处理
|
|
|
|
if exit_flag:
|
|
break
|
|
|
|
# 构建最终返回的字典
|
|
final_data = {
|
|
'data': all_data_batches # 'data' 键对应的值是列表的列表
|
|
}
|
|
if replace:
|
|
print("进行了替换")
|
|
final_data = self.field_replacement(data, final_data) # 字段替换，由id替换为标签名
|
|
|
|
return final_data
|
|
|
|
@staticmethod
|
|
def entry_widget_list(data: dict) -> Optional[Dict[str, Any]]: # 获取表单字段
|
|
"""
|
|
获取表单字段
|
|
:param data: 简道云插件发送过来的data，包含应用id、表单id、数据id等信息
|
|
:return:
|
|
"""
|
|
|
|
url = 'https://api.jiandaoyun.com/api/v5/app/entry/widget/list'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 app_key
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
payload = json.dumps({
|
|
"app_id": data['api_key'],
|
|
"entry_id": data['entry_id'],
|
|
})
|
|
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
# print(type(res.json()))
|
|
return res.json()
|
|
|
|
def field_replacement(self, data: dict, data_get: dict) -> dict: # 字段替换（id替换为标签名） json_str：请求后的返回数据
|
|
"""
|
|
字段替换，将id替换为标签名，即唯一值替换为表单中显示字段的名字
|
|
:param data: 简道云插件发送过来的data，包含表单id、数据id、应用id
|
|
:param data_get: 简道云请求的数据，一般是根据数据id获取到表单的数据
|
|
:return: 将根据数据id获取到的表单数据，进行替换，返回替换后的数据
|
|
"""
|
|
# 获取表单对应字段标签名称
|
|
widget_list = self.entry_widget_list(data)
|
|
|
|
# 检查widget_list是否有效
|
|
if not widget_list or 'widgets' not in widget_list or not isinstance(widget_list['widgets'], list):
|
|
raise ValueError("映射表没有接受到数据")
|
|
|
|
# 创建一个映射表，将_widget_名称映射到label
|
|
name_to_label = {widget['name']: widget['label'] for widget in widget_list['widgets']}
|
|
print(name_to_label)
|
|
|
|
# 遍历data_get对象中的"data"部分，替换键名
|
|
data = data_get.get('data', {})
|
|
new_data = {}
|
|
|
|
if isinstance(data, list): # 多数据查询返回list
|
|
# 当 data 是列表时执行的代码
|
|
for item in data:
|
|
new_data = {}
|
|
for key, value in item.items():
|
|
if key in name_to_label:
|
|
new_key = name_to_label[key]
|
|
else:
|
|
new_key = key
|
|
new_data[new_key] = value
|
|
item['data'] = new_data # 对每个item进行更新
|
|
|
|
elif isinstance(data, dict): # 单数据查询返回dict
|
|
# 当 data 是字典时执行的代码
|
|
for key, value in data.items():
|
|
if key in name_to_label:
|
|
new_key = name_to_label[key]
|
|
else:
|
|
new_key = key
|
|
new_data[new_key] = value
|
|
data_get['data'] = new_data # 更新JSON对象中的"data"部分
|
|
|
|
# 将更新后的JSON对象转换回JSON字符串
|
|
new_json = json.dumps(data_get, ensure_ascii=False, indent=2)
|
|
|
|
new_json = json.loads(new_json)
|
|
return new_json
|
|
|
|
@staticmethod
|
|
def data_batch_create(data: dict, max_retries: int = 20) -> Optional[requests.Response]: # 新建单条数据
|
|
"""
|
|
新建单条表单数据
|
|
:param max_retries: 最大重试次数
|
|
:param data: 应该包含应用id、表单id，以及新建的数据data['data']
|
|
:return: 返回创建后简道云返回的信息
|
|
"""
|
|
url = 'https://api.jiandaoyun.com/api/v5/app/entry/data/create'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 app_key
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
"""
|
|
data 样式 # 后续优化发送数据样式 目前输入字段，后续优化输入表单名称
|
|
jiandaoyun_data['data'] = {"_widget_1731650067055":{"value":f'{username}{password}'},
|
|
"_widget_1731650067056":{"value": f"{group}"}}
|
|
"""
|
|
|
|
payload = json.dumps({
|
|
"app_id": data['api_key'], # 应用ID
|
|
"entry_id": data['entry_id'], # 表单ID
|
|
"data": data['data'],
|
|
"is_start_workflow": data.get('is_start_workflow', "false"),
|
|
"is_start_trigger": data.get('is_start_trigger', "false"),
|
|
"transaction_id": data.get('transaction_id', "")
|
|
}
|
|
)
|
|
retries = 0
|
|
while retries <= max_retries:
|
|
try:
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
res.raise_for_status() # 检查HTTP响应状态码，如果不等于200会抛出异常
|
|
data_get = res.json()
|
|
print("返回结果:", data_get)
|
|
if res.status_code == 200:
|
|
return data_get
|
|
else:
|
|
print("请求失败, 将重新请求")
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"请求异常: {e}, 将重新请求")
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
if retries > max_retries:
|
|
print(f"超过最大重试次数({max_retries})，放弃此次请求")
|
|
|
|
@staticmethod
|
|
def entry_data_batch_create(
|
|
data: dict,
|
|
chunk_size: int = 90,
|
|
max_retries: int = 20
|
|
) -> List[Optional[requests.Response]]: # 新建多条数据 注意简道云限制1次最多100条数据
|
|
"""
|
|
新建多条数据
|
|
:param max_retries: 最大重试次数，此处设置20次
|
|
:param data:应包含数据id、表单id、以及需要新建的信息，新建信息应该是一个列表
|
|
:param chunk_size: 简道云限制批量新建一次最多100条，这里默认值设置为90条一次
|
|
:return:返回请求后的结果
|
|
"""
|
|
data = replace_decimals(data)
|
|
url = 'https://api.jiandaoyun.com/api/v5/app/entry/data/batch_create'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 appKey
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
"""
|
|
data_list 样式 # 后续优化发送数据样式 目前输入字段，后续优化输入表单名称
|
|
jiandaoyun_data_list['data_list'] = [{"_widget_1731650067055":{"value":f'{username}{password}'},
|
|
"_widget_1731650067056":{"value": f"{group}"}},
|
|
{"_widget_1731650067055":{"value":f'{username}{password}'},
|
|
"_widget_1731650067056":{"value": f"{group}"}}]
|
|
"""
|
|
|
|
# 获取data_list长度
|
|
total_length = len(data['data_list'])
|
|
print(f"多数据写入行数: {total_length}")
|
|
|
|
# 计算需要发送的次数
|
|
num_chunks = (total_length + chunk_size - 1) // chunk_size # //整除向下取证，需要加上chunk_size - 1保证不会有缺失数据
|
|
print(num_chunks)
|
|
data_get_list = []
|
|
for i in range(num_chunks):
|
|
start_index = i * chunk_size
|
|
end_index = min(start_index + chunk_size, total_length)
|
|
payload = json.dumps({
|
|
"app_id": data['api_key'], # 应用ID
|
|
"entry_id": data['entry_id'], # 表单ID
|
|
"data_list": data['data_list'][start_index:end_index],
|
|
"is_start_workflow": data.get('is_start_workflow', "false"),
|
|
"is_start_trigger": data.get('is_start_trigger', "false"),
|
|
}, cls=NpEncoder)
|
|
retries = 0
|
|
while retries <= max_retries:
|
|
try:
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
res.raise_for_status() # 检查HTTP响应状态码，如果不等于200会抛出异常
|
|
data_get = res.json()
|
|
print(i, "返回结果:", data_get)
|
|
if data_get["status"] == "success":
|
|
data_get_list.append(data_get)
|
|
break # 成功则跳出循环
|
|
else:
|
|
print("请求失败, 将重新请求")
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"请求异常: {e}, 将重新请求")
|
|
retries += 1
|
|
time.sleep(0.1) # 在重试之间稍作停顿
|
|
if retries > max_retries:
|
|
print(f"超过最大重试次数({max_retries})，放弃此次请求")
|
|
data_get_list.append(None) # 或者可以选择记录失败的payload以便后续处理
|
|
|
|
return data_get_list
|
|
|
|
@staticmethod
|
|
def entry_data_update(data: dict, max_retries: int = 20) -> dict: # 修改数据
|
|
"""
|
|
修改数据
|
|
:param max_retries: 最大重试次数，此处设置100次
|
|
:param data: 简道云插件发送过来的data，包含应用id、表单id、数据id等信息
|
|
:return: 修改数据后简道云返回的结果
|
|
"""
|
|
url = 'https://api.jiandaoyun.com/api/v5/app/entry/data/update'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 appKey
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
payload = json.dumps({
|
|
"app_id": data['api_key'], # 应用ID
|
|
"entry_id": data['entry_id'], # 表单ID
|
|
"data_id": data['data_id'], # 数据ID
|
|
"data": data['data']
|
|
}
|
|
)
|
|
print(payload)
|
|
data_get = None
|
|
retries = 0
|
|
while retries <= max_retries:
|
|
try:
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
res.raise_for_status() # 检查HTTP响应状态码，如果不等于200会抛出异常
|
|
data_get = res.json()
|
|
print("返回结果:", data_get)
|
|
if res.status_code == 200:
|
|
break # 成功则跳出循环
|
|
else:
|
|
print("请求失败, 将重新请求")
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"请求异常: {e}, 将重新请求")
|
|
retries += 1
|
|
time.sleep(10) # 在重试之间稍作停顿
|
|
if retries > max_retries:
|
|
print(f"超过最大重试次数({max_retries})，放弃此次请求")
|
|
|
|
continue
|
|
return data_get
|
|
|
|
@staticmethod
|
|
def entry_data_delete(data: dict, max_retries: int = 20, ) -> dict:
|
|
"""
|
|
删除单条数据
|
|
:param data: 应包含应用ID、表单ID、数据ID
|
|
:param max_retries: 最大重试次数，默认20
|
|
:return:
|
|
"""
|
|
url = 'https://api.jiandaoyun.com/api/v5/app/entry/data/delete'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 appKey
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
payload = json.dumps({
|
|
"app_id": data['api_key'], # 应用ID
|
|
"entry_id": data['entry_id'], # 表单ID
|
|
"data_id": data['data_id'], # 数据ID
|
|
}
|
|
)
|
|
retries = 0
|
|
delete_status = None
|
|
|
|
while retries <= max_retries:
|
|
try:
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
delete_status = res.json()
|
|
# 手动处理状态码 4001
|
|
if delete_status == {
|
|
"code": 4001,
|
|
"msg": "Data does not exist."
|
|
}:
|
|
print("返回结果:", delete_status)
|
|
break # 成功则跳出循环
|
|
|
|
# 检查其他状态码
|
|
res.raise_for_status() # 只对非 4001 的状态码进行检查
|
|
|
|
print("返回结果:", delete_status)
|
|
if res.status_code == 200:
|
|
break # 成功则跳出循环
|
|
else:
|
|
print("请求失败, 将重新请求")
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"请求异常: {e}, 将重新请求")
|
|
retries += 1
|
|
time.sleep(10) # 在重试之间稍作停顿
|
|
if retries > max_retries:
|
|
print(f"超过最大重试次数({max_retries})，放弃此次请求")
|
|
|
|
continue
|
|
return delete_status
|
|
|
|
@staticmethod
|
|
def entry_data_batch_delete(
|
|
data: dict,
|
|
chunk_size: int = 90,
|
|
max_retries: int = 20
|
|
) -> List[Optional[requests.Response]]: # 新建多条数据 注意简道云限制1次最多100条数据
|
|
"""
|
|
批量删除数据
|
|
:param data: 应包含应用ID、表单ID、数据ID列表
|
|
:param chunk_size:单词删除最大条数，默认90
|
|
:param max_retries:重试次数，默认20
|
|
:return:
|
|
"""
|
|
|
|
data = replace_decimals(data)
|
|
url = 'https://api.jiandaoyun.com/api/v5/app/entry/data/batch_delete'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 appKey
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
# 获取data_list长度
|
|
total_length = len(data['data_ids'])
|
|
print(f"多数据写入行数: {total_length}")
|
|
|
|
# 计算需要发送的次数
|
|
num_chunks = (total_length + chunk_size - 1) // chunk_size # //整除向下取证，需要加上chunk_size - 1保证不会有缺失数据
|
|
print(num_chunks)
|
|
data_get_list = []
|
|
|
|
for i in range(num_chunks):
|
|
start_index = i * chunk_size
|
|
end_index = min(start_index + chunk_size, total_length)
|
|
payload = json.dumps({
|
|
"app_id": data['api_key'], # 应用ID
|
|
"entry_id": data['entry_id'], # 表单ID
|
|
"data_ids": data['data_ids'][start_index:end_index],
|
|
}, cls=NpEncoder)
|
|
|
|
retries = 0
|
|
while retries <= max_retries:
|
|
try:
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
res.raise_for_status() # 检查HTTP响应状态码，如果不等于200会抛出异常
|
|
data_get = res.json()
|
|
print(i, "返回结果:", data_get)
|
|
if data_get["status"] == "success":
|
|
data_get_list.append(data_get)
|
|
break # 成功则跳出循环
|
|
else:
|
|
print("请求失败, 将重新请求")
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"请求异常: {e}, 将重新请求")
|
|
retries += 1
|
|
time.sleep(0.1) # 在重试之间稍作停顿
|
|
if retries > max_retries:
|
|
print(f"超过最大重试次数({max_retries})，放弃此次请求")
|
|
|
|
data_get_list.append(None) # 或者可以选择记录失败的payload以便后续处理
|
|
|
|
return data_get_list
|
|
|
|
@staticmethod
|
|
def workflow_instance_get(data: dict) -> dict:
|
|
"""
|
|
查询实例流程信息
|
|
:param data: 简道云插件发送过来的data，包含应用id
|
|
:return: 查询简道云流程实例信息返回的结果
|
|
"""
|
|
url = 'https://api.jiandaoyun.com/api/v5/workflow/instance/get'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 appKey
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
payload = json.dumps({
|
|
"instance_id": data['data_id'],
|
|
"tasks_type": 1
|
|
}
|
|
)
|
|
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
return res.json()
|
|
|
|
@staticmethod
|
|
def workflow_task_approve(data: dict) -> dict:
|
|
"""
|
|
流程待办提交
|
|
:param data:应包含username、instance_id(data_id)、task_id等信息
|
|
:return:返回简道云流程待办提交的结果
|
|
"""
|
|
url = 'https://api.jiandaoyun.com/api/v1/workflow/task/approve'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 appKey
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
payload = json.dumps({
|
|
"username": data["username"],
|
|
"instance_id": data["instance_id"],
|
|
"task_id": data['task_id'],
|
|
"comment": data.get("comment", "自动提交")
|
|
}
|
|
)
|
|
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
return res.json()
|
|
|
|
@staticmethod
|
|
def workflow_task_hand_over(data: dict, max_retries: int = 10) -> dict | None:
|
|
"""
|
|
流程待办转交
|
|
:param max_retries: 最大重试次数
|
|
:param data:应包含username、instance_id(data_id)、task_id等信息
|
|
:return:返回简道云流程待办转交的结果
|
|
"""
|
|
url = 'https://api.jiandaoyun.com/api/v1/workflow/task/transfer'
|
|
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 appKey
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
payload = json.dumps({
|
|
"username": data["username"], # 当前节点的负责人
|
|
"instance_id": data["instance_id"],
|
|
"task_id": data['task_id'],
|
|
"transfer_username": data['transfer_username'], # 转交人
|
|
"comment": "转交"
|
|
}
|
|
)
|
|
|
|
retries = 0
|
|
while retries <= max_retries:
|
|
try:
|
|
res: requests.Response = requests.post(url=url, data=payload, headers=headers, timeout=10)
|
|
res.raise_for_status() # 检查HTTP响应状态码，如果不等于200会抛出异常
|
|
if res.status_code == 200:
|
|
return res.json()
|
|
else:
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
except requests.exceptions.RequestException as e:
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
if retries > max_retries:
|
|
print(f"超过最大重试次数({max_retries})，放弃此次请求")
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_upload_token(data: dict, max_retries: int = 10) -> dict:
|
|
"""
|
|
获取文件上传凭证
|
|
:param data: 应包含应用ID、表单ID、事务ID
|
|
:return: 返回upload_url、upload_token
|
|
"""
|
|
url = 'https://api.jiandaoyun.com/api/v5/app/entry/file/get_upload_token'
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 appKey
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
payload = json.dumps({
|
|
"app_id": data['api_key'], # 应用ID
|
|
"entry_id": data['entry_id'], # 表单ID
|
|
"transaction_id": data['transaction_id'], # 事务ID
|
|
})
|
|
retries = 0
|
|
|
|
while retries <= max_retries:
|
|
try:
|
|
res = requests.post(url=url, data=payload, headers=headers)
|
|
res.raise_for_status() # 检查HTTP响应状态码，如果不等于200会抛出异常
|
|
res_j = res.json()
|
|
upload_url = res_j['token_and_url_list'][0]['url']
|
|
upload_token = res_j['token_and_url_list'][0]['token']
|
|
print("返回结果:", upload_url, upload_token)
|
|
if res.status_code == 200:
|
|
return {
|
|
'upload_url': upload_url,
|
|
'upload_token': upload_token
|
|
}
|
|
else:
|
|
print("请求失败, 将重新请求")
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"请求异常: {e}, 将重新请求")
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
if retries > max_retries:
|
|
print(f"超过最大重试次数({max_retries})，放弃此次请求")
|
|
|
|
@staticmethod
|
|
def upload_file(data: dict, max_retries: int = 10) -> dict:
|
|
"""
|
|
上传文件
|
|
:param data: 应包含上传文件路径、上传文件url、上传文件token
|
|
:return: 返回上传文件结果
|
|
"""
|
|
url = data['upload_url']
|
|
headers = {
|
|
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 appKey
|
|
# 'Content-Type': 'application/json'
|
|
}
|
|
file_path = data['file_path'] # 上传文件路径
|
|
payload = {
|
|
"token": data['upload_token'], # 上传文件token
|
|
}
|
|
f = open(file_path, 'rb')
|
|
files = {
|
|
"file": f
|
|
}
|
|
|
|
retries = 0
|
|
while retries <= max_retries:
|
|
try:
|
|
res = requests.post(url=url, data=payload, headers=headers, files=files)
|
|
res.raise_for_status() # 检查HTTP响应状态码，如果不等于200会抛出异常
|
|
data_get = res.json()
|
|
print("返回结果:", data_get)
|
|
if res.status_code == 200:
|
|
return data_get
|
|
else:
|
|
print("请求失败, 将重新请求")
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"请求异常: {e}, 将重新请求")
|
|
retries += 1
|
|
time.sleep(3) # 在重试之间稍作停顿
|
|
if retries > max_retries:
|
|
print(f"超过最大重试次数({max_retries})，放弃此次请求")
|
|
|
|
f.close()
|