panda/jdy_fastapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.客户信息修改，将硬编码改为动态取值

Browse Source 
2.新增项目批量停用、材料批量修改功能
This commit is contained in:
panda
2026-01-08 10:18:18 +08:00
parent 11e4151395
commit c4c4ccc7e9
7 changed files with 300 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/module/F6_Plugin_module.py
+115 -27
View File
@@ -1,5 +1,5 @@
"""
F6 插件模块
F6 后台执行模块
本模块提供 F6 插件相关的功能，包括：
- 文件上传和校验
@@ -7,6 +7,8 @@ F6 插件模块
- 历史记录删除
- 客户信息管理
- 车辆信息管理
- 项目信息批量启停
- 材料信息批量修改
依赖:
- requests: HTTP 请求
@@ -31,6 +33,10 @@ from app.tasks.delete_tasks import (
delete_customer_background,
delete_car_background
)
from app.tasks.material_tasks import (
batch_disable_projects,
batch_modify_materials,
)
from app.tasks.customer_tasks import modify_customer_info_background
from app.tasks.bi_tasks import bi_task_background
@@ -65,26 +71,26 @@ class F6PluginModule:
Returns:
tuple: 包含文件保存路径和处理后的数据的元组。如果文件保存成功，返回保存路径和数据；如果失败，返回 None 和数据。
"""
data = api_instance.entry_data_get(data=data,replace= True)
data = api_instance.entry_data_get(data=data, replace=True)
print(data)
try:
# 安全地访问附件信息
data_dict = data.get('data', {})
attachments = data_dict.get('附件', [])
if not attachments or len(attachments) == 0:
print('上传url未读取到，或无上传文件: 附件列表为空')
save_path = ''
return save_path, data
first_attachment = attachments[0]
url = first_attachment.get('url')
if not url:
print('上传url未读取到，或无上传文件: URL为空')
save_path = ''
return save_path, data
print(url)
except (KeyError, IndexError, TypeError) as e:
print(f'上传url未读取到，或无上传文件:{e}')
@@ -118,8 +124,7 @@ class F6PluginModule:
else:
return None, data
def check_file(self, data: Dict[str, Any]) -> Dict[str, str]: # 校验上传文件
def check_file(self, data: Dict[str, Any]) -> dict[str, str] | None: # 校验上传文件
"""
校验上传文件。
@@ -143,7 +148,7 @@ class F6PluginModule:
# 安全地获取 Action 字段
data_dict = data1.get('data', {})
action = data_dict.get('Action（隐藏）')
if not action:
return {'msg': '缺少Action字段，无法校验文件'}
@@ -176,7 +181,6 @@ class F6PluginModule:
else:
return {'msg': '当前节点无附件上传', 'check': '是'}
@staticmethod
def create_brand(data: Dict[str, Any]) -> Dict[str, str]:
"""
@@ -191,7 +195,7 @@ class F6PluginModule:
Returns:
Dict[str, str]: 包含执行状态的字典，{'msg': '正在执行', 'msg_details': '正在执行，请稍后看结果'}
"""
entry_data = api_instance.entry_data_get(data=data,replace= True)
entry_data = api_instance.entry_data_get(data=data, replace=True)
print('执行 品牌批量新建')
username = entry_data['data']['账号']
password = entry_data['data']['密码']
@@ -232,7 +236,7 @@ class F6PluginModule:
Returns:
Dict[str, str]: 包含执行状态的字典
"""
entry_data = api_instance.entry_data_get(data=data,replace= True)
entry_data = api_instance.entry_data_get(data=data, replace=True)
username = entry_data['data']['账号']
password = entry_data['data']['密码']
company_name = entry_data['data']['公司名称']
@@ -279,7 +283,7 @@ class F6PluginModule:
Dict[str, str]: 包含执行状态的字典
"""
print('执行 删除客户')
entry_data = api_instance.entry_data_get(data=data,replace= True)
entry_data = api_instance.entry_data_get(data=data, replace=True)
username = entry_data['data']['账号']
password = entry_data['data']['密码']
company_name = entry_data['data']['公司名称']
@@ -298,7 +302,8 @@ class F6PluginModule:
thread = threading.Thread(target=delete_customer_background,
args=(data, cookies, total,))
thread.start()
return {'msg': '正在执行中', 'msg_details': f'总计{total}条数据，8-20点3.5s一条数据，其余时间1.5s一条数据'}
return {'msg': '正在执行中',
'msg_details': f'总计{total}条数据，8-20点3.5s一条数据，其余时间1.5s一条数据'}
else:
return {'msg': '未执行', 'msg_details': '无客户信息'}
else:
@@ -318,7 +323,7 @@ class F6PluginModule:
Returns:
Dict[str, str]: 包含执行状态的字典
"""
entry_data = api_instance.entry_data_get(data=data,replace= True)
entry_data = api_instance.entry_data_get(data=data, replace=True)
username = entry_data['data']['账号']
password = entry_data['data']['密码']
company_name = entry_data['data']['公司名称']
@@ -351,20 +356,21 @@ class F6PluginModule:
else:
return {'msg': '未执行', 'msg_details': '登录失败'}
def modify_customer_info(self, data: Dict[str, str]):
@staticmethod
def modify_customer_info(data: Dict[str, str]):
"""
修改客户信息
从简道云获取修改客户信息请求，读取 Excel 文件，并在后台线程中批量修改客户信息。
立即返回"正在执行中"的提示，实际修改在后台线程中执行。
Args:
data: 包含表单ID(api_key)、表单ID(entry_id)、数据ID(data_id)的字典
Returns:
Dict[str, str]: 包含执行状态的字典
"""
entry_data = api_instance.entry_data_get(data=data,replace= True)
entry_data = api_instance.entry_data_get(data=data, replace=True)
username = entry_data['data']['账号']
password = entry_data['data']['密码']
company_name = entry_data['data']['公司名称']
@@ -389,6 +395,90 @@ class F6PluginModule:
else:
return {'msg': '未执行', 'msg_details': 'cookies获取失败'}
@staticmethod
def disable_projects(data: Dict[str, Any]) -> Dict[str, str]:
"""
项目批量启停
从简道云获取项目批量启停请求，读取 Excel 文件，并在后台线程中批量启停项目。
立即返回"正在执行"的提示，实际创建在后台线程中执行。
Args:
data: 包含表单ID(api_key)、表单ID(entry_id)、数据ID(data_id)的字典
Returns:
Dict[str, str]: 包含执行状态的字典，{'msg': '正在执行', 'msg_details': '正在执行，请稍后看结果'}
"""
entry_data = api_instance.entry_data_get(data=data, replace=True)
print('执行 项目批量停用/启用')
username = entry_data['data']['账号']
password = entry_data['data']['密码']
company_name = entry_data['data']['公司名称']
save_path = entry_data['data']['文件保存地址']
option = entry_data['data']['项目材料批量操作']
login_response = F6Module.login_in(username, password, company_name)
if login_response is None:
return {'msg': '登录失败'}
try:
df = pd.read_excel(save_path, sheet_name=0, dtype='string')
except Exception as e:
return {'msg': f'读取Excel文件失败: {str(e)},文件路径：{save_path}'}
cookies = requests.utils.dict_from_cookiejar(login_response.cookies)
try:
thread = threading.Thread(target=batch_disable_projects,
args=(data, cookies, df, save_path,option))
thread.start()
except Exception as e:
print(f'创建线程失败: {str(e)}')
return {'msg': '正在执行', 'msg_details': '正在执行，请稍后看结果'}
@staticmethod
def disable_material(data: Dict[str, Any]) -> Dict[str, str]:
"""
材料批量启停
从简道云获取材料批量启停请求，读取 Excel 文件，并在后台线程中批量启停材料。
立即返回"正在执行"的提示，实际创建在后台线程中执行。
Args:
data: 包含表单ID(api_key)、表单ID(entry_id)、数据ID(data_id)的字典
Returns:
Dict[str, str]: 包含执行状态的字典，{'msg': '正在执行', 'msg_details': '正在执行，请稍后看结果'}
"""
entry_data = api_instance.entry_data_get(data=data, replace=True)
print('执行 材料批量停用/启用')
username = entry_data['data']['账号']
password = entry_data['data']['密码']
company_name = entry_data['data']['公司名称']
save_path = entry_data['data']['文件保存地址']
option = entry_data['data']['项目材料批量操作']
login_response = F6Module.login_in(username, password, company_name)
if login_response is None:
return {'msg': '登录失败'}
try:
df = pd.read_excel(save_path, sheet_name=0, dtype='string')
except Exception as e:
return {'msg': f'读取Excel文件失败: {str(e)},文件路径：{save_path}'}
cookies = requests.utils.dict_from_cookiejar(login_response.cookies)
try:
thread = threading.Thread(target=batch_modify_materials,
args=(data, cookies, df, save_path, option))
thread.start()
except Exception as e:
print(f'创建线程失败: {str(e)}')
return {'msg': '正在执行', 'msg_details': '正在执行，请稍后看结果'}
@staticmethod
def bi_task(data: Dict[str, Any]) -> Dict[str, str]:
"""
@@ -403,15 +493,15 @@ class F6PluginModule:
Returns:
Dict[str, str]: 包含执行状态的字典，{'msg': '正在执行', 'msg_details': '正在执行，请稍后看结果'}
"""
entry_data = api_instance.entry_data_get(data=data,replace= True)
entry_data = api_instance.entry_data_get(data=data, replace=True)
print('执行 BI任务')
# 获取必要的参数（根据实际需求调整）
username = entry_data['data'].get('账号')
password = entry_data['data'].get('密码')
company_name = entry_data['data'].get('公司名称')
save_path = entry_data['data'].get('文件保存地址')
# 如果需要登录F6系统
cookies = None
if username and password and company_name:
@@ -419,7 +509,7 @@ class F6PluginModule:
if login_response is None:
return {'msg': '登录失败', 'msg_details': '无法登录F6系统'}
cookies = requests.utils.dict_from_cookiejar(login_response.cookies)
# 如果需要读取Excel文件
df = None
if save_path:
@@ -427,7 +517,7 @@ class F6PluginModule:
df = pd.read_excel(save_path, sheet_name=0, dtype='string')
except Exception as e:
return {'msg': f'读取Excel文件失败: {str(e)},文件路径：{save_path}'}
# 启动后台线程执行BI任务
try:
thread = threading.Thread(target=bi_task_background,
@@ -438,5 +528,3 @@ class F6PluginModule:
return {'msg': '任务启动失败', 'msg_details': f'无法启动后台任务: {str(e)}'}
return {'msg': '正在执行', 'msg_details': '正在执行，请稍后看结果'}