# 读取配置 from lib.dataread import * from config_jingbo_yuedu import * import datetime # 导入Graphs类用于生成PDF from lib.tools import Graphs from reportlab.platypus import SimpleDocTemplate from reportlab.lib.pagesizes import letter as letter_page global_config.update({ # 核心参数 'logger': logger, 'dataset': dataset, 'y': y, # 'offsite_col': offsite_col, # 'avg_cols': avg_cols, # 'offsite': offsite, 'edbcodenamedict': edbcodenamedict, 'is_debug': is_debug, 'is_train': is_train, 'is_fivemodels': is_fivemodels, 'is_update_report': is_update_report, 'settings': settings, 'bdwdname': bdwdname, 'columnsrename': columnsrename, 'price_columns': price_columns, # 模型参数 'data_set': data_set, 'input_size': input_size, 'horizon': horizon, 'train_steps': train_steps, 'val_check_steps': val_check_steps, 'val_size': val_size, 'test_size': test_size, 'modelsindex': modelsindex, 'rote': rote, 'bdwd_items': bdwd_items, # 特征工程开关 'is_del_corr': is_del_corr, 'is_del_tow_month': is_del_tow_month, 'is_eta': is_eta, 'is_update_eta': is_update_eta, 'is_fivemodels': is_fivemodels, 'is_update_predict_value': is_update_predict_value, 'early_stop_patience_steps': early_stop_patience_steps, # 时间参数 'start_year': start_year, 'end_time': end_time or datetime.datetime.now().strftime("%Y-%m-%d"), 'freq': freq, # 保持列表结构 # 接口配置 'login_pushreport_url': login_pushreport_url, 'login_data': login_data, 'upload_url': upload_url, 'upload_data': upload_data, 'upload_warning_url': upload_warning_url, 'warning_data': warning_data, # 'upload_report_audit_url': upload_report_audit_url, # 'upload_report_audit_data': upload_report_audit_data, # 查询接口 'query_data_list_item_nos_url': query_data_list_item_nos_url, 'query_data_list_item_nos_data': query_data_list_item_nos_data, # 上传数据项 'push_data_value_list_url': push_data_value_list_url, 'push_data_value_list_data': push_data_value_list_data, 'push_png_report_url': push_png_report_url, 'push_png_report_data': push_png_report_data, # eta 配置 'APPID': APPID, 'SECRET': SECRET, 'etadata': data, 'edbcodelist': edbcodelist, 'ClassifyId': ClassifyId, 'edbcodedataurl': edbcodedataurl, 'classifyidlisturl': classifyidlisturl, 'edbdatapushurl': edbdatapushurl, 'edbdeleteurl': edbdeleteurl, 'edbbusinessurl': edbbusinessurl, 'ClassifyId': ClassifyId, 'classifylisturl': classifylisturl, # 数据库配置 'sqlitedb': sqlitedb, 'is_bdwd': is_bdwd, 'db_mysql': db_mysql, 'DEFAULT_CONFIG': DEFAULT_CONFIG, }) def generate_pdf_report(image_path, pdf_path, title): """生成PDF报告""" # 创建内容对应的空列表 content = list() # 添加标题 content.append(Graphs.draw_title(title)) # 检查图片文件是否存在 if not os.path.exists(image_path): logger.error(f'图片文件不存在: {image_path}') # 添加错误信息到PDF content.append(Graphs.draw_text(f'错误：图片文件不存在: {image_path}')) else: try: # 添加图片 content.append(Graphs.draw_img(image_path)) logger.info(f'成功添加图片到PDF: {image_path}') except Exception as e: logger.error(f'添加图片到PDF时出错: {e}') content.append(Graphs.draw_text(f'错误：添加图片到PDF时出错: {str(e)}')) try: # 生成pdf文件 doc = SimpleDocTemplate(pdf_path, pagesize=letter_page) doc.build(content) logger.info(f'成功生成PDF报告: {pdf_path}') except Exception as e: logger.error(f'生成PDF报告时出错: {e}') raise e def push_png_report(): current_end_time = global_config['end_time'] # 使用 datetime.datetime 来避免类型问题 current_dt = datetime.datetime.strptime(current_end_time, '%Y-%m-%d') previous_trading_day = ( current_dt - datetime.timedelta(days=1)).strftime('%Y-%m-%d') png_report_files = [ # 'yuanyou_zhouducorrelation.png', # 'yuanyou_yueducorrelation.png', 'yuanyou_combined_correlation.png' ] for png_report_file in png_report_files: logger.info(f'发送图片{png_report_file}到钉钉工作组') # 根据图片文件名确定推送内容标题 if png_report_file == 'yuanyou_zhouducorrelation.png': push_content = f'{current_end_time}Brent连续合约日、周维度预测价格走势' else: push_content = f'{current_end_time}Brent连续合约月维度预测价格走势' # 生成PDF报告 pdf_title = push_content # 使用您指定的标题格式 # 清理文件名中的非法字符 safe_filename = push_content.replace(':', '-').replace('/', '-').replace('\\', '-').replace( '*', '').replace('?', '').replace('"', '').replace('<', '').replace('>', '').replace('|', '') pdf_filename = f'{safe_filename}.pdf' # 使用您指定的文件名格式 pdf_path = os.path.join(global_config['dataset'], pdf_filename) image_path = os.path.join(global_config['dataset'], png_report_file) # 检查数据集目录是否存在，不存在则创建 if not os.path.exists(global_config['dataset']): os.makedirs(global_config['dataset']) logger.info(f'创建数据集目录: {global_config["dataset"]}') try: generate_pdf_report(image_path, pdf_path, pdf_title) logger.info(f'生成PDF报告 {pdf_filename} 成功') except Exception as e: logger.error(f'生成PDF报告 {pdf_filename} 失败：{e}') # 即使PDF生成失败，也继续执行后续操作 continue try: # 检查PDF文件是否存在 if not os.path.exists(pdf_path): logger.error(f'PDF文件未生成: {pdf_path}') continue # with open(pdf_path, 'rb') as f: # pdf_base64_data = base64.b64encode(f.read()).decode('utf-8') with open(os.path.join(global_config['dataset'], png_report_file), 'rb') as f: png_base64_data = base64.b64encode(f.read()).decode('utf-8') # 直接推送报告 # config.upload_data["data"]["fileBase64"] = base64_data data = global_config['push_png_report_data'] data['data']['fileBase64'] = png_base64_data data['data']['billNo'] = str(time.time()) data['data']['pushContent'] = push_content # 报告需审核后推送 # config.upload_report_audit_data["data"]["fileBase64"] = pdf_base64_data # data = global_config['upload_report_audit_data'] # data['data']['imgFileBase64'] = png_base64_data # data['data']['fileImgName'] = push_content # data['data']['fileName'] = pdf_filename # if png_report_file == 'yuanyou_zhouducorrelation.png': # data['data']['pushContent'] = f'{end_time}Brent连续合约日、周维度预测价格走势' # else: # data['data']['pushContent'] = f'{end_time}Brent连续合约月维度预测价格走势' pngreportdata = push_png_report_to_market(data) logger.info(f'{png_report_file}推送图片报告到钉钉成功{pngreportdata}') except Exception as e: logger.error(f'{png_report_file}推送图片报告到钉钉失败：{e}') if __name__ == '__main__': # 图片报告 try: logger.info('图片报告ing') # global_config['end_time'] = '2025-10-24' # previous_trading_day = (pd.Timestamp(global_config['end_time']) - # pd.tseries.offsets.BusinessDay(1)).strftime('%Y-%m-%d') # global_config['end_time'] = previous_trading_day from models.nerulforcastmodels import yuanyou_bdwd_png, yuanyou_imgreport_twoinone # yuanyou_bdwd_png(global_config=global_config) yuanyou_imgreport_twoinone() # logger.info('图片报告end') except Exception as e: logger.info(f'图片报告失败：{e}') time.sleep(5) push_png_report()