Build w/ love

src/streamlit_app.py

@@ -1,94 +1,182 @@
+import os
 import time
 import requests
 import pandas as pd
 import streamlit as st
-import urllib3
 
 API_URL = "https://taic.moda.gov.tw/api/v1/dataset.search.export"
 
-# 關閉 SSL 驗證 warning（HF Space 需要）
-urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-
+# ---- Config ----
 st.set_page_config(page_title="TAIC Pulse", layout="wide")
 
-st.title("臺灣主權AI訓練語料庫 Explorer")
-st.caption(
-    "⚡ 即時資料：本頁面在啟動時會從來源 API 抓取一次最新 JSON，"
-)
-
-# ---------------------------
-# Fetch once, cache forever (per Space runtime)
-# ---------------------------
+APP_TITLE = "臺灣主權 AI 訓練語料庫 Explorer"
+st.title(APP_TITLE)
+st.caption("⚡ 即時資料：本頁面在啟動時向來源 API 抓取一次最新 JSON，並提供互動式篩選與檢視（非持續輪詢）。")
 
+# SSL 驗證開關：遇到憑證鏈問題時可設 0
+# HF Spaces 可在 Settings -> Variables 設定
+VERIFY_SSL = os.getenv("TAIC_VERIFY_SSL", "1") == "1"
 
-@st.cache_data
-def fetch_data_once():
-    r = requests.get(API_URL, timeout=30, verify=False)
-    r.raise_for_status()
-    fetched_at = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
-    return r.json(), fetched_at
+# ---- Helpers ----
 
 
-with st.spinner("載入資料中..."):
-    data, fetched_at = fetch_data_once()
+def fetch_json_once() -> dict | list:
+    # 不提供 timeout slider：這邊給一個合理預設即可
+    # 若想改 timeout，請直接改數字或改用 env
+    timeout_sec = int(os.getenv("TAIC_TIMEOUT_SEC", "20"))
 
-st.metric("資料抓取時間", fetched_at)
+    r = requests.get(API_URL, timeout=timeout_sec, verify=VERIFY_SSL)
+    r.raise_for_status()
+    return r.json()
+
+
+@st.cache_data(show_spinner=True)
+def cached_fetch_json() -> dict | list:
+    # cache 版本（session 重啟仍可快取命中）
+    return fetch_json_once()
+
+
+def load_data_once():
+    """
+    只抓一次：
+    - 若 session_state 已有資料：永遠使用，不再打 API
+    - 若沒有：從 st.cache_data 取（可能命中 cache 或實際打一次）
+    """
+    if "taic_data" not in st.session_state:
+        data = cached_fetch_json()
+        st.session_state.taic_data = data
+        st.session_state.fetched_at = time.strftime(
+            "%Y-%m-%d %H:%M:%S", time.localtime())
+    return st.session_state.taic_data, st.session_state.fetched_at
+
+
+def extract_items(data):
+    # 依你先前寫法：list 就直接用；dict 優先取 data，其次整包
+    if isinstance(data, list):
+        return data
+    if isinstance(data, dict):
+        return data.get("data", data)
+    return data
+
+
+def normalize_df(items) -> pd.DataFrame:
+    df = pd.json_normalize(items)
+
+    # 避免欄位全空造成後續選單/表格爆炸
+    if df.empty:
+        return df
+
+    # 如果有欄位型態很怪（list/dict），先轉字串，確保能顯示/篩選
+    for c in df.columns:
+        if df[c].map(lambda x: isinstance(x, (list, dict))).any():
+            df[c] = df[c].apply(lambda x: str(x) if pd.notna(x) else x)
+
+    return df
+
+
+def pick_candidate_fields(df: pd.DataFrame) -> list[str]:
+    # 常見欄位名對應（你可按 TAIC 實際欄位補更多）
+    preferred = {"category", "theme", "publisher",
+                 "organization", "org", "format", "license", "city"}
+    candidates = [c for c in df.columns if c.lower() in preferred]
+
+    # 如果沒命中，就提供讓使用者選（但仍是「選一次」的互動，不會重抓）
+    if not candidates:
+        st.sidebar.info("找不到預設欄位，請自行選擇要做成下拉選單的欄位。")
+        candidates = st.sidebar.multiselect(
+            "選擇要做成下拉選單的欄位（連動順序=顯示順序）",
+            df.columns.tolist(),
+            default=df.columns[:2].tolist() if len(
+                df.columns) >= 2 else df.columns.tolist(),
+        )
+    return candidates
+
+
+def cascading_filters(df: pd.DataFrame, fields: list[str]) -> tuple[pd.DataFrame, dict]:
+    """
+    連動式 filters：
+    - 依 fields 的順序逐一生成 selectbox
+    - 每個 selectbox 的選項都來自「前面已套用 filters 的資料」
+    """
+    filtered_tmp = df.copy()
+    selected = {}
+
+    for field in fields:
+        if field not in filtered_tmp.columns:
+            continue
+
+        # 以目前 filtered_tmp 生成可選值
+        values = (
+            filtered_tmp[field]
+            .dropna()
+            .astype(str)
+            .unique()
+            .tolist()
+        )
+        values = sorted(values)
+
+        if not values:
+            # 這欄在目前條件下已無可用值
+            continue
+
+        choice = st.sidebar.selectbox(
+            f"{field} 篩選",
+            ["(全部)"] + values,
+            index=0,
+            key=f"filter_{field}",
+        )
+
+        if choice != "(全部)":
+            selected[field] = choice
+            filtered_tmp = filtered_tmp[filtered_tmp[field].astype(
+                str) == choice]
+
+    return filtered_tmp, selected
+
+
+# ---- Load ----
+with st.sidebar:
+    st.header("篩選條件（連動選單）")
+
+data, fetched_at = load_data_once()
+items = extract_items(data)
+df = normalize_df(items)
+
+# ---- Header metrics ----
+col1, col2 = st.columns([2, 1])
+with col1:
+    st.subheader("資料預覽與篩選")
+with col2:
+    st.metric("資料抓取時間", fetched_at)
 
 st.divider()
 
-# ---------------------------
-# Normalize JSON to DataFrame
-# ---------------------------
-items = data if isinstance(data, list) else data.get("data", data)
-df = pd.json_normalize(items)
-
-# ---------------------------
-# Sidebar filters
-# ---------------------------
-st.sidebar.header("篩選條件")
-
-candidate_fields = [
-    c for c in df.columns
-    if c.lower() in {"category", "theme", "publisher", "organization", "org", "format", "license", "city"}
-]
-
-if not candidate_fields:
-    candidate_fields = st.sidebar.multiselect(
-        "選擇要做成下拉選單的欄位",
-        options=df.columns.tolist(),
-        default=['授權方式', '是否為開放資料', '資料提供機關']
-    )
-
-filters = {}
-for field in candidate_fields:
-    values = sorted(df[field].dropna().astype(str).unique().tolist())
-    if not values:
-        continue
-    choice = st.sidebar.selectbox(f"{field}", ["(全部)"] + values)
-    if choice != "(全部)":
-        filters[field] = choice
-
-filtered = df.copy()
-for k, v in filters.items():
-    filtered = filtered[filtered[k].astype(str) == v]
-
-q = st.sidebar.text_input("全文關鍵字搜尋")
+if df.empty:
+    st.warning("資料是空的，或 JSON 結構不符合預期（items 解析後沒有表格資料）。")
+    st.stop()
+
+# ---- Filters ----
+candidate_fields = pick_candidate_fields(df)
+filtered, selected_filters = cascading_filters(df, candidate_fields)
+
+# 文字搜尋（可選）
+q = st.sidebar.text_input("全文關鍵字（contains）", "")
 if q.strip():
     mask = filtered.astype(str).apply(
         lambda row: row.str.contains(q, case=False, na=False)
     ).any(axis=1)
     filtered = filtered[mask]
 
-# ---------------------------
-# Main view
-# ---------------------------
-st.subheader("資料預覽")
-st.write(f"共 {len(filtered):,} 筆（原始 {len(df):,} 筆）")
+# ---- Table ----
+st.write(f"共 **{len(filtered):,}** 筆（原始 **{len(df):,}** 筆）")
+if selected_filters:
+    st.caption(
+        "已套用條件：" + "、".join([f"{k}={v}" for k, v in selected_filters.items()]))
+
 st.dataframe(filtered, use_container_width=True)
 
-# ---------------------------
-# Download (simple & clean)
-# ---------------------------
+# ---- Download ----
+# 不要 progress bar：直接準備 bytes
 csv_bytes = filtered.to_csv(index=False).encode("utf-8-sig")
 
 st.download_button(