aging_tool/service.py

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"]}