Update process_report.py

process_report.py

@@ -13,8 +13,7 @@ import json
 import base64
 from io import BytesIO
 from typing import Optional, Tuple, List
-
-from datetime import datetime, date, timedelta
+from datetime import datetime, date
 
 import pandas as pd
 import requests
@@ -30,15 +29,14 @@ FROM_EMAIL = os.environ.get("FROM_EMAIL")
 TO_EMAIL = os.environ.get("TO_EMAIL")
 TIMEZONE = os.environ.get("TZ", "Asia/Shanghai")
 
-# ====== 业务相关字段（尽量兼容括号全角/半角差异）======
-# 一些系统表可能列名略有差异，这里做一个“同义列名”匹配表
+# ====== 业务相关字段 ======
 ALIASES = {
     "请购日期": ["请购日期", "请购日", "申请日期"],
     "请购单号": ["请购单号", "请购单编号", "申请单号"],
     "物料编码": ["物料编码", "物料号", "物料代码"],
     "物料名称": ["物料名称", "品名", "名称"],
     "纱支密度": ["纱支密度", "纱支/密度", "纱支 密度"],
-    "门幅(CM)": ["门幅（CM)", "门幅(CM)", "门幅cm", "门幅"], # 已修正
+    "门幅(CM)": ["门幅（CM)", "门幅(CM)", "门幅cm", "门幅"],
     "颜色": ["颜色", "色号/颜色", "色号"],
     "主单位": ["主单位", "单位"],
     "主数量": ["主数量", "数量", "请购数量"],
@@ -51,31 +49,16 @@ ALIASES = {
     "计划到货日期": ["计划到货日期", "预计到货日期", "承诺到货日期", "计划到货日"],
 }
 
-# “邮件发送的格式”列顺序，如果检测到模板文件，会按模板优先排序
 EMAIL_COLS_DEFAULT = [
-    "请购日期","请购单号","物料编码","物料名称","纱支密度","门幅(CM)","颜色","主单位", # 已修正
+    "请购日期","请购单号","物料编码","物料名称","纱支密度","门幅(CM)","颜色","主单位",
     "主数量","需求日期","供应商","到货日期","到货主数量","入库日期","入库主数量","目前进度"
 ]
 
-TEMPLATE_CANDIDATES = [
-    # 若你把模板Excel放进仓库根目录或 templates 目录，可被自动识别
-    "/workspace/邮件发送的格式.xlsx",
-    "/workspace/templates/邮件发送的格式.xlsx",
-    "/app/邮件发送的格式.xlsx",
-    "/app/templates/邮件发送的格式.xlsx",
-]
-
-
 # ====== 工具函数 ======
 def _today() -> date:
-    # 用本地系统日期即可（Space容器时区一般是UTC；你可在 Space 里设 TZ 环境变量 + tzdata 以保证正确）
     return datetime.now().date()
 
-
 def _normalize_columns(df: pd.DataFrame) -> pd.DataFrame:
-    """
-    将 DataFrame 的列名映射到标准中文名（按 ALIASES）。
-    """
     mapped = {}
     for std_name, variants in ALIASES.items():
         for v in df.columns:
@@ -86,12 +69,7 @@ def _normalize_columns(df: pd.DataFrame) -> pd.DataFrame:
     df = df.rename(columns=mapped)
     return df
 
-
 def _find_header_row(path: str, must_have: List[str] = None, try_rows: int = 10) -> int:
-    """
-    尝试在前 try_rows 行中找到包含关键列（如“物料编码”、“主数量”）的表头行。
-    找不到则返回 0。
-    """
     must_have = must_have or ["物料编码", "主数量"]
     for r in range(try_rows):
         try:
@@ -103,23 +81,16 @@ def _find_header_row(path: str, must_have: List[str] = None, try_rows: int = 10)
             return r
     return 0
 
-
 def read_system_export(path: str) -> pd.DataFrame:
-    """
-    读取“系统导出格式”Excel，并做列名标准化、空列丢弃、日期/数字类型转换。
-    """
     header_row = _find_header_row(path)
     try:
         df = pd.read_excel(path, header=header_row)
     except Exception:
         df = pd.read_excel(path, header=0)
 
-    # 丢掉全空列
     df = df.dropna(axis=1, how="all")
-    # 标准化列名
     df = _normalize_columns(df)
 
-    # --- 筛选逻辑 ---
     if "物料名称" in df.columns:
         df["物料名称"] = df["物料名称"].astype(str).fillna('')
         contains_e = df["物料名称"].str.contains("鹅")
@@ -128,86 +99,59 @@ def read_system_export(path: str) -> pd.DataFrame:
         contains_huazhu_special = df["物料名称"].str.contains("华住专用")
         to_remove = (contains_e | contains_ya | contains_huazhu) & ~contains_huazhu_special
         df = df[~to_remove]
-    # --- 筛选逻辑结束 ---
 
-    # 转日期
     for c in ["请购日期","需求日期","到货日期","入库日期","计划到货日期"]:
         if c in df.columns:
             df[c] = pd.to_datetime(df[c], errors="coerce")
-
-    # 转数字
+            
     for c in ["主数量","到货主数量","入库主数量"]:
         if c in df.columns:
             df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0)
 
-    # 去掉全空行
     df = df.dropna(how="all")
     return df
 
-
 def _first_nonnull(series: pd.Series):
-    for v in series:
-        if pd.notna(v):
-            return v
-    return None
-
+    return series.dropna().iloc[0] if not series.dropna().empty else None
 
 def aggregate_for_email(df: pd.DataFrame) -> pd.DataFrame:
-    """
-    业务汇总规则
-    """
     group_keys = [k for k in [
-        "请购单号","物料编码","物料名称","纱支密度","门幅(CM)","颜色","主单位","供应商" # 已修正
+        "请购单号","物料编码","物料名称","纱支密度","门幅(CM)","颜色","主单位","供应商"
     ] if k in df.columns]
 
     if not group_keys:
         raise RuntimeError("找不到用于分组的关键字段（如 请购单号/物料编码 等），请检查导入的表头。")
 
-    agg_map = {}
-    if "主数量" in df.columns:
-        agg_map["主数量"] = _first_nonnull
-    if "请购日期" in df.columns:
-        agg_map["请购日期"] = _first_nonnull
-    if "需求日期" in df.columns:
-        agg_map["需求日期"] = _first_nonnull
-    if "到货日期" in df.columns:
-        agg_map["到货日期"] = "max"
-    if "入库日期" in df.columns:
-        agg_map["入库日期"] = "max"
-    if "计划到货日期" in df.columns:
-        agg_map["计划到货日期"] = "max"
-    if "到货主数量" in df.columns:
-        agg_map["到货主数量"] = "sum"
-    if "入库主数量" in df.columns:
-        agg_map["入库主数量"] = "sum"
-
-    grouped = df.groupby(group_keys, dropna=False).agg(agg_map).reset_index()
-
-    # 计算 “目前进度”
-    grouped["目前进度"] = grouped.apply(_calc_progress_row, axis=1)
+    agg_map = {
+        "主数量": _first_nonnull,
+        "请购日期": _first_nonnull,
+        "需求日期": _first_nonnull,
+        "到货日期": "max",
+        "入库日期": "max",
+        "计划到货日期": "max",
+        "到货主数量": "sum",
+        "入库主数量": "sum"
+    }
+    
+    final_agg_map = {k: v for k, v in agg_map.items() if k in df.columns}
 
-    # --- 新增修改 1：移除“计划到货日期”列 ---
-    # 在计算完“目前进度”后，再从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 = _order_like_template(grouped)
-    
-    # --- 新增修改 2：格式化所有日期列为短日期 ---
+    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:
-            # NaT值（空日期）会被保留，非空日期会被格式化为 YYYY/M/D
             final[col] = pd.to_datetime(final[col], errors='coerce').dt.strftime('%Y/%-m/%-d')
 
     return final
 
-
 def _calc_progress_row(row: pd.Series) -> str:
-    """
-    目前进度的业务口径
-    """
     today = pd.Timestamp(_today())
     main_qty = float(row.get("主数量", 0) or 0)
     arr_qty = float(row.get("到货主数量", 0) or 0)
@@ -215,7 +159,9 @@ def _calc_progress_row(row: pd.Series) -> str:
     arrival_date = row.get("到货日期", pd.NaT)
     plan_arrival = row.get("计划到货日期", pd.NaT)
 
-    if main_qty > 0 and arr_qty >= main_qty:
+    # --- 逻辑修改：调整“完全到货”的判断标准 ---
+    # 当到货数量达到主数量的97%时，也视为完全到货
+    if main_qty > 0 and arr_qty >= main_qty * 0.97:
         return "完全到货"
 
     parts: List[str] = []
@@ -223,9 +169,8 @@ def _calc_progress_row(row: pd.Series) -> str:
     
     if arr_qty > 0:
         parts.append(f"部分到货 缺货{shortage:g}米")
-    else:
-        if pd.isna(arrival_date) or arr_qty == 0:
-            parts.append("未到货")
+    elif pd.isna(arrival_date) or arr_qty == 0:
+        parts.append("未到货")
 
     if (pd.isna(arrival_date) or arr_qty < main_qty) and pd.notna(demand_date):
         overdue_days = (today - demand_date).days
@@ -244,30 +189,6 @@ def _calc_progress_row(row: pd.Series) -> str:
         
     return "；".join(parts)
 
-
-def _order_like_template(df: pd.DataFrame) -> pd.DataFrame:
-    """
-    若能找到“邮件发送的格式.xlsx”，按其表头顺序输出；否则用 EMAIL_COLS_DEFAULT。
-    """
-    template_cols = None
-    for p in TEMPLATE_CANDIDATES:
-        if os.path.exists(p):
-            try:
-                tdf = pd.read_excel(p, nrows=0)
-                template_cols = list(map(str, tdf.columns))
-                break
-            except Exception:
-                continue
-
-    if template_cols is None:
-        template_cols = EMAIL_COLS_DEFAULT
-
-    front = [c for c in template_cols if c in df.columns]
-    tail = [c for c in df.columns if c not in front]
-    cols = front + tail
-    return df[cols].copy()
-
-
 def _find_latest_input(input_dir: str) -> Optional[str]:
     files = []
     for pat in ("*.xlsx", "*.xls"):
@@ -277,70 +198,42 @@ 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()
 
-
 def _build_html_body(df: pd.DataFrame, title: str) -> str:
-    # fillna('') 可以在生成html时将 NaN/NaT 显示为空字符串
-    table_html = df.fillna('').to_html(index=False, escape=False)
-    html = f"""<html>
-  <head>
-    <meta charset="utf-8" />
-    <style>
-      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}
+    table_html = df.fillna('').to_html(index=False, escape=False, na_rep="")
+    html = f"""<html><head><meta charset="utf-8" /><style>
+    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>"""
+    </body></html>"""
     return html
 
-
-def _send_email_via_resend(subject: str, html_body: str,
-                           attachment_bytes: Optional[bytes],
-                           attachment_name: str) -> Tuple[bool, str]:
+def _send_email_via_resend(subject: str, html_body: str, attachment_bytes: Optional[bytes], attachment_name: str) -> Tuple[bool, str]:
     if not (RESEND_API_KEY and FROM_EMAIL and TO_EMAIL):
         return False, "缺少 Resend 配置（RESEND_API_KEY / FROM_EMAIL / TO_EMAIL）"
-
+    
     url = "https://api.resend.com/emails"
-    headers = {
-        "Authorization": f"Bearer {RESEND_API_KEY}",
-        "Content-Type": "application/json",
-    }
-    payload = {
-        "from": FROM_EMAIL,
-        "to": [TO_EMAIL],
-        "subject": subject,
-        "html": html_body,
-    }
+    headers = {"Authorization": f"Bearer {RESEND_API_KEY}", "Content-Type": "application/json"}
+    payload = {"from": FROM_EMAIL, "to": [TO_EMAIL], "subject": subject, "html": html_body}
     if attachment_bytes is not None:
-        payload["attachments"] = [{
-            "filename": attachment_name,
-            "content": base64.b64encode(attachment_bytes).decode("utf-8"),
-        }]
-
-    resp = requests.post(url, headers=headers, data=json.dumps(payload))
-    if resp.ok:
+        payload["attachments"] = [{"filename": attachment_name, "content": base64.b64encode(attachment_bytes).decode("utf-8")}]
+    
+    try:
+        resp = requests.post(url, headers=headers, data=json.dumps(payload), timeout=20)
+        resp.raise_for_status()
         return True, resp.text
-    return False, f"HTTP {resp.status_code}: {resp.text}"
-
+    except requests.exceptions.RequestException as e:
+        return False, f"邮件发送请求失败: {e}"
 
 # ====== 主流程 ======
 def run_once(file_path: Optional[str] = None) -> dict:
-    """
-    单次处理：读入Excel -> 汇总 -> 生成输出 -> 发邮件。
-    返回一个 dict 给上层（便于 app.py 返回给前端）。
-    """
     try:
         if file_path is None:
             file_path = _find_latest_input(INPUT_DIR)
@@ -349,10 +242,9 @@ def run_once(file_path: Optional[str] = None) -> dict:
 
         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)
-        os.makedirs(OUTPUT_DIR, exist_ok=True)
         final.to_excel(out_path, index=False)
 
         subject = f"采购执行表自动推送 {datetime.now().date()}"
@@ -360,27 +252,19 @@ def run_once(file_path: Optional[str] = None) -> dict:
         attach = _df_to_excel_bytes(final)
         ok, info = _send_email_via_resend(subject, html, attachment_bytes=attach, attachment_name=out_name)
 
-        return {
-            "ok": ok,
-            "msg": "邮件发送成功" if ok else f"邮件发送失败：{info}",
-            "input": file_path,
-            "output": out_path,
-            "rows": len(final),
-        }
+        return {"ok": ok, "msg": "邮件发送成功" if ok else f"邮件发送失败：{info}", "input": file_path, "output": out_path, "rows": len(final)}
+    
     except Exception as e:
         import traceback
         return {"ok": False, "msg": f"处理文件时发生严重错误: {e}", "traceback": traceback.format_exc()}
 
-
-def main(trigger_file: Optional[str] = None):
-    result = run_once(trigger_file)
-    print(json.dumps(result, ensure_ascii=False, indent=2))
-    # 返回码：成功0，失败1（便于将来做cron/健康检查）
+def main():
+    arg_file = sys.argv[1] if len(sys.argv) > 1 else None
+    result = run_once(arg_file)
+    print(json.dumps(result, ensure_ascii=False))
+    
     if not result.get("ok"):
         sys.exit(1)
 
-
 if __name__ == "__main__":
-    # 允许命令行传入具体文件路径
-    arg_file = sys.argv[1] if len(sys.argv) > 1 else None
-    main(arg_file)
+    main()