diff --git a/app.py b/app.py index 3b3d3f8..b7aeeb4 100644 --- a/app.py +++ b/app.py @@ -1048,408 +1048,6 @@ def system_settings_page(): """系統設定與匯入頁面""" return render_template('system_settings.html', system_version=SYSTEM_VERSION) -@app.route('/edm') -def edm_dashboard(): - """🚩 新增：MOMO 限時搶購 (EDM) 專屬儀表板""" - db = DatabaseManager() - session = db.get_session() - - # V-New: 排序參數 - sort_by = request.args.get('sort_by', 'default') - order = request.args.get('order', 'desc') - - try: - # 1. 基礎統計 - # 取得最後更新時間 - last_update = session.query(PromoProduct.crawled_at).filter(PromoProduct.page_type == 'edm').order_by(desc(PromoProduct.crawled_at)).first() - last_update_str = last_update[0].strftime('%Y-%m-%d %H:%M') if last_update else "尚無資料" - - # 🚩 V9.29 新增：取得最新的活動時間文字 - latest_entry = session.query(PromoProduct).filter(PromoProduct.page_type == 'edm').order_by(desc(PromoProduct.crawled_at)).first() - activity_time = getattr(latest_entry, 'activity_time_text', '限時搶購') if latest_entry else "限時搶購" - - # 2. 查詢資料 (V9.44: 只顯示最新批次的資料) - # 找出最新的 batch_id - latest_batch = session.query(PromoProduct.batch_id).filter(PromoProduct.page_type == 'edm').order_by(desc(PromoProduct.crawled_at)).first() - current_batch_id = latest_batch[0] if latest_batch else None - - # 🚩 V9.55 修正：改為查詢「全商品的最新狀態快照」，而非僅查詢最新批次 - # 因為 Scheduler 現在只記錄異動，若只查最新批次，未變動的商品會消失 - subq = session.query( - func.max(PromoProduct.id).label('max_id') - ).filter(PromoProduct.page_type == 'edm').group_by(PromoProduct.i_code, PromoProduct.time_slot).subquery() - - latest_records = session.query(PromoProduct).join(subq, PromoProduct.id == subq.c.max_id).all() - - # 過濾顯示列表：顯示「上架中」、「本批次剛下架」或「今日結束時段」的商品 - items_in_batch = [] - today_start = datetime.now(TAIPEI_TZ).replace(hour=0, minute=0, second=0, microsecond=0) - - for item in latest_records: - # V9.60: 隱藏自然結束的時段商品 - # V-New: 如果商品是今天才結束的，則依然顯示在儀表板上，方便回查 - # V-Fix: 確保時區比較一致 - item_crawled_at = item.crawled_at - if item_crawled_at and item_crawled_at.tzinfo is None: - # V-Fix: 使用 replace 而非 localize (datetime.timezone 不支援 localize 方法) - item_crawled_at = item_crawled_at.replace(tzinfo=TAIPEI_TZ) - - if item.status_change == 'SLOT_END' and item_crawled_at < today_start: - continue - - # V-New: 如果是下架狀態，只有當它是「今天」下架的才顯示 - if item.status_change == 'DELISTED' and item_crawled_at < today_start: - continue - items_in_batch.append(item) - - # V9.45: 按時段分組 - grouped_items = {} - for item in items_in_batch: - if item.time_slot not in grouped_items: - grouped_items[item.time_slot] = [] - grouped_items[item.time_slot].append(item) - - # 按時段鍵值排序 (e.g., 00:00, 07:00, ...) - sorted_grouped_items = dict(sorted(grouped_items.items())) - - # V9.45: 決定預設顯示的頁籤 - def get_current_time_slot(): - hour = datetime.now(TAIPEI_TZ).hour - available_slots = sorted([int(s.split(':')[0]) for s in sorted_grouped_items.keys() if s and ':' in s]) if sorted_grouped_items else [0, 7, 11, 14, 18, 22] - current_slot_hour = 0 - for s in available_slots: - if hour >= s: - current_slot_hour = s - return f"{current_slot_hour:02d}:00" - - active_tab = get_current_time_slot() - if active_tab not in sorted_grouped_items and sorted_grouped_items: - active_tab = next(iter(sorted_grouped_items)) - - # V-New: 計算在架天數與總銷量 - all_icodes_in_batch = [item.i_code for item in items_in_batch] - product_categories = {} - days_on_shelf_map = {} - total_sold_map = {} - - if all_icodes_in_batch: - # 從主商品表 (products) 查詢這些 i_code 對應的分類 - main_products = session.query(Product.i_code, Product.category).filter(Product.i_code.in_(all_icodes_in_batch)).all() - product_categories = {p.i_code: p.category for p in main_products} - - # 計算上架天數 (days_on_shelf) - # V-Fix: 使用 CAST 轉換為 DATE，兼容 PostgreSQL 和 SQLite - from sqlalchemy import cast, Date - days_on_shelf_q = session.query( - PromoProduct.i_code, - func.count(func.distinct(cast(PromoProduct.crawled_at, Date))) - ).filter( # V-New: 增加 page_type 過濾 - PromoProduct.i_code.in_(all_icodes_in_batch), - PromoProduct.page_type == 'edm' - ).group_by(PromoProduct.i_code).all() - days_on_shelf_map = {r[0]: r[1] for r in days_on_shelf_q} - - # 計算總銷量 - # 1. 找出每個商品第一次有庫存紀錄的 ID - first_qty_subq = session.query( - PromoProduct.i_code, - func.min(PromoProduct.id).label('min_id') - ).filter( - PromoProduct.i_code.in_(all_icodes_in_batch), - PromoProduct.remain_qty.isnot(None), - PromoProduct.page_type == 'edm' - ).group_by(PromoProduct.i_code).subquery() - - # 2. 根據 ID 取得當時的庫存 - first_qty_records = session.query( - PromoProduct.i_code, PromoProduct.remain_qty - ).join(first_qty_subq, PromoProduct.id == first_qty_subq.c.min_id).all() - first_qty_map = {r[0]: r[1] for r in first_qty_records} - - # 3. 計算總銷量 (初始庫存 - 當前庫存) - for item in items_in_batch: - # 確保該商品有初始庫存紀錄，且當前庫存也存在 - if item.i_code in first_qty_map and item.remain_qty is not None: - initial_qty = first_qty_map[item.i_code] - current_qty = item.remain_qty - # 只有在初始庫存大於當前庫存時才計算，避免負數 - if initial_qty > current_qty: - total_sold_map[item.i_code] = initial_qty - current_qty - - # V-Fix: 修正 NameError: name 'history_map' is not defined - # 準備銷售歷程資料 - history_map = {} - if all_icodes_in_batch: - all_history_records = session.query( - PromoProduct.i_code, - PromoProduct.time_slot, - PromoProduct.remain_qty, - PromoProduct.crawled_at - ).filter( - PromoProduct.i_code.in_(all_icodes_in_batch), - PromoProduct.crawled_at >= today_start - ).order_by(PromoProduct.crawled_at).all() - - for rec in all_history_records: - key = (rec.i_code, rec.time_slot) - if key not in history_map: - history_map[key] = [] - - if rec.remain_qty is not None: - if not history_map[key] or (history_map[key] and history_map[key][-1]['qty'] != rec.remain_qty): - history_map[key].append({'time': rec.crawled_at.strftime('%H:%M'), 'qty': rec.remain_qty}) - - # 將查到的分類資訊附加到每個 item 物件上 - for item in items_in_batch: - item.main_category = product_categories.get(item.i_code) - if item.main_category: - item.category_color = get_color_for_string(item.main_category) - # V-New: 附加在架天數與總銷量 - item.days_on_shelf = days_on_shelf_map.get(item.i_code, 1) - item.total_sold = total_sold_map.get(item.i_code, 0) - # V-New: Attach quantity history - item.qty_history = history_map.get((item.i_code, item.time_slot), []) - - # V9.46: 排序邏輯優化 (中文註解) - # 排序規則: - # 1. 有貼標 (main_category 存在) 的商品優先 - # 2. 有狀態變更 (NEW, 漲價, 降價) 的商品次之 - # 3. 已下架的商品再次之 - # 4. 最後按價格由高到低排序 - reverse = (order == 'desc') - for time_slot in sorted_grouped_items: - if sort_by == 'name': - sorted_grouped_items[time_slot].sort(key=lambda x: x.name or '', reverse=reverse) - elif sort_by == 'remain_qty': - # 將 None 視為 -1，確保排序時在最下方 - sorted_grouped_items[time_slot].sort(key=lambda x: x.remain_qty if x.remain_qty is not None else -1, reverse=reverse) - elif sort_by == 'price': - sorted_grouped_items[time_slot].sort(key=lambda x: x.price if x.price is not None else -1, reverse=reverse) - else: # 預設排序 - sorted_grouped_items[time_slot].sort(key=lambda x: ( - 1 if x.main_category else 0, - 2 if x.status_change in ['NEW', 'PRICE_UP', 'PRICE_DOWN'] else (1 if x.status_change == 'DELISTED' else 0), - x.price if x.price is not None else -1 - ), reverse=True) - - # 🚩 V-Fix: 修正時段統計，使其能區分「當前狀態」與「上次異動」 - # V-New: 重構時段統計邏輯，確保統計所有今日異動 - slot_stats = {} - today_start = datetime.now(TAIPEI_TZ).replace(hour=0, minute=0, second=0, microsecond=0).replace(tzinfo=None) - - # 1. 取得今日所有異動紀錄 - today_change_records = session.query(PromoProduct).filter(PromoProduct.crawled_at >= today_start, PromoProduct.page_type == 'edm').all() - - # 2. 取得所有相關時段的鍵 (今日異動的 + 當前在架的) - slots_from_changes = {rec.time_slot for rec in today_change_records} - slots_from_display = set(sorted_grouped_items.keys()) - all_relevant_slots = sorted(list(slots_from_changes.union(slots_from_display))) - - # 3. 初始化所有相關時段的統計數據 - for slot in all_relevant_slots: - slot_stats[slot] = {'new': 0, 'up': 0, 'down': 0, 'delisted_last_run': 0, 'on_shelf': 0, 'delisted_total': 0} - - # 4. 累加新品、漲價、降價、下架的數量 (從今日歷史紀錄) - for rec in today_change_records: - if rec.time_slot in slot_stats: - if rec.status_change == 'NEW': - slot_stats[rec.time_slot]['new'] += 1 - elif rec.status_change == 'PRICE_UP': - slot_stats[rec.time_slot]['up'] += 1 - elif rec.status_change == 'PRICE_DOWN': - slot_stats[rec.time_slot]['down'] += 1 - elif rec.status_change in ['DELISTED', 'SLOT_END']: - slot_stats[rec.time_slot]['delisted_last_run'] += 1 - - # 5. 計算在架與下架總數 (從當前顯示的商品快照) - for slot, items in sorted_grouped_items.items(): - if slot in slot_stats: - on_shelf_count = sum(1 for item in items if item.status_change not in ['DELISTED', 'SLOT_END']) - delisted_total_count = len(items) - on_shelf_count - slot_stats[slot]['on_shelf'] = on_shelf_count - slot_stats[slot]['delisted_total'] = delisted_total_count - - # V-New: 建立儀表板頁籤 - promo_pages = [ - {'url': url_for('edm_dashboard'), 'name': '限時搶購', 'id': 'edm'}, - {'url': url_for('festival_dashboard'), 'name': '1.1狂歡購物節', 'id': 'festival'} - ] - - scheduler_stats = load_scheduler_stats() - - return render_template('edm_dashboard.html', - promo_pages=promo_pages, - current_promo_page='edm', - page_title='MOMO 限時搶購', - grouped_items=sorted_grouped_items, - slot_stats=slot_stats, - total_edm_products=len(items_in_batch), - last_update=last_update_str, - activity_time=activity_time, - active_tab=active_tab, - current_batch_id=current_batch_id, - public_url=public_url, - scheduler_stats=scheduler_stats, - current_sort=sort_by, - current_order=order, - slugify=slugify) - except Exception as e: - sys_log.error(f"🚨 EDM Dashboard 渲染錯誤: {e}") - return f"系統錯誤: {e}" - finally: - session.close() - -@app.route('/festival') -def festival_dashboard(): - """🚩 新增：1.1 狂歡購物節專屬儀表板""" - db = DatabaseManager() - session = db.get_session() - - PAGE_TYPE = "festival" - PAGE_NAME = "1.1狂歡購物節" - - sort_by = request.args.get('sort_by', 'default') - order = request.args.get('order', 'desc') - - try: - # 1. 基礎統計 - last_update = session.query(PromoProduct.crawled_at).filter(PromoProduct.page_type == PAGE_TYPE).order_by(desc(PromoProduct.crawled_at)).first() - last_update_str = last_update[0].strftime('%Y-%m-%d %H:%M') if last_update else "尚無資料" - - latest_entry = session.query(PromoProduct).filter(PromoProduct.page_type == PAGE_TYPE).order_by(desc(PromoProduct.crawled_at)).first() - activity_time = getattr(latest_entry, 'activity_time_text', PAGE_NAME) if latest_entry else PAGE_NAME - - # 2. 查詢資料 - subq = session.query( - func.max(PromoProduct.id).label('max_id') - ).filter(PromoProduct.page_type == PAGE_TYPE).group_by(PromoProduct.i_code, PromoProduct.time_slot).subquery() - - latest_records = session.query(PromoProduct).join(subq, PromoProduct.id == subq.c.max_id).all() - - items_in_batch = [] - today_start = datetime.now(TAIPEI_TZ).replace(hour=0, minute=0, second=0, microsecond=0).replace(tzinfo=None) - - for item in latest_records: - if item.status_change == 'SLOT_END' and item.crawled_at < today_start: - continue - if item.status_change == 'DELISTED' and item.crawled_at < today_start: - continue - items_in_batch.append(item) - - # 此頁面使用區塊標題作為分組依據 - grouped_items = {} - for item in items_in_batch: - if item.time_slot not in grouped_items: - grouped_items[item.time_slot] = [] - grouped_items[item.time_slot].append(item) - - sorted_grouped_items = dict(sorted(grouped_items.items())) - - # 預設顯示第一個頁籤 - active_tab = next(iter(sorted_grouped_items)) if sorted_grouped_items else "" - - all_icodes_in_batch = [item.i_code for item in items_in_batch] - product_categories = {} - days_on_shelf_map = {} - total_sold_map = {} - - if all_icodes_in_batch: - main_products = session.query(Product.i_code, Product.category).filter(Product.i_code.in_(all_icodes_in_batch)).all() - product_categories = {p.i_code: p.category for p in main_products} - - # V-Fix: 使用 CAST 轉換為 DATE，兼容 PostgreSQL 和 SQLite - from sqlalchemy import cast, Date - days_on_shelf_q = session.query( - PromoProduct.i_code, - func.count(func.distinct(cast(PromoProduct.crawled_at, Date))) - ).filter( - PromoProduct.i_code.in_(all_icodes_in_batch), - PromoProduct.page_type == PAGE_TYPE - ).group_by(PromoProduct.i_code).all() - days_on_shelf_map = {r[0]: r[1] for r in days_on_shelf_q} - - # 將查到的分類資訊附加到每個 item 物件上 - for item in items_in_batch: - item.main_category = product_categories.get(item.i_code) - if item.main_category: - item.category_color = get_color_for_string(item.main_category) - item.days_on_shelf = days_on_shelf_map.get(item.i_code, 1) - # V-Fix: 為 festival 頁面提供預設值，避免共用模板在渲染 total_sold 和 qty_history 時出錯 - item.total_sold = 0 - item.qty_history = [] - - # 排序邏輯 - reverse = (order == 'desc') - for time_slot in sorted_grouped_items: - if sort_by == 'name': - sorted_grouped_items[time_slot].sort(key=lambda x: x.name or '', reverse=reverse) - elif sort_by == 'price': - sorted_grouped_items[time_slot].sort(key=lambda x: x.price if x.price is not None else -1, reverse=reverse) - else: # 預設排序 - sorted_grouped_items[time_slot].sort(key=lambda x: ( - 1 if x.main_category else 0, - 2 if x.status_change in ['NEW', 'PRICE_UP', 'PRICE_DOWN'] else (1 if x.status_change == 'DELISTED' else 0), - x.price if x.price is not None else -1 - ), reverse=True) - - # 時段統計 - slot_stats = {} - today_start = datetime.now(TAIPEI_TZ).replace(hour=0, minute=0, second=0, microsecond=0).replace(tzinfo=None) - - today_change_records = session.query(PromoProduct).filter(PromoProduct.crawled_at >= today_start, PromoProduct.page_type == PAGE_TYPE).all() - - slots_from_changes = {rec.time_slot for rec in today_change_records} - slots_from_display = set(sorted_grouped_items.keys()) - all_relevant_slots = sorted(list(slots_from_changes.union(slots_from_display))) - - for slot in all_relevant_slots: - slot_stats[slot] = {'new': 0, 'up': 0, 'down': 0, 'delisted_last_run': 0, 'on_shelf': 0, 'delisted_total': 0} - - for rec in today_change_records: - if rec.time_slot in slot_stats: - if rec.status_change == 'NEW': slot_stats[rec.time_slot]['new'] += 1 - elif rec.status_change == 'PRICE_UP': slot_stats[rec.time_slot]['up'] += 1 - elif rec.status_change == 'PRICE_DOWN': slot_stats[rec.time_slot]['down'] += 1 - elif rec.status_change in ['DELISTED', 'SLOT_END']: slot_stats[rec.time_slot]['delisted_last_run'] += 1 - - for slot, items in sorted_grouped_items.items(): - if slot in slot_stats: - on_shelf_count = sum(1 for item in items if item.status_change not in ['DELISTED', 'SLOT_END']) - delisted_total_count = len(items) - on_shelf_count - slot_stats[slot]['on_shelf'] = on_shelf_count - slot_stats[slot]['delisted_total'] = delisted_total_count - - scheduler_stats = load_scheduler_stats() - - # 建立儀表板頁籤 - promo_pages = [ - {'url': url_for('edm_dashboard'), 'name': '限時搶購', 'id': 'edm'}, - {'url': url_for('festival_dashboard'), 'name': '1.1狂歡購物節', 'id': 'festival'} - ] - - # 注意：這裡我們重複使用 edm_dashboard.html 範本 - # 您需要建立一個它的複本，命名為 festival.html - return render_template('edm_dashboard.html', - promo_pages=promo_pages, - current_promo_page='festival', - page_title=PAGE_NAME, - grouped_items=sorted_grouped_items, - slot_stats=slot_stats, - total_edm_products=len(items_in_batch), - last_update=last_update_str, - activity_time=activity_time, - active_tab=active_tab, - public_url=public_url, - scheduler_stats=scheduler_stats, - current_sort=sort_by, - current_order=order, - slugify=slugify) - except Exception as e: - sys_log.error(f"🚨 {PAGE_NAME} Dashboard 渲染錯誤: {e}") - return f"系統錯誤: {e}" - finally: - session.close() - @app.route('/abc_analysis/detail') def abc_analysis_detail(): """ABC 分析詳細報表頁面"""