Update process_report.py

process_report.py

@@ -3,24 +3,7 @@
 """
 process_report.py
 读取最新或指定的“系统导出格式”Excel，按业务口径聚合为“邮件发送的格式”，
-在“目前进度”列写入：
-  - 完全到货
-  - 部分到货 缺货X米（X=主数量-到货主数量汇总）
-  - 未到货（含逾期天数）
-  - 未来7天到货（依据 计划到货日期）
 并通过 Resend 发送邮件（HTML表格 + Excel附件）。
-
-环境变量（Hugging Face → Settings → Variables and secrets → Repository secrets）：
-  RESEND_API_KEY  : Resend 的 API Key（必须）
-  FROM_EMAIL      : 发件人，例如 "采购机器人 <bot@your-domain.com>"（必须）
-  TO_EMAIL        : 收件人，例如 "you@your-company.com"（必须）
-  INPUT_DIR       : 输入目录（默认 /tmp/uploads）
-  OUTPUT_DIR      : 输出目录（默认 /tmp/outputs）
-  TZ              : 时区（默认 Asia/Shanghai）
-
-使用方式：
-  1) 上传文件到 INPUT_DIR 后，调用 main(trigger_file=该文件路径)
-  2) 或命令行：python process_report.py [可选:具体文件路径]
 """
 
 import os
@@ -30,15 +13,14 @@ 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
 
 # ====== 目录 & 环境 ======
-INPUT_DIR = os.environ.get("INPUT_DIR", "/tmp/uploads")
-OUTPUT_DIR = os.environ.get("OUTPUT_DIR", "/tmp/outputs")
+INPUT_DIR = os.environ.get("INPUT_DIR", "/data/uploads")
+OUTPUT_DIR = os.environ.get("OUTPUT_DIR", "/data/outputs")
 os.makedirs(INPUT_DIR, exist_ok=True)
 os.makedirs(OUTPUT_DIR, exist_ok=True)
 
@@ -47,378 +29,233 @@ 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", "门幅"],
+    "颜色": ["颜色", "色号/颜色", "色号"],
+    "主单位": ["主单位", "单位"],
+    "主数量": ["主数量", "数量", "请购数量"],
+    "需求日期": ["需求日期", "需求日", "交期", "要求到货日期"],
+    "供应商": ["供应商", "供货商", "供应商名称"],
+    "到货日期": ["到货日期", "实到日期", "收货日期"],
+    "到货主数量": ["到货主数量", "到货数量", "实到数量"],
+    "入库日期": ["入库日期", "入库日"],
+    "入库主数量": ["入库主数量", "入库数量"],
+    "计划到货日期": ["计划到货日期", "预计到货日期", "承诺到货日期", "计划到货日"],
 }
 
-# “邮件发送的格式”列顺序，如果检测到模板文件，会按模板优先排序
 EMAIL_COLS_DEFAULT = [
-    "请购日期","请购单号","物料编码","物料名称","纱支密度","门幅(CM)","颜色","主单位", # 已修正
-    "主数量","需求日期","供应商","到货日期","到货主数量","入库日期","入库主数量","目前进度"
-]
-
-TEMPLATE_CANDIDATES = [
-    # 若你把模板Excel放进仓库根目录或 templates 目录，可被自动识别
-    "/workspace/邮件发送的格式.xlsx",
-    "/workspace/templates/邮件发送的格式.xlsx",
-    "/app/邮件发送的格式.xlsx",
-    "/app/templates/邮件发送的格式.xlsx",
+    "请购日期","请购单号","物料编码","物料名称","纱支密度","门幅(CM)","颜色","主单位",
+    "主数量","需求日期","供应商","到货日期","到货主数量","入库日期","入库主数量","目前进度"
 ]
 
-
 # ====== 工具函数 ======
 def _today() -> date:
-    # 用本地系统日期即可（Space容器时区一般是UTC；你可在 Space 里设 TZ 环境变量 + tzdata 以保证正确）
-    return datetime.now().date()
-
+    return datetime.now().date()
 
 def _normalize_columns(df: pd.DataFrame) -> pd.DataFrame:
-    """
-    将 DataFrame 的列名映射到标准中文名（按 ALIASES）。
-    例如：'门幅(CM)' -> '门幅(CM)'
-    """
-    mapped = {}
-    for std_name, variants in ALIASES.items():
-        for v in df.columns:
-            v_clean = str(v).strip()
-            if v_clean in variants:
-                mapped[v] = std_name
-                break
-    df = df.rename(columns=mapped)
-    return df
-
+    mapped = {}
+    for std_name, variants in ALIASES.items():
+        for v in df.columns:
+            v_clean = str(v).strip()
+            if v_clean in variants:
+                mapped[v] = std_name
+                break
+    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:
-            df_try = pd.read_excel(path, header=r, nrows=1)
-        except Exception:
-            continue
-        cols = [str(c).strip() for c in df_try.columns]
-        if all(any(m in c for c in cols) or m in cols for m in must_have):
-            return r
-    return 0
-
+    must_have = must_have or ["物料编码", "主数量"]
+    for r in range(try_rows):
+        try:
+            df_try = pd.read_excel(path, header=r, nrows=1)
+        except Exception:
+            continue
+        cols = [str(c).strip() for c in df_try.columns]
+        if all(any(m in c for c in cols) or m in cols for m in must_have):
+            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)
-
-        # 定义需要排除和保留的条件
-        contains_e = df["物料名称"].str.contains("鹅", na=False)
-        contains_ya = df["物料名称"].str.contains("鸭", na=False)
-        contains_huazhu = df["物料名称"].str.contains("华住", na=False)
-        contains_huazhu_special = df["物料名称"].str.contains("华住专用", na=False)
-
-        # 组合条件：
-        # 需要移除的行 = (包含“鹅” 或 “鸭” 或 “华住”) 并且 (不包含“华住专用”)
-        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
-
+    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)
+        contains_e = df["物料名称"].str.contains("鹅", na=False)
+        contains_ya = df["物料名称"].str.contains("鸭", na=False)
+        contains_huazhu = df["物料名称"].str.contains("华住", na=False)
+        contains_huazhu_special = df["物料名称"].str.contains("华住专用", na=False)
+        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:
-    """
-    业务汇总规则：
-      - “主数量”不做分类汇总（保留原单的主数量），
-      - “到货主数量”、“入库主数量”需要汇总求和，
-      - “到货日期/入库日期/计划到货日期”取最大（最近），
-      - 分组键：不包含数量列，常用维度如下（尽量稳定，避免将数量/日期放入分组键）：
-        请购单号、物料编码、物料名称、纱支密度、门幅（CM)、颜色、主单位、供应商
-      - “请购日期/需求日期”保留“首个非空”
-    """
-    group_keys = [k for k in [
-        "请购单号","物料编码","物料名称","纱支密度","门幅(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)
-
-    # 最终列顺序：优先按模板，其次按默认顺序
-    final = _order_like_template(grouped)
-    return final
-
+    group_keys = [k for k in [
+        "请购单号","物料编码","物料名称","纱支密度","门幅(CM)","颜色","主单位","供应商"
+    ] if k in df.columns]
+
+    if not group_keys:
+        raise RuntimeError("找不到用于分组的关键字段（如 请购单号/物料编码 等），请检查导入的表头。")
+
+    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}
+
+    grouped = df.groupby(group_keys, dropna=False).agg(final_agg_map).reset_index()
+    grouped["目前进度"] = grouped.apply(_calc_progress_row, axis=1)
+    
+    final_cols = [col for col in EMAIL_COLS_DEFAULT if col in grouped.columns]
+    return grouped[final_cols]
 
 def _calc_progress_row(row: pd.Series) -> str:
-    """
-    目前进度的业务口径：
-      1) 到货主数量 ≥ 主数量 => “完全到货”
-      2) 否则：
-         - 若到货主数量 > 0 => “部分到货 缺货X米”
-         - 若到货主数量 == 0 且到货日期为空 => “未到货”
-         - 逾期：需求日期存在且 today > 需求日期，追加 “已逾期Y天”
-         - 未来7天到货：计划到货日期在 [today, today+7] 区间，追加 “未来7天到货（计划YYYY-MM-DD）”
-    """
-    today = pd.Timestamp(_today())
-    main_qty = float(row.get("主数量", 0) or 0)
-    arr_qty = float(row.get("到货主数量", 0) or 0)
-
-    demand_date = row.get("需求日期", pd.NaT)
-    arrival_date = row.get("到货日期", pd.NaT)
-    plan_arrival = row.get("计划到货日期", pd.NaT)
-
-    # 完全到货
-    if main_qty > 0 and arr_qty >= main_qty:
-        return "完全到货"
-
-    parts: List[str] = []
-
-    # 部分/未到
-    shortage = max(0.0, main_qty - arr_qty)
-    if arr_qty > 0:
-        parts.append(f"部分到货 缺货{shortage:g}米")
-    else:
-        # 到货日期为空或数量为0都视为未到货
-        if 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 - pd.Timestamp(demand_date.date())).days
-        if overdue_days > 0:
-            parts.append(f"已逾期{overdue_days}天")
-
-    # 未来7天到货（计划）
-    if pd.notna(plan_arrival):
-        days_ahead = (pd.Timestamp(plan_arrival.date()) - today).days
-        if 0 <= days_ahead <= 7:
-            parts.append(f"未来7天到货（计划{str(plan_arrival.date())}）")
-
-    # 如果什么都没有匹配，给一个保底描述
-    if not parts:
-        # 例如：需求未到期且无计划到货
-        if pd.notna(demand_date) and today <= pd.Timestamp(demand_date.date()):
-            return "未到货（未到期）"
-        return "处理中"
-
-    return "；".join(parts)
-
-
-def _order_like_template(df: pd.DataFrame) -> pd.DataFrame:
-    """
-    若能找到“邮件发送的格式.xlsx”，按其表头顺序输出；否则用 EMAIL_COLS_DEFAULT。
-    模板里缺的列会自动从 df 里补；df 有但模板没有的列会追加在后面。
-    """
-    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]
-    # 再把 df 里其余列追加在后
-    tail = [c for c in df.columns if c not in front]
-    cols = front + tail
-    return df[cols].copy()
-
+    today = pd.Timestamp(_today())
+    main_qty = float(row.get("主数量", 0) or 0)
+    arr_qty = float(row.get("到货主数量", 0) or 0)
+    demand_date = row.get("需求日期", pd.NaT)
+    arrival_date = row.get("到货日期", pd.NaT)
+    plan_arrival = row.get("计划到货日期", pd.NaT)
+
+    if main_qty > 0 and arr_qty >= main_qty:
+        return "完全到货"
+
+    parts: List[str] = []
+    shortage = max(0.0, main_qty - arr_qty)
+    
+    if arr_qty > 0:
+        parts.append(f"部分到货 缺货{shortage:g}米")
+    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 - pd.Timestamp(demand_date.date())).days
+        if overdue_days > 0:
+            parts.append(f"已逾期{overdue_days}天")
+
+    if pd.notna(plan_arrival):
+        days_ahead = (pd.Timestamp(plan_arrival.date()) - today).days
+        if 0 <= days_ahead <= 7:
+            parts.append(f"未来7天到货（计划{str(plan_arrival.date())}）")
+    
+    if not parts:
+        if pd.notna(demand_date) and today <= pd.Timestamp(demand_date.date()):
+            return "未到货（未到期）"
+        return "处理中"
+        
+    return "；".join(parts)
 
 def _find_latest_input(input_dir: str) -> Optional[str]:
-    files = []
-    for pat in ("*.xlsx", "*.xls"):
-        files.extend(glob.glob(os.path.join(input_dir, pat)))
-    if not files:
-        return None
-    files.sort(key=os.path.getmtime, reverse=True)
-    return files[0]
-
+    files = []
+    for pat in ("*.xlsx", "*.xls"):
+        files.extend(glob.glob(os.path.join(input_dir, pat)))
+    if not files:
+        return None
+    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()
-
+    bio = BytesIO()
+    df.to_excel(bio, index=False)
+    bio.seek(0)
+    return bio.read()
 
 def _build_html_body(df: pd.DataFrame, title: str) -> str:
-    table_html = df.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}
-    <p style="color:#666;">备注：此邮件由自动化系统生成。</p>
-  </body>
-</html>"""
-    return html
-
-
-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,
-    }
-    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:
-        return True, resp.text
-    return False, f"HTTP {resp.status_code}: {resp.text}"
-
+    table_html = df.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>"""
+    return html
+
+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}
+    if attachment_bytes is not None:
+        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
+    except requests.exceptions.RequestException as e:
+        return False, f"邮件发送请求失败: {e}"
 
 # ====== 主流程 ======
 def run_once(file_path: Optional[str] = None) -> dict:
-    """
-    单次处理：读入Excel -> 汇总 -> 生成输出 -> 发邮件。
-    返回一个 dict 给上层（便于 app.py 返回给前端）。
-    """
-    if file_path is None:
-        file_path = _find_latest_input(INPUT_DIR)
-        if not file_path:
-            return {"ok": False, "msg": f"未在 {INPUT_DIR} 找到Excel输入文件"}
-
-    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()}"
-    html = _build_html_body(final, title=f"采购执行表（{datetime.now().date()}）")
-    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),
-    }
-
-
-def main(trigger_file: Optional[str] = None):
-    result = run_once(trigger_file)
-    print(json.dumps(result, ensure_ascii=False, indent=2))
-    # 返回码：成功0，失败1（便于将来做cron/健康检查）
-    if not result.get("ok"):
-        sys.exit(1)
-
+    try:
+        if file_path is None:
+            file_path = _find_latest_input(INPUT_DIR)
+            if not file_path:
+                return {"ok": False, "msg": f"未在 {INPUT_DIR} 找到Excel输入文件"}
+
+        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)
+        final.to_excel(out_path, index=False)
+
+        subject = f"采购执行表自动推送 {datetime.now().date()}"
+        html = _build_html_body(final, title=f"采购执行表（{datetime.now().date()}）")
+        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)}
+    
+    except Exception as e:
+        # 捕获处理流程中的任何其他错误，并以详细的 traceback 形式返回
+        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)
+    # 打印 JSON 结果，便于上层调用者（如 app.py）解析
+    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()