from __future__ import annotations import gc import os import traceback from contextlib import contextmanager from datetime import datetime from pathlib import Path from threading import Thread from typing import Any from uuid import uuid4 import pandas as pd from werkzeug.datastructures import FileStorage from werkzeug.utils import secure_filename ALLOWED_EXTENSIONS = {"xlsx", "xls", "xlsm", "csv"} analysis_tasks: dict[str, dict[str, Any]] = {} CHANNEL_INDEX = 1 SGUID_INDEX = 3 AVAILABLE_INDEX = 18 INV_0_90_INDEX = 19 INV_91_180_INDEX = 20 INV_181_270_INDEX = 21 INV_271_365_INDEX = 22 INV_365_PLUS_INDEX = 23 TRACE_CN_MAN_INDEX = 24 SOLD_30_DAYS_INDEX = 34 OVERAGE_STORAGE_INDEX = 36 MONTHLY_STORAGE_INDEX = 37 INCLUDE_RAW_SHEET = os.environ.get("MRO_INCLUDE_RAW_SHEET", "1") != "0" class AgingSummaryError(Exception): pass class AnalysisTaskNotFound(Exception): pass def allowed_file(filename: str) -> bool: return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_EXTENSIONS @contextmanager def safe_table_reader(file_path: Path): dataframe: pd.DataFrame | None = None suffix = file_path.suffix.lower() try: if suffix == ".csv": dataframe = None last_error = None for encoding in ("utf-8-sig", "gb18030", "gbk"): try: dataframe = pd.read_csv(file_path, encoding=encoding) break except Exception as exc: last_error = exc if dataframe is None: raise AgingSummaryError(f"CSV 读取失败: {last_error}") else: dataframe = read_excel_file(file_path, sheet_name=0) yield dataframe finally: if dataframe is not None: del dataframe gc.collect() def read_excel_file(file_path: Path, **kwargs) -> pd.DataFrame: try: return pd.read_excel(file_path, engine="calamine", **kwargs) except ImportError: return pd.read_excel(file_path, **kwargs) def start_analysis_task(uploaded_file: FileStorage, upload_dir: Path, output_dir: Path) -> str: filename = uploaded_file.filename or "" if not allowed_file(filename): raise AgingSummaryError("仅支持上传 .xlsx、.xls、.xlsm 或 .csv 文件。") timestamp = datetime.now().strftime("%Y%m%d%H%M%S%f") safe_name = secure_filename(filename) or f"aging_{timestamp}.xlsx" upload_path = upload_dir / f"{timestamp}_{safe_name}" uploaded_file.save(upload_path) task_id = uuid4().hex analysis_tasks[task_id] = { "status": "processing", "result": None, "filename": upload_path.name, "start_time": datetime.now().isoformat(), } Thread( target=_process_analysis, kwargs={"task_id": task_id, "file_path": upload_path, "output_dir": output_dir}, daemon=True, ).start() return task_id def _process_analysis(task_id: str, file_path: Path, output_dir: Path) -> None: try: result = analyze_aging_stock_data(file_path=file_path, output_dir=output_dir) analysis_tasks[task_id]["status"] = "completed" analysis_tasks[task_id]["result"] = result analysis_tasks[task_id]["end_time"] = datetime.now().isoformat() except Exception as exc: traceback.print_exc() analysis_tasks[task_id]["status"] = "error" analysis_tasks[task_id]["result"] = {"success": False, "error": f"分析过程中出错: {exc}"} analysis_tasks[task_id]["end_time"] = datetime.now().isoformat() finally: file_path.unlink(missing_ok=True) def _collapse_people(series: pd.Series) -> str: values = [str(value).strip() for value in series.fillna("") if str(value).strip()] unique_values = [] for value in values: if value not in unique_values: unique_values.append(value) return " / ".join(unique_values) def analyze_aging_stock_data(file_path: Path, output_dir: Path) -> dict[str, Any]: with safe_table_reader(file_path) as dataframe: if dataframe.shape[1] <= MONTHLY_STORAGE_INDEX: return {"success": False, "error": "源文件列数不足，至少需要包含到 AL 列。"} working = dataframe.copy() extracted = pd.DataFrame( { "channel": working.iloc[:, CHANNEL_INDEX], "SGU": working.iloc[:, SGUID_INDEX], "available": working.iloc[:, AVAILABLE_INDEX], "inv_age_0_to_90_days": working.iloc[:, INV_0_90_INDEX], "inv_age_91_to_180_days": working.iloc[:, INV_91_180_INDEX], "inv_age_181_to_270_days": working.iloc[:, INV_181_270_INDEX], "inv_age_271_to_365_days": working.iloc[:, INV_271_365_INDEX], "inv_age_365_plus_days": working.iloc[:, INV_365_PLUS_INDEX], "trace_cn_man": working.iloc[:, TRACE_CN_MAN_INDEX], "sold_30days": working.iloc[:, SOLD_30_DAYS_INDEX], "超龄仓储": working.iloc[:, OVERAGE_STORAGE_INDEX], "月度仓储": working.iloc[:, MONTHLY_STORAGE_INDEX], } ) extracted["channel"] = extracted["channel"].fillna("").astype(str).str.strip() extracted["SGU"] = extracted["SGU"].fillna("").astype(str).str.strip() extracted["trace_cn_man"] = extracted["trace_cn_man"].fillna("").astype(str).str.strip() numeric_columns = [ "available", "inv_age_0_to_90_days", "inv_age_91_to_180_days", "inv_age_181_to_270_days", "inv_age_271_to_365_days", "inv_age_365_plus_days", "sold_30days", "超龄仓储", "月度仓储", ] for column in numeric_columns: extracted[column] = pd.to_numeric(extracted[column], errors="coerce").fillna(0) extracted = extracted[(extracted["SGU"] != "") & (extracted["channel"] != "")].copy() if extracted.empty: return {"success": False, "error": "未识别到有效的 SGU 和 channel 数据。"} extracted["inv_age_271_plus_days"] = ( extracted["inv_age_271_to_365_days"] + extracted["inv_age_365_plus_days"] ) summary = ( extracted.groupby(["SGU", "channel"], as_index=False, dropna=False) .agg( trace_cn_man=("trace_cn_man", _collapse_people), available=("available", "sum"), inv_age_0_to_90_days=("inv_age_0_to_90_days", "sum"), inv_age_91_to_180_days=("inv_age_91_to_180_days", "sum"), inv_age_181_to_270_days=("inv_age_181_to_270_days", "sum"), inv_age_271_plus_days=("inv_age_271_plus_days", "sum"), sold_30days=("sold_30days", "sum"), 超龄仓储=("超龄仓储", "sum"), 月度仓储=("月度仓储", "sum"), ) .sort_values(["超龄仓储", "available"], ascending=[True, False]) .reset_index(drop=True) ) summary.insert(0, "排名", range(1, len(summary) + 1)) output_filename = f"超龄库存汇总_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" output_path = output_dir / output_filename summary_stats = pd.DataFrame( { "指标": [ "有效原始行数", "SGU+channel 组合数", "总available", "总271天以上库存", "总sold_30days", "总超龄仓储", "总月度仓储", ], "数值": [ f"{len(extracted):,}", f"{len(summary):,}", f"{summary['available'].sum():,.2f}", f"{summary['inv_age_271_plus_days'].sum():,.2f}", f"{summary['sold_30days'].sum():,.2f}", f"{summary['超龄仓储'].sum():,.2f}", f"{summary['月度仓储'].sum():,.2f}", ], } ) with pd.ExcelWriter(output_path, engine="openpyxl") as writer: if INCLUDE_RAW_SHEET: dataframe.to_excel(writer, sheet_name="原始数据", index=False) summary.to_excel(writer, sheet_name="超龄库存汇总", index=False) summary_stats.to_excel(writer, sheet_name="数据摘要", index=False) return { "success": True, "output_file": str(output_path), "output_filename": output_filename, "stats": { "total_rows": int(len(extracted)), "total_groups": int(len(summary)), "total_available": float(summary["available"].sum()), "total_271_plus": float(summary["inv_age_271_plus_days"].sum()), "total_overage_storage": float(summary["超龄仓储"].sum()), "total_monthly_storage": float(summary["月度仓储"].sum()), }, "top_groups": summary.head(10).to_dict("records"), } def get_task_status(task_id: str) -> dict[str, Any]: if task_id not in analysis_tasks: raise AnalysisTaskNotFound("任务不存在。") task = analysis_tasks[task_id] return {"status": task["status"], "result": task["result"]}