feat: 多帳號切換 + whoami,並修正版本/跨年日期比對

README.md

@@ -61,6 +61,15 @@ The best MCP server for Ptt. Proudly built by <a href="https://pyptt.cc/">PyPtt<
         *   `ghcr.io/pyptt/ptt_mcp_server:latest`: 指定要運行的 Docker 映像檔。
     *   `"env"`: 將 `PTT_ID` 和 `PTT_PW` 直接設定為環境變數。**請務必替換為您自己的 PTT 帳號和密碼。**
 
+    **多帳號設定（選用）：**
+    若您有多組 PTT 帳號，可改用環境變數 `PTT_ACCOUNTS`（格式：`名稱=帳號:密碼`，多組以逗號分隔）一次設定多組，之後即可用 `switch_account("名稱")` 切換帳號、`list_accounts()` 查詢可用名稱：
+
+    ```
+    PTT_ACCOUNTS=default=acc1:pw1,alt=acc2:pw2
+    ```
+
+    名稱不可含 `=`、帳號不可含 `:`、密碼不可含 `,`（PTT 帳密實務上不會用到這些字元）。使用 Docker 時，請在 `args` 中同時加上 `"-e", "PTT_ACCOUNTS"`。原本的單一 `PTT_ID`/`PTT_PW` 設定仍然相容，會自動成為名為 `default` 的帳號。
+
 3.  **啟動與測試**：
     您的 MCP 客戶端現在應該能自動啟動 PTT MCP 伺服器了。您可以嘗試一個簡單的指令來測試連線，例如要求它登入 PTT。
 

README_ENG.md

@@ -58,6 +58,15 @@ Using Docker is the most recommended way to deploy the PTT MCP Server, as it pro
      * `-e PTT_ID` and `-e PTT_PW`: Tells Docker to pass the `PTT_ID` and `PTT_PW` environment variables to the container.  
      * `ghcr.io/PyPtt/ptt_mcp_server:latest`: Specifies the Docker image to run.  
    * `env`: Sets `PTT_ID` and `PTT_PW` directly as environment variables. **Be sure to replace these with your own PTT account ID and password.**  
+
+   **Multiple accounts (optional):**  
+   If you have more than one PTT account, you can configure them all at once with the `PTT_ACCOUNTS` environment variable (format: `name=id:password`, comma-separated), then switch between them with `switch_account("name")` and list the available names with `list_accounts()`:
+
+   ```
+   PTT_ACCOUNTS=default=acc1:pw1,alt=acc2:pw2
+   ```
+
+   Names must not contain `=`, IDs must not contain `:`, and passwords must not contain `,` (PTT credentials never use these characters in practice). When using Docker, also add `"-e", "PTT_ACCOUNTS"` to `args`. The single `PTT_ID`/`PTT_PW` setup remains supported and becomes an account named `default`.
 3. Launch and Test:  
    Your MCP client should now be able to start the PTT MCP server automatically. You can try a simple command, such as asking it to log into PTT, to test the connection.
 

src/api_post.py

@@ -7,6 +7,22 @@
 from utils import _call_ptt_service
 
 
+def parse_date_str(date_str: str) -> datetime:
+    # Add a dummy year to make it a full date for comparison.
+    # Assuming all dates are within the current year for simplicity.
+    # For cross-year comparisons, more complex logic would be needed.
+
+    if date_str.count('/') == 2:
+        return datetime.strptime(date_str, "%Y/%m/%d")
+
+    current_year = datetime.now().year
+    return datetime.strptime(f"{current_year}/{date_str}", "%Y/%m/%d")
+
+
+def _md(d: datetime):
+    return (d.month, d.day)
+
+
 def register_tools(mcp: FastMCP, memory_storage: Dict[str, Any], version: str):
     @mcp.tool()
     def get_board_rules() -> Dict[str, Any]:
@@ -53,24 +69,14 @@ def get_post_index_range(board: str, target_date_str: str) -> Dict[str, Any]:
                             失敗時: {'success': False, 'message': str}
         """
 
-        # Helper to parse and compare dates
-        def parse_date_str(date_str: str) -> datetime:
-            # Add a dummy year to make it a full date for comparison.
-            # Assuming all dates are within the current year for simplicity.
-            # For cross-year comparisons, more complex logic would be needed.
-
-            if date_str.count('/') == 2:
-                return datetime.strptime(date_str, "%Y/%m/%d")
-
-            current_year = datetime.now().year
-            return datetime.strptime(f"{current_year}/{date_str}", "%Y/%m/%d")
-
         try:
             target_date = parse_date_str(target_date_str)
         except ValueError:
             return {"success": False,
                     "message": f"Invalid target_date_str format: {target_date_str}. Expected 'YYYY/MM/DD'."}
 
+        # ponytail: 只比對 month/day（PTT list_date 無年份）。若目標日期區間跨年（12月→1月），binary search 的單調性會被打破、結果可能不對；需要跨年支援時得改用文章 header 的完整日期。
+
         # 1. Get the newest index for the board
         newest_index_response = _call_ptt_service(
             memory_storage,
@@ -115,9 +121,9 @@ def parse_date_str(date_str: str) -> datetime:
                 low = mid + 1
                 continue
 
-            if post_date < target_date:
+            if _md(post_date) < _md(target_date):
                 low = mid + 1
-            elif post_date == target_date:
+            elif _md(post_date) == _md(target_date):
                 start_index = mid
                 high = mid - 1  # Try to find an earlier one
             else:  # post_date > target_date
@@ -148,9 +154,9 @@ def parse_date_str(date_str: str) -> datetime:
                 low = mid + 1
                 continue
 
-            if post_date > target_date:
+            if _md(post_date) > _md(target_date):
                 high = mid - 1
-            elif post_date == target_date:
+            elif _md(post_date) == _md(target_date):
                 end_index = mid
                 low = mid + 1  # Try to find a later one
             else:  # post_date < target_date
@@ -168,8 +174,8 @@ def parse_date_str(date_str: str) -> datetime:
                 index=start_index,
                 query=True,
             )
-            if not start_post_response.get('success') or not start_post_response.get('data') or parse_date_str(
-                    start_post_response['data'].get('list_date', '')) != target_date:
+            if not start_post_response.get('success') or not start_post_response.get('data') or _md(parse_date_str(
+                    start_post_response['data'].get('list_date', ''))) != _md(target_date):
                 return {"success": False,
                         "message": f"在 {board} 板找不到日期 {target_date} 的任何文章 (start_index verification failed)."}
 
@@ -181,8 +187,8 @@ def parse_date_str(date_str: str) -> datetime:
                 index=end_index,
                 query=True,
             )
-            if not end_post_response.get('success') or not end_post_response.get('data') or parse_date_str(
-                    end_post_response['data'].get('list_date', '')) != target_date:
+            if not end_post_response.get('success') or not end_post_response.get('data') or _md(parse_date_str(
+                    end_post_response['data'].get('list_date', ''))) != _md(target_date):
                 return {"success": False,
                         "message": f"在 {board} 板找不到日期 {target_date} 的任何文章 (end_index verification failed)."}
 

src/api_ptt.py

@@ -1,10 +1,80 @@
+import threading
 from typing import Dict, Any, Optional, List, Tuple
 
 import PyPtt
 from fastmcp import FastMCP
 
 from utils import _call_ptt_service, _handle_ptt_exception
 
+# ponytail: module-level 鎖序列化池變動，防 FastMCP 併發呼叫造成 double-login/race。
+_pool_lock = threading.Lock()
+
+
+def _login_new_service(
+    ptt_id: str, ptt_pw: str
+) -> Tuple[Optional[Any], Dict[str, Any]]:
+    """建一個新 Service 並登入。成功回 (svc, 成功dict)；失敗先 close() 掉失敗的 svc
+    再回 (None, 錯誤dict)。此 helper 不碰 active session 或池。"""
+    ptt_service = PyPtt.Service({})
+    try:
+        ptt_service.call(
+            "login",
+            {
+                "ptt_id": ptt_id,
+                "ptt_pw": ptt_pw,
+                "kick_other_session": True,
+            },
+        )
+        return ptt_service, {"success": True, "message": "登入成功"}
+    except Exception as e:
+        # 洩漏防護：登入失敗的 Service 內有 daemon thread + 連線，離開前必須 close。
+        try:
+            ptt_service.close()
+        except Exception:
+            pass
+        return None, _handle_ptt_exception(e, {})
+
+
+def _is_alive(svc: Any) -> bool:
+    """驗活：對池裡的 Service 做一個便宜的 authenticated 呼叫探測，丟例外即視為 dead。"""
+    # ponytail: PTT 會斷閒置連線，光看 _is_login flag 偵測不到 server 端斷線，故主動用 get_time 探測。
+    try:
+        svc.call("get_time")
+        return True
+    except Exception:
+        return False
+
+
+def _activate(memory_storage: Dict[str, Any], name: str) -> Dict[str, Any]:
+    """把帳號 name 設為 active（登入並放進池）。原子性：新登入失敗不動 active session。"""
+    accounts = memory_storage["accounts"]
+    with _pool_lock:
+        pool = memory_storage["pool"]
+        svc = pool.get(name)
+
+        # 池裡有但已死 → close 汰換（離開池的 Service 都要 close 防洩漏）。
+        if svc is not None and not _is_alive(svc):
+            try:
+                svc.close()
+            except Exception:
+                pass
+            pool.pop(name, None)
+            svc = None
+
+        if svc is None:
+            svc, result = _login_new_service(accounts[name]["id"], accounts[name]["pw"])
+            if svc is None:
+                # 原子性關鍵：登入失敗，不動 ptt_bot / current_account / ptt_id / ptt_pw。
+                return result
+            pool[name] = svc
+
+        memory_storage["ptt_bot"] = svc
+        memory_storage["current_account"] = name
+        # 同步憑證，讓 bare login() 等仍與 active 帳號一致。
+        memory_storage["ptt_id"] = accounts[name]["id"]
+        memory_storage["ptt_pw"] = accounts[name]["pw"]
+        return {"success": True, "message": "登入成功"}
+
 
 def register_tools(mcp: FastMCP, memory_storage: Dict[str, Any], version: str):
     @mcp.tool()
@@ -34,8 +104,20 @@ def logout() -> Dict[str, Any]:
         if ptt_service is None:
             return {"success": False, "message": "尚未登入，無需登出"}
 
-        result = _call_ptt_service(memory_storage, "logout", success_message="登出成功")
-        memory_storage["ptt_bot"] = None
+        # 只登出 active 帳號：logout 後 close、從池移除、ptt_bot=None；其他池內帳號不動。
+        with _pool_lock:
+            result = _call_ptt_service(
+                memory_storage, "logout", success_message="登出成功"
+            )
+            try:
+                ptt_service.close()
+            except Exception:
+                pass
+            pool = memory_storage["pool"]
+            for pool_name, pool_svc in list(pool.items()):
+                if pool_svc is ptt_service:
+                    del pool[pool_name]
+            memory_storage["ptt_bot"] = None
         return result
 
     @mcp.tool()
@@ -68,31 +150,83 @@ def login() -> Dict[str, Any]:
                             - 'NEED_MODERATOR_PERMISSION': 需要看板管理員權限。
                             - 'UNKNOWN_ERROR': 操作時發生未知錯誤。
         """
-        # 如果已經有一個 bot 實例，先登出舊的
-        if memory_storage["ptt_bot"] is not None:
-            try:
-                memory_storage["ptt_bot"].call("logout")
-                memory_storage["ptt_bot"] = None  # 清除 session
-            except Exception:
-                pass
+        return _activate(memory_storage, memory_storage["current_account"])
 
-        ptt_service = PyPtt.Service({})
-        try:
-            ptt_service.call(
-                "login",
-                {
-                    "ptt_id": memory_storage["ptt_id"],
-                    "ptt_pw": memory_storage["ptt_pw"],
-                    "kick_other_session": True,
-                },
-            )
-            # 登入成功後，將 bot 實例存起來
-            memory_storage["ptt_bot"] = ptt_service
+    @mcp.tool()
+    def list_accounts() -> Dict[str, Any]:
+        """列出已在環境變數設定的 PTT 帳號名稱（不含密碼），以及目前登入中的帳號名稱。
 
-            return {"success": True, "message": "登入成功"}
-        except Exception as e:
-            memory_storage["ptt_bot"] = None
-            return _handle_ptt_exception(e, {})
+        Returns:
+            Dict[str, Any]: {'success': True,
+                             'accounts': ['default', 'alt', ...],
+                             'current': 'default' 或 None}
+        """
+        # ponytail: 用 whoami 那套真實狀態判定（讀 PyPtt 私有 _is_login），沒 active 登入才不謊報。
+        api = getattr(memory_storage.get("ptt_bot"), "_api", None)
+        current = (
+            memory_storage.get("current_account")
+            if getattr(api, "_is_login", False)
+            else None
+        )
+        return {
+            "success": True,
+            "accounts": list(memory_storage.get("accounts", {}).keys()),
+            "current": current,
+        }
+
+    @mcp.tool()
+    def switch_account(name: str) -> Dict[str, Any]:
+        """切換到指定名稱的 PTT 帳號並登入。
+
+        帳號需事先透過環境變數 PTT_ACCOUNTS 設定（見 README）。切換後會記住此帳號，
+        後續 login() 也會使用它。可先用 list_accounts() 查詢可用名稱。
+
+        Args:
+            name (str): 帳號名稱。
+
+        Returns:
+            Dict[str, Any]: 成功 {'success': True,
+                                  'message': "已切換到帳號 'xxx' 並登入成功",
+                                  'current': 'xxx'}
+                            找不到名稱 {'success': False,
+                                        'message': "找不到帳號 'xxx'",
+                                        'available': [...]}
+                            登入失敗則回傳 login 相同的錯誤格式。
+        """
+        accounts = memory_storage.get("accounts", {})
+        if name not in accounts:
+            return {
+                "success": False,
+                "message": f"找不到帳號 '{name}'",
+                "available": list(accounts.keys()),
+            }
+        result = _activate(memory_storage, name)
+        if result.get("success"):
+            result["message"] = f"已切換到帳號 '{name}' 並登入成功"
+            result["current"] = name
+        return result
+
+    @mcp.tool()
+    def whoami() -> Dict[str, Any]:
+        """回傳目前登入中的 PTT 帳號資訊，實際登入狀態直接向 PyPtt 查詢。
+
+        Returns:
+            Dict[str, Any]: {'success': True,
+                             'account': 'default',   # 帳號名稱（來自 PTT_ACCOUNTS 標籤）
+                             'ptt_id': 'CodingMan',  # PyPtt 目前登入的實際 PTT ID；未登入為 None
+                             'logged_in': True}      # PyPtt 回報是否已登入
+        """
+        # ponytail: 讀 PyPtt 私有屬性取真實狀態；PyPtt 無公開存取方法，改版壞了再換
+        api = getattr(memory_storage.get("ptt_bot"), "_api", None)
+        logged_in = bool(getattr(api, "_is_login", False))
+        # PyPtt 登出不清 ptt_id，未登入時不回報殘留值
+        ptt_id = (getattr(api, "ptt_id", "") or None) if logged_in else None
+        return {
+            "success": True,
+            "account": memory_storage.get("current_account"),
+            "ptt_id": ptt_id,
+            "logged_in": logged_in,
+        }
 
     @mcp.tool()
     def get_post(

src/auto_version.py

@@ -38,14 +38,18 @@ def get_version() -> tuple[Optional[str], Optional[str]]:
     return None, None  # Should not be reached
 
 
+def _version_key(version_str: str) -> tuple:
+    return tuple(int(part) for part in version_str.split('.'))
+
+
 def main():
     remote_version, current_version = get_version()
 
     if remote_version is None or current_version is None:
         print("Failed to retrieve version information.")
         return
 
-    if int(remote_version.replace('.', '')) <= int(current_version.replace('.', '')):
+    if _version_key(remote_version) <= _version_key(current_version):
         print(current_version)
     else:
         print(remote_version)