Spaces:

Gabriel00A
/

purchase-report

Paused

App Files Files Community

Gabriel00A commited on Sep 22, 2025

Commit

2b0fa1b

verified ·

1 Parent(s): 795f7ca

Update process_report.py

Browse files

Files changed (1) hide show

process_report.py +59 -56

process_report.py CHANGED Viewed

@@ -13,7 +13,6 @@ import json
 import base64
 from io import BytesIO
 from typing import Optional, Tuple, List
 from datetime import datetime, date
 import pandas as pd
@@ -55,6 +54,14 @@ EMAIL_COLS_DEFAULT = [
     "主数量","需求日期","供应商","到货日期","到货主数量","入库日期","入库主数量","目前进度"
 ]
 # ====== 工具函数 ======
 def _today() -> date:
     return datetime.now().date()
@@ -139,18 +146,30 @@ def aggregate_for_email(df: pd.DataFrame) -> pd.DataFrame:
     grouped = df.groupby(group_keys, dropna=False).agg(final_agg_map).reset_index()
     grouped["目前进度"] = grouped.apply(_calc_progress_row, axis=1)
     if "计划到货日期" in grouped.columns:
         grouped = grouped.drop(columns=["计划到货日期"])
     final_cols = [col for col in EMAIL_COLS_DEFAULT if col in grouped.columns]
     final = grouped[final_cols]
-    # --- 唯一的修改在此处 ---
-    # 将日期格式化为 yyyy-mm-dd 标准格式
     date_cols_to_format = ["请购日期", "需求日期", "到货日期", "入库日期"]
     for col in date_cols_to_format:
         if col in final.columns:
-            final[col] = pd.to_datetime(final[col], errors='coerce').dt.strftime('%Y-%m-%d') # 已更新为 yyyy-mm-dd 格式
     return final
@@ -199,9 +218,40 @@ def _find_latest_input(input_dir: str) -> Optional[str]:
     files.sort(key=os.path.getmtime, reverse=True)
     return files[0]
-def _df_to_excel_bytes(df: pd.DataFrame) -> bytes:
     bio = BytesIO()
-    df.to_excel(bio, index=False)
     bio.seek(0)
     return bio.read()
@@ -211,8 +261,8 @@ def _build_html_body(df: pd.DataFrame, title: str) -> str:
     table {{ border-collapse: collapse; font-size: 13px; }}
     table, th, td {{ border: 1px solid #ccc; padding: 6px; }}
     th {{ background:#f6f6f6; }}
     </style></head><body><h3>{title}</h3>{table_html}
-    <p style="color:#666;">备注：此邮件由自动化系统生成。</p>
     </body></html>"""
     return html
@@ -244,12 +294,10 @@ def run_once(file_path: Optional[str] = None) -> dict:
         raw = read_system_export(file_path)
         final = aggregate_for_email(raw)
-        # --- 使用带样式的Excel生成函数 ---
-        attach = _df_to_styled_excel_bytes(final) # 假设这个函数存在
         out_name = f"邮件发送的格式_{datetime.now().strftime('%Y%m%d')}.xlsx"
         out_path = os.path.join(OUTPUT_DIR, out_name)
-        # 仍然可以在服务器上保存一份不带样式的Excel
         final.to_excel(out_path, index=False)
         subject = f"采购执行表自动推��� {datetime.now().date()}"
@@ -262,51 +310,6 @@ def run_once(file_path: Optional[str] = None) -> dict:
         import traceback
         return {"ok": False, "msg": f"处理文件时发生严重错误: {e}", "traceback": traceback.format_exc()}
-def _df_to_styled_excel_bytes(df: pd.DataFrame) -> bytes:
-    """
-    生成带样式的Excel文件。
-    使用XlsxWriter引擎来添加字体、边框、条件格式等。
-    """
-    bio = BytesIO()
-    # fillna('') 可以在生成excel时将 <NA> 显示为空字符串
-    writer = pd.ExcelWriter(bio, engine='xlsxwriter')
-    sheet_name = '采购执行表'
-    df.fillna('').to_excel(writer, sheet_name=sheet_name, index=False)
-    workbook = writer.book
-    worksheet = writer.sheets[sheet_name]
-    # 定义格式
-    header_format = workbook.add_format({'bold': True, 'font_name': 'Arial', 'font_size': 10, 'border': 1, 'align': 'center', 'valign': 'vcenter'})
-    default_format = workbook.add_format({'font_name': 'Arial', 'font_size': 10, 'border': 1})
-    overdue_format = workbook.add_format({'font_name': 'Arial', 'font_size': 10, 'border': 1, 'font_color': 'red'})
-    # 应用表头格式
-    for col_num, value in enumerate(df.columns.values):
-        worksheet.write(0, col_num, value, header_format)
-    # 应用内容格式和条件格式
-    worksheet.conditional_format(1, 0, len(df), len(df.columns)-1, {'type': 'no_blanks', 'format': default_format})
-    try:
-        progress_col_idx = df.columns.get_loc('目前进度')
-        for row_num in range(len(df)):
-            status_text = df.iloc[row_num, progress_col_idx]
-            if isinstance(status_text, str) and "逾期" in status_text:
-                worksheet.set_row(row_num + 1, None, overdue_format)
-    except KeyError:
-        pass
-    # 自动调整列宽
-    for i, col in enumerate(df.columns):
-        column_len = max(df[col].astype(str).str.len().max(), len(col)) + 2
-        worksheet.set_column(i, i, min(column_len, 40))
-    writer.close()
-    bio.seek(0)
-    return bio.read()
 def main():
     arg_file = sys.argv[1] if len(sys.argv) > 1 else None
     result = run_once(arg_file)

 import base64
 from io import BytesIO
 from typing import Optional, Tuple, List
 from datetime import datetime, date
 import pandas as pd
     "主数量","需求日期","供应商","到货日期","到货主数量","入库日期","入库主数量","目前进度"
 ]
+TEMPLATE_CANDIDATES = [
+    "/workspace/邮件发送的格式.xlsx",
+    "/workspace/templates/邮件发送的格式.xlsx",
+    "/app/邮件发送的格式.xlsx",
+    "/app/templates/邮件发送的格式.xlsx",
+]
 # ====== 工具函数 ======
 def _today() -> date:
     return datetime.now().date()
     grouped = df.groupby(group_keys, dropna=False).agg(final_agg_map).reset_index()
     grouped["目前进度"] = grouped.apply(_calc_progress_row, axis=1)
+    def get_sort_key(status_text):
+        if "逾期" in status_text and "未到货" in status_text:
+            return 1
+        if "逾期" in status_text and "部分到货" in status_text:
+            return 2
+        if "未来7天" in status_text:
+            return 3
+        if "完全到货" in status_text:
+            return 4
+        return 5
+    grouped['sort_key'] = grouped['目前进度'].apply(get_sort_key)
+    grouped = grouped.sort_values(by='sort_key').drop(columns=['sort_key'])
     if "计划到货日期" in grouped.columns:
         grouped = grouped.drop(columns=["计划到货日期"])
     final_cols = [col for col in EMAIL_COLS_DEFAULT if col in grouped.columns]
     final = grouped[final_cols]
     date_cols_to_format = ["请购日期", "需求日期", "到货日期", "入库日期"]
     for col in date_cols_to_format:
         if col in final.columns:
+            final[col] = pd.to_datetime(final[col], errors='coerce').dt.strftime('%Y-%m-%d')
     return final
     files.sort(key=os.path.getmtime, reverse=True)
     return files[0]
+def _df_to_styled_excel_bytes(df: pd.DataFrame) -> bytes:
     bio = BytesIO()
+    writer = pd.ExcelWriter(bio, engine='xlsxwriter')
+    sheet_name = '采购执行表'
+    df.fillna('').to_excel(writer, sheet_name=sheet_name, index=False)
+    workbook = writer.book
+    worksheet = writer.sheets[sheet_name]
+    header_format = workbook.add_format({'bold': True, 'font_name': 'Arial', 'font_size': 10, 'border': 1, 'align': 'center', 'valign': 'vcenter'})
+    default_format = workbook.add_format({'font_name': 'Arial', 'font_size': 10, 'border': 1})
+    overdue_format = workbook.add_format({'font_name': 'Arial', 'font_size': 10, 'border': 1, 'font_color': 'red'})
+    for col_num, value in enumerate(df.columns.values):
+        worksheet.write(0, col_num, value, header_format)
+    worksheet.conditional_format(1, 0, len(df), len(df.columns)-1, {'type': 'no_blanks', 'format': default_format})
+    try:
+        progress_col_idx = df.columns.get_loc('目前进度')
+        for row_num in range(len(df)):
+            status_text = df.iloc[row_num, progress_col_idx]
+            if isinstance(status_text, str) and "逾期" in status_text:
+                worksheet.set_row(row_num + 1, None, overdue_format)
+    except KeyError:
+        pass
+    for i, col in enumerate(df.columns):
+        column_len = df[col].astype(str).str.len().max()
+        column_len = max(column_len, len(col) * 2)
+        worksheet.set_column(i, i, min(column_len, 40))
+    writer.close()
     bio.seek(0)
     return bio.read()
     table {{ border-collapse: collapse; font-size: 13px; }}
     table, th, td {{ border: 1px solid #ccc; padding: 6px; }}
     th {{ background:#f6f6f6; }}
+    tr:nth-child(even) {{background-color: #f2f2f2;}}
     </style></head><body><h3>{title}</h3>{table_html}
     </body></html>"""
     return html
         raw = read_system_export(file_path)
         final = aggregate_for_email(raw)
         out_name = f"邮件发送的格式_{datetime.now().strftime('%Y%m%d')}.xlsx"
         out_path = os.path.join(OUTPUT_DIR, out_name)
+        attach = _df_to_styled_excel_bytes(final)
         final.to_excel(out_path, index=False)
         subject = f"采购执行表自动推��� {datetime.now().date()}"
         import traceback
         return {"ok": False, "msg": f"处理文件时发生严重错误: {e}", "traceback": traceback.format_exc()}
 def main():
     arg_file = sys.argv[1] if len(sys.argv) > 1 else None
     result = run_once(arg_file)