319 lines
11 KiB
Python
319 lines
11 KiB
Python
"""Pure builders for the sales-analysis chart payload contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Iterable
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
|
|
DOW_LABELS = ["週一", "週二", "週三", "週四", "週五", "週六", "週日"]
|
|
HOUR_LABELS = [f"{hour:02d}:00" for hour in range(24)]
|
|
|
|
|
|
def _has_columns(frame: pd.DataFrame, *columns: str | None) -> bool:
|
|
return bool(columns) and all(column and column in frame.columns for column in columns)
|
|
|
|
|
|
def _to_number(value: Any) -> float:
|
|
try:
|
|
number = float(value)
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
return number if np.isfinite(number) else 0.0
|
|
|
|
|
|
def build_sales_filter_options(
|
|
frame: pd.DataFrame,
|
|
columns: dict[str, str | None],
|
|
) -> dict[str, list[str]]:
|
|
"""Build period-scoped filter choices without another full-table query."""
|
|
result = {
|
|
"categories": [],
|
|
"brands": [],
|
|
"vendors": [],
|
|
"activities": [],
|
|
"payments": [],
|
|
"months": [],
|
|
}
|
|
if frame is None or frame.empty:
|
|
return result
|
|
|
|
column_map = {
|
|
"categories": columns.get("category"),
|
|
"brands": columns.get("brand"),
|
|
"vendors": columns.get("vendor"),
|
|
"activities": columns.get("activity"),
|
|
"payments": columns.get("payment"),
|
|
}
|
|
for key, column in column_map.items():
|
|
if not column or column not in frame.columns:
|
|
continue
|
|
values = frame[column].dropna().astype(str).str.strip()
|
|
result[key] = sorted(value for value in values.unique().tolist() if value)
|
|
|
|
if "_month_str" in frame.columns:
|
|
result["months"] = sorted({
|
|
month
|
|
for value in frame["_month_str"].dropna().tolist()
|
|
if (month := _normalise_month(value))
|
|
})
|
|
return result
|
|
|
|
|
|
def build_analysis_scope(
|
|
period: dict[str, Any],
|
|
*,
|
|
month: Any = "all",
|
|
dow: Any = "all",
|
|
hour: Any = "all",
|
|
) -> dict[str, Any]:
|
|
"""Build the concise, visible scope label shared by every chart and table."""
|
|
parts = [str(period.get("label") or "全部資料")]
|
|
month_value = _normalise_month(month) if str(month or "").lower() != "all" else None
|
|
if month_value:
|
|
parts.append(f"月份 {month_value}")
|
|
|
|
dow_raw = str(dow or "all")
|
|
if dow_raw != "all" and dow_raw.isdigit() and 0 <= int(dow_raw) < len(DOW_LABELS):
|
|
parts.append(DOW_LABELS[int(dow_raw)])
|
|
|
|
hour_raw = str(hour or "all")
|
|
if hour_raw != "all" and hour_raw.isdigit() and 0 <= int(hour_raw) <= 23:
|
|
parts.append(f"{int(hour_raw):02d}:00")
|
|
|
|
return {
|
|
"label": " · ".join(parts),
|
|
"start_date": str(period.get("start_date") or ""),
|
|
"end_date": str(period.get("end_date") or ""),
|
|
}
|
|
|
|
|
|
def build_treemap_chart_data(
|
|
frame: pd.DataFrame,
|
|
category_col: str | None,
|
|
product_name_col: str | None,
|
|
amount_col: str | None,
|
|
*,
|
|
category_limit: int = 10,
|
|
product_limit: int = 5,
|
|
) -> list[dict[str, Any]]:
|
|
"""Return the category/name/value contract consumed by Chart.js treemap."""
|
|
if frame is None or frame.empty or not _has_columns(frame, category_col, product_name_col, amount_col):
|
|
return []
|
|
|
|
grouped = frame[[category_col, product_name_col, amount_col]].copy()
|
|
grouped[amount_col] = pd.to_numeric(grouped[amount_col], errors="coerce").fillna(0)
|
|
grouped[category_col] = grouped[category_col].fillna("未分類").astype(str)
|
|
grouped[product_name_col] = grouped[product_name_col].fillna("未命名商品").astype(str)
|
|
grouped = (
|
|
grouped.groupby([category_col, product_name_col], as_index=False, dropna=False)[amount_col]
|
|
.sum()
|
|
)
|
|
grouped = grouped[grouped[amount_col] > 0]
|
|
if grouped.empty:
|
|
return []
|
|
|
|
category_order = (
|
|
grouped.groupby(category_col)[amount_col]
|
|
.sum()
|
|
.nlargest(category_limit)
|
|
.index.tolist()
|
|
)
|
|
result: list[dict[str, Any]] = []
|
|
for category in category_order:
|
|
products = grouped[grouped[category_col] == category].nlargest(product_limit, amount_col)
|
|
result.extend(
|
|
{
|
|
"category": str(row[category_col]),
|
|
"name": str(row[product_name_col]),
|
|
"value": _to_number(row[amount_col]),
|
|
}
|
|
for _, row in products.iterrows()
|
|
)
|
|
return result
|
|
|
|
|
|
def build_bcg_chart_data(
|
|
frame: pd.DataFrame,
|
|
product_id_col: str | None,
|
|
product_name_col: str | None,
|
|
qty_col: str | None,
|
|
amount_col: str | None,
|
|
profit_col: str = "calculated_profit",
|
|
*,
|
|
limit: int = 400,
|
|
) -> dict[str, Any]:
|
|
"""Aggregate SKU rows and return points plus weighted-margin medians."""
|
|
empty = {"points": [], "x_median": 0.0, "y_median": 0.0}
|
|
if frame is None or frame.empty or not _has_columns(
|
|
frame, product_name_col, qty_col, amount_col, profit_col
|
|
):
|
|
return empty
|
|
|
|
group_cols = [product_name_col]
|
|
if product_id_col and product_id_col in frame.columns:
|
|
group_cols.insert(0, product_id_col)
|
|
columns = group_cols + [qty_col, amount_col, profit_col]
|
|
working = frame[columns].copy()
|
|
for column in (qty_col, amount_col, profit_col):
|
|
working[column] = pd.to_numeric(working[column], errors="coerce").fillna(0)
|
|
|
|
grouped = working.groupby(group_cols, as_index=False, dropna=False).agg(
|
|
_qty=(qty_col, "sum"),
|
|
_amount=(amount_col, "sum"),
|
|
_profit=(profit_col, "sum"),
|
|
)
|
|
grouped = grouped[(grouped["_qty"] > 0) & (grouped["_amount"] > 0)]
|
|
if grouped.empty:
|
|
return empty
|
|
grouped["_margin"] = grouped["_profit"] * 100.0 / grouped["_amount"]
|
|
grouped["_margin"] = grouped["_margin"].replace([np.inf, -np.inf], np.nan).fillna(0)
|
|
grouped = grouped.nlargest(limit, "_amount")
|
|
|
|
names = grouped[product_name_col].fillna("未命名商品").astype(str)
|
|
duplicate_names = names.value_counts()
|
|
points = []
|
|
for _, row in grouped.iterrows():
|
|
raw_name = row.get(product_name_col)
|
|
name = str(raw_name) if raw_name is not None and not pd.isna(raw_name) else "未命名商品"
|
|
product_id = row.get(product_id_col) if product_id_col else None
|
|
if duplicate_names.get(name, 0) > 1 and product_id is not None and not pd.isna(product_id):
|
|
name = f"{name} [{product_id}]"
|
|
points.append(
|
|
{
|
|
"x": _to_number(row["_qty"]),
|
|
"y": _to_number(row["_margin"]),
|
|
"name": name,
|
|
"amount": _to_number(row["_amount"]),
|
|
}
|
|
)
|
|
|
|
return {
|
|
"points": points,
|
|
"x_median": _to_number(grouped["_qty"].median()),
|
|
"y_median": _to_number(grouped["_margin"].median()),
|
|
}
|
|
|
|
|
|
def build_heatmap_chart_data(
|
|
frame: pd.DataFrame,
|
|
amount_col: str | None,
|
|
*,
|
|
dow_col: str = "_dow",
|
|
hour_col: str = "_hour",
|
|
) -> dict[str, Any]:
|
|
"""Return a stable 7 x 24 weekday/hour matrix."""
|
|
empty = {"dows": DOW_LABELS, "hours": HOUR_LABELS, "matrix": []}
|
|
if frame is None or frame.empty or not _has_columns(frame, dow_col, hour_col, amount_col):
|
|
return empty
|
|
|
|
working = frame[[dow_col, hour_col, amount_col]].copy()
|
|
for column in (dow_col, hour_col, amount_col):
|
|
working[column] = pd.to_numeric(working[column], errors="coerce")
|
|
working = working[
|
|
working[dow_col].between(0, 6)
|
|
& working[hour_col].between(0, 23)
|
|
& working[amount_col].notna()
|
|
]
|
|
if working.empty:
|
|
return empty
|
|
|
|
matrix = [[0.0 for _ in range(24)] for _ in range(7)]
|
|
grouped = working.groupby([dow_col, hour_col])[amount_col].sum()
|
|
for (dow, hour), value in grouped.items():
|
|
matrix[int(dow)][int(hour)] = _to_number(value)
|
|
return {"dows": DOW_LABELS, "hours": HOUR_LABELS, "matrix": matrix}
|
|
|
|
|
|
def build_seasonality_chart_data(
|
|
frame: pd.DataFrame,
|
|
category_col: str | None,
|
|
amount_col: str | None,
|
|
*,
|
|
month_col: str = "_month_str",
|
|
category_limit: int = 10,
|
|
) -> dict[str, Any]:
|
|
"""Return categories x months matrix for the selected period only."""
|
|
empty = {"categories": [], "months": [], "matrix": []}
|
|
if frame is None or frame.empty or not _has_columns(frame, month_col, category_col, amount_col):
|
|
return empty
|
|
|
|
working = frame[[month_col, category_col, amount_col]].copy()
|
|
working[month_col] = working[month_col].map(_normalise_month)
|
|
working[category_col] = working[category_col].fillna("未分類").astype(str)
|
|
working[amount_col] = pd.to_numeric(working[amount_col], errors="coerce").fillna(0)
|
|
working = working[working[month_col].notna() & (working[amount_col] > 0)]
|
|
if working.empty:
|
|
return empty
|
|
|
|
categories = (
|
|
working.groupby(category_col)[amount_col]
|
|
.sum()
|
|
.nlargest(category_limit)
|
|
.index.tolist()
|
|
)
|
|
months = sorted(working[month_col].dropna().unique().tolist())
|
|
pivot = (
|
|
working[working[category_col].isin(categories)]
|
|
.pivot_table(index=category_col, columns=month_col, values=amount_col, aggfunc="sum", fill_value=0)
|
|
.reindex(index=categories, columns=months, fill_value=0)
|
|
)
|
|
return {
|
|
"categories": [str(category) for category in categories],
|
|
"months": months,
|
|
"matrix": [[_to_number(value) for value in row] for row in pivot.to_numpy().tolist()],
|
|
}
|
|
|
|
|
|
def build_marketing_chart_data(summary: dict[str, Any] | None, metric: str) -> dict[str, Any]:
|
|
"""Convert marketing summary records into the chart labels/values contract."""
|
|
value_key = {"amount": "revenue", "qty": "qty", "profit": "profit"}.get(metric, "revenue")
|
|
result: dict[str, Any] = {}
|
|
for key in ("discount", "coupon", "bonus", "click"):
|
|
records = summary.get(key, []) if isinstance(summary, dict) else []
|
|
if not isinstance(records, list):
|
|
records = []
|
|
labels = []
|
|
values = []
|
|
for record in records:
|
|
if not isinstance(record, dict):
|
|
continue
|
|
labels.append(str(record.get("name") or "未命名活動"))
|
|
values.append(_to_number(record.get(value_key)))
|
|
result[key] = {"labels": labels, "values": values}
|
|
result["metric"] = value_key
|
|
return result
|
|
|
|
|
|
def resolve_yoy_year_options(
|
|
months: Iterable[Any],
|
|
analysis_period: dict[str, Any] | None,
|
|
*,
|
|
fallback_year: int | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Derive comparison-year controls from live data and the active period."""
|
|
years = {
|
|
int(month[:4])
|
|
for value in (months if months is not None else [])
|
|
if (month := _normalise_month(value))
|
|
}
|
|
period_end = str((analysis_period or {}).get("end_date") or "")
|
|
current_year = int(period_end[:4]) if period_end[:4].isdigit() else (fallback_year or datetime.now().year)
|
|
years.update({current_year - 1, current_year})
|
|
return {
|
|
"years": sorted(years, reverse=True),
|
|
"year1": current_year - 1,
|
|
"year2": current_year,
|
|
}
|
|
|
|
|
|
def _normalise_month(value: Any) -> str | None:
|
|
raw = str(value or "").strip().replace("/", "-")[:7]
|
|
try:
|
|
return datetime.strptime(raw, "%Y-%m").strftime("%Y-%m")
|
|
except ValueError:
|
|
return None
|