402 lines
16 KiB
Python
402 lines
16 KiB
Python
from yd_api import YDAPI
|
||
from tqdm import tqdm
|
||
import hashlib
|
||
from datetime import datetime, timedelta
|
||
import pandas as pd
|
||
import mysql.connector
|
||
import json
|
||
from back_ground_module import CommonModule
|
||
import numpy as np
|
||
from config import Config
|
||
from log_config import configure_task_logger, configure_error_task_logger
|
||
|
||
# 获取已经配置好的常规日志记录器
|
||
logger = configure_task_logger()
|
||
|
||
# 获取已经配置好的错误任务日志记录器
|
||
error_task_logger = configure_error_task_logger()
|
||
common_module = CommonModule()
|
||
|
||
# 初始化 API 实例和 Token
|
||
api_instance = YDAPI()
|
||
TOKEN = api_instance.generateToken()
|
||
|
||
FORMID = "FORM-XHA66881FHMAR0F07GT4Y59GGA972DD6B5OHLB"
|
||
appType = "APP_RTPWHV37ENXPQUZHTL25"
|
||
systemToken = "IA766O61SHFZT6UB0WNOB58GI5RW2K58KCU1LL6"
|
||
|
||
|
||
class TimeConsumingProcess():
|
||
"""
|
||
获取流程表单数据耗时
|
||
"""
|
||
|
||
def fetch_process_data(self):
|
||
"""获取所有流程实例"""
|
||
today_midnight = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
|
||
yesterday_midnight = today_midnight - timedelta(days=1)
|
||
|
||
start_time = yesterday_midnight.strftime("%Y-%m-%d") # 昨天0点
|
||
end_time = today_midnight.strftime("%Y-%m-%d") # 今天0点
|
||
form_data_two = api_instance.read_processes_instances(
|
||
token=TOKEN, formUuid=FORMID, page=1, n=100,
|
||
appType=appType, systemToken=systemToken, instanceStatus="",
|
||
createFromTimeGMT=start_time, createToTimeGMT=end_time
|
||
) # 之后添加增量更新
|
||
|
||
all_process_list = []
|
||
|
||
PAGES_two = form_data_two.get('totalCount') // 100 + 1
|
||
|
||
for a in tqdm(range(1, PAGES_two + 1)):
|
||
try:
|
||
form_data_two = api_instance.read_processes_instances(
|
||
token=TOKEN, formUuid=FORMID, page=a, n=100,
|
||
appType=appType, systemToken=systemToken, instanceStatus="",
|
||
createFromTimeGMT=start_time, createToTimeGMT=end_time
|
||
)
|
||
all_process_list = all_process_list + form_data_two.get("data")
|
||
except Exception as e:
|
||
logger.warning(f"获取流程实例数据时出错: {e}")
|
||
continue
|
||
|
||
return all_process_list
|
||
|
||
def extract_approval_records(self, process_instances: list):
|
||
"""提取每条流程的审批记录"""
|
||
all_data_list = []
|
||
for data in tqdm(process_instances, desc="处理流程实例"):
|
||
processInstanceId = data.get("processInstanceId")
|
||
version = data.get("version")
|
||
|
||
res_new = api_instance.get_approval_records(
|
||
token=TOKEN, processInstanceId=processInstanceId,
|
||
appType=appType, systemToken=systemToken
|
||
)
|
||
records_new = res_new.get('result', [])
|
||
for record in records_new:
|
||
operateTimeGMT = record.get('operateTimeGMT')
|
||
# if operateTimeGMT is not None:
|
||
# operateTime = datetime.fromtimestamp(operateTimeGMT / 1000).strftime('%Y-%m-%d %H:%M:%S')
|
||
# else:
|
||
# operateTime = operateTimeGMT
|
||
showName = record.get('showName')
|
||
operatorName = record.get('operatorName')
|
||
action = record.get('action')
|
||
# data_id = record.get('dataId')
|
||
activity_id = record.get('activityId')
|
||
all_data_list.append(
|
||
[operateTimeGMT, showName, operatorName, action, processInstanceId, version, activity_id])
|
||
|
||
df = pd.DataFrame(all_data_list)
|
||
# df.to_csv("审批记录.csv", index=False)
|
||
|
||
return all_data_list
|
||
|
||
def group_by_process(self, all_data_list):
|
||
"""按 '提交申请' 分组，一个流程为一组"""
|
||
result_groups = []
|
||
current_group = []
|
||
j = 1
|
||
for record in all_data_list:
|
||
showName = record[1]
|
||
if showName == "提交申请":
|
||
record.append(0)
|
||
j = 1
|
||
if current_group:
|
||
result_groups.append(current_group)
|
||
current_group = []
|
||
current_group.append(record)
|
||
else:
|
||
record.append(j)
|
||
j += 1
|
||
current_group.append(record)
|
||
if current_group:
|
||
result_groups.append(current_group)
|
||
return result_groups
|
||
|
||
def transform_to_wide_table(self, result_groups):
|
||
"""将审批记录从长表转为宽表"""
|
||
flattened_rows = []
|
||
|
||
for group in result_groups:
|
||
row_data = {}
|
||
|
||
# 遍历其余审批节点
|
||
for i, item in enumerate(group, start=1):
|
||
operateTimeGMT, showName, operatorName, action, dataId, version, activity_id, index, = item
|
||
if action == "已撤销":
|
||
showName = "该节点已撤销"
|
||
row_data.update({
|
||
f'审批{i}时间': operateTimeGMT,
|
||
f'审批{i}节点名': showName,
|
||
f'审批{i}人': operatorName,
|
||
f'审批{i}动作': action,
|
||
f'序号{i}': index,
|
||
f'审批{i}数据id': dataId,
|
||
f'审批{i}流程版本': version,
|
||
f'审批{i}流程节点id': activity_id
|
||
})
|
||
|
||
flattened_rows.append(row_data)
|
||
|
||
# 转换为DataFrame
|
||
df_final = pd.DataFrame(flattened_rows)
|
||
|
||
# 计算最大审批步骤
|
||
max_steps = max(len(group) - 1 for group in result_groups) # 减去提交节点
|
||
|
||
# 构建所有列名
|
||
all_columns = [
|
||
'审批时间', '审批节点名', '审批人', '审批动作', '序号', '数据id', '流程版本', '流程节点id'
|
||
]
|
||
for i in range(1, max_steps + 1):
|
||
all_columns.extend([
|
||
f'审批{i}时间',
|
||
f'审批{i}节点名',
|
||
f'审批{i}人',
|
||
f'审批{i}动作',
|
||
f'序号{i}',
|
||
f'审批{i}数据id',
|
||
f'审批{i}流程版本',
|
||
f'审批{i}流程节点id'
|
||
])
|
||
|
||
# 统一列结构并填充缺失值
|
||
df_final = df_final.reindex(columns=all_columns)
|
||
# number填充为0string填充为-
|
||
numeric_cols = df_final.select_dtypes(include=['number']).columns
|
||
str_cols = df_final.select_dtypes(include=['object']).columns
|
||
|
||
df_final[numeric_cols] = df_final[numeric_cols].fillna(0)
|
||
df_final[str_cols] = df_final[str_cols].fillna("-")
|
||
df_final = df_final[df_final['审批时间'].notna()]
|
||
|
||
# 导出CSV
|
||
# df_final.to_csv("审批流程行转列结果_with_node_name.csv", index=False)
|
||
|
||
return df_final, max_steps
|
||
|
||
def classify_flows(self, df_final, max_steps):
|
||
"""根据审批节点名 + 动作组合进行流程分组"""
|
||
|
||
def extract_signature(row):
|
||
signature = []
|
||
i = 1
|
||
while f'审批{i}节点名' in row:
|
||
node_name = row[f'审批{i}节点名']
|
||
action = row[f'审批{i}动作']
|
||
if node_name == "-":
|
||
break
|
||
signature.append((node_name, action))
|
||
i += 1
|
||
return signature
|
||
|
||
def has_special_action(signature):
|
||
special_actions = {"已撤销", "已转交", "已退回", "已拒绝"}
|
||
for _, action in signature:
|
||
if action in special_actions:
|
||
return True
|
||
return False
|
||
|
||
def get_hash(signature):
|
||
sig_str = str(signature)
|
||
return hashlib.md5(sig_str.encode('utf-8')).hexdigest()
|
||
|
||
df_final['signature'] = df_final.apply(extract_signature, axis=1)
|
||
|
||
group_map = {}
|
||
current_group_id = 1
|
||
df_final['group_id'] = 0
|
||
|
||
for idx, row in df_final.iterrows():
|
||
sig = row['signature']
|
||
if has_special_action(sig):
|
||
df_final.at[idx, 'group_id'] = 5000
|
||
# current_group_id += 1
|
||
else:
|
||
sig_hash = get_hash(sig)
|
||
if sig_hash not in group_map:
|
||
group_map[sig_hash] = current_group_id
|
||
current_group_id += 1
|
||
df_final.at[idx, 'group_id'] = group_map[sig_hash]
|
||
|
||
# df_final.to_csv("审批流程分类结果.csv", index=False)
|
||
# print("✅ 分组完成，已保存至 '审批流程分类结果.csv'")
|
||
|
||
result_rows = []
|
||
for index, row in df_final.iterrows():
|
||
base_info = {'group_id': row["group_id"]}
|
||
process_id_list = []
|
||
process_list = []
|
||
for i in range(1, max_steps):
|
||
prefix = f'审批{i}'
|
||
if row[f'{prefix}流程节点id'] != "-":
|
||
process_id = row[f'{prefix}流程节点id']
|
||
process_id_list.append(process_id)
|
||
if row[f'{prefix}节点名'] != "-":
|
||
process = row[f'{prefix}节点名']
|
||
process_list.append(process)
|
||
|
||
for i in range(1, max_steps): # 审批1到审批n
|
||
prefix = f'审批{i}'
|
||
approval_data = {}
|
||
if f'{prefix}时间' in df_final.columns and pd.notna(row[f'{prefix}时间']) and row[
|
||
f'{prefix}时间'] != '-':
|
||
approval_data = {
|
||
'审批时间': row[f'{prefix}时间'],
|
||
'审批节点名': row[f'{prefix}节点名'],
|
||
'审批人': row[f'{prefix}人'],
|
||
'审批动作': row[f'{prefix}动作'],
|
||
'序号': row[f'序号{i}'] if f'序号{i}' in df_final.columns else '-',
|
||
f'审批数据id': row[f'审批{i}数据id'] if f'审批{i}数据id' in df_final.columns else '-',
|
||
f'审批流程版本': row[f'审批{i}流程版本'] if f'审批{i}流程版本' in df_final.columns else '-',
|
||
f'审批流程节点id': row[
|
||
f'审批{i}流程节点id'] if f'审批{i}流程节点id' in df_final.columns else '-',
|
||
f'审批节点id合并': process_id_list,
|
||
f'审批节点名合并': process_list
|
||
}
|
||
# 合并基础数据和审批数据
|
||
result_row = {**base_info, **approval_data}
|
||
result_rows.append(result_row)
|
||
dfn = pd.DataFrame(result_rows)
|
||
|
||
# dfn.to_csv("审批流程分类结果_with_node_name.csv", index=False)
|
||
|
||
return dfn
|
||
|
||
def time_calculate(self, df_final):
|
||
"""计算每个审批步骤之间的耗时（秒），并设置"提交申请"的耗时为0"""
|
||
# 创建副本避免SettingWithCopyWarning
|
||
df_final = df_final.copy()
|
||
|
||
# 确保审批时间为 datetime 类型
|
||
df_final["审批时间"] = pd.to_datetime(df_final["审批时间"])
|
||
|
||
# 处理空值（修复FutureWarning）
|
||
numeric_cols = df_final.select_dtypes(include=['number']).columns
|
||
df_final[numeric_cols] = df_final[numeric_cols].fillna(0)
|
||
str_cols = df_final.select_dtypes(include=['object']).columns
|
||
df_final[str_cols] = df_final[str_cols].fillna("-")
|
||
|
||
# 处理JSON列（修复其他SettingWithCopyWarning）
|
||
df_final.loc[:, '审批节点id合并'] = df_final['审批节点id合并'].apply(
|
||
lambda x: json.dumps(x, ensure_ascii=False) if isinstance(x, (list, dict)) else x
|
||
)
|
||
df_final.loc[:, '审批节点名合并'] = df_final['审批节点名合并'].apply(
|
||
lambda x: json.dumps(x, ensure_ascii=False) if isinstance(x, (list, dict)) else x
|
||
)
|
||
|
||
# 计算耗时
|
||
df_final.loc[:, '耗时'] = df_final['审批时间'].diff().dt.total_seconds().div(60)
|
||
df_final.loc[df_final['审批动作'] == '提交申请', '耗时'] = 0
|
||
df_final['耗时'] = df_final['耗时'].fillna(0)
|
||
|
||
# 处理时间转换（修复第一个SettingWithCopyWarning）
|
||
df_final.loc[:, "审批时间"] = pd.to_datetime(df_final["审批时间"]).dt.strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
# 拼接url
|
||
df_final[
|
||
"url"] = "https://f6car.aliwork.com/APP_RTPWHV37ENXPQUZHTL25/formDetail/FORM-XHA66881FHMAR0F07GT4Y59GGA972DD6B5OHLB?formInstId=" + \
|
||
df_final["审批数据id"].astype(str)
|
||
return df_final
|
||
|
||
def write_to_bi(self, df_final):
|
||
# 连接信息
|
||
|
||
table_name = "yida_process_time_statistics"
|
||
|
||
# 连接数据库
|
||
connection = mysql.connector.connect(
|
||
host=Config.BI_CONN_host,
|
||
user=Config.BI_CONN_INFO_user,
|
||
password=Config.BI_CONN_INFO_password,
|
||
database=Config.BI_CONN_INFO_database
|
||
)
|
||
print(f"成功连接 {Config.BI_CONN_INFO_database}")
|
||
cursor = connection.cursor()
|
||
|
||
# 数据预处理
|
||
df = df_final.copy()
|
||
|
||
# 1. 转换时间戳为字符串
|
||
for col in df.select_dtypes(include=['datetime64[ns]', 'datetimetz']).columns:
|
||
df[col] = pd.to_datetime(df[col]).dt.strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
# 2. 增强空值处理（包含更多可能的空值表示）
|
||
empty_values = [np.nan, None, '', 'null', 'NULL', 'NA', 'NaN', 'None', '-']
|
||
df = df.replace(empty_values, None)
|
||
|
||
# 3. 严格过滤条件（确保所有关键字段都有有效值）
|
||
required_columns = ['审批节点名', '审批时间', 'group_id', '审批人', '审批动作']
|
||
df = df.dropna(subset=required_columns, how='any')
|
||
|
||
# 4. 额外检查：确保时间格式正确
|
||
df = df[df['审批时间'].str.match(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$') | df['审批时间'].isnull()]
|
||
|
||
# 生成插入语句
|
||
try:
|
||
columns = ', '.join(df.columns)
|
||
placeholders = ', '.join(['%s'] * len(df.columns))
|
||
insert_query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
|
||
|
||
# 批量插入数据
|
||
records = [tuple(row) for row in df.values]
|
||
cursor.executemany(insert_query, records)
|
||
connection.commit()
|
||
print(f"成功导入 {cursor.rowcount} 条记录到 {table_name} 表")
|
||
|
||
except Exception as e:
|
||
connection.rollback()
|
||
print(f"写入失败: {str(e)}")
|
||
# 打印前5条问题数据
|
||
print("问题数据示例:")
|
||
print(df.head().to_string())
|
||
|
||
finally:
|
||
cursor.close()
|
||
connection.close()
|
||
|
||
def main(self):
|
||
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
try:
|
||
logger.info("开始执行宜搭流程耗时写入BI任务。")
|
||
# Step 1: 获取流程实例
|
||
process_instances = self.fetch_process_data()
|
||
logger.info("获取流程实例成功。")
|
||
|
||
# Step 2: 提取审批记录
|
||
all_data_list = self.extract_approval_records(process_instances)
|
||
logger.info("提取审批记录成功。")
|
||
|
||
# Step 3: 按 '提交申请' 分组
|
||
result_groups = self.group_by_process(all_data_list)
|
||
logger.info("按 '提交申请' 分组成功。")
|
||
|
||
# Step 4: 转换为宽表
|
||
df_final, max_steps = self.transform_to_wide_table(result_groups)
|
||
logger.info("转换为宽表成功。")
|
||
|
||
# Step 5: 对流程进行分类并保存结果
|
||
df_final1 = self.classify_flows(df_final, max_steps)
|
||
logger.info("对流程进行分类并保存结果成功。")
|
||
|
||
# Step 6: 耗时计算
|
||
df_final2 = self.time_calculate(df_final1)
|
||
logger.info("耗时计算成功。")
|
||
|
||
# Step 7: 向BI写入数据
|
||
self.write_to_bi(df_final2)
|
||
logger.info("向BI写入数据成功。")
|
||
|
||
common_module.send_task_status(task_start_time, "宜搭流程耗时写入BI")
|
||
logger.info("宜搭流程耗时写入BI任务执行成功。")
|
||
except Exception as e:
|
||
error_task_logger.error(f"宜搭流程耗时写入BI执行出错: {e}")
|
||
common_module.send_task_error(task_start_time, "宜搭流程耗时写入BI", str(e))
|
||
|
||
|
||
if __name__ == '__main__':
|
||
start = TimeConsumingProcess()
|
||
start.main()
|