diff --git a/app.py b/app.py index 6b3c338..0dcc7f0 100644 --- a/app.py +++ b/app.py @@ -5908,402 +5908,6 @@ def growth_analysis(): sys_log.error(f"Growth Analysis Error: {e}") return f"系統錯誤: {e}" -# ================= 📅 V-New: 當日業績看板 ================= - -@app.route('/daily_sales') -def daily_sales(): - """當日業績看板 (Day-over-Day 與 Week-over-Week 分析)""" - try: - db = DatabaseManager() - engine = db.engine - table_name = 'daily_sales_snapshot' - - # 1. 檢查資料表是否存在 - inspector = inspect(engine) - if table_name not in inspector.get_table_names(): - return render_template('daily_sales.html', - error="尚未匯入當日業績資料，請先至系統設定頁面匯入 Excel。", - selected_date=None, available_dates=[], current=None, dod=None, wow=None, - chart_data=None, categories=None, calendar_data=None, selected_month=None) - - # 2. 讀取資料（使用快取） - cache_key = f'{table_name}_daily' - if cache_key in _SALES_PROCESSED_CACHE: - df = _SALES_PROCESSED_CACHE[cache_key]['df'] - else: - df = pd.read_sql(f"SELECT * FROM {table_name}", engine) - if df.empty: - return render_template('daily_sales.html', - error="資料表為空，請先匯入當日業績資料。", - selected_date=None, available_dates=[], current=None, dod=None, wow=None, - chart_data=None, categories=None, calendar_data=None, selected_month=None) - - # 3. 資料前處理（欄位識別、型別轉換） - df = preprocess_daily_sales_data(df) - _SALES_PROCESSED_CACHE[cache_key] = {'df': df} - - # 4. 取得可用日期列表 - available_dates = sorted(df['snapshot_date'].unique(), reverse=True) - available_dates_str = [d.strftime('%Y-%m-%d') if isinstance(d, pd.Timestamp) else str(d) for d in available_dates] - - # 5. 取得選擇的日期（從 URL 參數或使用最新日期） - selected_date_param = request.args.get('date') - if selected_date_param: - selected_date = pd.to_datetime(selected_date_param) - else: - selected_date = df['snapshot_date'].max() - - # 6. 取得選擇的月份（用於行事曆顯示） - selected_month_param = request.args.get('month') - if selected_month_param: - selected_month = pd.to_datetime(selected_month_param) - else: - selected_month = selected_date - - # V-New 2026-01-15: 判斷是否為月概覽模式（沒有選擇特定日期） - is_month_view = not selected_date_param and not request.args.get('month') - # 如果只有 month 參數沒有 date 參數，也是月概覽模式 - if selected_month_param and not selected_date_param: - is_month_view = True - - # 7. 計算 KPI - current_kpi = calculate_daily_kpis(df, selected_date) - dod_kpi = calculate_dod(df, selected_date) - wow_kpi = calculate_wow(df, selected_date) - - # V-New 2026-01-15: 計算月度總計 KPI - month_start = selected_month.replace(day=1) - month_end = (month_start + pd.DateOffset(months=1)) - pd.Timedelta(days=1) - month_df = df[(df['snapshot_date'] >= month_start) & (df['snapshot_date'] <= month_end)] - - # V-Fix 2026-01-15: 使用 find_col 動態獲取正確欄位名稱 - cols = month_df.columns.tolist() - col_amount = find_col(cols, ['銷售金額', '業績', '金額', '總業績']) - col_cost = find_col(cols, ['成本', 'Cost', '總成本']) - col_profit = find_col(cols, ['毛利', 'Profit']) - col_qty = find_col(cols, ['銷售數量', '銷量', '數量']) - col_name = find_col(cols, ['商品名稱', '品名', 'Name']) - - month_kpi = { - 'total_revenue': float(month_df[col_amount].sum()) if col_amount else 0, - 'total_cost': float(month_df[col_cost].sum()) if col_cost else 0, - 'gross_margin': float(month_df[col_profit].sum()) if col_profit else 0, - 'total_qty': float(month_df[col_qty].sum()) if col_qty else 0, - 'sku_count': int(month_df[col_name].nunique()) if col_name else 0, - 'days_with_data': int(month_df['snapshot_date'].nunique()) - } - # 若無毛利欄位，用業績減成本計算 - if not col_profit and col_amount and col_cost: - month_kpi['gross_margin'] = month_kpi['total_revenue'] - month_kpi['total_cost'] - # 計算月度毛利率 - if month_kpi['total_revenue'] > 0: - month_kpi['margin_rate'] = month_kpi['gross_margin'] / month_kpi['total_revenue'] * 100 - else: - month_kpi['margin_rate'] = 0 - # 計算月度客單價 - if month_kpi['total_qty'] > 0: - month_kpi['avg_price'] = month_kpi['total_revenue'] / month_kpi['total_qty'] - else: - month_kpi['avg_price'] = 0 - - # 8. 準備圖表數據（根據選擇的日期） - chart_data = prepare_daily_charts(df, selected_date, days=30) - - # 9. 準備分類聚合列表 - # V-Fix 2026-01-15: 根據檢視模式（單日/月度）決定聚合範圍 - category_list = prepare_category_summary( - df, - date_str=selected_date, - is_month_view=is_month_view, - month_start=month_start if is_month_view else None, - month_end=month_end if is_month_view else None - ) - - # 10. 準備行事曆數據 - calendar_data = prepare_calendar_data(df, selected_month) - - # 11. V-New: 準備行銷活動業績數據 - marketing_data = prepare_marketing_summary( - df, - selected_date=selected_date if not is_month_view else None, - is_month_view=is_month_view, - month_start=month_start if is_month_view else None, - month_end=month_end if is_month_view else None - ) - - # 12. 回傳模板 - return render_template('daily_sales.html', - selected_date=selected_date.strftime('%Y-%m-%d') if isinstance(selected_date, pd.Timestamp) else selected_date, - available_dates=available_dates_str, - current=current_kpi, - dod=dod_kpi, - wow=wow_kpi, - month_kpi=month_kpi, # V-New: 月度總計 - is_month_view=is_month_view, # V-New: 月概覽模式標誌 - chart_data=chart_data, - categories=category_list, - calendar_data=calendar_data, - marketing_data=marketing_data, # V-New: 行銷活動數據 - selected_month=selected_month.strftime('%Y-%m') if isinstance(selected_month, pd.Timestamp) else selected_month) - - except Exception as e: - sys_log.error(f"[Web] [DailySales] Error: {e}") - import traceback - traceback.print_exc() - return render_template('daily_sales.html', - error=f"系統錯誤: {str(e)}", - selected_date=None, - available_dates=[], - current=None, - dod=None, - wow=None, - month_kpi=None, - is_month_view=False, - chart_data=None, - categories=None, - calendar_data=None, - marketing_data=None, - selected_month=None) - -@app.route('/daily_sales/export') -def export_daily_sales_category(): - """匯出當日業績分類明細為 Excel""" - try: - from datetime import datetime - import io - from flask import send_file - - db = DatabaseManager() - engine = db.engine - table_name = 'daily_sales_snapshot' - - # 檢查資料表是否存在 - inspector = inspect(engine) - if table_name not in inspector.get_table_names(): - return "資料表不存在", 404 - - # 讀取資料 - cache_key = f'{table_name}_daily' - if cache_key in _SALES_PROCESSED_CACHE: - df = _SALES_PROCESSED_CACHE[cache_key]['df'] - else: - df = pd.read_sql(f"SELECT * FROM {table_name}", engine) - df = preprocess_daily_sales_data(df) - _SALES_PROCESSED_CACHE[cache_key] = {'df': df} - - # 取得選擇的日期 - selected_date = request.args.get('date') - if not selected_date: - available_dates = sorted(df['snapshot_date'].unique(), reverse=True) - if available_dates: - selected_date = str(available_dates[0]) - else: - return "無可用日期", 404 - - # 準備分類資料 - categories = prepare_category_summary(df, selected_date) - - if not categories: - return "無資料可匯出", 404 - - # 轉為 DataFrame - export_df = pd.DataFrame(categories) - - # 重新排列欄位順序並重新命名為中文 - column_mapping = { - 'category': '分類', - 'vendor': '廠商', - 'revenue': '總業績', - 'cost': '總成本', - 'profit': '毛利', - 'margin_rate': '毛利率(%)', - 'qty': '總銷量', - 'sku_count': 'SKU數', - 'avg_price': '平均單價' - } - - # 只保留存在的欄位 - export_columns = [col for col in column_mapping.keys() if col in export_df.columns] - export_df = export_df[export_columns] - export_df = export_df.rename(columns=column_mapping) - - # 格式化數值欄位 - for col in export_df.columns: - if col in ['總業績', '總成本', '毛利', '總銷量', 'SKU數', '平均單價']: - export_df[col] = export_df[col].apply(lambda x: f"{x:,.0f}" if pd.notna(x) else "0") - elif col == '毛利率(%)': - export_df[col] = export_df[col].apply(lambda x: f"{x:.1f}" if pd.notna(x) else "0.0") - - # 產生檔案名稱 - filename = f"當日業績_分類明細_{selected_date}.xlsx" - - # 寫入 Excel - output = io.BytesIO() - with pd.ExcelWriter(output, engine='openpyxl') as writer: - export_df.to_excel(writer, index=False, sheet_name='分類業績明細') - - # 調整欄寬 - worksheet = writer.sheets['分類業績明細'] - for idx, col in enumerate(export_df.columns, 1): - max_length = max( - export_df[col].astype(str).apply(len).max(), - len(col) - ) + 2 - worksheet.column_dimensions[chr(64 + idx)].width = min(max_length, 50) - - output.seek(0) - - sys_log.info(f"[Web] [DailySales] Excel 匯出成功: {filename}") - - return send_file( - output, - mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', - as_attachment=True, - download_name=filename - ) - - except Exception as e: - sys_log.error(f"[Web] [DailySales] Excel 匯出失敗: {e}") - import traceback - traceback.print_exc() - return f"匯出失敗: {str(e)}", 500 - -# V-New 2026-01-15: 行銷活動業績匯出 API -@app.route('/daily_sales/export_marketing') -def export_marketing_summary_excel(): - """匯出行銷活動業績明細為 Excel""" - try: - import io - from flask import send_file - - db = DatabaseManager() - engine = db.engine - table_name = 'daily_sales_snapshot' - - # 讀取資料 - cache_key = f'{table_name}_daily' - if cache_key in _SALES_PROCESSED_CACHE: - df = _SALES_PROCESSED_CACHE[cache_key]['df'] - else: - df = pd.read_sql(f"SELECT * FROM {table_name}", engine) - df = preprocess_daily_sales_data(df) - _SALES_PROCESSED_CACHE[cache_key] = {'df': df} - - # 取得參數 - activity_type = request.args.get('type', 'all') # coupon, discount, bonus, click, all - start_date = request.args.get('start_date') - end_date = request.args.get('end_date') - selected_date = request.args.get('date') - - # 額外篩選參數 (與 sales_analysis 同步) - selected_category = request.args.get('category', 'all') - selected_brand = request.args.get('brand', 'all') - selected_vendor = request.args.get('vendor', 'all') - keyword = request.args.get('keyword', '') - - # 決定日期範圍 - if start_date and end_date: - df = df[(df['snapshot_date'] >= pd.to_datetime(start_date)) & - (df['snapshot_date'] <= pd.to_datetime(end_date))] - date_label = f"{start_date}_{end_date}" - elif selected_date: - df = df[df['snapshot_date'] == pd.to_datetime(selected_date)] - date_label = selected_date - else: - date_label = "全部" - - # 應用額外篩選 - cols = df.columns.tolist() - col_category = find_col(cols, ['館別', '商品館', '分類', 'Category']) - col_brand = find_col(cols, ['品牌', 'Brand']) - col_vendor = find_col(cols, ['廠商名稱', 'Vendor Name', '廠商', '供應商', 'Vendor', 'Supplier']) - col_name = find_col(cols, ['商品名稱', '品名']) - col_amount = find_col(cols, ['銷售金額', '業績', '金額', '總業績']) - col_qty = find_col(cols, ['銷售數量', '銷量', '數量']) - - if selected_category != 'all' and col_category: - df = df[df[col_category] == selected_category] - if selected_brand != 'all' and col_brand: - df = df[df[col_brand] == selected_brand] - if selected_vendor != 'all' and col_vendor: - df = df[df[col_vendor] == selected_vendor] - if keyword and col_name: - df = df[df[col_name].str.contains(keyword, case=False, na=False)] - - # 定義行銷活動欄位 - marketing_cols = { - 'coupon': ('折價券活動名稱', '折價券活動'), - 'discount': ('折扣活動名稱', '折扣活動'), - 'bonus': ('滿額再折扣活動名稱', '滿額再折扣'), - 'click': ('點我再折扣', '點我再折扣') - } - - # 準備 Excel 輸出 - output = io.BytesIO() - with pd.ExcelWriter(output, engine='openpyxl') as writer: - # 如果是 all，循環所有類型 - types_to_export = [activity_type] if activity_type != 'all' else ['coupon', 'discount', 'bonus', 'click'] - - summary_rows = [] - - for t in types_to_export: - if t not in marketing_cols: continue - col_internal, sheet_label = marketing_cols[t] - if col_internal not in df.columns: - continue - - # 聚合數據 - # V-Fix: 排除空值和 0 - m_df = df[df[col_internal].notna() & (df[col_internal] != '') & (df[col_internal] != '0') & (df[col_internal] != 0)] - - if m_df.empty: - continue - - grouped = m_df.groupby(col_internal).agg({ - col_amount: 'sum', - col_qty: 'sum', - col_name: 'count' # 訂單筆數/商品筆數 - }).reset_index() - - # 重命名 - grouped.columns = ['活動名稱', '總業績', '總銷量', '項目筆數'] - grouped = grouped.sort_values(by='總業績', ascending=False) - - # 寫入工作表 - grouped.to_excel(writer, sheet_name=sheet_label[:31], index=False) - - # 加入到總表數據 - grouped['活動類型'] = sheet_label - summary_rows.append(grouped) - - # 建立總表工作表 (如果有多個類型) - if len(summary_rows) > 1: - all_m_df = pd.concat(summary_rows).sort_values(by='總業績', ascending=False) - all_m_df = all_m_df[['活動類型', '活動名稱', '總業績', '總銷量', '項目筆數']] - all_m_df.to_excel(writer, sheet_name='合併總表', index=False) - - output.seek(0) - output.seek(0) - - filename = f"行銷活動分析_{date_label}.xlsx" - # 處理中文檔名編碼 - from urllib.parse import quote - encoded_filename = quote(filename) - - return send_file( - output, - mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', - as_attachment=True, - download_name=filename, - conditional=True - ) - - except Exception as e: - sys_log.error(f"[Web] [Marketing] Excel 匯出失敗: {e}") - import traceback - traceback.print_exc() - return f"匯出失敗: {str(e)}", 500 - def preprocess_daily_sales_data(df): """前處理當日業績資料：欄位識別、型別轉換""" cols = df.columns.tolist() diff --git a/services/import_service.py b/services/import_service.py index 8c8d70d..8382d57 100644 --- a/services/import_service.py +++ b/services/import_service.py @@ -637,6 +637,22 @@ class ImportService: session.close() logger.info(f"任務 {job_id} 匯入成功: {total_rows} 筆") + + # 跨 worker 清 daily_sales cache（gunicorn N worker 各持獨立 in-memory cache， + # 需打 N 次 internal endpoint 確保每個 worker 的快取都被清掉） + try: + import requests + base = os.getenv('INTERNAL_BASE_URL', 'http://127.0.0.1:5000') + n = int(os.getenv('GUNICORN_WORKERS', '4')) + for _ in range(n): + try: + requests.post(f"{base}/api/daily_sales/clear_cache", timeout=2) + except Exception: + break + logger.info(f"任務 {job_id} 已清 daily_sales cache（{n} workers）") + except Exception as e: + logger.warning(f"任務 {job_id} 清 cache 失敗（不影響主流程）: {e}") + return True except Exception as e: