Update get_msg.py

get_msg.py

@@ -25,7 +25,8 @@ ACCOUNT_WS_URL = (
     "?platform=web&EIO=4&transport=websocket"
 )
 
-HEARTBEAT_INTERVAL = 30.0
+# heartbeat 応答を待つタイムアウト
+HEARTBEAT_RESPONSE_TIMEOUT = 5.0
 TOKEN_CHECK_INTERVAL = 1.0
 REQUEST_TIMEOUT = 30.0
 
@@ -348,6 +349,7 @@ def send_account_connect(ws: LoggingWS, session_token: str) -> None:
 
 
 def send_refresh(ws: LoggingWS, session_token: str) -> None:
+    """account WS 用の refresh（channelId は送らない）"""
     message = (
         '42/desk/account,0'
         + json.dumps(
@@ -360,6 +362,25 @@ def send_refresh(ws: LoggingWS, session_token: str) -> None:
     ws.send(message)
 
 
+def send_channel_refresh(
+    ws: LoggingWS, session_token: str, channel_id: str
+) -> None:
+    """channel WS 用の refresh（channelId を含める）"""
+    message = (
+        '42/desk/channel,3'
+        + json.dumps(
+            ["refresh", {"jwt": session_token, "channelId": channel_id}],
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+    )
+    log(
+        "SOCKET",
+        f"refresh を送信します（channel WS, channelId={channel_id}）。",
+    )
+    ws.send(message)
+
+
 def send_heartbeat(ws: LoggingWS, manager_id: str) -> None:
     message = (
         '42/desk/channel,'
@@ -450,6 +471,7 @@ def channel_worker(
     channel_id: str,
     manager_id: str,
     stop_event: threading.Event,
+    token_holder: dict[str, str],
 ) -> None:
     try:
         ws = open_channel_ws(session_token, channel_id)
@@ -457,14 +479,36 @@ def channel_worker(
         log("CHANNEL", f"接続失敗: {exc!r}")
         return
 
-    last_heartbeat = time.monotonic()
+    current_token = session_token
+    last_token_check = time.monotonic()
+
+    # 最初の heartbeat を送信（以降は応答が来るたびに ping-pong で送り続ける）
+    send_heartbeat(ws, manager_id)
+    last_heartbeat_sent = time.monotonic()
+
     log("SOCKET", "Channel 待機中...")
     try:
         while not stop_event.is_set():
             now = time.monotonic()
-            if now - last_heartbeat >= HEARTBEAT_INTERVAL:
+
+            # heartbeat 応答が 5 秒以内に来なければ再送する
+            if now - last_heartbeat_sent >= HEARTBEAT_RESPONSE_TIMEOUT:
+                log(
+                    "SOCKET",
+                    f"heartbeat 応答が {HEARTBEAT_RESPONSE_TIMEOUT}s 以内に"
+                    "来ませんでした。再送します。",
+                )
                 send_heartbeat(ws, manager_id)
-                last_heartbeat = now
+                last_heartbeat_sent = now
+
+            # x-account の変更チェック（channel refresh 用）
+            if now - last_token_check >= TOKEN_CHECK_INTERVAL:
+                new_token = token_holder.get("token")
+                if new_token and new_token != current_token:
+                    log("AUTH", "x-account 変更検出 -> Channel refresh 送信")
+                    send_channel_refresh(ws, new_token, channel_id)
+                    current_token = new_token
+                last_token_check = now
 
             try:
                 packet = ws.recv()
@@ -487,8 +531,15 @@ def channel_worker(
             if namespace == "/desk/channel" and payload:
                 event_name = payload[0]
                 log("SOCKET", f"event={event_name}")
-                if event_name == "create":
+                if event_name == "heartbeat":
+                    # 応答が返ってきたので、次の heartbeat を送信
+                    send_heartbeat(ws, manager_id)
+                    last_heartbeat_sent = time.monotonic()
+                elif event_name == "create":
                     print_create_message(payload)
+                elif event_name == "expired":
+                    log("CHANNEL", "expired を受信しました。接続を終了します。")
+                    break
     except Exception as exc:
         log("CHANNEL", f"エラーで終了: {exc!r}")
     finally:
@@ -514,7 +565,10 @@ def account_worker(
             if now - last_token_check >= TOKEN_CHECK_INTERVAL:
                 current_tokens = client.get_tokens() or {}
                 current_session_token = current_tokens.get("x-account")
-                if current_session_token and current_session_token != token_holder["token"]:
+                if (
+                    current_session_token
+                    and current_session_token != token_holder["token"]
+                ):
                     log("AUTH", "x-account 変更検出 -> refresh 送信")
                     send_refresh(ws, current_session_token)
                     token_holder["token"] = current_session_token
@@ -572,20 +626,33 @@ def run() -> None:
         tokens = client.get_tokens()
         if not tokens:
             raise RuntimeError("get_tokens() が空でした。")
-        required = ("ch-session-1", "ch-veil-id", "x-account", "x-account-refresh")
+        required = (
+            "ch-session-1",
+            "ch-veil-id",
+            "x-account",
+            "x-account-refresh",
+        )
         missing = [k for k in required if not tokens.get(k)]
         if missing:
-            raise RuntimeError("必要なトークンが取得できません: " + ", ".join(missing))
+            raise RuntimeError(
+                "必要なトークンが取得できません: " + ", ".join(missing)
+            )
 
         manager_id, channel_id = fetch_ids(tokens)
         token_holder = {"token": tokens["x-account"]}
 
         stop_event = threading.Event()
 
-        # channel 側を先に起動（account の接続待ちで channel の connect が遅延しないように）
+        # channel 側を先に起動
         channel_thread = threading.Thread(
             target=channel_worker,
-            args=(token_holder["token"], channel_id, manager_id, stop_event),
+            args=(
+                token_holder["token"],
+                channel_id,
+                manager_id,
+                stop_event,
+                token_holder,
+            ),
             name="channel-ws",
             daemon=True,
         )