wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(routes): 刪除 app.py edm festival 重複路由

Browse Source 
ADR-017 Phase 3f-1 edm sprint
This commit is contained in:
OoO
2026-04-29 21:07:16 +08:00
parent 1f88c2817b
commit 8fad8ebff0
1 changed files with 0 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files

402
app.py
View File

@@ -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 分析詳細報表頁面"""