placement_tool/service.py

from __future__ import annotations

import gc
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]] = {}

KEYWORD_INDEX = 3
MATCH_TYPE_INDEX = 5
SPEND_INDEX = 14
SALES_INDEX = 15


class PlacementSummaryError(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 = pd.read_csv(file_path)
        else:
            dataframe = pd.read_excel(file_path, sheet_name=0)
        yield dataframe
    finally:
        if dataframe is not None:
            del dataframe
        gc.collect()


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 PlacementSummaryError("仅支持上传 .xlsx、.xls、.xlsm 或 .csv 文件。")

    timestamp = datetime.now().strftime("%Y%m%d%H%M%S%f")
    safe_name = secure_filename(filename) or f"placement_{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_placement_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 analyze_placement_data(file_path: Path, output_dir: Path) -> dict[str, Any]:
    with safe_table_reader(file_path) as dataframe:
        if dataframe.shape[1] <= SALES_INDEX:
            return {
                "success": False,
                "error": "源文件列数不足，至少需要包含到 P 列。",
            }

        working = dataframe.copy()
        column_names = [str(col).strip() for col in working.columns]
        working.columns = column_names

        extracted = pd.DataFrame(
            {
                "关键词": working.iloc[:, KEYWORD_INDEX],
                "匹配方式": working.iloc[:, MATCH_TYPE_INDEX],
                "花费": working.iloc[:, SPEND_INDEX],
                "广告销售额": working.iloc[:, SALES_INDEX],
            }
        )

        extracted["关键词"] = extracted["关键词"].fillna("").astype(str).str.strip()
        extracted["匹配方式"] = extracted["匹配方式"].fillna("").astype(str).str.strip()
        extracted["花费"] = pd.to_numeric(extracted["花费"], errors="coerce").fillna(0)
        extracted["广告销售额"] = pd.to_numeric(extracted["广告销售额"], errors="coerce").fillna(0)

        extracted = extracted[extracted["关键词"] != ""].copy()
        if extracted.empty:
            return {
                "success": False,
                "error": "未识别到有效的关键词数据，请检查 D 列是否有内容。",
            }

        summary = (
            extracted.groupby(["关键词", "匹配方式"], as_index=False, dropna=False)
            .agg({"花费": "sum", "广告销售额": "sum"})
            .sort_values(["花费", "广告销售额"], ascending=[False, False])
            .reset_index(drop=True)
        )
        summary.insert(0, "排名", range(1, len(summary) + 1))
        summary["ACOS"] = summary.apply(
            lambda row: "有点击不出单"
            if row["广告销售额"] == 0
            else f"{(row['花费'] / row['广告销售额']):.2%}",
            axis=1,
        )

        enriched = extracted.merge(
            summary[["关键词", "匹配方式", "花费", "广告销售额", "ACOS"]],
            on=["关键词", "匹配方式"],
            how="left",
            suffixes=("_原始", "_汇总"),
        )

        total_spend = float(summary["花费"].sum())
        total_sales = float(summary["广告销售额"].sum())
        valid_acos = summary[summary["广告销售额"] > 0].copy()
        average_acos = (
            (valid_acos["花费"].sum() / valid_acos["广告销售额"].sum())
            if not valid_acos.empty and valid_acos["广告销售额"].sum() > 0
            else None
        )

        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_filename = f"投放词汇总_{timestamp}.xlsx"
        output_path = output_dir / output_filename

        summary_stats = pd.DataFrame(
            {
                "指标": [
                    "有效原始行数",
                    "关键词+匹配方式组合数",
                    "总花费",
                    "总广告销售额",
                    "平均ACOS",
                ],
                "数值": [
                    f"{len(extracted):,}",
                    f"{len(summary):,}",
                    f"¥{total_spend:,.2f}",
                    f"¥{total_sales:,.2f}",
                    f"{average_acos:.2%}" if average_acos is not None else "有点击不出单",
                ],
            }
        )

        with pd.ExcelWriter(output_path, engine="openpyxl") as writer:
            dataframe.to_excel(writer, sheet_name="原始数据", index=False)
            summary.to_excel(writer, sheet_name="投放词汇总", index=False)
            enriched.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_spend": total_spend,
                "total_sales": total_sales,
                "avg_acos": f"{average_acos:.2%}" if average_acos is not None else "有点击不出单",
            },
            "top_terms": 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"],
    }