diff --git a/config.py b/config.py index c4bac84..f75d0b6 100644 --- a/config.py +++ b/config.py @@ -414,7 +414,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.814" +SYSTEM_VERSION = "V10.815" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/routes/sales_routes.py b/routes/sales_routes.py index 4f2ddc7..56bfd53 100644 --- a/routes/sales_routes.py +++ b/routes/sales_routes.py @@ -34,6 +34,16 @@ from services.analysis_period_service import ( parse_iso_date, parse_month, ) +from services.sales_analysis_chart_service import ( + build_analysis_scope, + build_bcg_chart_data, + build_heatmap_chart_data, + build_marketing_chart_data, + build_sales_filter_options, + build_seasonality_chart_data, + build_treemap_chart_data, + resolve_yoy_year_options, +) from services.cache_manager import ( _SALES_PROCESSED_CACHE, _SALES_OPTIONS_CACHE, @@ -41,7 +51,6 @@ from services.cache_manager import ( _SALES_ANALYSIS_PAGE_CACHE_DIR, set_sales_processed_cache, ) -from utils.text_helpers import get_color_for_string # 時區設定 TAIPEI_TZ = timezone(timedelta(hours=8)) @@ -55,7 +64,6 @@ sales_bp = Blueprint('sales', __name__) _TABLE_DATA_CACHE = {} _TABLE_DATA_CACHE_TTL = 60 _SALES_PREVIEW_CACHE_TTL = 600 -_SALES_OPTIONS_CACHE_TTL = 1800 _SALES_PAGE_CONTEXT_CACHE_TTL = 180 _SALES_PAGE_CONTEXT_CACHE_MAX = 24 _SALES_SHARED_PAGE_CONTEXT_CACHE_TTL = 1800 @@ -75,7 +83,14 @@ def _safe_period_date(year, month, day): return date(year, month, min(day, monthrange(year, month)[1])) -def _project_yoy_period(target_year, start_value=None, end_value=None, month_value=None): +def _project_yoy_period( + target_year, + start_value=None, + end_value=None, + month_value=None, + data_range_value=None, + anchor_date=None, +): start = parse_iso_date(start_value) end = parse_iso_date(end_value) month_raw = str(month_value or '').strip().lower() @@ -96,11 +111,32 @@ def _project_yoy_period(target_year, start_value=None, end_value=None, month_val return projected_start, projected_end if month_raw and month_raw != 'all': - if not month_raw.isdigit() or not 1 <= int(month_raw) <= 12: - raise ValueError('month 必須介於 1 到 12') - month = int(month_raw) + selected_month = parse_month(month_raw) + if selected_month: + month = selected_month.month + elif month_raw.isdigit() and 1 <= int(month_raw) <= 12: + month = int(month_raw) + else: + raise ValueError('month 必須是 1 到 12 或 YYYY-MM') return date(target_year, month, 1), date(target_year, month, monthrange(target_year, month)[1]) + range_raw = str(data_range_value or '').strip() + if range_raw: + if not range_raw.isdigit() or int(range_raw) not in {0, 1, 3, 6, 12}: + raise ValueError('data_range 必須是 0、1、3、6 或 12') + range_months = int(range_raw) + if range_months > 0: + anchor = parse_iso_date(anchor_date) or datetime.now(TAIPEI_TZ).date() + range_start = anchor - timedelta(days=range_months * 30) + projected_end = _safe_period_date(target_year, anchor.month, anchor.day) + projected_start_year = target_year - int(range_start.year < anchor.year) + projected_start = _safe_period_date( + projected_start_year, + range_start.month, + range_start.day, + ) + return projected_start, projected_end + return date(target_year, 1, 1), date(target_year, 12, 31) @@ -339,67 +375,6 @@ def _preview_sales_filter_options(engine, table_name): return options -def _get_sales_filter_options(engine, table_name, cols_map): - """完整篩選選項共用快取；資料匯入後 clear_sales_cache() 會一起清掉。""" - validate_table_name(table_name) - col_category = cols_map.get('category') - col_brand = cols_map.get('brand') - col_vendor = cols_map.get('vendor') - col_activity = cols_map.get('activity') - col_payment = cols_map.get('payment') - col_date = cols_map.get('date') - option_cols = (col_category, col_brand, col_vendor, col_activity, col_payment, col_date) - digest = hashlib.md5(repr(option_cols).encode('utf-8')).hexdigest() - cache_key = f"sales_analysis:filter_options:{table_name}:{digest}" - cached = _get_timed_cache(_SALES_OPTIONS_CACHE, cache_key, _SALES_OPTIONS_CACHE_TTL) - if cached: - return cached - - options = { - 'categories': [], - 'brands': [], - 'vendors': [], - 'activities': [], - 'payments': [], - 'months': [], - } - - def read_distinct(conn, col): - if not col: - return [] - sql = f'SELECT DISTINCT "{col}" FROM "{table_name}" WHERE "{col}" IS NOT NULL AND "{col}" <> \'\' ORDER BY "{col}"' - result = conn.execute(text(sql)).fetchall() - return [str(row[0]) for row in result if row[0]] - - with engine.connect() as conn: - options['categories'] = read_distinct(conn, col_category) - options['brands'] = read_distinct(conn, col_brand) - options['vendors'] = read_distinct(conn, col_vendor) - options['activities'] = read_distinct(conn, col_activity) - options['payments'] = read_distinct(conn, col_payment) - - if col_date: - date_fields = ['日期', '訂單日期', '時間'] - for field in date_fields: - try: - result = conn.execute(text(f""" - SELECT DISTINCT replace(substr("{field}", 1, 7), '/', '-') as month - FROM "{table_name}" - WHERE "{field}" IS NOT NULL AND "{field}" != '' - ORDER BY month - """)).fetchall() - months = [row[0] for row in result if row[0] and '-' in str(row[0])] - if months: - options['months'] = months - sys_log.info(f"[Sales Analysis] 從欄位 {field} 提取到 {len(months)} 個月份: {months}") - break - except Exception as ex: - sys_log.warning(f"[Sales Analysis] 從欄位 {field} 提取月份失敗: {ex}") - - _set_timed_cache(_SALES_OPTIONS_CACHE, cache_key, options) - return options - - def _growth_empty_payload(now_taipei=None): now_taipei = now_taipei or datetime.now(TAIPEI_TZ) return ( @@ -1081,9 +1056,12 @@ def _get_filtered_sales_data(cache_key): if selected_activity != 'all' and col_activity: target_df = target_df[target_df[col_activity] == selected_activity] if selected_payment != 'all' and col_payment: target_df = target_df[target_df[col_payment] == selected_payment] - if selected_dow != 'all' and col_date: target_df = target_df[target_df['_dow'] == int(selected_dow)] - if selected_hour != 'all' and col_date: target_df = target_df[target_df['_hour'] == int(selected_hour)] - if selected_month != 'all' and col_date: target_df = target_df[target_df['_month_str'] == selected_month] + if selected_dow != 'all' and col_date and '_dow' in target_df.columns: + target_df = target_df[target_df['_dow'] == int(selected_dow)] + if selected_hour != 'all' and cols_map.get('has_hour') and '_hour' in target_df.columns: + target_df = target_df[target_df['_hour'] == int(selected_hour)] + if selected_month != 'all' and col_date and '_month_str' in target_df.columns: + target_df = target_df[target_df['_month_str'] == selected_month] if keyword: target_df = target_df[target_df[col_name].astype(str).str.contains(keyword, case=False, na=False)] @@ -1106,8 +1084,17 @@ def _build_top_product_chart_data( limit=20, ): """Aggregate product rows before ranking so one SKU occupies one chart slot.""" - metric_label = '銷售金額 ($)' if selected_metric == 'amount' else '銷售數量' - empty = {'labels': [], 'chart_values': [], 'metric_label': metric_label} + metric_label = { + 'amount': '銷售金額 ($)', + 'qty': '銷售數量', + 'profit': '毛利金額 ($)', + }.get(selected_metric, '銷售金額 ($)') + empty = { + 'labels': [], + 'chart_values': [], + 'metric_label': metric_label, + 'is_money': selected_metric in {'amount', 'profit'}, + } if target_df is None or target_df.empty or not product_name_col or not value_col: return empty @@ -1137,6 +1124,7 @@ def _build_top_product_chart_data( 'labels': labels, 'chart_values': [float(value) for value in grouped['_chart_value']], 'metric_label': metric_label, + 'is_money': selected_metric in {'amount', 'profit'}, } @@ -1259,8 +1247,25 @@ def sales_analysis(): _set_sales_shared_page_context_cache(preview_cache_key, preview_context) return render_template('sales_analysis.html', **preview_context) - # 解析 data_range_months（有篩選時才處理） - data_range_months = int(data_range_param or '0') + # 解析並正規化期間；所有圖、表與 API 共用同一組 canonical 值。 + try: + data_range_months = int(data_range_param or '0') + except (TypeError, ValueError) as exc: + raise ValueError('data_range 必須是 0、1、3、6 或 12') from exc + if data_range_months not in {0, 1, 3, 6, 12}: + raise ValueError('data_range 必須是 0、1、3、6 或 12') + + parsed_start = parse_iso_date(start_date) + parsed_end = parse_iso_date(end_date) + if start_date and not parsed_start: + raise ValueError('start_date 必須是 YYYY-MM-DD') + if end_date and not parsed_end: + raise ValueError('end_date 必須是 YYYY-MM-DD') + if parsed_start and parsed_end and parsed_start > parsed_end: + parsed_start, parsed_end = parsed_end, parsed_start + start_date = parsed_start.isoformat() if parsed_start else '' + end_date = parsed_end.isoformat() if parsed_end else '' + if start_date or end_date: period_start = start_date or end_date period_end = end_date or start_date @@ -1487,14 +1492,14 @@ def sales_analysis(): col_brand = find_col(['品牌', 'Brand']) # V-New: 品牌欄位 col_vendor = find_col(['廠商名稱', 'Vendor Name', '廠商', '供應商', 'Vendor', 'Supplier']) # V-Opt: 優先抓取名稱 col_activity = find_col(['活動', '折扣', 'Activity', 'Campaign', 'Promotion', '專案']) # V-New: 活動欄位 - col_payment = find_col(['付款方式', 'Payment', 'Pay']) # V-New: 付款方式欄位 + col_payment = find_col(['付款方式', '付款', 'Payment', 'Pay']) # V-New: 付款方式欄位 col_price = find_col(['單價', 'Price', '價格', 'Avg Price']) # V-New: 嘗試尋找單價欄位 col_cost = find_col(['成本', 'Cost', '進價', 'Cost Price', 'Wholesale']) # V-New: 成本欄位 col_profit = find_col(['毛利', 'Profit', '利潤']) # V-New: 直接尋找毛利欄位 (若有) col_return_qty = find_col(['退貨數量', 'Return Qty', '退貨']) # V-New: 退貨欄位 col_amount = find_col(['銷售金額', '業績', '金額', 'Amount', 'Sales', 'Total']) col_qty = find_col(['銷售數量', '銷量', '數量', 'Qty', 'Quantity']) - col_category = find_col(['館別', '分類', 'Category']) + col_category = find_col(['商品館', '館別', '分類', 'Category']) if not col_name or not col_amount: return render_template('sales_analysis.html', @@ -1525,25 +1530,29 @@ def sales_analysis(): if col_return_qty: df[col_return_qty] = pd.to_numeric(df[col_return_qty], errors='coerce').fillna(0) - # V-Fix: 智慧日期時間合併邏輯 (聚合模式下跳過) + # 日期維度是所有期間聯動圖表的共同契約；聚合視圖也必須保留日期。 col_date = None - if not is_aggregated_mode: + has_hour_dimension = False + if is_aggregated_mode and col_date_part: + df[col_date_part] = pd.to_datetime(df[col_date_part], errors='coerce') + col_date = col_date_part + else: if col_date_part and col_time_part: - # 兩者都有，嘗試合併 try: df['combined_dt'] = pd.to_datetime(df[col_date_part].astype(str) + ' ' + df[col_time_part].astype(str), errors='coerce') col_date = 'combined_dt' + has_hour_dimension = df['combined_dt'].notna().any() except: - # 合併失敗，退回使用時間欄位 (假設包含日期) 或日期欄位 col_date = col_time_part or col_date_part elif col_time_part: - # 只有時間欄位 (可能包含日期) df[col_time_part] = pd.to_datetime(df[col_time_part], errors='coerce') col_date = col_time_part + has_hour_dimension = df[col_time_part].notna().any() elif col_date_part: - # 只有日期欄位 + raw_date_values = df[col_date_part].astype(str) df[col_date_part] = pd.to_datetime(df[col_date_part], errors='coerce') col_date = col_date_part + has_hour_dimension = raw_date_values.str.contains(':', regex=False, na=False).any() # V-New: 若無明確單價欄位，則自動計算 (金額 / 數量) if not col_price and col_amount and col_qty: @@ -1568,13 +1577,12 @@ def sales_analysis(): df[col_margin_rate] = df[col_margin_rate].replace([np.inf, -np.inf, np.nan], 0) # === V-Opt: 效能優化預計算 (V9.98) === - # 1. 日期維度 (加速篩選與聚合，避免重複呼叫 .dt 存取器) - # 聚合模式下跳過日期維度計算 - if col_date and not is_aggregated_mode: + # 1. 日期維度 (所有期間圖表共用；小時僅在來源真的有時間時開放) + if col_date: df['_dow'] = df[col_date].dt.dayofweek - df['_hour'] = df[col_date].dt.hour df['_week'] = df[col_date].dt.strftime('%G-W%V') - df['_month_str'] = df[col_date].dt.strftime('%Y-%m') # V-New: 月份維度 (YYYY-MM) + df['_month_str'] = df[col_date].dt.strftime('%Y-%m') + df['_hour'] = df[col_date].dt.hour if has_hour_dimension else np.nan # 2. 毛利額 (加速 Top 3 分析，避免 runtime 計算) if col_profit: @@ -1596,7 +1604,8 @@ def sales_analysis(): 'vendor': col_vendor, 'activity': col_activity, 'payment': col_payment, 'price': col_price, 'cost': col_cost, 'profit': col_profit, 'return_qty': col_return_qty, - 'pid': col_pid # V-New: 儲存商品ID欄位 + 'pid': col_pid, + 'has_hour': bool(has_hour_dimension), }, 'pid': col_pid, # V-New: 儲存商品ID欄位 'time': time.time() @@ -1625,43 +1634,24 @@ def sales_analysis(): col_return_qty = cols_map.get('return_qty') col_date = cols_map.get('date') col_pid = cols_map.get('pid') + has_hour_dimension = bool(cols_map.get('has_hour')) - all_categories = [] - all_brands = [] - all_vendors = [] - all_activities = [] - all_payments = [] - all_months = [] - - try: - filter_options = _get_sales_filter_options(db.engine, table_name, cols_map) - all_categories = filter_options['categories'] - all_brands = filter_options['brands'] - all_vendors = filter_options['vendors'] - all_activities = filter_options['activities'] - all_payments = filter_options['payments'] - all_months = filter_options['months'] - except Exception as e: - sys_log.warning(f"[Sales Analysis] 從數據庫查詢下拉選項失敗: {e}") - # 如果查詢失敗，回退到從快取讀取 - if cache_key in _SALES_PROCESSED_CACHE: - original_df = _SALES_PROCESSED_CACHE[cache_key]['df'] - elif table_name in _SALES_PROCESSED_CACHE: - original_df = _SALES_PROCESSED_CACHE[table_name]['df'] - else: - original_df = pd.DataFrame() - - if not original_df.empty: - all_categories = sorted(original_df[col_category].dropna().astype(str).unique().tolist()) if col_category else [] - all_brands = sorted(original_df[col_brand].dropna().astype(str).unique().tolist()) if col_brand else [] - all_vendors = sorted(original_df[col_vendor].dropna().astype(str).unique().tolist()) if col_vendor else [] - all_activities = sorted(original_df[col_activity].dropna().astype(str).unique().tolist()) if col_activity else [] - all_payments = sorted(original_df[col_payment].dropna().astype(str).unique().tolist()) if col_payment else [] - all_months = sorted(original_df['_month_str'].dropna().unique().tolist()) if col_date and '_month_str' in original_df.columns else [] + # 篩選選項直接取自已載入的期間資料，避免再掃一次完整正式資料表。 + filter_options = build_sales_filter_options(df, cols_map) + all_categories = filter_options['categories'] + all_brands = filter_options['brands'] + all_vendors = filter_options['vendors'] + all_activities = filter_options['activities'] + all_payments = filter_options['payments'] + all_months = filter_options['months'] # 取得前端參數供模板回填 selected_category = request.args.get('category', 'all') selected_metric = request.args.get('metric', 'amount') + if selected_metric not in {'amount', 'qty', 'profit'}: + selected_metric = 'amount' + if selected_metric == 'profit' and not (col_cost or col_profit): + selected_metric = 'amount' selected_brand = request.args.get('brand', 'all') selected_vendor = request.args.get('vendor', 'all') selected_activity = request.args.get('activity', 'all') @@ -1679,6 +1669,8 @@ def sales_analysis(): sort_col = col_amount if selected_metric == 'qty' and col_qty: sort_col = col_qty + elif selected_metric == 'profit' and (col_cost or col_profit): + sort_col = 'calculated_profit' target_df = target_df.sort_values(by=sort_col, ascending=False) @@ -1801,51 +1793,26 @@ def sales_analysis(): 'amt': float(amt_val) # 用於 tooltip 顯示金額 }) - # 📊 V-New: BCG 矩陣分析 (BCG Matrix) - # X軸: 銷量 (Qty), Y軸: 毛利率 (Margin %) - bcg_data = {'datasets': [], 'thresholds': {'x': 0, 'y': 0}} - # V-Fix: 確保 calculated_margin_rate 欄位存在 - if col_qty and (col_cost or col_profit) and not target_df.empty and 'calculated_margin_rate' in target_df.columns: - # 1. 計算閾值 (使用中位數，避免極端值影響) - # 過濾掉銷量為 0 的商品，避免干擾閾值計算 - active_products = target_df[target_df[col_qty] > 0] - if not active_products.empty and 'calculated_margin_rate' in active_products.columns: - median_qty = active_products[col_qty].median() - median_margin = active_products['calculated_margin_rate'].median() - - # 若中位數為 0 (例如大部分商品沒銷量)，則給一個預設值以利顯示 - if median_qty == 0: median_qty = 1 - - bcg_data['thresholds'] = {'x': float(median_qty), 'y': float(median_margin)} - - # 2. 分類商品 (四象限) - # Stars (明星): High Qty, High Margin - stars = active_products[(active_products[col_qty] >= median_qty) & (active_products['calculated_margin_rate'] >= median_margin)] - # Cows (金牛): High Qty, Low Margin - cows = active_products[(active_products[col_qty] >= median_qty) & (active_products['calculated_margin_rate'] < median_margin)] - # Questions (問題): Low Qty, High Margin - questions = active_products[(active_products[col_qty] < median_qty) & (active_products['calculated_margin_rate'] >= median_margin)] - # Dogs (瘦狗): Low Qty, Low Margin - dogs = active_products[(active_products[col_qty] < median_qty) & (active_products['calculated_margin_rate'] < median_margin)] - - def format_bcg_points(df_segment): - # 限制點數，避免前端卡頓 (各象限最多 100 點) - return [{'x': float(row[col_qty]), 'y': float(row['calculated_margin_rate']), 'name': str(row[col_name]), 'amt': float(row[col_amount])} for _, row in df_segment.head(100).iterrows()] - - bcg_data['datasets'] = [ - {'label': '明星商品 (Stars)', 'data': format_bcg_points(stars), 'backgroundColor': 'rgba(255, 206, 86, 0.8)', 'borderColor': 'rgba(255, 206, 86, 1)'}, # Yellow - {'label': '金牛商品 (Cows)', 'data': format_bcg_points(cows), 'backgroundColor': 'rgba(75, 192, 192, 0.8)', 'borderColor': 'rgba(75, 192, 192, 1)'}, # Green - {'label': '問題商品 (Questions)', 'data': format_bcg_points(questions), 'backgroundColor': 'rgba(54, 162, 235, 0.8)', 'borderColor': 'rgba(54, 162, 235, 1)'}, # Blue - {'label': '瘦狗商品 (Dogs)', 'data': format_bcg_points(dogs), 'backgroundColor': 'rgba(201, 203, 207, 0.8)', 'borderColor': 'rgba(201, 203, 207, 1)'} # Grey - ] + # BCG 契約以 SKU 聚合後的銷量與加權毛利率為準。 + bcg_data = build_bcg_chart_data( + target_df, + col_pid, + col_name, + col_qty, + col_amount, + ) # 📊 V-New: 時間維度分析 (Time Analysis) dow_data = {'labels': ['週一', '週二', '週三', '週四', '週五', '週六', '週日'], 'chart_values': [0]*7} hourly_data = {'labels': [f"{i:02d}:00" for i in range(24)], 'chart_values': [0]*24} weekly_data = {'labels': [], 'chart_values': []} # V-New: 每週趨勢 monthly_data = {'labels': [], 'chart_values': []} # V-New: 每月趨勢 - heatmap_data = [] # V-New: 多維度熱力圖 (Day x Hour) - treemap_data = [] # V-New: 板塊圖數據 + heatmap_data = ( + build_heatmap_chart_data(target_df, col_amount) + if has_hour_dimension + else {'dows': [], 'hours': [], 'matrix': []} + ) + treemap_data = [] if col_date: # 過濾掉日期無效的資料 @@ -1876,52 +1843,12 @@ def sales_analysis(): # V-Fix (2026-01-23): 處理 NaN 值避免 JSON 序列化失敗 weekly_data['chart_values'] = [float(x) if not np.isnan(x) else 0 for x in week_group.tolist()] - # 4. 多維度熱力圖 (Day x Hour) - V-Fix: 確保數據完整性 - dh_group = target_df.groupby(['_dow', '_hour'])[col_amount].sum() - # V-Opt: 正規化氣泡大小 (Normalize Bubble Size) 以提升可讀性 - max_val = dh_group.max() if not dh_group.empty else 1 - - for (day, hour), val in dh_group.items(): - # V-Fix (2026-01-23): 處理 NaN 值 - if np.isnan(val): - val = 0 - # 將數值映射到 3~25px 的半徑範圍，確保視覺可辨識 - radius = 3 + (math.sqrt(val) / math.sqrt(max_val)) * 22 if val > 0 else 0 - heatmap_data.append({ - 'x': int(hour), # X軸: 小時 (0-23) - 'y': int(day), # Y軸: 星期 (0-6) - 'r': float(radius) if not np.isnan(radius) else 0, # V-Adj: 正規化後半徑 - 'v': float(val) # 實際數值 (用於 Tooltip) - }) - - # 📊 V-New: 板塊圖 (Treemap) 數據準備 - # 結構: Root -> Category -> Product (Top 5 per cat) - if col_category and col_name and col_amount and not target_df.empty: - # V-Opt: 優化聚合邏輯，先聚合再篩選，避免在迴圈中重複過濾大表 - # 1. 先聚合 Category + Product (大幅減少資料量) - cat_prod_group = target_df.groupby([col_category, col_name])[col_amount].sum().reset_index() - - # 2. 找出前 10 大分類 - top_cats = cat_prod_group.groupby(col_category)[col_amount].sum().nlargest(10).index.tolist() - - # 3. 針對前 10 大分類，各取前 5 大商品 - for cat in top_cats: - if not cat: continue - # 在縮減後的資料中篩選，速度極快 - cat_subset = cat_prod_group[cat_prod_group[col_category] == cat] - top_prods = cat_subset.nlargest(5, col_amount) - - for _, row in top_prods.iterrows(): - # V-Fix (2026-01-23): 處理 NaN 值 - amount_val = row[col_amount] - if pd.isna(amount_val): - amount_val = 0 - treemap_data.append({ - 'category': str(cat), - 'product': str(row[col_name]) if pd.notna(row[col_name]) else '', - 'value': float(amount_val), - 'color': get_color_for_string(str(cat)) # V-Fix: 增加顏色參數，確保與分類顏色一致且清晰 - }) + treemap_data = build_treemap_chart_data( + target_df, + col_category, + col_name, + col_amount, + ) # 📊 V-New: ABC 分析 (Pareto Analysis) - TODO #8 # A類: 累積營收 0-80% (核心商品) @@ -2012,59 +1939,29 @@ def sales_analysis(): 'sku_count': int(row[col_name]) }) - # 📊 V-New: 淡旺季熱力圖 (Seasonality Analysis) - TODO #10 - seasonality_data = None - if col_date and col_category and col_amount and not target_df.empty: - # 1. 取得前 10 大分類 (避免圖表過大) - # 使用 target_df (受篩選影響)，這樣可以看特定品牌下的分類季節性 - top_cats_season = target_df.groupby(col_category)[col_amount].sum().nlargest(10).index.tolist() - - # 2. 聚合數據 (Month x Category) - season_group = target_df[target_df[col_category].isin(top_cats_season)].groupby(['_month_str', col_category])[col_amount].sum().reset_index() - - # 3. 轉換為 Bubble Chart 格式 - # X軸: 月份 (需解析 _month_str 取得順序) - # Y軸: 分類 (使用 top_cats_season 的索引) - - # 取得所有月份並排序 - all_months_sorted = _sorted_valid_month_labels(target_df['_month_str'].unique()) - month_map = {m: i for i, m in enumerate(all_months_sorted)} - cat_map = {c: i for i, c in enumerate(top_cats_season)} - - points = [] - max_val_season = season_group[col_amount].max() if not season_group.empty else 1 - - for _, row in season_group.iterrows(): - m_str = row['_month_str'] - cat = row[col_category] - val = row[col_amount] - - if m_str in month_map and cat in cat_map: - # 正規化大小 (3~25px) - radius = 3 + (math.sqrt(val) / math.sqrt(max_val_season)) * 25 if val > 0 else 0 - points.append({ - 'x': month_map[m_str], - 'y': cat_map[cat], - 'r': radius, - 'v': float(val), - 'm': m_str, - 'c': cat - }) - - seasonality_data = { - 'datasets': [{ - 'label': '淡旺季熱點', - 'data': points, - # 顏色將在前端動態生成 - }], - 'yLabels': top_cats_season, - 'xLabels': all_months_sorted - } + seasonality_data = build_seasonality_chart_data( + target_df, + col_category, + col_amount, + ) if col_date else {'categories': [], 'months': [], 'matrix': []} # 📊 V-New 2026-01-15: 行銷活動業績貢獻 (Marketing Campaign Contribution) - marketing_data = None - if not target_df.empty: - marketing_data = prepare_marketing_summary(target_df, sort_by=selected_metric) + marketing_summary = prepare_marketing_summary( + target_df, + sort_by=selected_metric, + ) if not target_df.empty else {} + marketing_data = build_marketing_chart_data(marketing_summary, selected_metric) + yoy_options = resolve_yoy_year_options( + all_months, + analysis_period, + fallback_year=datetime.now(TAIPEI_TZ).year, + ) + analysis_scope = build_analysis_scope( + analysis_period, + month=selected_month, + dow=selected_dow, + hour=selected_hour if has_hour_dimension else 'all', + ) context = { 'marketing_data': marketing_data, @@ -2127,15 +2024,20 @@ def sales_analysis(): 'brand': col_brand, 'return_qty': col_return_qty, 'pid': col_pid, + 'hour': has_hour_dimension, }, 'table_name': table_name, 'data_range_months': data_range_months, 'start_date': start_date, 'end_date': end_date, - 'total_records': len(df), + 'total_records': len(target_df), 'active_page': 'sales', 'db_data_range': db_data_range, 'analysis_period': analysis_period, + 'analysis_scope': analysis_scope, + 'yoy_years': yoy_options['years'], + 'yoy_year1': yoy_options['year1'], + 'yoy_year2': yoy_options['year2'], } _set_sales_page_context_cache(page_cache_key, context) _set_sales_shared_page_context_cache(page_cache_key, context) @@ -3433,8 +3335,13 @@ def api_yoy_comparison(): start_value = request.args.get('start_date') end_value = request.args.get('end_date') month_value = request.args.get('month') - year1_start, year1_end = _project_yoy_period(year1, start_value, end_value, month_value) - year2_start, year2_end = _project_yoy_period(year2, start_value, end_value, month_value) + data_range_value = request.args.get('data_range') + year1_start, year1_end = _project_yoy_period( + year1, start_value, end_value, month_value, data_range_value + ) + year2_start, year2_end = _project_yoy_period( + year2, start_value, end_value, month_value, data_range_value + ) aggregate_sql, metric_label = metric_options[metric] engine = DatabaseManager().engine @@ -3480,7 +3387,12 @@ def api_yoy_comparison(): 'metric_label': metric_label, 'monthly_breakdown': monthly_breakdown, 'period': { - 'linked': bool(start_value or end_value or (month_value and month_value != 'all')), + 'linked': bool( + start_value + or end_value + or (month_value and month_value != 'all') + or (data_range_value and data_range_value != '0') + ), 'year1': { 'start_date': year1_start.isoformat(), 'end_date': year1_end.isoformat(), diff --git a/services/sales_analysis_chart_service.py b/services/sales_analysis_chart_service.py new file mode 100644 index 0000000..e2e5a8d --- /dev/null +++ b/services/sales_analysis_chart_service.py @@ -0,0 +1,318 @@ +"""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 diff --git a/templates/sales_analysis.html b/templates/sales_analysis.html index 8140b46..d628856 100644 --- a/templates/sales_analysis.html +++ b/templates/sales_analysis.html @@ -13,6 +13,12 @@ - Marketing chart titles：text-primary 藍 / text-success 綠 → page-accent / olive #} +{% macro period_badge(period) -%} + + {{ period.label }} + +{%- endmacro %} + {% block extra_css %} @@ -141,7 +147,7 @@ - {% for opt in [(1,'最近 1 個月 (推薦)'), (3,'最近 3 個月'), (6,'最近 6 個月'), (12,'最近 12 個月'), (0,'全部資料')] %} @@ -279,7 +285,7 @@
-
+
{% set dow_labels = {'all':'全部星期','0':'週一','1':'週二','2':'週三','3':'週四','4':'週五','5':'週六','6':'週日'} %} -
+
+ {% if cols.hour %}
@@ -332,6 +339,7 @@
+ {% endif %}
@@ -479,21 +487,18 @@ 年度成長對照
+ {{ period_badge(analysis_scope) }} vs +
+ {{ period_badge(analysis_scope) }} + +
@@ -776,46 +808,64 @@
-
每月業績趨勢
+
+ 每月業績趨勢 + {{ period_badge(analysis_scope) }} +
-
每週業績趨勢（{{ analysis_period.label }}）
+
+ 每週業績趨勢 + {{ period_badge(analysis_scope) }} +
-
+
-
每日業績趨勢 (週一至週日)
+
+ 星期別業績分佈 + {{ period_badge(analysis_scope) }} +
+ {% if cols.hour %}
-
每小時業績熱點
+
+ 每小時業績熱點 + {{ period_badge(analysis_scope) }} +
-
- 多維度熱點 (星期 × 小時) - +
+ + 星期 × 小時業績熱點 + + + {{ period_badge(analysis_scope) }}
+ {% endif %}
{% endif %}
-
- 商品作戰清單（{{ analysis_period.label }}） +
+ 商品作戰清單 + {{ period_badge(analysis_scope) }}
@@ -840,6 +890,43 @@ +
+ + + + + + + + + + +
排名商品分類數值毛利率
+
+
+
+ + 匯出明細 + + +
+
+
+
+ {% endif %}{# /no_filter else #}
{# /sales-analysis-page #} {% endif %}{# /error #} diff --git a/tests/test_analysis_period_linkage.py b/tests/test_analysis_period_linkage.py index 3faff66..9809fc8 100644 --- a/tests/test_analysis_period_linkage.py +++ b/tests/test_analysis_period_linkage.py @@ -296,12 +296,22 @@ def test_sales_frontend_uses_live_period_linked_api_routes(): assert "new URLSearchParams(window.location.search)" in script assert "/api/sales_analysis/yoy_comparison" in script assert "/api/sales_analysis/table_data" in script + assert "/api/sales_analysis/top_detail" in script + assert "topDetailModal" in script + assert "window.open(url.toString()" not in script + assert "form.requestSubmit()" in script + assert "if (start) start.value = '';" in script + assert "if (end) end.value = '';" in script assert "d.growth_rate" in script assert "d.monthly_breakdown" in script assert "columns," in script assert 'data-field="product_id"' in template assert 'data-field="amount"' in template - assert "每週業績趨勢（{{ analysis_period.label }}）" in template + assert "data-analysis-period" in template + assert 'id="topDetailModal"' in template + assert template.count("period_badge(analysis_scope)") >= 10 + assert "yoy_years" in template + assert '