Update process_report.py

process_report.py

@@ -1,243 +1,405 @@
 #!/usr/bin/env python3
+# -*- coding: utf-8 -*-
 """
 process_report.py
-核心逻辑：读取最新上传的“系统导出格式”Excel，按业务口径聚合并生成“邮件发送的格式”Excel，
-然后通过 Resend 发邮件（HTML 表格 + 附件）。
-
-注意：请在环境变量中配置：
-  RESEND_API_KEY, FROM_EMAIL, TO_EMAIL, INPUT_DIR, OUTPUT_DIR, TZ
+读取最新或指定的“系统导出格式”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
 import sys
-import base64
 import glob
 import json
-from datetime import datetime, timedelta
+import base64
+from io import BytesIO
+from typing import Optional, Tuple, List
+
+from datetime import datetime, date, timedelta
+
 import pandas as pd
 import requests
 
-# 环境变量和默认路径
-INPUT_DIR = os.environ.get('INPUT_DIR', '/data/uploads')
-OUTPUT_DIR = os.environ.get('OUTPUT_DIR', '/data/outputs')
-RESEND_API_KEY = os.environ.get('RESEND_API_KEY')
-FROM_EMAIL = os.environ.get('FROM_EMAIL')
-TO_EMAIL = os.environ.get('TO_EMAIL')
-TIMEZONE = os.environ.get('TZ', 'Asia/Shanghai')
-
+# ====== 目录 & 环境 ======
+INPUT_DIR = os.environ.get("INPUT_DIR", "/tmp/uploads")
+OUTPUT_DIR = os.environ.get("OUTPUT_DIR", "/tmp/outputs")
 os.makedirs(INPUT_DIR, exist_ok=True)
 os.makedirs(OUTPUT_DIR, exist_ok=True)
 
+RESEND_API_KEY = os.environ.get("RESEND_API_KEY")
+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", "门幅"],
+    "颜色": ["颜色", "色号/颜色", "色号"],
+    "主单位": ["主单位", "单位"],
+    "主数量": ["主数量", "数量", "请购数量"],
+    "需求日期": ["需求日期", "需求日", "交期", "要求到货日期"],
+    "供应商": ["供应商", "供货商", "供应商名称"],
+    "到货日期": ["到货日期", "实到日期", "收货日期"],
+    "到货主数量": ["到货主数量", "到货数量", "实到数量"],
+    "入库日期": ["入库日期", "入库日"],
+    "入库主数量": ["入库主数量", "入库数量"],
+    "计划到货日期": ["计划到货日期", "预计到货日期", "承诺到货日期", "计划到货日"],
+}
+
+# “邮件发送的格式”列顺序，如果检测到模板文件，会按模板优先排序
+EMAIL_COLS_DEFAULT = [
+    "请购日期","请购单号","物料编码","物料名称","纱支密度","门幅（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）。
+    例如：'门幅(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
+
+
+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
+
+
+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)
+
+    # 转日期
+    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
+
+
+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)
 
-def find_latest_input():
-    # 支持多种文件名，匹配 Excel 后缀
-    patterns = [os.path.join(INPUT_DIR, '*.xlsx'), os.path.join(INPUT_DIR, '*.xls')]
+    # 最终列顺序：优先按模板，其次按默认顺序
+    final = _order_like_template(grouped)
+    return final
+
+
+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()
+
+
+def _find_latest_input(input_dir: str) -> Optional[str]:
     files = []
-    for p in patterns:
-        files.extend(glob.glob(p))
+    for pat in ("*.xlsx", "*.xls"):
+        files.extend(glob.glob(os.path.join(input_dir, pat)))
     if not files:
         return None
-    files = sorted(files, key=os.path.getmtime, reverse=True)
+    files.sort(key=os.path.getmtime, reverse=True)
     return files[0]
 
 
-def read_system_export(path):
-    # 根据你给出的示例，表头在第5行（index=4），并且有若干空列
-    try:
-        df = pd.read_excel(path, header=4)
-    except Exception as e:
-        # 备用尝试：不指定 header
-        df = pd.read_excel(path, header=0)
-    # 丢弃全空列
-    df = df.dropna(axis=1, how='all')
-    return df
+def _df_to_excel_bytes(df: pd.DataFrame) -> bytes:
+    bio = BytesIO()
+    df.to_excel(bio, index=False)
+    bio.seek(0)
+    return bio.read()
 
 
-def normalize_and_aggregate(df):
-    # 需要的列名字（与模板保持一致）
-    cols_needed = [
-        '请购日期','请购单号','物料编码','物料名称','纱支密度','门幅（CM)',
-        '颜色','主单位','主数量','需求日期','供应商','到货日期','到货主数量','入库日期','入库主数量','计划到货日期'
-    ]
-    # 取交集
-    present = [c for c in cols_needed if c in df.columns]
-    proc = df[present].copy()
-
-    # 转换日期列
-    date_cols = ['请购日期','需求日期','到货日期','入库日期','计划到货日期']
-    for c in date_cols:
-        if c in proc.columns:
-            proc[c] = pd.to_datetime(proc[c], errors='coerce')
-
-    # 数字列
-    num_cols = ['主数量','到货主数量','入库主数量']
-    for c in num_cols:
-        if c in proc.columns:
-            proc[c] = pd.to_numeric(proc[c], errors='coerce').fillna(0)
-
-    # 聚合键
-    group_keys = [
-        k for k in [
-            '请购日期','请购单号','物料编码','物料名称','纱支密度','门幅（CM)',
-            '颜色','主单位','主数量','需求日期','供应商'
-        ] if k in proc.columns
-    ]
+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
 
-    agg_map = {}
-    if '到货日期' in proc.columns:
-        agg_map['到货日期'] = 'max'
-    if '到货主数量' in proc.columns:
-        agg_map['到货主数量'] = 'sum'
-    if '入库日期' in proc.columns:
-        agg_map['入库日期'] = 'max'
-    if '入库主数量' in proc.columns:
-        agg_map['入库主数量'] = 'sum'
-    if '计划到货日期' in proc.columns:
-        agg_map['计划到货日期'] = 'max'
 
-    if not group_keys:
-        raise RuntimeError('无法找到合适的分组键，请检查上传的模板列名是否正确')
-
-    agg = proc.groupby(group_keys, dropna=False).agg(agg_map).reset_index()
-
-    # 计算“目前进度”
-    today = pd.Timestamp(datetime.now().date())
-
-    def calc_progress(row):
-        main_qty = row.get('主数量', 0) or 0
-        arr_qty = 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 '完全到货'
-
-        shortage = max(0, main_qty - arr_qty)
-
-        # 未到或部分到
-        if pd.isna(arrival_date) or arr_qty < main_qty:
-            # 逾期判断
-            if pd.notna(demand_date) and today > demand_date:
-                overdue_days = (today - demand_date).days
-                if arr_qty > 0:
-                    return f'部分到货 缺货{shortage:g}（已逾期{overdue_days}天）'
-                else:
-                    return f'未到货（已逾期{overdue_days}天）'
-            else:
-                # 未来7天到货
-                if pd.notna(plan_arrival) and today <= plan_arrival <= today + pd.Timedelta(days=7):
-                    return f'未来7天到货（计划{plan_arrival.date()}）'
-                return f'部分到货 缺货{shortage:g}' if arr_qty > 0 else '未到货（未到期）'
-        return '完全到货'
-
-    agg['目前进度'] = agg.apply(calc_progress, axis=1)
-
-    # 按邮件模板列排序（若缺列，保持存在列）
-    email_cols = [
-        '请购日期','请购单号','物料编码','物料名称','纱支密度','门幅（CM)','颜色','主单位',
-        '主数量','需求日期','供应商','到货日期','到货主数量','入库日期','入库主数量','目前进度'
-    ]
-    final_cols = [c for c in email_cols if c in agg.columns]
-    out = agg[final_cols].copy()
-    return out
-
-
-def df_to_excel_bytes(df):
-    from io import BytesIO
-    b = BytesIO()
-    df.to_excel(b, index=False)
-    b.seek(0)
-    return b.read()
-
-
-def send_email_via_resend(subject, html_body, attachment_bytes=None, attachment_name='report.xlsx'):
-    if not RESEND_API_KEY or not FROM_EMAIL or not TO_EMAIL:
-        print('缺少 Resend 配置：RESEND_API_KEY / FROM_EMAIL / TO_EMAIL', file=sys.stderr)
-        return False
-
-    url = 'https://api.resend.com/emails'
+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'
+        "Authorization": f"Bearer {RESEND_API_KEY}",
+        "Content-Type": "application/json",
     }
-
     payload = {
-        'from': FROM_EMAIL,
-        'to': [TO_EMAIL],
-        'subject': subject,
-        'html': html_body
+        "from": FROM_EMAIL,
+        "to": [TO_EMAIL],
+        "subject": subject,
+        "html": html_body,
     }
-
     if attachment_bytes is not None:
-        b64 = base64.b64encode(attachment_bytes).decode('utf-8')
-        payload['attachments'] = [
-            {
-                'filename': attachment_name,
-                'content': b64
-            }
-        ]
+        payload["attachments"] = [{
+            "filename": attachment_name,
+            "content": base64.b64encode(attachment_bytes).decode("utf-8"),
+        }]
 
     resp = requests.post(url, headers=headers, data=json.dumps(payload))
-    try:
-        resp.raise_for_status()
-    except Exception as e:
-        print('发送邮件失败，状态码：', resp.status_code, resp.text, file=sys.stderr)
-        return False
-    print('邮件发送成功，响应：', resp.text)
-    return True
+    if resp.ok:
+        return True, resp.text
+    return False, f"HTTP {resp.status_code}: {resp.text}"
 
 
-def build_html_body(df, title=None):
-    title = title or f'采购执行表 — {datetime.now().date()}'
-    # 简单的 HTML: 标题 + 表格
-    table_html = df.to_html(index=False, escape=False)
-    html = f"""
-    <html>
-      <head>
-        <meta charset='utf-8' />
-        <style>
-          table {{ border-collapse: collapse; }}
-          table, th, td {{ border: 1px solid #ccc; padding: 6px; }}
-          th {{ background:#f3f3f3; }}
-        </style>
-      </head>
-      <body>
-        <h3>{title}</h3>
-        {table_html}
-        <p>备注：此邮件由自动化系统生成。</p>
-      </body>
-    </html>
+# ====== 主流程 ======
+def run_once(file_path: Optional[str] = None) -> dict:
     """
-    return html
-
-
-def main():
-    latest = find_latest_input()
-    if not latest:
-        print('未找到输入文件，请将“系统导出格式”上传到', INPUT_DIR)
-        return
-    print('使用输入文件：', latest)
+    单次处理：读入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(latest)
-    out = normalize_and_aggregate(raw)
+    raw = read_system_export(file_path)
+    final = aggregate_for_email(raw)
 
-    # 输出文件名
-    basename = os.path.splitext(os.path.basename(latest))[0]
     out_name = f"邮件发送的格式_{datetime.now().strftime('%Y%m%d')}.xlsx"
     out_path = os.path.join(OUTPUT_DIR, out_name)
-    out.to_excel(out_path, index=False)
-    print('已生成：', out_path)
+    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),
+    }
 
-    # 生成邮件正文并发送
-    html = build_html_body(out, title=f'采购执行表（{datetime.now().date()}）')
-    attachment = df_to_excel_bytes(out)
-    subject = f'采购执行表自动推送 {datetime.now().date()}'
 
-    ok = send_email_via_resend(subject, html, attachment_bytes=attachment, attachment_name=out_name)
-    if not ok:
-        print('邮件发送失败，请检查 Resend 配置或网络')
+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)
 
 
-if __name__ == '__main__':
-    main()
+if __name__ == "__main__":
+    # 允许命令行传入具体文件路径
+    arg_file = sys.argv[1] if len(sys.argv) > 1 else None
+    main(arg_file)