From 26958e0c0f49686e4b5243b461845a86c34f0b7a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 14:06:10 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=96=92=E7=BD=AE=E6=96=B7=E7=B7=9A?= =?UTF-8?q?=E8=87=AA=E5=8B=95=E9=87=8D=E7=99=BB=20+=20=E7=99=BB=E5=85=A5?= =?UTF-8?q?=E9=A0=BB=E7=8E=87=E9=99=90=E5=88=B6=EF=BC=8860=20=E7=A7=92?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 同一帳號 60 秒內只允許一次登入嘗試（失敗也算），超過回 LOGIN_THROTTLED + retry_after - 閒置超過 60 秒的下一次呼叫前先用 get_time 探測，斷線才重登；先探測再呼叫，避免寫入操作重試重複送出 - 主動登出或從未登入時不自動登入 - login/logout docstring 提示 agent 不需做完事就登出、不要反覆 login Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011od43ZvAoaP413L1eYeWSD --- src/api_ptt.py | 55 ++++++++++++- src/test_auto_relogin.py | 135 ++++++++++++++++++++++++++++++++ src/test_switch_account_pool.py | 2 + src/utils.py | 7 ++ 4 files changed, 198 insertions(+), 1 deletion(-) create mode 100644 src/test_auto_relogin.py diff --git a/src/api_ptt.py b/src/api_ptt.py index bd74151..57e76e1 100644 --- a/src/api_ptt.py +++ b/src/api_ptt.py @@ -1,4 +1,5 @@ import threading +import time from typing import Dict, Any, Optional, List, Tuple import PyPtt @@ -9,6 +10,14 @@ # ponytail: module-level 鎖序列化池變動，防 FastMCP 併發呼叫造成 double-login/race。 _pool_lock = threading.Lock() +# 同一帳號兩次登入嘗試的最小間隔（秒）。PTT 會擋太頻繁的登入，失敗的嘗試也算。 +LOGIN_MIN_INTERVAL = 60 +# 距離上次呼叫超過這麼久，下次呼叫前先探測連線，斷了就自動重登。 +IDLE_PROBE_SECONDS = 60 + +# 時鐘抽出來，方便測試快轉時間。 +_clock = time.monotonic + def _bad_post_type_from_name(name: Optional[str]) -> Optional["PyPtt.BadPostType"]: """劣退／惡退分類的字串名 → PyPtt.BadPostType。 @@ -76,6 +85,18 @@ def _activate(memory_storage: Dict[str, Any], name: str) -> Dict[str, Any]: svc = None if svc is None: + last_login_at = memory_storage.setdefault("last_login_at", {}) + now = _clock() + last = last_login_at.get(name) + if last is not None and now - last < LOGIN_MIN_INTERVAL: + retry_after = int(LOGIN_MIN_INTERVAL - (now - last)) + 1 + return { + "success": False, + "message": f"登入太頻繁，請 {retry_after} 秒後再試", + "code": "LOGIN_THROTTLED", + "retry_after": retry_after, + } + last_login_at[name] = now svc, result = _login_new_service(accounts[name]["id"], accounts[name]["pw"]) if svc is None: # 原子性關鍵：登入失敗，不動 ptt_bot / current_account / ptt_id / ptt_pw。 @@ -87,10 +108,33 @@ def _activate(memory_storage: Dict[str, Any], name: str) -> Dict[str, Any]: # 同步憑證，讓 bare login() 等仍與 active 帳號一致。 memory_storage["ptt_id"] = accounts[name]["id"] memory_storage["ptt_pw"] = accounts[name]["pw"] + memory_storage["last_used_at"] = _clock() return {"success": True, "message": "登入成功"} +def _ensure_alive(memory_storage: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """閒置一段時間後，在下次呼叫前先探測連線，斷了就自動重登。 + + 回 None 代表可以繼續呼叫；回 dict 代表重登失敗（含 LOGIN_THROTTLED），呼叫端直接回傳它。 + """ + # ponytail: 先探測再呼叫，而非失敗後重試：發文、推文、轉帳等寫入操作重試可能重複送出。 + if memory_storage.get("ptt_bot") is None: + # 從未登入或已主動登出，不自動登入。 + return None + now = _clock() + last_used_at = memory_storage.get("last_used_at") + if last_used_at is not None and now - last_used_at < IDLE_PROBE_SECONDS: + memory_storage["last_used_at"] = now + return None + # _activate 會用 get_time 探測：活著就沿用，死了才重登（受 LOGIN_MIN_INTERVAL 限制）。 + result = _activate(memory_storage, memory_storage["current_account"]) + return None if result.get("success") else result + + def register_tools(mcp: FastMCP, memory_storage: Dict[str, Any], version: str): + # 讓 utils._call_ptt_service 在每次呼叫前做閒置探測；放在 storage 裡避免 utils ↔ api_ptt 循環 import。 + memory_storage["ensure_alive"] = lambda: _ensure_alive(memory_storage) + @mcp.tool() def get_version() -> Dict[str, Any]: """ @@ -107,6 +151,8 @@ def logout() -> Dict[str, Any]: """Logs out from the PTT service. This function terminates the current PTT session if one is active. + 不需要在每次任務完成後登出：閒置斷線時 server 會在下次呼叫前自動重新登入， + 頻繁登出再登入反而容易被 PTT 擋下。只有在使用者明確要求時才呼叫。 Returns: Dict[str, Any]: A dictionary indicating the result of the logout attempt. @@ -120,8 +166,12 @@ def logout() -> Dict[str, Any]: # 只登出 active 帳號：logout 後 close、從池移除、ptt_bot=None；其他池內帳號不動。 with _pool_lock: + # 已持有 _pool_lock，不能再走閒置探測（_activate 會再拿鎖而死結）。 result = _call_ptt_service( - memory_storage, "logout", success_message="登出成功" + memory_storage, + "logout", + success_message="登出成功", + ensure_alive=False, ) try: ptt_service.close() @@ -140,6 +190,8 @@ def login() -> Dict[str, Any]: This function initializes a connection to PTT and attempts to log in. The login status is maintained on the server for subsequent calls. + 若已登入且連線正常，不會重複登入。閒置斷線時 server 會在下次呼叫前自動重新登入， + 不需要反覆呼叫 login。同一帳號 60 秒內只允許一次登入嘗試。 Returns: Dict[str, Any]: A dictionary containing the result of the login attempt. @@ -153,6 +205,7 @@ def login() -> Dict[str, Any]: - 'NO_PERMISSION': 沒有權限。 - 'LOGIN_FAILED': 登入失敗。 - 'TWO_FACTOR_AUTH_REQUIRED': 需要二階段驗證，請先用一般 BBS 客戶端從這台機器手動登入完成驗證。 + - 'LOGIN_THROTTLED': 登入太頻繁，請依回傳的 'retry_after' 秒數等待後再試。 - 'WRONG_CREDENTIALS': 帳號或密碼錯誤。 - 'CANT_RESPONSE': 已結案並標記, 不得回應。 - 'NO_FAST_COMMENT': 推文間隔太短。 diff --git a/src/test_auto_relogin.py b/src/test_auto_relogin.py new file mode 100644 index 0000000..d75e533 --- /dev/null +++ b/src/test_auto_relogin.py @@ -0,0 +1,135 @@ +"""閒置自動重登與登入頻率限制測試。 + +沿用 test_switch_account_pool 的假 Service 工廠；以替換 api_ptt._clock 快轉時間。 +""" + +import pytest + +from test_switch_account import _load_api_ptt, _register_tools, _FakeService +from test_switch_account_pool import _setup + + +class _Clock: + def __init__(self): + self.now = 1000.0 + + def __call__(self): + return self.now + + +@pytest.fixture +def env(monkeypatch): + api_ptt = _load_api_ptt() + factory = _setup(api_ptt) + # 死連線的 get_time 會走到 _handle_ptt_exception 的 ParameterError 分支。 + if not hasattr(api_ptt.PyPtt, "ParameterError"): + monkeypatch.setattr( + api_ptt.PyPtt, + "ParameterError", + type("ParameterError", (Exception,), {}), + raising=False, + ) + clock = _Clock() + monkeypatch.setattr(api_ptt, "_clock", clock) + tools, ms = _register_tools(api_ptt) + return api_ptt, factory, clock, tools, ms + + +def test_login_throttled_within_interval(env): + api_ptt, factory, clock, tools, ms = env + svc1 = _FakeService(alive=True) + svc2 = _FakeService(alive=True) + factory.prepare(svc1, svc2) + + assert tools["login"]()["success"] is True + svc1.alive = False + + clock.now += 10 + result = tools["login"]() + assert result["success"] is False + assert result["code"] == "LOGIN_THROTTLED" + assert 0 < result["retry_after"] <= api_ptt.LOGIN_MIN_INTERVAL + assert len(factory.handed_out) == 1 # 沒有真的去連 PTT + + clock.now += api_ptt.LOGIN_MIN_INTERVAL + assert tools["login"]()["success"] is True + assert ms["ptt_bot"] is svc2 + + +def test_failed_login_also_throttled(env): + api_ptt, factory, clock, tools, ms = env + wrong = api_ptt.PyPtt.WrongIDorPassword("bad pw") + factory.prepare(_FakeService(login_error=wrong)) + + assert tools["login"]()["success"] is False + assert tools["login"]()["code"] == "LOGIN_THROTTLED" + assert len(factory.handed_out) == 1 + + +def test_login_when_alive_is_not_throttled(env): + # 連線活著時 login() 不會重登，自然也不受頻率限制。 + api_ptt, factory, clock, tools, ms = env + svc = _FakeService(alive=True) + factory.prepare(svc) + + tools["login"]() + assert tools["login"]()["success"] is True + assert svc.login_calls == 1 + + +def test_idle_disconnect_relogs_before_call(env): + api_ptt, factory, clock, tools, ms = env + svc1 = _FakeService(alive=True) + svc2 = _FakeService(alive=True) + factory.prepare(svc1, svc2) + + tools["login"]() + svc1.alive = False # 閒置被 PTT 斷線 + clock.now += api_ptt.LOGIN_MIN_INTERVAL + api_ptt.IDLE_PROBE_SECONDS + + assert tools["get_time"]()["success"] is True + assert svc1.close_calls == 1 # 死連線被汰換 + assert svc2.login_calls == 1 # 自動重登一次 + assert ms["ptt_bot"] is svc2 + + +def test_recent_call_skips_probe(env): + # 剛用過就不探測：連線即使死了也不會在呼叫前重登。 + api_ptt, factory, clock, tools, ms = env + svc = _FakeService(alive=True) + factory.prepare(svc) + + tools["login"]() + svc.alive = False + clock.now += api_ptt.IDLE_PROBE_SECONDS - 1 + + tools["get_time"]() + assert len(factory.handed_out) == 1 + + +def test_idle_relogin_throttled_returns_error_without_calling(env): + api_ptt, factory, clock, tools, ms = env + svc1 = _FakeService(alive=True) + factory.prepare(svc1) + + tools["login"]() + svc1.alive = False + # 超過閒置探測門檻，但還在登入間隔內 → 回 LOGIN_THROTTLED，不送出原本的操作。 + clock.now += api_ptt.IDLE_PROBE_SECONDS + ms["last_login_at"]["default"] = clock.now - 1 + + result = tools["get_time"]() + assert result["code"] == "LOGIN_THROTTLED" + assert len(factory.handed_out) == 1 + + +def test_no_auto_login_when_logged_out(env): + api_ptt, factory, clock, tools, ms = env + factory.prepare(_FakeService(alive=True)) + + tools["login"]() + assert tools["logout"]()["success"] is True + clock.now += api_ptt.LOGIN_MIN_INTERVAL + api_ptt.IDLE_PROBE_SECONDS + + assert tools["get_time"]()["code"] == "NOT_LOGGED_IN" + assert len(factory.handed_out) == 1 diff --git a/src/test_switch_account_pool.py b/src/test_switch_account_pool.py index a5e57f7..5eb9e9b 100644 --- a/src/test_switch_account_pool.py +++ b/src/test_switch_account_pool.py @@ -126,6 +126,8 @@ def test_switch_to_dead_relogs(): tools["switch_account"]("default") tools["switch_account"]("alt") svc_default.alive = False # 模擬 server 端閒置斷線 + # 退回上次登入時間，避開 LOGIN_MIN_INTERVAL 的登入頻率限制。 + ms["last_login_at"]["default"] -= api_ptt.LOGIN_MIN_INTERVAL result = tools["switch_account"]("default") assert result["success"] is True diff --git a/src/utils.py b/src/utils.py index 033a870..2634e5b 100644 --- a/src/utils.py +++ b/src/utils.py @@ -69,8 +69,15 @@ def _call_ptt_service( success_message: Optional[str] = None, empty_data_message: Optional[str] = None, empty_data_code: Optional[str] = None, + ensure_alive: bool = True, **kwargs, ) -> Dict[str, Any]: + if ensure_alive: + hook = session_storage_instance.get("ensure_alive") + error = hook() if hook is not None else None + if error is not None: + return error + ptt_service = session_storage_instance.get("ptt_bot") if ptt_service is None: return {