# -*- coding: utf-8 -*- import re import os import logging import pandas as pd from docx.oxml.ns import qn from fastapi import HTTPException from fastapi.responses import JSONResponse from docx import Document from pathlib import Path from datetime import datetime from docx.shared import Inches, Pt from app.tools.deal_excels import deal_excel, top5_dod_analysis, transform_data, deal_excel_over_load from app.tools.get_time import get_time from app.tools.replace_text import replace_text_in_docx from app.tools.replace_table import copy_table, copy_sta_table from app.tools.style import table_style from app.tools.effective_date import get_next_day from app.tools.find_before_word import extract_overload_info_from_previous_day from app.tools.count_data import count_change_outage, count_outage_sentiment from app.tools.draw_picture import plot_electricity_comparison # 获取日志记录器 logger = logging.getLogger(__name__) def deal_docx(folder_path, save_path=None, time_type=0): """ :param folder_path: 文件上传后保存的路径 :param save_path: 最终生成的日报/简报的保存路径 :param time_type: 判断时间的统计方式,0代表前一天17点之后到当天17点之前,1代表当天00:00:00到当天23:59:59 :return: 返回生成日报的存储路径,保存到mysql """ # 拿到文件夹下所有文件名 # folder_path = 'E:/work_data/work/三工单日报/20250308/源数据/源数据' # folder_path = 'E:/work_data/work/三工单日报/20250309/20250309' # folder_path = 'E:/work_data/work/三工单日报/20250310/20250310' try: logger.info("进入日报生成方法") files = os.listdir(folder_path) file_path_dict = {} # 拿到需要分析的三个文档 for file in files: # 停电word if file.endswith(".docx") and "投诉服务" in file: file_path_dict["power_off_doc"] = folder_path + "/" + file continue # 舆情word if file.endswith(".docx") and "抢修投诉舆情" in file: file_path_dict["sentiment_doc"] = folder_path + "/" + file print(f"舆情文件路径{file_path_dict['sentiment_doc']}") continue # 投诉excel if file.endswith(".xlsx") and "投诉统计表" in file: file_path_dict["power_off_excel"] = folder_path + "/" + file continue if file.endswith(".xlsx") and "停电报表配变明细" in file: file_path_dict["over_load_excel"] = folder_path + "/" + file continue # 如果传入的文件不对,抛出异常 if len(file_path_dict) != 4: logger.exception("文件格式错误") raise HTTPException( status_code=400, detail="文档无法正确解析,请确认上传的生成日报的资料是否完整", ) # ————————————————————————处理word文档————————————————————————— # 读取停电word文件信息 doc_poweroff = Document(file_path_dict["power_off_doc"]) # 读取舆情word文件 doc_sentiment = Document(file_path_dict["sentiment_doc"]) # 日报拼接数据字典 doc_dict = {} if time_type == 0: # 旧版正则 # update:2025-07-04 格式维06和7的匹配 time_re = re.compile(r"^(\d+年)?\d+月\d+日\d+时至.{7,15}期间[,,]") elif time_type == 1: # 20250429过滤时间正则 time_re = re.compile( r"^(\d+年)?\d+月\d+日[^,,。\.;;“”\']{0,10}至[^,,。\.;;“”\']{0,15}期间[,,]" ) # 避免拿错段落,则进行遍历 paragraphs_poweroff = doc_poweroff.paragraphs for para in paragraphs_poweroff: # 第一点内容 if re.match(r".*全网累计停电.*", para.text): # print(para.text) doc_dict["first_point_para1"] = re.sub(time_re, "", para.text) continue if re.match(r".*全网故障抢修工单.*", para.text): # print(para.text) doc_dict["first_point_para2"] = re.sub( r"[,,]整体抢修工作态势正常。+$", "", para.text ) continue if "first_point_para2" not in doc_dict: if re.match(r".*抢修工单.*", para.text): # print(para.text) doc_dict["first_point_para2"] = re.sub( r"[,,]整体抢修工作态势正常。+$", "", para.text ) continue # 第二点过载台数 if re.search(r"过载(\d+)(?:[条台]|\s*)?", para.text): doc_dict["over_load"] = ( re.search(r"过载(\d+)(?:[条台]|\s*)?", para.text).group().replace("过载", "") ).replace("条", "").replace("台", "") # print(doc_dict['over_load']) continue """ if re.search(r"过载\d+台", para.text): doc_dict["over_load"] = ( re.search(r"过载\d+台", para.text).group().replace("过载", "") ).replace("台", "") # print(doc_dict['over_load']) continue """ # 拿到舆情的段落 paragraphs_sentiment = doc_sentiment.paragraphs for para in paragraphs_sentiment: if re.match(r".*舆情风险信息\d+条.*", para.text): text_temp = re.sub(time_re, "", para.text) doc_dict["sentiment_trend"] = re.search( r"[^,\\.。]*[,,]舆情态势[^,\\.。]*[\\.。]$", text_temp ).group() doc_dict["sentiment_para"] = re.sub( r"[^,\\.。,]*[,,]舆情态势[^,\\.。]*[\\.。]$", "", text_temp ) continue # 获取所有表格 print("获取投诉服务的表格") tables = doc_poweroff.tables # 舆情直接取第一个表格 print("舆情的表格") table_sentiment = doc_sentiment.tables[0] # 表1 “抢修、投诉、舆情”三工单监测汇总表 table1 = tables[0] # 表2配变过载监测汇总表——配变过载集成到了一张表table1 #table2 = tables[1] # 表3停电用户前五供电局——停电前五移动到了第二张表 table3 = tables[1] #table3 = tables[2] # 新增表4 95598供电类投诉前五供电局统计表 # table4 = doc_poweroff.add_table(6, 5) # ————————————————————————处理word文档————————————————————————— # ----------------------------------------------------------------------------------------- # ————————————————————————表格环比统计—————————————————————————— # 首先拿到分析时间,明确要分析哪天的数据 ( start_time, end_time, before_start_time, year, month, day, day_before, month_before, ) = get_time(files, time_type) # 获取后一天的时间 year_now, month_now, day_now = get_next_day(int(year), int(month), int(day)) # 通过上述时间,统计停电excel的情况 # 当天情况 province_statistics, district_statistics = deal_excel( start_time, end_time, file_path_dict["power_off_excel"] ) print(f"省份统计{province_statistics}") print(f"地市统计{district_statistics}") province_statistics_list = list(province_statistics.values()) # 当天省份总投诉 province_statistics_total = sum(province_statistics.values()) print(f"省份总投诉{province_statistics_total}") # 昨天情况 province_stat_before, district_stat_before = deal_excel( before_start_time, start_time, file_path_dict["power_off_excel"] ) print(f"省份昨日情况{province_stat_before}") # 昨天省份总投诉 province_stat_be_total = sum(province_stat_before.values()) print(f"省份昨日总投诉{province_stat_be_total}") # 省份环比 province_dod = { k: province_statistics[k] - province_stat_before[k] for k in province_statistics.keys() } # 最终省份环比结果 for key, value in province_dod.items(): if int(value) > 0: province_dod[key] = "+" + str(value) elif int(value) == 0: province_dod[key] = "持平" print(f"省份环比{province_dod}") province_dod_list = list(province_dod.values()) # 表1中剩余的省份统计数据及舆情的统计数据、环比情况 table1_extra_data = transform_data( [province_statistics_list, province_dod_list] ) logger.info( f"表1中剩余的省份统计数据及舆情的统计数据、环比情况:{table1_extra_data}" ) # 将昨天的地市统计转成字典 district_stat_before = dict(district_stat_before) # 查看今天的前五在昨天的情况 """ 情况1:今天的数据大于5,则可以直接用现有逻辑 情况2:今天的数据小于5,值判断小于5的这几条,比如只有1条,就判断这一条的情况 """ top_dod_dict = {} # 需要判断地市停电的有没有5个,分小于5或者大于等于5 top5_name_list = [] top5_poweroff_list = [] # update:2025-07-04 修改供电类投诉前五供电局统计表的同排行 need_district_statistics = ( district_statistics[0:5] if len(district_statistics) > 5 else district_statistics ) other_district_statistic = ( district_statistics[5:] if len(district_statistics) > 5 else [] ) other_count = 0 if ( len(other_district_statistic) > 0 and district_statistics[4][1] == district_statistics[5][1] ): for i in range(len(other_district_statistic)): if other_district_statistic[i][1] == district_statistics[4][1]: other_count += 1 poweroff_value = need_district_statistics[len(need_district_statistics) - 1][1] count = 0 for i in range(len(need_district_statistics)): current_poweroff_value = need_district_statistics[i][1] if current_poweroff_value == poweroff_value: count += 1 else: top5_name_list.append(need_district_statistics[i][0]) top5_poweroff_list.append(need_district_statistics[i][1]) top_dod_dict[need_district_statistics[i][0]] = top5_dod_analysis( need_district_statistics[i], district_stat_before ) if count == 1 and other_count==0: top5_name_list.append( need_district_statistics[len(need_district_statistics) - 1][0] ) top5_poweroff_list.append( need_district_statistics[len(need_district_statistics) - 1][1] ) top_dod_dict[ need_district_statistics[len(need_district_statistics) - 1][0] ] = top5_dod_analysis( need_district_statistics[len(need_district_statistics) - 1], district_stat_before, ) else: top5_name_list.append(f"其他{count + other_count}家单位") top5_poweroff_list.append(poweroff_value) top_dod_dict["其他单位"] = "—" # old_version """ if len(district_statistics) >= 5: # 地市前五统计 # print(district_statistics) top1 = district_statistics[0] top2 = district_statistics[1] top3 = district_statistics[2] top4 = district_statistics[3] top5 = district_statistics[4] print(f'地市前五{top1}{top2}{top3}{top4}{top5}') top5_name_list = [top1[0], top2[0], top3[0], top4[0], top5[0]] top5_poweroff_list = [top1[1], top2[1], top3[1], top4[1], top5[1]] top_dod_dict[top1[0]] = top5_dod_analysis(top1, district_stat_before) top_dod_dict[top2[0]] = top5_dod_analysis(top2, district_stat_before) top_dod_dict[top3[0]] = top5_dod_analysis(top3, district_stat_before) top_dod_dict[top4[0]] = top5_dod_analysis(top4, district_stat_before) top_dod_dict[top5[0]] = top5_dod_analysis(top5, district_stat_before) elif 0 < len(district_statistics) < 5: for i in range(len(district_statistics)): top5_name_list.append(district_statistics[i][0]) top5_poweroff_list.append(district_statistics[i][1]) top_dod_dict[district_statistics[i][0]] = top5_dod_analysis(district_statistics[i], district_stat_before) print(f"地市前五名称{top5_name_list}") print(f"地市前五数据{top5_poweroff_list}") # top_dod_dict[top1[0]] = top5_dod_analysis(top1, district_stat_before) # top_dod_dict[top2[0]] = top5_dod_analysis(top2, district_stat_before) # top_dod_dict[top3[0]] = top5_dod_analysis(top3, district_stat_before) # top_dod_dict[top4[0]] = top5_dod_analysis(top4, district_stat_before) # top_dod_dict[top5[0]] = top5_dod_analysis(top5, district_stat_before) """ print(f"地市环比{top_dod_dict}") top5_stat_list = list(top_dod_dict.values()) # 地市前5的名称、数据、环比放入列表并转至,方便写入表格4 top5_list = transform_data([top5_name_list, top5_poweroff_list, top5_stat_list]) # 省总的投诉情况及环比 complain_dod = int(province_statistics_total) - int(province_stat_be_total) logger.info(f"省份总量环比{complain_dod}") # 计算省份总量环比 if complain_dod > 0: # 使用 f-string 进行格式化 complain_dod = f"增加{complain_dod / province_stat_be_total * 100:.2f}%" elif complain_dod < 0: # 使用 f-string 进行格式化 complain_dod = f"减少{-complain_dod / province_stat_be_total * 100:.2f}%" else: complain_dod = "持平" # 异常处置情况 electricity_exception = "停电、投诉、舆情整体平稳,无异常情况。" standardize_date = None if time_type == 0: # !!!旧版 前面已经过滤掉了时间信息,此处对时间进行单独赋值操作 standardize_date = ( f"{month_before}月{day_before}日17时至{month}月{day}日17时期间," ) elif time_type == 1: # -------------------------20250429更新,修改开始和结束时间--------------------------------- #standardize_date = f"{year}年{month}月{day}日0时至24时期间," standardize_date = f"{month}月{day}日0时至24时期间," # standardize_date = '' # -------------------------20250429更新,修改开始和结束时间--------------------------------- # ————————————————————————表格环比统计—————————————————————————— # 获取当前文件夹路径 current_path = Path(__file__).parent templates_path = str(os.path.join(current_path.parent, "templates")).replace( "\\", "/" ) # 默认标题 # 注意,标题会根据不同时期进行调整 report_title = r"南方电网公司“停电抢修、投诉服务、舆情管控”三工单联动监测日报" # ————————————————————————组装完整简报—————————————————————————— if time_type == 0: # 旧版正则 sample_first_para = ( f"{month_before}月{day_before}日17时至{month}月{day}日17时" ) elif time_type == 1: # 20250429过滤时间正则 sample_first_para = f"{year}年{month}月{day}日0时至24时" # 简报舆情信息 doc_dict["sentiment_para_simple"] = doc_dict["sentiment_para"].replace( "全网监测到", "" ) if re.search(r"重要用户停电[^0]户", doc_dict["first_point_para1"]): doc_dict["have_important"] = re.sub( "[,,]用户停电情况总体平稳", "", re.sub("其中[,,]", "", doc_dict["first_point_para1"]), ) else: doc_dict["have_important"] = ( re.sub( r"[,,]其中.{0,3}重要用户停电0户.{0,5}停电情况总体平稳[\\.。]", "", doc_dict["first_point_para1"], ) + ",无重要用户停电。" ) # 获取停电数字信息 ( total_outage, short_term_outage, change_outage, percentage, short_precentage, important_stop_outage, type, ) = count_change_outage(doc_dict["have_important"]) # 获取舆情数字信息 today_sentiment, type_sentiment, yesterday_sentiment, result_sentiment = ( count_outage_sentiment(doc_dict["sentiment_para_simple"]) ) # 简报的舆情信息只要总数和环比 complain_simple = ( f"95598供电类投诉{province_statistics_total}条,环比{complain_dod}" ) print(doc_dict["have_important"]) print(doc_dict["sentiment_para_simple"]) current_doc_name = f"南方电网公司停电抢修投诉服务舆情管控三工单联动监测日报{year}{int(month):02d}{int(day):02d}.docx" if "over_load" not in doc_dict: raise Exception("过载数据不符合标准未解析,请查看文档") doc_dict_over_load = doc_dict["over_load"] over_load_before = extract_overload_info_from_previous_day( current_word=current_doc_name ) if over_load_before: # 将字符串转换为浮点数 over_load_before = float(over_load_before) doc_dict_over_load = float(doc_dict_over_load) if over_load_before > doc_dict_over_load: over_load_percent = ( (over_load_before - doc_dict_over_load) / over_load_before * 100 ) over_load_percent = f"{over_load_percent:.2f}%" over_load_type = "减少" elif over_load_before < doc_dict_over_load: over_load_percent = ( (doc_dict_over_load - over_load_before) / over_load_before * 100 ) over_load_percent = f"{over_load_percent:.2f}%" over_load_type = "增加" else: over_load_percent = "" over_load_type = "持平" else: over_load_before = "" over_load_percent = ",缺少上一天数据" over_load_type = "无法估计" # 组装替换的文本 replacements_simple = { "{{standardize_date}}": standardize_date, "{{total_outage}}": str(total_outage), "{{short_term_outage}}": str(short_term_outage), "{{change_outage}}": str(change_outage), "{{percentage}}": str(percentage), "{{short_precentage}}": str(short_precentage), "{{important_stop_outage}}": str(important_stop_outage), "{{type}}": type, "{{have_important}}": doc_dict["have_important"], "{{over_load}}": doc_dict["over_load"], "{{over_load_percent}}": str(over_load_percent), "{{over_load_type}}": over_load_type, "{{complain}}": complain_simple, "{{sample_first_para}}": sample_first_para, "{{today_sentiment}}": str(today_sentiment), "{{type_sentiment}}": type_sentiment, "{{yesterday_sentiment}}": str(yesterday_sentiment), "{{result_sentiment}}": str(result_sentiment) if result_sentiment!="" else "", "{{year}}": year, "{{month}}": month, "{{day}}": day, } # 组装简报 electricity_daily_simple = Document(f"{templates_path}/简报模板.docx") # 替换模板字符串 replace_text_in_docx(electricity_daily_simple, replacements_simple) datas = { "停电用户\n(万户)": { "昨天": total_outage + change_outage, "今天": total_outage, }, "过载配变\n(台)": {"昨天": over_load_before, "今天": doc_dict_over_load}, "95598供电类\n投诉(条)": { "昨天": province_stat_be_total, "今天": province_statistics_total, }, "涉电力供应类舆情\n风险信息(条)": { "昨天": yesterday_sentiment, "今天": today_sentiment, }, } # 将数据转换为DataFrame df = pd.DataFrame(datas) # 遍历 datas 中的每个值,将 None 或 空字符串替换为 0 for key, value in datas.items(): for sub_key, sub_value in value.items(): if sub_value is None or sub_value == "": datas[key][sub_key] = None # 将不存在或为空的值设置为 0 else: datas[key][sub_key] = int(sub_value) # 确保值是整数 # 生成柱状图 img_path = plot_electricity_comparison(year, month, day, datas) # 查找插入图片的位置(假设模板中有"{{IMG_PLACEHOLDER}}"作为占位符) img_placeholder = "{{IMG_PLACEHOLDER}}" img_inserted = False for paragraph in electricity_daily_simple.paragraphs: if img_placeholder in paragraph.text: # 删除占位符文本 paragraph.text = paragraph.text.replace(img_placeholder, "") # 插入图片 run = paragraph.add_run() run.add_picture(img_path, width=Inches(6.0)) img_inserted = True break if not img_inserted: # 如果未找到占位符,则在文档末尾添加图片 p = electricity_daily_simple.add_paragraph() run = p.add_run() run.add_picture(img_path, width=Inches(6.0)) # 将表格写入简报 # 设置全局样式 style = electricity_daily_simple.styles["Normal"] style.font.name = "Times New Roman" # 按照月份分门别类的文件夹 save_folder = f"{year}{str(month).zfill(2)}" # 创建子文件夹 if not os.path.exists(f"{save_path}/{save_folder}"): os.makedirs(f"{save_path}/{save_folder}", exist_ok=True) final_file = None final_sim_file = None # 最终保存文件的路径情况 if time_type == 0: final_file = f"{save_path}/{save_folder}/{report_title}-{year}{str(month).zfill(2)}{str(day).zfill(2)}.docx" final_sim_file = f"{save_path}/{save_folder}/【简版】{report_title}-{year}{str(month).zfill(2)}{str(day).zfill(2)}.docx" elif time_type == 1: final_file = f"{save_path}/{save_folder}/{report_title}-{year_now}{str(month_now).zfill(2)}{str(day_now).zfill(2)}.docx" final_sim_file = f"{save_path}/{save_folder}/【简版】{report_title}-{year_now}{str(month_now).zfill(2)}{str(day_now).zfill(2)}.docx" # 删除旧文件,方便文件更新 delete_old_file(final_file) delete_old_file(final_sim_file) # 生成简报 # 接口保存路径地址 # 保存为Excel文件 path = f"{save_path}/{save_folder}/{year}{str(month).zfill(2)}{str(day).zfill(2)}电力统计数据.xlsx" df.to_excel(path, index=True) electricity_daily_simple.save(final_sim_file) # 测试保存路径 # electricity_daily_simple.save(f'【简版】公司全国“两会”保供电期间配网设备运行及三工单监测日报-{year}{str(month).zfill(2)}{str(day).zfill(2)}.docx') # # ————————————————————————组装完整简报—————————————————————————— # ----------------------------------------------------------------------------------------- # ————————————————————————组装完整日报—————————————————————————— # 将数据组装相关的时间内容 doc_dict["first_point_para1"] = standardize_date + doc_dict["first_point_para1"] #doc_dict["sentiment_para"] = standardize_date + doc_dict["sentiment_para"] doc_dict["sentiment_para"] = doc_dict["sentiment_para"] # {{standardize_date}}全网收到{{complain_num}}条供电类投诉,环比{{complain_dod}}条; complain_text = ( standardize_date + f"全网收到{str(province_statistics_total)}条供电类投诉,环比{complain_dod}" ) # update:2025-07-04 备注,增加过载环比 replacements = {} if time_type == 0: # 组装替换的文本 replacements = { "{{year}}": year, "{{month}}": month, "{{day}}": day, "{{power_off_one}}": doc_dict["first_point_para1"], "{{power_off_two}}": doc_dict["first_point_para2"], "{{over_load}}": doc_dict["over_load"], "{{over_load_percent}}": str(over_load_percent), "{{over_load_type}}": over_load_type, "{{complain}}": complain_simple, "{{sentiment}}": doc_dict["sentiment_para"], "{{sentiment_trend}}": doc_dict["sentiment_trend"], "{{exception}}": electricity_exception, } elif time_type == 1: # 组装替换的文本 replacements = { "{{year}}": str(year_now), "{{month}}": str(month_now), "{{day}}": str(day_now), "{{power_off_one}}": doc_dict["first_point_para1"], "{{power_off_two}}": doc_dict["first_point_para2"], "{{over_load}}": doc_dict["over_load"], "{{over_load_percent}}": str(over_load_percent), "{{over_load_type}}": over_load_type, "{{complain}}": complain_simple, "{{sentiment}}": doc_dict["sentiment_para"], "{{sentiment_trend}}": doc_dict["sentiment_trend"], "{{exception}}": electricity_exception, } # 组装日报 electricity_daily = Document(f"{templates_path}/日报模板.docx") # replace_text_in_docx(electricity_daily, replacements) # 将表格添加到新的文档里 # 组装表1的数据 # 此处缺少省份统计数据和舆情数据 logger.info("将投诉服务的表格1写入日报的表格1,从2-8行,2-9列写入") # 定义要查看的区域范围 start_row1 = 2 # 起始行索引(从0开始) end_row1 = 8 # 结束行索引(不包括) start_col1 = 2 # 起始列索引(从0开始) end_col1 = 9 # 结束列索引(不包括) copy_table( table1, electricity_daily.tables[0], start_row1, end_row1, start_col1, end_col1, 0, ) # 插入各个省份的投诉数据及环比 logger.info("将投诉数据写入到表1,第二行开始,第11列开始") # 省份统计的表格数据在表格中的起始位置 start_row_pro_sta = 2 start_col_pro_sta = 11 copy_sta_table( electricity_daily.tables[0], table1_extra_data, start_row_pro_sta, start_col_pro_sta, ) # 放入舆情的数据 logger.info("将舆情数据写入到表1,第二行开始,第11列开始") # 定义要查看的区域范围 start_row1_1 = 2 # 起始行索引(从0开始) end_row1_1 = 8 # 结束行索引(不包括) start_col1_1 = 10 # 起始列索引(从0开始) end_col1_1 = 13 # 结束列索引(不包括) copy_table( table_sentiment, electricity_daily.tables[0], start_row1_1, end_row1_1, start_col1_1, end_col1_1, 3, ) # 复制表1的数据 logger.info("将配变过载写入表0,第二行开始,第9列开始") # 定义要查看的区域范围 start_row2 = 2 # 起始行索引(从0开始) end_row2 = 8 # 结束行索引(不包括) start_col2 = 9 # 起始列索引(从0开始) end_col2 = 11 # 结束列索引(不包括) #target_row=1 #target_col=7 copy_table( table1, electricity_daily.tables[0], start_row2, end_row2, start_col2, end_col2, 0 ) """ copy_table( table2, electricity_daily.tables[1], start_row2, end_row2, start_col2, end_col2, 0, ) """ # 复制表3的数据 logger.info("将停电前五插入表格2") # 定义要查看的区域范围 start_row3 = 2 # 起始行索引(从0开始) end_row3 = 7 # 结束行索引(不包括) start_col3 = 1 # 起始列索引(从0开始) end_col3 = 5 # 结束列索引(不包括) copy_table( table3, electricity_daily.tables[1], start_row3, end_row3, start_col3, end_col3, 0, ) """ copy_table( table3, electricity_daily.tables[2], start_row3, end_row3, start_col3, end_col3, 0, ) """ # 填充表格4 # 需要判断是否前五数据不存在 logger.info("将自行统计的数据插入表格5") # 表4中的插入位置 start_tb4_row = 2 start_tb4_col = 5 if top5_list: copy_sta_table( electricity_daily.tables[1], top5_list, start_tb4_row, start_tb4_col, is_dynamics=False ) logger.info("将配变重过载的数据插入表格5") # 统计配变过载前五的数据 top_5_results = deal_excel_over_load(file_path_dict["over_load_excel"]) # 表4中的插入位置 start_tb4_row = 2 start_tb4_col = 8 if not top_5_results.empty: data = top_5_results.values copy_sta_table( electricity_daily.tables[1], data, start_tb4_row, start_tb4_col, ) # copy_sta_table(electricity_daily.tables[3], top5_list, start_tb4_row, start_tb4_col) # 将表格中的字体中文设置成仿宋,英文数字设置成新罗马,均为11号大小 for table in electricity_daily.tables: table_style(table) # 设置英文数字样式 # 设置全局样式 # 显式设置每个段落的字体 for paragraph in electricity_daily.paragraphs: for run in paragraph.runs: run.font.name = "Times New Roman" #run.font.name = "仿宋" #run._element.rPr.rFonts.set(qn('w:eastAsia'), '仿宋') #run.font.size = Pt(16) # 接口保存路径 electricity_daily.save(final_file) # 返回doc、年月日,然后在接口代码里进行分析后,提取表1的数据保存到数据库 # return electricity_daily, year, month, day # 日报本身的时间 statistics_time = None if time_type == 0: statistics_time = datetime(int(year), int(month), int(day)) elif time_type == 1: statistics_time = datetime(int(year_now), int(month_now), int(day_now)) # 返回值保存到数据库,以二进制保存 if time_type == 0: return { "report_title": f"{report_title}-{year}{str(month).zfill(2)}{str(day).zfill(2)}.docx", "daily_report": final_file, "daily_repo_simple": final_sim_file, "statistics_time": statistics_time, "save_folder": save_folder, # 'excel':path, # 'img':img_path } elif time_type == 1: return { "report_title": f"{report_title}-{year_now}{str(month_now).zfill(2)}{str(day_now).zfill(2)}.docx", "daily_report": final_file, "daily_repo_simple": final_sim_file, "statistics_time": statistics_time, "save_folder": save_folder, # 'excel':path, # 'img':img_path } # 测试保存路径 # electricity_daily.save(f'公司全国“两会”保供电期间配网设备运行及三工单监测日报-{year}{str(month).zfill(2)}{str(day).zfill(2)}.docx') # # ————————————————————————组装完整日报—————————————————————————— except Exception as e: logger.exception("最终渲染阶段失败") return JSONResponse( content={"status_code": 500, "detail": f"word解析异常:{e}"} ) # 从磁盘删除旧文件方法 def delete_old_file(file): try: if os.path.exists(file): os.remove(file) logger.info("磁盘里的旧文件删除成功") except Exception as e: logger.info(f"删除旧文件失败:{e}") # if __name__ == '__main__': # folder_path = 'E:/work_data/work/三工单日报/20250310/20250310' # # deal_docx(folder_path)