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",
)