WindFarms.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493
  1. # -*- coding: utf-8 -*-
  2. # @Time : 2024/5/15
  3. # @Author : 魏志亮
  4. import datetime
  5. import multiprocessing
  6. import tempfile
  7. from etl.base.TranseParam import TranseParam
  8. from service.plt_service import get_all_wind, update_trans_status_error, update_trans_status_running, \
  9. update_trans_status_success
  10. from service.trans_service import creat_table_and_add_partition, rename_table, save_file_to_db
  11. from utils.file.trans_methods import *
  12. from utils.log.trans_log import logger
  13. from utils.zip.unzip import unzip, unrar, get_desc_path
  14. class WindFarms(object):
  15. def __init__(self, name, batch_no=None, field_code=None, params: TranseParam = None, wind_full_name=None,
  16. schedule_exec=True):
  17. self.name = name
  18. self.batch_no = batch_no
  19. self.field_code = field_code
  20. self.wind_full_name = wind_full_name
  21. self.save_zip = False
  22. self.trans_param = params
  23. self.exist_wind_names = multiprocessing.Manager().list()
  24. self.wind_col_trans = get_all_wind(self.field_code)
  25. self.batch_count = 50000
  26. self.save_path = None
  27. self.schedule_exec = schedule_exec
  28. self.lock = multiprocessing.Manager().Lock()
  29. self.statistics_map = multiprocessing.Manager().dict()
  30. def set_trans_param(self, params: TranseParam):
  31. self.trans_param = params
  32. read_path = str(params.read_path)
  33. if read_path.find(self.wind_full_name) == -1:
  34. message = "读取路径与配置路径不匹配:" + self.trans_param.read_path + ",配置文件为:" + self.wind_full_name
  35. update_trans_status_error(self.batch_no, self.trans_param.read_type, message, self.schedule_exec)
  36. raise ValueError(message)
  37. self.save_path = os.path.join(read_path[0:read_path.find(self.wind_full_name)], self.wind_full_name, "清理数据")
  38. def params_valid(self, not_null_list=list()):
  39. for arg in not_null_list:
  40. if arg is None or arg == '':
  41. raise Exception("Invalid param set :" + arg)
  42. def get_save_path(self):
  43. return os.path.join(self.save_path, self.batch_no, self.trans_param.read_type)
  44. def get_save_tmp_path(self):
  45. return os.path.join(tempfile.gettempdir(), self.wind_full_name, self.batch_no, self.trans_param.read_type)
  46. def get_excel_tmp_path(self):
  47. return os.path.join(self.get_save_tmp_path(), 'excel_tmp' + os.sep)
  48. def get_read_tmp_path(self):
  49. return os.path.join(self.get_save_tmp_path(), 'read_tmp')
  50. def df_save_to_tmp_file(self, df=pd.DataFrame(), file=None):
  51. if self.trans_param.is_vertical_table:
  52. pass
  53. else:
  54. # 转换字段
  55. if self.trans_param.cols_tran:
  56. cols_tran = self.trans_param.cols_tran
  57. real_cols_trans = dict()
  58. for k, v in cols_tran.items():
  59. if v and not v.startswith("$"):
  60. real_cols_trans[v] = k
  61. trans_print("包含转换字段,开始处理转换字段")
  62. df.rename(columns=real_cols_trans, inplace=True)
  63. del_keys = set(df.columns) - set(cols_tran.keys())
  64. for key in del_keys:
  65. df.drop(key, axis=1, inplace=True)
  66. df = del_blank(df, ['wind_turbine_number'])
  67. self.save_to_tmp_csv(df, file)
  68. def get_and_remove(self, file):
  69. to_path = self.get_excel_tmp_path()
  70. if str(file).endswith("zip"):
  71. if str(file).endswith("csv.zip"):
  72. copy_to_new(file, file.replace(self.trans_param.read_path, to_path).replace("csv.zip", 'csv.gz'))
  73. else:
  74. desc_path = file.replace(self.trans_param.read_path, to_path)
  75. is_success, e = unzip(file, get_desc_path(desc_path))
  76. self.trans_param.has_zip = True
  77. if not is_success:
  78. raise e
  79. elif str(file).endswith("rar"):
  80. desc_path = file.replace(self.trans_param.read_path, to_path)
  81. is_success, e = unrar(file, get_desc_path(desc_path))
  82. self.trans_param.has_zip = True
  83. if not is_success:
  84. raise e
  85. else:
  86. copy_to_new(file, file.replace(self.trans_param.read_path, to_path))
  87. def read_excel_to_df(self, file):
  88. read_cols = [v for k, v in self.trans_param.cols_tran.items() if v and not v.startswith("$")]
  89. trans_dict = {}
  90. for k, v in self.trans_param.cols_tran.items():
  91. if v and not str(v).startswith("$"):
  92. trans_dict[v] = k
  93. if self.trans_param.is_vertical_table:
  94. vertical_cols = self.trans_param.vertical_cols
  95. df = read_file_to_df(file, vertical_cols)
  96. df = df[df[self.trans_param.vertical_key].isin(read_cols)]
  97. df.rename(columns={self.trans_param.cols_tran['wind_turbine_number']: 'wind_turbine_number',
  98. self.trans_param.cols_tran['time_stamp']: 'time_stamp'}, inplace=True)
  99. df[self.trans_param.vertical_key] = df[self.trans_param.vertical_key].map(trans_dict).fillna(
  100. df[self.trans_param.vertical_key])
  101. return df
  102. else:
  103. trans_dict = dict()
  104. for k, v in self.trans_param.cols_tran.items():
  105. if v and v.startswith("$"):
  106. trans_dict[v] = k
  107. if self.trans_param.merge_columns:
  108. df = read_file_to_df(file)
  109. else:
  110. if self.trans_param.need_valid_cols:
  111. df = read_file_to_df(file, read_cols)
  112. else:
  113. df = read_file_to_df(file)
  114. # 处理列名前缀问题
  115. if self.trans_param.resolve_col_prefix:
  116. columns_dict = dict()
  117. for column in df.columns:
  118. columns_dict[column] = eval(self.trans_param.resolve_col_prefix)
  119. df.rename(columns=columns_dict, inplace=True)
  120. for k, v in trans_dict.items():
  121. if k.startswith("$file"):
  122. file_name = ".".join(os.path.basename(file).split(".")[0:-1])
  123. if k == "$file":
  124. df[v] = str(file_name)
  125. else:
  126. datas = str(k.replace("$file", "").replace("[", "").replace("]", "")).split(":")
  127. if len(datas) != 2:
  128. raise Exception("字段映射出现错误 :" + str(trans_dict))
  129. df[v] = str(file_name[int(datas[0]):int(datas[1])]).strip()
  130. elif k.startswith("$folder"):
  131. folder = file
  132. cengshu = int(str(k.replace("$folder", "").replace("[", "").replace("]", "")))
  133. for i in range(cengshu):
  134. folder = os.path.dirname(folder)
  135. df[v] = str(str(folder).split(os.sep)[-1]).strip()
  136. return df
  137. def _save_to_tmp_csv_by_name(self, df, name):
  138. save_name = str(name) + '.csv'
  139. save_path = os.path.join(self.get_read_tmp_path(), save_name)
  140. create_file_path(save_path, is_file_path=True)
  141. with self.lock:
  142. if name in self.exist_wind_names:
  143. contains_name = True
  144. else:
  145. contains_name = False
  146. self.exist_wind_names.append(name)
  147. if contains_name:
  148. df.to_csv(save_path, index=False, encoding='utf8', mode='a',
  149. header=False)
  150. else:
  151. df.to_csv(save_path, index=False, encoding='utf8')
  152. def save_to_tmp_csv(self, df, file):
  153. trans_print("开始保存", str(file), "到临时文件")
  154. names = set(df['wind_turbine_number'].values)
  155. with multiprocessing.Pool(6) as pool:
  156. pool.starmap(self._save_to_tmp_csv_by_name,
  157. [(df[df['wind_turbine_number'] == name], name) for name in names])
  158. del df
  159. trans_print("保存", str(names), "到临时文件成功, 风机数量", len(names))
  160. def set_statistics_data(self, df):
  161. if not df.empty:
  162. min_date = pd.to_datetime(df['time_stamp']).min()
  163. max_date = pd.to_datetime(df['time_stamp']).max()
  164. with self.lock:
  165. if 'min_date' in self.statistics_map.keys():
  166. if self.statistics_map['min_date'] > min_date:
  167. self.statistics_map['min_date'] = min_date
  168. else:
  169. self.statistics_map['min_date'] = min_date
  170. if 'max_date' in self.statistics_map.keys():
  171. if self.statistics_map['max_date'] < max_date:
  172. self.statistics_map['max_date'] = max_date
  173. else:
  174. self.statistics_map['max_date'] = max_date
  175. if 'total_count' in self.statistics_map.keys():
  176. self.statistics_map['total_count'] = self.statistics_map['total_count'] + df.shape[0]
  177. else:
  178. self.statistics_map['total_count'] = df.shape[0]
  179. def save_statistics_file(self):
  180. save_path = os.path.join(os.path.dirname(self.get_save_path()),
  181. self.trans_param.read_type + '_statistics.txt')
  182. create_file_path(save_path, is_file_path=True)
  183. with open(save_path, 'w', encoding='utf8') as f:
  184. f.write("总数据量:" + str(self.statistics_map['total_count']) + "\n")
  185. f.write("最小时间:" + str(self.statistics_map['min_date']) + "\n")
  186. f.write("最大时间:" + str(self.statistics_map['max_date']) + "\n")
  187. f.write("风机数量:" + str(len(read_excel_files(self.get_read_tmp_path()))) + "\n")
  188. def save_to_csv(self, filename):
  189. df = read_file_to_df(filename)
  190. if self.trans_param.is_vertical_table:
  191. df = df.pivot_table(index=['time_stamp', 'wind_turbine_number'], columns=self.trans_param.vertical_key,
  192. values=self.trans_param.vertical_value,
  193. aggfunc='max')
  194. # 重置索引以得到普通的列
  195. df.reset_index(inplace=True)
  196. for k in self.trans_param.cols_tran.keys():
  197. if k not in df.columns:
  198. df[k] = None
  199. df = df[self.trans_param.cols_tran.keys()]
  200. # 添加年月日
  201. trans_print("包含时间字段,开始处理时间字段,添加年月日", filename)
  202. df['time_stamp'] = pd.to_datetime(df['time_stamp'])
  203. df['year'] = df['time_stamp'].dt.year
  204. df['month'] = df['time_stamp'].dt.month
  205. df['day'] = df['time_stamp'].dt.day
  206. df.sort_values(by='time_stamp', inplace=True)
  207. df['time_stamp'] = df['time_stamp'].apply(
  208. lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))
  209. trans_print("处理时间字段结束")
  210. # 转化风机名称
  211. trans_print("开始转化风机名称")
  212. if self.trans_param.wind_name_exec:
  213. exec_str = f"df['wind_turbine_number'].apply(lambda wind_name: {self.trans_param.wind_name_exec} )"
  214. df['wind_turbine_number'] = eval(exec_str)
  215. df['wind_turbine_number'] = df['wind_turbine_number'].map(
  216. self.wind_col_trans).fillna(
  217. df['wind_turbine_number'])
  218. trans_print("转化风机名称结束")
  219. wind_col_name = str(df['wind_turbine_number'].values[0])
  220. if self.save_zip:
  221. save_path = os.path.join(self.get_save_path(), str(wind_col_name) + '.csv.gz')
  222. else:
  223. save_path = os.path.join(self.get_save_path(), str(wind_col_name) + '.csv')
  224. create_file_path(save_path, is_file_path=True)
  225. if self.save_zip:
  226. df.to_csv(save_path, compression='gzip', index=False, encoding='utf-8')
  227. else:
  228. df.to_csv(save_path, index=False, encoding='utf-8')
  229. self.set_statistics_data(df)
  230. del df
  231. trans_print("保存" + str(filename) + ".csv成功")
  232. def remove_file_to_tmp_path(self):
  233. # 读取文件
  234. try:
  235. if os.path.isfile(self.trans_param.read_path):
  236. all_files = [self.trans_param.read_path]
  237. else:
  238. all_files = read_files(self.trans_param.read_path)
  239. with multiprocessing.Pool(6) as pool:
  240. pool.starmap(self.get_and_remove, [(i,) for i in all_files])
  241. all_files = read_excel_files(self.get_excel_tmp_path())
  242. trans_print('读取文件数量:', len(all_files))
  243. except Exception as e:
  244. logger.exception(e)
  245. message = "读取文件列表错误:" + self.trans_param.read_path + ",系统返回错误:" + str(e)
  246. update_trans_status_error(self.batch_no, self.trans_param.read_type, message, self.schedule_exec)
  247. raise e
  248. return all_files
  249. def read_file_and_save_tmp(self):
  250. all_files = read_excel_files(self.get_excel_tmp_path())
  251. if self.trans_param.merge_columns:
  252. dfs_list = list()
  253. index_keys = [self.trans_param.cols_tran['time_stamp']]
  254. wind_col = self.trans_param.cols_tran['wind_turbine_number']
  255. if str(wind_col).startswith("$"):
  256. wind_col = 'wind_turbine_number'
  257. index_keys.append(wind_col)
  258. df_map = dict()
  259. with multiprocessing.Pool(6) as pool:
  260. dfs = pool.starmap(self.read_excel_to_df, [(file,) for file in all_files])
  261. for df in dfs:
  262. key = '-'.join(df.columns)
  263. if key in df_map.keys():
  264. df_map[key] = pd.concat([df_map[key], df])
  265. else:
  266. df_map[key] = df
  267. for k, df in df_map.items():
  268. df.drop_duplicates(inplace=True)
  269. df.set_index(keys=index_keys, inplace=True)
  270. df = df[~df.index.duplicated(keep='first')]
  271. dfs_list.append(df)
  272. df = pd.concat(dfs_list, axis=1)
  273. df.reset_index(inplace=True)
  274. try:
  275. self.df_save_to_tmp_file(df, "")
  276. except Exception as e:
  277. logger.exception(e)
  278. message = "合并列出现错误:" + str(e)
  279. update_trans_status_error(self.batch_no, self.trans_param.read_type, message, self.schedule_exec)
  280. raise e
  281. else:
  282. all_arrays = split_array(all_files, 6)
  283. for arr in all_arrays:
  284. with multiprocessing.Pool(6) as pool:
  285. dfs = pool.starmap(self.read_excel_to_df, [(ar,) for ar in arr])
  286. try:
  287. for df in dfs:
  288. self.df_save_to_tmp_file(df)
  289. except Exception as e:
  290. logger.exception(e)
  291. message = "整理临时文件,系统返回错误:" + str(e)
  292. update_trans_status_error(self.batch_no, self.trans_param.read_type, message,
  293. self.schedule_exec)
  294. raise e
  295. def mutiprocessing_to_save_file(self):
  296. # 开始保存到正式文件
  297. trans_print("开始保存到excel文件")
  298. all_tmp_files = read_excel_files(self.get_read_tmp_path())
  299. try:
  300. with multiprocessing.Pool(6) as pool:
  301. pool.starmap(self.save_to_csv, [(file,) for file in all_tmp_files])
  302. except Exception as e:
  303. logger.exception(e)
  304. message = "保存文件错误,系统返回错误:" + str(e)
  305. update_trans_status_error(self.batch_no, self.trans_param.read_type, message, self.schedule_exec)
  306. raise e
  307. trans_print("结束保存到excel文件")
  308. def mutiprocessing_to_save_db(self):
  309. # 开始保存到SQL文件
  310. trans_print("开始保存到数据库文件")
  311. all_saved_files = read_excel_files(self.get_save_path())
  312. table_name = self.batch_no + "_" + self.trans_param.read_type
  313. creat_table_and_add_partition(table_name, len(all_saved_files), self.trans_param.read_type)
  314. try:
  315. with multiprocessing.Pool(6) as pool:
  316. pool.starmap(save_file_to_db,
  317. [(table_name, file, self.batch_count) for file in all_saved_files])
  318. except Exception as e:
  319. logger.exception(e)
  320. message = "保存到数据库错误,系统返回错误:" + str(e)
  321. update_trans_status_error(self.batch_no, self.trans_param.read_type, message, self.schedule_exec)
  322. raise e
  323. trans_print("结束保存到数据库文件")
  324. def _rename_file(self):
  325. save_path = self.get_save_path()
  326. files = os.listdir(save_path)
  327. files.sort(key=lambda x: int(str(x).split(os.sep)[-1].split(".")[0][1:]))
  328. for index, file in enumerate(files):
  329. file_path = os.path.join(save_path, 'F' + str(index + 1).zfill(3) + ".csv.gz")
  330. os.rename(os.path.join(save_path, file), file_path)
  331. def delete_batch_files(self):
  332. trans_print("开始删除已存在的批次文件夹")
  333. if os.path.exists(self.get_save_path()):
  334. shutil.rmtree(self.get_save_path())
  335. trans_print("删除已存在的批次文件夹")
  336. def delete_tmp_files(self):
  337. trans_print("开始删除临时文件夹")
  338. if os.path.exists(self.get_excel_tmp_path()):
  339. shutil.rmtree(self.get_excel_tmp_path())
  340. if os.path.exists(self.get_read_tmp_path()):
  341. shutil.rmtree(self.get_read_tmp_path())
  342. if os.path.exists(self.get_save_tmp_path()):
  343. shutil.rmtree(self.get_save_tmp_path())
  344. trans_print("删除临时文件夹删除成功")
  345. def delete_batch_db(self):
  346. table_name = "_".join([self.batch_no, self.trans_param.read_type])
  347. renamed_table_name = "del_" + table_name + "_" + datetime.datetime.now().strftime('%Y%m%d%H%M%S')
  348. rename_table(table_name, renamed_table_name)
  349. def run(self, step=0, end=3):
  350. begin = datetime.datetime.now()
  351. trans_print("开始执行", self.name, self.trans_param.read_type)
  352. update_trans_status_running(self.batch_no, self.trans_param.read_type, self.schedule_exec)
  353. if step <= 0 and end >= 0:
  354. tmp_begin = datetime.datetime.now()
  355. trans_print("开始初始化字段")
  356. self.delete_batch_files()
  357. self.delete_batch_db()
  358. self.params_valid([self.name, self.batch_no, self.field_code, self.save_path, self.trans_param.read_type,
  359. self.trans_param.read_path, self.wind_full_name])
  360. if self.trans_param.resolve_col_prefix:
  361. column = "测试"
  362. eval(self.trans_param.resolve_col_prefix)
  363. if self.trans_param.wind_name_exec:
  364. wind_name = "测试"
  365. eval(self.trans_param.wind_name_exec)
  366. trans_print("初始化字段结束,耗时:", str(datetime.datetime.now() - tmp_begin), ",总耗时:",
  367. str(datetime.datetime.now() - begin))
  368. if step <= 1 and end >= 1:
  369. # 更新运行状态到运行中
  370. tmp_begin = datetime.datetime.now()
  371. self.delete_tmp_files()
  372. trans_print("开始保存到临时路径")
  373. # 开始读取数据并分类保存临时文件
  374. self.remove_file_to_tmp_path()
  375. trans_print("保存到临时路径结束,耗时:", str(datetime.datetime.now() - tmp_begin), ",总耗时:",
  376. str(datetime.datetime.now() - begin))
  377. if step <= 2 and end >= 2:
  378. # 更新运行状态到运行中
  379. tmp_begin = datetime.datetime.now()
  380. trans_print("开始保存到临时文件")
  381. # 开始读取数据并分类保存临时文件
  382. self.read_file_and_save_tmp()
  383. trans_print("保存到临时文件结束,耗时:", str(datetime.datetime.now() - tmp_begin), ",总耗时:",
  384. str(datetime.datetime.now() - begin))
  385. if step <= 3 and end >= 3:
  386. tmp_begin = datetime.datetime.now()
  387. trans_print("开始保存到文件")
  388. self.mutiprocessing_to_save_file()
  389. self.save_statistics_file()
  390. trans_print("保存到文件结束,耗时:", str(datetime.datetime.now() - tmp_begin), ",总耗时:",
  391. str(datetime.datetime.now() - begin))
  392. if step <= 4 and end >= 4:
  393. tmp_begin = datetime.datetime.now()
  394. trans_print("开始保存到数据库")
  395. self.mutiprocessing_to_save_db()
  396. trans_print("保存到数据库结束,耗时:", str(datetime.datetime.now() - tmp_begin), ",总耗时:",
  397. str(datetime.datetime.now() - begin))
  398. # 如果end==0 则说明只是进行了验证
  399. if end != 0:
  400. update_trans_status_success(self.batch_no, self.trans_param.read_type,
  401. len(read_excel_files(self.get_read_tmp_path())), self.schedule_exec)
  402. trans_print("开始执行", self.name, self.trans_param.read_type, ",,总耗时:",
  403. str(datetime.datetime.now() - begin))
  404. self.delete_tmp_files()