Update process_report.py

process_report.py

@@ -1,130 +1,48 @@
 #!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 """
 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
-import sys
-import glob
-import json
-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", "/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",
-]
+import datetime as dt
+from pathlib import Path
 
 
-# ====== 工具函数 ======
-def _today() -> date:
-    # 用本地系统日期即可（Space容器时区一般是UTC；你可在 Space 里设 TZ 环境变量 + tzdata 以保证正确）
-    return datetime.now().date()
-
-
-def _normalize_columns(df: pd.DataFrame) -> pd.DataFrame:
+def _find_header_row(path: str) -> int:
     """
-    将 DataFrame 的列名映射到标准中文名（按 ALIASES）。
-    例如：'门幅(CM)' -> '门幅（CM)'
+    找到 Excel 中的表头行（假设含“物料名称”那一行就是表头）。
     """
-    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
+    for i in range(5):
+        row = pd.read_excel(path, header=i, nrows=1)
+        if "物料名称" in row.columns:
+            return i
+    return 0
 
 
-def _find_header_row(path: str, must_have: List[str] = None, try_rows: int = 10) -> int:
+def _normalize_columns(df: pd.DataFrame) -> pd.DataFrame:
     """
-    尝试在前 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
+    rename_map = {
+        "物料名称": "物料名称",
+        "需求日期": "需求日期",
+        "请购日期": "请购日期",
+        "计划到货日期": "计划到货日期",
+        "到货日期": "到货日期",
+        "入库日期": "入库日期",
+        "主数量": "主数量",
+        "到货主数量": "到货主数量",
+        "入库主数量": "入库主数量",
+    }
+    df = df.rename(columns=lambda x: str(x).strip())
+    df = df.rename(columns=rename_map)
+    return df
 
 
 def read_system_export(path: str) -> pd.DataFrame:
     """
-    读取“系统导出格式”Excel，并做列名标准化、空列丢弃、日期/数字类型转换。
+    读取“系统导出格式”Excel，并做清洗。
     """
     header_row = _find_header_row(path)
     try:
@@ -134,272 +52,96 @@ def read_system_export(path: str) -> pd.DataFrame:
 
     # 丢掉全空列
     df = df.dropna(axis=1, how="all")
-    # 标准化列名
+    # 列名统一
     df = _normalize_columns(df)
 
     # 转日期
-    for c in ["请购日期","需求日期","到货日期","入库日期","计划到货日期"]:
+    for c in ["请购日期", "需求日期", "到货日期", "入库日期", "计划到货日期"]:
         if c in df.columns:
             df[c] = pd.to_datetime(df[c], errors="coerce")
 
     # 转数字
-    for c in ["主数量","到货主数量","入库主数量"]:
+    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)
 
-    # 最终列顺序：优先按模板，其次按默认顺序
-    final = _order_like_template(grouped)
-    return final
+    # ========= 新增过滤逻辑 =========
+    if "物料名称" in df.columns:
+        mask_remove = df["物料名称"].str.contains("鹅|鸭|华住", na=False)
+        mask_keep = df["物料名称"].str.contains("华住专用", na=False)
+        df = df[~mask_remove | mask_keep]
+    # ===============================
 
-
-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)
+    return df
 
 
-def _order_like_template(df: pd.DataFrame) -> pd.DataFrame:
+def analyze_report(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 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()
-
-
-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}"
-
-
-# ====== 主流程 ======
-def run_once(file_path: Optional[str] = None) -> dict:
+    today = dt.datetime.now().date()
+
+    # 汇总数量
+    grouped = df.groupby("物料名称", as_index=False).agg({
+        "主数量": "sum",
+        "到货主数量": "sum",
+        "入库主数量": "sum",
+        "需求日期": "max",
+        "到货日期": "max",
+    })
+
+    def calc_progress(row):
+        demand = row["主数量"]
+        received = row["到货主数量"]
+        due_date = row["需求日期"]
+        arrival_date = row["到货日期"]
+
+        if received >= demand and demand > 0:
+            return "完全到货"
+
+        # 部分到货
+        if 0 < received < demand:
+            missing = demand - received
+            return f"部分到货，缺 {missing:.0f} 米"
+
+        # 未到货
+        if received == 0:
+            if pd.isna(due_date):
+                return "未到货"
+            days_diff = (today - due_date.date()).days
+            if days_diff > 0:
+                return f"逾期 {days_diff} 天未到货"
+            elif 0 <= (due_date.date() - today).days <= 7:
+                return "未来7天要到货"
+            else:
+                return "未到货"
+
+        return "未到货"
+
+    grouped["目前进度"] = grouped.apply(calc_progress, axis=1)
+
+    return grouped
+
+
+def save_to_excel(df: pd.DataFrame, out_path: str):
     """
-    单次处理：读入Excel -> 汇总 -> 生成输出 -> 发邮件。
-    返回一个 dict 给上层（便于 app.py 返回给前端）。
+    结果保存到 Excel
     """
-    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),
-    }
+    Path(out_path).parent.mkdir(parents=True, exist_ok=True)
+    df.to_excel(out_path, index=False)
 
 
-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)
+def main():
+    src = "系统导出格式.xlsx"
+    out = "输出结果.xlsx"
+    df = read_system_export(src)
+    result = analyze_report(df)
+    save_to_excel(result, out)
+    print(f"已生成：{out}")
 
 
 if __name__ == "__main__":
-    # 允许命令行传入具体文件路径
-    arg_file = sys.argv[1] if len(sys.argv) > 1 else None
-    main(arg_file)
+    main()