import requests import pandas as pd import streamlit as st import plotly.express as px import plotly.graph_objects as go import matplotlib.font_manager as fm import os import platform # ── 頁面設定 ────────────────────────────────────────────── st.set_page_config( page_title="MOMO 商品價格分析", page_icon="🛒", layout="wide", ) st.title("🛒 MOMO 電商商品價格分析儀表板") st.markdown("輸入關鍵字,即時抓取 MOMO 商品資料並視覺化分析。") # ── 中文字型安裝(不使用 wget) ──────────────────────────── @st.cache_resource def install_chinese_font(): """ 嘗試安裝系統中文字型;若找不到則下載 Noto Sans TC。 回傳字型路徑(供 matplotlib 使用,Plotly 另行設定)。 """ # 常見系統中文字型路徑 system_fonts = { "Linux": [ "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc", "/usr/share/fonts/opentype/noto/NotoSansCJKtc-Regular.otf", ], "Darwin": [ # macOS "/System/Library/Fonts/PingFang.ttc", "/Library/Fonts/Arial Unicode.ttf", ], "Windows": [ "C:/Windows/Fonts/msjh.ttc", "C:/Windows/Fonts/mingliu.ttc", ], } sys_name = platform.system() for path in system_fonts.get(sys_name, []): if os.path.exists(path): return path # 找不到系統字型 → 用 requests 下載 Noto Sans TC font_path = "NotoSansTC-Regular.ttf" if not os.path.exists(font_path): url = ( "https://github.com/googlefonts/noto-cjk/raw/main/Sans/OTF/TraditionalChinese/" "NotoSansCJKtc-Regular.otf" ) try: r = requests.get(url, timeout=30) r.raise_for_status() with open(font_path, "wb") as f: f.write(r.content) fm.fontManager.addfont(font_path) except Exception as e: st.warning(f"字型下載失敗:{e},圖表中文可能顯示異常。") return None return font_path install_chinese_font() # ── 抓取 MOMO 資料 ───────────────────────────────────────── def fetch_momo(keyword: str, max_items: int) -> pd.DataFrame: url = "https://apisearch.momoshop.com.tw/momoSearchCloud/moec/textSearch" headers = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/127.0.0.0 Safari/537.36" ) } all_products = [] page = 1 per_page = 24 # MOMO 預設每頁約 24 筆 while len(all_products) < max_items: payload = { "host": "momoshop", "flag": "searchEngine", "data": { "specialGoodsType": "", "isBrandSeriesPage": "false", "authorNo": "", "originalCateCode": "", "cateType": "", "searchValue": keyword, "cateCode": "", "cateLevel": "-1", "cp": "N", "NAM": "N", "first": "N", "freeze": "N", "superstore": "N", "tvshop": "N", "china": "N", "tomorrow": "N", "stockYN": "N", "prefere": "N", "threeHours": "N", "video": "N", "cycle": "N", "cod": "N", "superstorePay": "N", "showType": "chessboardType", "curPage": str(page), "priceS": "0", "priceE": "9999999", "searchType": "1", "reduceKeyword": "", "isFuzzy": "0", "rtnCateDatainfo": { "cateCode": "", "cateLv": "-1", "keyword": keyword, "curPage": str(page), "historyDoPush": "false", "timestamp": 1723036027826, }, "flag": 2018, "serviceCode": "MT01", "addressSearchData": {}, "adSource": "tenmax", }, } try: resp = requests.post(url, headers=headers, json=payload, timeout=15) resp.raise_for_status() except Exception as e: st.error(f"第 {page} 頁請求失敗:{e}") break data = resp.json() products = data.get("rtnSearchData", {}).get("goodsInfoList", []) if not products: break for p in products: name = p.get("goodsName", "") price_raw = p.get("goodsPrice", "0") try: price = int( str(price_raw).replace("$", "").replace(",", "").strip() ) except ValueError: price = 0 all_products.append({"品名": name, "價格": price}) page += 1 if len(products) < per_page: break # 已無更多頁 df = pd.DataFrame(all_products[:max_items]) return df # ── 側邊欄:使用者輸入 ──────────────────────────────────── with st.sidebar: st.header("⚙️ 查詢設定") keyword = st.text_input("搜尋關鍵字", value="耳機") max_items = st.slider("最多筆數", min_value=10, max_value=200, value=70, step=10) search_btn = st.button("🔍 開始抓取", use_container_width=True) st.divider() st.markdown("資料來源:[MOMO 購物網](https://www.momoshop.com.tw/)") # ── 主流程 ──────────────────────────────────────────────── if search_btn: if not keyword.strip(): st.warning("請輸入搜尋關鍵字!") st.stop() with st.spinner(f"正在抓取「{keyword}」的商品資料,請稍候..."): df = fetch_momo(keyword, max_items) if df.empty: st.error("未抓到任何資料,請檢查網路或更換關鍵字。") st.stop() # 統計資訊 avg_price = df["價格"].mean() max_price = df["價格"].max() min_price = df["價格"].min() st.success(f"共抓取 **{len(df)}** 筆資料") col1, col2, col3 = st.columns(3) col1.metric("💰 平均價格", f"${avg_price:,.0f}") col2.metric("📈 最高價格", f"${max_price:,.0f}") col3.metric("📉 最低價格", f"${min_price:,.0f}") st.divider() # ── 圖表區 ──────────────────────────────────────────── tab1, tab2, tab3 = st.tabs(["📈 折線圖", "🥧 圓餅圖", "☀️ 旭日圖"]) # ---- Tab1:折線圖 ---- with tab1: st.subheader(f"「{keyword}」商品售價折線圖") df_line = df.copy().reset_index() df_line.columns = ["商品序號", "品名", "價格"] fig_line = go.Figure() fig_line.add_trace( go.Scatter( x=df_line["商品序號"], y=df_line["價格"], mode="lines+markers", name="售價", line=dict(color="#3B82F6", width=2), marker=dict(size=5), hovertext=df_line["品名"], hovertemplate="%{hovertext}
價格:$%{y:,}", ) ) fig_line.add_hline( y=avg_price, line_dash="dash", line_color="red", annotation_text=f"平均 ${avg_price:,.0f}", annotation_position="bottom right", ) fig_line.update_layout( xaxis_title="商品序號", yaxis_title="價格 (NTD)", height=500, hovermode="x unified", plot_bgcolor="rgba(0,0,0,0)", paper_bgcolor="rgba(0,0,0,0)", ) st.plotly_chart(fig_line, use_container_width=True) # ---- Tab2:圓餅圖(價格區間分布) ---- with tab2: st.subheader(f"「{keyword}」價格區間分布圓餅圖") bins = [0, 500, 1000, 3000, 5000, 10000, float("inf")] labels = ["≤500", "501~1000", "1001~3000", "3001~5000", "5001~10000", ">10000"] df["價格區間"] = pd.cut(df["價格"], bins=bins, labels=labels, right=True) pie_data = df["價格區間"].value_counts().reset_index() pie_data.columns = ["價格區間", "數量"] fig_pie = px.pie( pie_data, names="價格區間", values="數量", title=f"{keyword} 各價格區間商品數量", hole=0.35, color_discrete_sequence=px.colors.qualitative.Pastel, ) fig_pie.update_traces(textposition="inside", textinfo="percent+label") fig_pie.update_layout( height=500, plot_bgcolor="rgba(0,0,0,0)", paper_bgcolor="rgba(0,0,0,0)", ) st.plotly_chart(fig_pie, use_container_width=True) # ---- Tab3:旭日圖(品牌 × 價格區間) ---- with tab3: st.subheader(f"「{keyword}」品牌 × 價格區間旭日圖") st.caption("品牌由品名前幾個字推斷,僅供參考。") # 從品名前 N 字切出「品牌」(簡易規則) def extract_brand(name: str, n: int = 4) -> str: return name[:n] if len(name) >= n else name df["品牌"] = df["品名"].apply(extract_brand) sun_data = ( df.groupby(["價格區間", "品牌"]) .size() .reset_index(name="數量") ) fig_sun = px.sunburst( sun_data, path=["價格區間", "品牌"], values="數量", title=f"{keyword} 價格區間 × 品牌旭日圖", color="數量", color_continuous_scale="RdBu", ) fig_sun.update_layout( height=600, plot_bgcolor="rgba(0,0,0,0)", paper_bgcolor="rgba(0,0,0,0)", ) st.plotly_chart(fig_sun, use_container_width=True) st.divider() # ── 原始資料表 ───────────────────────────────────────── with st.expander("📋 查看原始資料", expanded=False): st.dataframe(df[["品名", "價格"]].reset_index(drop=True), use_container_width=True) # ── 下載 CSV ────────────────────────────────────────── csv = df[["品名", "價格"]].to_csv(index=False, encoding="utf-8-sig") st.download_button( label="⬇️ 下載 CSV", data=csv, file_name=f"MOMO_{keyword}.csv", mime="text/csv", )