改用 WebSocket 驅動實時更新

- 新增 websocket_api.py：連接 Universalis WebSocket API
- 首次查詢用 REST API，之後用 WebSocket 緩存
- 自動刷新改為 5 秒
- 物品資訊與市場數據並行請求（aiohttp）
- 新增更新紀錄頁籤

app.py

@@ -20,6 +20,7 @@ from src.watchlist import (
     remove_item_from_list,
 )
 from src.ai_analysis import analyze_item_with_ai, get_market_summary
+from src.websocket_api import start_websocket
 
 
 def refresh_watchlist_with_notify(watchlist: list):
@@ -59,6 +60,8 @@ def create_app() -> gr.Blocks:
         **支援伺服器:** 伊弗利特、迦樓羅、利維坦、鳳凰、奧汀、巴哈姆特、拉姆、泰坦
 
         > **搜尋提示:** 可輸入繁體中文、英文名稱、物品 ID，或貼上 Universalis 網址
+
+        🔗 **WebSocket 已連線** - 使用異步 API 加速查詢
         """)
 
         with gr.Tabs():
@@ -68,6 +71,7 @@ def create_app() -> gr.Blocks:
             _build_watchlist_tab(watchlist_state)
             _build_tax_tab()
             _build_stats_tab()
+            _build_changelog_tab()
 
     return app
 
@@ -117,7 +121,7 @@ def _build_market_tab() -> None:
                 with gr.Row():
                     search_btn = gr.Button("查詢市場", variant="primary")
                     auto_refresh = gr.Checkbox(
-                        label="自動刷新 (60秒)",
+                        label="自動刷新 (5秒)",
                         value=False,
                     )
 
@@ -152,8 +156,8 @@ def _build_market_tab() -> None:
             comparison_table = gr.Dataframe(interactive=False)
             comparison_chart = gr.Plot()
 
-        # 自動刷新用的隱藏計時器
-        timer = gr.Timer(value=60, active=False)
+        # 自動刷新用的計時器 (5秒，使用 WebSocket 緩存)
+        timer = gr.Timer(value=5, active=False)
 
         # 事件綁定
         search_input.change(
@@ -466,7 +470,50 @@ def _build_ai_tab() -> None:
         )
 
 
+def _build_changelog_tab() -> None:
+    """建立更新紀錄頁籤."""
+    with gr.TabItem("更新紀錄"):
+        gr.Markdown("""
+### v1.3.0 (2024-12)
+- 改用 WebSocket 驅動實時更新
+- 首次查詢用 REST API，之後用 WebSocket 緩存
+- 自動刷新改為 5 秒（使用緩存時幾乎無延遲）
+- 物品資訊與市場數據並行請求
+
+### v1.2.0 (2024-12)
+- 新增 AI 分析功能
+- 支援跨服套利判斷
+- 手機版面優化
+
+### v1.1.0 (2024-12)
+- 新增監看清單功能
+- 支援設定目標價格提醒
+- 資料儲存於瀏覽器 LocalStorage
+
+### v1.0.0 (2024-12)
+- 首次發布
+- 支援繁體中文搜尋物品
+- 市場價格查詢、交易紀錄
+- 跨伺服器比價
+- 稅率資訊、上傳統計
+
+---
+資料來源: [Universalis API](https://universalis.app/)
+        """)
+
+
 if __name__ == "__main__":
+    # 啟動 WebSocket 連線（背景執行）
+    print("啟動 WebSocket 連線...")
+    ws_client = start_websocket()
+
+    # 訂閱陸行鳥資料中心的所有伺服器更新
+    from src.config import WORLDS
+    for world_id in WORLDS.keys():
+        ws_client.subscribe("listings/add", world_id)
+        ws_client.subscribe("sales/add", world_id)
+    print("已訂閱陸行鳥資料中心的市場更新")
+
     application = create_app()
     application.launch(
         server_name="0.0.0.0",

requirements.txt

@@ -5,3 +5,6 @@ plotly>=5.15.0
 opencc-python-reimplemented>=0.1.7
 huggingface_hub>=0.20.0
 python-dotenv>=1.0.0
+aiohttp>=3.9.0
+websockets>=12.0
+pymongo>=4.6.0

src/api.py

@@ -1,7 +1,10 @@
 """Universalis 和 XIVAPI 的 API 呼叫函數."""
 
+import asyncio
 import re
+from typing import Optional
 
+import aiohttp
 import requests
 from opencc import OpenCC
 
@@ -285,3 +288,215 @@ def get_recent_activity(world_or_dc: str = None, limit: int = 15) -> list:
     activity_list.sort(key=lambda x: x["last_update"], reverse=True)
 
     return activity_list
+
+
+# ============================================================
+# 異步 API 函數 (使用 aiohttp，速度更快)
+# ============================================================
+
+async def get_market_data_async(
+    item_id: int,
+    world_or_dc: str = None,
+    session: aiohttp.ClientSession = None,
+) -> dict:
+    """異步取得市場板數據.
+
+    Args:
+        item_id: 物品 ID
+        world_or_dc: 伺服器或資料中心名稱
+        session: 可選的 aiohttp session
+
+    Returns:
+        市場數據字典
+    """
+    if world_or_dc is None:
+        world_or_dc = DATA_CENTER
+
+    url = f"{UNIVERSALIS_BASE}/{world_or_dc}/{item_id}"
+    params = {"listings": 50, "entries": 50}
+
+    close_session = False
+    if session is None:
+        session = aiohttp.ClientSession()
+        close_session = True
+
+    try:
+        async with session.get(
+            url, params=params, timeout=aiohttp.ClientTimeout(total=MARKET_API_TIMEOUT)
+        ) as response:
+            if response.status == 200:
+                return await response.json()
+            return {}
+    except Exception as e:
+        print(f"異步取得市場數據錯誤: {e}")
+        return {}
+    finally:
+        if close_session:
+            await session.close()
+
+
+async def get_item_info_async(
+    item_id: int,
+    session: aiohttp.ClientSession = None,
+) -> dict:
+    """異步取得物品詳細資訊.
+
+    Args:
+        item_id: 物品 ID
+        session: 可選的 aiohttp session
+
+    Returns:
+        物品資訊字典
+    """
+    close_session = False
+    if session is None:
+        session = aiohttp.ClientSession()
+        close_session = True
+
+    try:
+        url = f"{CAFEMAKER_BASE}/item/{item_id}"
+        async with session.get(
+            url, timeout=aiohttp.ClientTimeout(total=API_TIMEOUT)
+        ) as response:
+            if response.status == 200:
+                data = await response.json()
+                if data.get("Name"):
+                    data["Name"] = _s2t_converter.convert(data["Name"])
+                return data
+    except Exception as e:
+        print(f"異步取得物品資訊錯誤: {e}")
+
+    # 備用：嘗試 XIVAPI
+    try:
+        url = f"{XIVAPI_BASE}/item/{item_id}"
+        params = {"language": "en"}
+        async with session.get(
+            url, params=params, timeout=aiohttp.ClientTimeout(total=API_TIMEOUT)
+        ) as response:
+            if response.status == 200:
+                return await response.json()
+    except Exception as e:
+        print(f"XIVAPI 異步取得物品資訊錯誤: {e}")
+    finally:
+        if close_session:
+            await session.close()
+
+    return {"ID": item_id, "Name": f"物品 {item_id}", "LevelItem": 0}
+
+
+async def get_multi_item_market_data_async(
+    item_ids: list[int],
+    world_or_dc: str = None,
+) -> dict:
+    """異步批量取得多個物品的市場數據.
+
+    Args:
+        item_ids: 物品 ID 列表
+        world_or_dc: 伺服器或資料中心名稱
+
+    Returns:
+        以物品 ID 為 key 的市場數據字典
+    """
+    if world_or_dc is None:
+        world_or_dc = DATA_CENTER
+
+    if not item_ids:
+        return {}
+
+    # Universalis 支援批量查詢，最多 100 個物品
+    ids_str = ",".join(str(i) for i in item_ids[:100])
+    url = f"{UNIVERSALIS_BASE}/{world_or_dc}/{ids_str}"
+    params = {"listings": 20, "entries": 20}
+
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(
+                url, params=params, timeout=aiohttp.ClientTimeout(total=MARKET_API_TIMEOUT)
+            ) as response:
+                if response.status == 200:
+                    data = await response.json()
+                    # 如果是單個物品，包裝成 items 格式
+                    if "items" not in data and "itemID" in data:
+                        return {data["itemID"]: data}
+                    return data.get("items", {})
+    except Exception as e:
+        print(f"批量取得市場數據錯誤: {e}")
+
+    return {}
+
+
+async def get_full_item_data_async(
+    item_id: int,
+    world_or_dc: str = None,
+) -> dict:
+    """異步取得物品的完整資料（物品資訊 + 市場數據）.
+
+    Args:
+        item_id: 物品 ID
+        world_or_dc: 伺服器或資料中心名稱
+
+    Returns:
+        包含 item_info 和 market_data 的字典
+    """
+    async with aiohttp.ClientSession() as session:
+        # 並行請求物品資訊和市場數據
+        item_info_task = get_item_info_async(item_id, session)
+        market_data_task = get_market_data_async(item_id, world_or_dc, session)
+
+        item_info, market_data = await asyncio.gather(
+            item_info_task, market_data_task
+        )
+
+        return {
+            "item_info": item_info,
+            "market_data": market_data,
+        }
+
+
+def get_market_data_fast(item_id: int, world_or_dc: str = None) -> dict:
+    """快速取得市場板數據（使用異步）.
+
+    這是 get_market_data 的快速版本，適合在 Gradio 中使用。
+
+    Args:
+        item_id: 物品 ID
+        world_or_dc: 伺服器或資料中心名稱
+
+    Returns:
+        市場數據字典
+    """
+    try:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        result = loop.run_until_complete(get_market_data_async(item_id, world_or_dc))
+        loop.close()
+        return result
+    except Exception as e:
+        print(f"快速取得市場數據錯誤: {e}")
+        # 回退到同步版本
+        return get_market_data(item_id, world_or_dc)
+
+
+def get_full_item_data_fast(item_id: int, world_or_dc: str = None) -> dict:
+    """快速取得物品完整資料（使用異步並行請求）.
+
+    Args:
+        item_id: 物品 ID
+        world_or_dc: 伺服器或資料中心名稱
+
+    Returns:
+        包含 item_info 和 market_data 的字典
+    """
+    try:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        result = loop.run_until_complete(get_full_item_data_async(item_id, world_or_dc))
+        loop.close()
+        return result
+    except Exception as e:
+        print(f"快速取得完整資料錯誤: {e}")
+        # 回退到同步版本
+        return {
+            "item_info": get_item_info(item_id),
+            "market_data": get_market_data(item_id, world_or_dc),
+        }

src/display.py

@@ -7,7 +7,10 @@ import gradio as gr
 import pandas as pd
 import plotly.graph_objects as go
 
+import time
+
 from .api import (
+    get_full_item_data_fast,
     get_item_info,
     get_market_data,
     get_recent_activity,
@@ -15,6 +18,7 @@ from .api import (
     get_upload_stats,
     search_items,
 )
+from .websocket_api import get_ws_client
 from .charts import (
     create_cross_world_comparison,
     create_price_chart,
@@ -91,17 +95,43 @@ def display_item_market(
         return "", empty_df, empty_df, empty_fig, empty_df, empty_fig
 
     item_id = item_selection
-
-    # 取得物品資訊
-    item_info = get_item_info(item_id)
-    item_name = item_info.get("Name", f"物品 {item_id}")
-    item_level = item_info.get("LevelItem", 0)
-
-    # 取得市場數據
     world_query = (
         selected_world if selected_world != "全部伺服器" else DATA_CENTER
     )
-    market_data = get_market_data(item_id, world_query)
+
+    # 開始關注此物品的 WebSocket 更新
+    ws_client = get_ws_client()
+    if ws_client:
+        ws_client.watch_item(item_id)
+
+    # 檢查 WebSocket 是否有此物品的緩存數據
+    ws_data = None
+    if ws_client:
+        ws_data = ws_client.get_cached_data(item_id)
+
+    if ws_data and ws_data.get("data", {}).get("listings"):
+        # 使用 WebSocket 緩存的數據（更快）
+        cached = ws_data["data"]
+        item_info = get_item_info(item_id)  # 物品資訊還是用 API
+        market_data = {
+            "listings": cached.get("listings", []),
+            "recentHistory": cached.get("recentHistory", []),
+            "currentAveragePrice": cached.get("currentAveragePrice", 0),
+            "averagePrice": cached.get("averagePrice", 0),
+            "minPrice": cached.get("minPrice", 0),
+            "maxPrice": cached.get("maxPrice", 0),
+            "listingsCount": len(cached.get("listings", [])),
+            "regularSaleVelocity": cached.get("regularSaleVelocity", 0),
+            "lastUploadTime": int(ws_data["timestamp"] * 1000),
+        }
+    else:
+        # 首次查詢，使用 REST API
+        full_data = get_full_item_data_fast(item_id, world_query)
+        item_info = full_data.get("item_info", {})
+        market_data = full_data.get("market_data", {})
+
+    item_name = item_info.get("Name", f"物品 {item_id}")
+    item_level = item_info.get("LevelItem", 0)
 
     if not market_data:
         return (

src/websocket_api.py

@@ -0,0 +1,314 @@
+"""Universalis WebSocket API 實現."""
+
+import asyncio
+import threading
+import time
+from typing import Callable, Optional
+from queue import Queue
+
+import bson
+import websockets
+
+from .config import DATA_CENTER, WORLD_IDS, WORLDS
+
+# WebSocket 設定
+UNIVERSALIS_WS_URL = "wss://universalis.app/api/ws"
+
+# 陸行鳥資料中心的所有伺服器 ID
+CHOCOBO_WORLD_IDS = list(WORLDS.keys())
+
+
+class UniversalisWebSocket:
+    """Universalis WebSocket 客戶端."""
+
+    def __init__(self):
+        self._ws: Optional[websockets.WebSocketClientProtocol] = None
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._thread: Optional[threading.Thread] = None
+        self._running = False
+        self._subscriptions: set = set()
+        self._callbacks: dict[str, list[Callable]] = {}
+        self._message_queue: Queue = Queue()
+        self._connected = False
+        # 物品數據緩存: {item_id: {"data": {...}, "timestamp": time}}
+        self._item_cache: dict[int, dict] = {}
+        # 當前訂閱的物品 ID
+        self._watched_items: set[int] = set()
+
+    def start(self):
+        """啟動 WebSocket 連線（在背景執行緒中運行）."""
+        if self._running:
+            return
+
+        self._running = True
+        self._thread = threading.Thread(target=self._run_event_loop, daemon=True)
+        self._thread.start()
+
+    def stop(self):
+        """停止 WebSocket 連線."""
+        self._running = False
+        self._connected = False
+        if self._ws and self._loop:
+            try:
+                asyncio.run_coroutine_threadsafe(
+                    self._ws.close(), self._loop
+                ).result(timeout=2)
+            except Exception:
+                pass
+        if self._loop and self._loop.is_running():
+            self._loop.call_soon_threadsafe(self._loop.stop)
+        if self._thread:
+            self._thread.join(timeout=3)
+
+    def _run_event_loop(self):
+        """在背景執行緒中運行事件循環."""
+        self._loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self._loop)
+        try:
+            self._loop.run_until_complete(self._connect_and_listen())
+        except Exception as e:
+            print(f"WebSocket 事件循環錯誤: {e}")
+        finally:
+            self._loop.close()
+
+    async def _connect_and_listen(self):
+        """連接並監聽 WebSocket."""
+        while self._running:
+            try:
+                async with websockets.connect(
+                    UNIVERSALIS_WS_URL,
+                    ping_interval=30,
+                    ping_timeout=10,
+                ) as ws:
+                    self._ws = ws
+                    self._connected = True
+                    print("WebSocket 已連接到 Universalis")
+
+                    # 重新訂閱之前的頻道
+                    for channel in self._subscriptions:
+                        await self._send_subscribe(channel)
+
+                    # 監聽消息
+                    async for message in ws:
+                        if not self._running:
+                            break
+                        await self._handle_message(message)
+
+            except websockets.exceptions.ConnectionClosed:
+                print("WebSocket 連線已關閉，嘗試重新連接...")
+            except Exception as e:
+                print(f"WebSocket 錯誤: {e}")
+
+            self._connected = False
+            if self._running:
+                await asyncio.sleep(5)  # 等待後重新連接
+
+    async def _send_subscribe(self, channel: str):
+        """發送訂閱消息."""
+        if self._ws and self._connected:
+            msg = bson.encode({"event": "subscribe", "channel": channel})
+            await self._ws.send(msg)
+            print(f"已訂閱頻道: {channel}")
+
+    async def _send_unsubscribe(self, channel: str):
+        """發送取消訂閱消息."""
+        if self._ws and self._connected:
+            msg = bson.encode({"event": "unsubscribe", "channel": channel})
+            await self._ws.send(msg)
+            print(f"已取消訂閱頻道: {channel}")
+
+    async def _handle_message(self, message: bytes):
+        """處理收到的消息."""
+        try:
+            data = bson.decode(message)
+            event = data.get("event", "")
+            item_id = data.get("item")
+            world_id = data.get("world")
+
+            # 檢查是否為陸行鳥資料中心的伺服器
+            if world_id and world_id not in CHOCOBO_WORLD_IDS:
+                return  # 忽略其他資料中心的消息
+
+            # 如果是我們關注的物品，更新緩存
+            if item_id and item_id in self._watched_items:
+                self._item_cache[item_id] = {
+                    "data": data,
+                    "timestamp": time.time(),
+                    "event": event,
+                    "world": world_id,
+                }
+
+            # 放入消息佇列
+            self._message_queue.put(data)
+
+            # 呼叫回調
+            if event in self._callbacks:
+                for callback in self._callbacks[event]:
+                    try:
+                        callback(data)
+                    except Exception as e:
+                        print(f"回調錯誤: {e}")
+
+        except Exception as e:
+            print(f"處理消息錯誤: {e}")
+
+    def subscribe(self, channel: str, world_id: int = None):
+        """訂閱頻道.
+
+        Args:
+            channel: 頻道名稱 (listings/add, listings/remove, sales/add)
+            world_id: 可選的伺服器 ID，用於過濾
+        """
+        if world_id:
+            full_channel = f"{channel}{{world={world_id}}}"
+        else:
+            full_channel = channel
+
+        self._subscriptions.add(full_channel)
+
+        if self._loop and self._connected:
+            asyncio.run_coroutine_threadsafe(
+                self._send_subscribe(full_channel), self._loop
+            )
+
+    def subscribe_item(self, item_id: int, world_or_dc: str = None):
+        """訂閱特定物品的更新.
+
+        Args:
+            item_id: 物品 ID
+            world_or_dc: 伺服器或資料中心名稱
+        """
+        # 訂閱該物品的上架和銷售更新
+        if world_or_dc and world_or_dc != "全部伺服器":
+            world_id = WORLD_IDS.get(world_or_dc)
+            if world_id:
+                self.subscribe("listings/add", world_id)
+                self.subscribe("sales/add", world_id)
+        else:
+            # 訂閱陸行鳥資料中心所有伺服器
+            for world_id in CHOCOBO_WORLD_IDS:
+                self.subscribe("listings/add", world_id)
+                self.subscribe("sales/add", world_id)
+
+    def unsubscribe(self, channel: str, world_id: int = None):
+        """取消訂閱頻道."""
+        if world_id:
+            full_channel = f"{channel}{{world={world_id}}}"
+        else:
+            full_channel = channel
+
+        self._subscriptions.discard(full_channel)
+
+        if self._loop and self._connected:
+            asyncio.run_coroutine_threadsafe(
+                self._send_unsubscribe(full_channel), self._loop
+            )
+
+    def on_event(self, event: str, callback: Callable):
+        """註冊事件回調.
+
+        Args:
+            event: 事件名稱 (listings/add, listings/remove, sales/add)
+            callback: 回調函數，接收事件數據
+        """
+        if event not in self._callbacks:
+            self._callbacks[event] = []
+        self._callbacks[event].append(callback)
+
+    def get_latest_messages(self, limit: int = 10) -> list:
+        """取得最新的消息.
+
+        Args:
+            limit: 最大數量
+
+        Returns:
+            消息列表
+        """
+        messages = []
+        while not self._message_queue.empty() and len(messages) < limit:
+            try:
+                messages.append(self._message_queue.get_nowait())
+            except:
+                break
+        return messages
+
+    def is_connected(self) -> bool:
+        """檢查是否已連接."""
+        return self._connected
+
+    def watch_item(self, item_id: int):
+        """開始關注某個物品的更新.
+
+        Args:
+            item_id: 物品 ID
+        """
+        self._watched_items.add(item_id)
+
+    def unwatch_item(self, item_id: int):
+        """停止關注某個物品."""
+        self._watched_items.discard(item_id)
+        self._item_cache.pop(item_id, None)
+
+    def get_cached_data(self, item_id: int) -> Optional[dict]:
+        """取得物品的緩存數據.
+
+        Args:
+            item_id: 物品 ID
+
+        Returns:
+            緩存數據，如果沒有則返回 None
+        """
+        return self._item_cache.get(item_id)
+
+    def has_update(self, item_id: int, since: float = 0) -> bool:
+        """檢查物品是否有新更新.
+
+        Args:
+            item_id: 物品 ID
+            since: 時間戳，檢查此時間之後是否有更新
+
+        Returns:
+            是否有新更新
+        """
+        cache = self._item_cache.get(item_id)
+        if cache and cache["timestamp"] > since:
+            return True
+        return False
+
+    def clear_cache(self, item_id: int = None):
+        """清除緩存.
+
+        Args:
+            item_id: 物品 ID，如果為 None 則清除所有
+        """
+        if item_id:
+            self._item_cache.pop(item_id, None)
+        else:
+            self._item_cache.clear()
+
+
+# 全域 WebSocket 客戶端實例
+_ws_client: Optional[UniversalisWebSocket] = None
+
+
+def get_ws_client() -> UniversalisWebSocket:
+    """取得全域 WebSocket 客戶端."""
+    global _ws_client
+    if _ws_client is None:
+        _ws_client = UniversalisWebSocket()
+    return _ws_client
+
+
+def start_websocket():
+    """啟動全域 WebSocket 連線."""
+    client = get_ws_client()
+    client.start()
+    return client
+
+
+def stop_websocket():
+    """停止全域 WebSocket 連線."""
+    global _ws_client
+    if _ws_client:
+        _ws_client.stop()
+        _ws_client = None