Update get_msg.py

get_msg.py

@@ -20,11 +20,18 @@ WS_URL = (
     "wss://desk-ws.channel.io/socket.io/"
     "?platform=web&EIO=4&transport=websocket"
 )
+ACCOUNT_WS_URL = (
+    "wss://account-ws.channel.io/socket.io/"
+    "?platform=web&EIO=4&transport=websocket"
+)
 
 HEARTBEAT_INTERVAL = 30.0
 TOKEN_CHECK_INTERVAL = 1.0
 REQUEST_TIMEOUT = 30.0
 
+# account 側は短いタイムアウトで recv し、トークン監視を素早く回す
+ACCOUNT_RECV_TIMEOUT = 1.0
+
 JST = timezone(timedelta(hours=9))
 
 
@@ -49,12 +56,12 @@ def log_block(tag: str, title: str, body: str) -> None:
 class LoggingWS:
     """websocket.WebSocket をラップして送受信を必ずログ出力する。"""
 
-    def __init__(self, ws: websocket.WebSocket) -> None:
+    def __init__(self, ws: websocket.WebSocket, label: str = "WS") -> None:
         self._ws = ws
+        self._label = label
         self._send_count = 0
         self._recv_count = 0
 
-    # --- 送信 ---
     def send(self, payload: str | bytes) -> None:
         self._send_count += 1
         text = (
@@ -63,18 +70,17 @@ class LoggingWS:
             else payload
         )
         log_block(
-            "WS-SEND",
+            f"{self._label}-SEND",
             f"#{self._send_count} ({len(text)} bytes)",
             text,
         )
         self._ws.send(payload)
 
-    # --- 受信 ---
     def recv(self) -> str | bytes | None:
         packet = self._ws.recv()
         self._recv_count += 1
         if packet is None:
-            log("WS-RECV", f"#{self._recv_count} (None / 切断)")
+            log(f"{self._label}-RECV", f"#{self._recv_count} (None / 切断)")
             return None
         text = (
             packet.decode("utf-8", errors="replace")
@@ -82,7 +88,7 @@ class LoggingWS:
             else packet
         )
         log_block(
-            "WS-RECV",
+            f"{self._label}-RECV",
             f"#{self._recv_count} ({len(text)} chars)",
             text,
         )
@@ -92,10 +98,9 @@ class LoggingWS:
         try:
             self._ws.close()
         finally:
-            log("WS", "WebSocket をクローズしました。")
+            log(self._label, "WebSocket をクローズしました。")
 
     def __getattr__(self, name: str) -> Any:
-        # 未定義の属性は元の WebSocket に委譲
         return getattr(self._ws, name)
 
 
@@ -158,7 +163,6 @@ def get_nested(data: Any, *path: str) -> Any:
 
 
 def find_first_key(data: Any, key: str) -> Any:
-    """レスポンス形式が多少変わっても id を拾えるように再帰検索する。"""
     if isinstance(data, dict):
         if key in data:
             return data[key]
@@ -219,10 +223,8 @@ def fetch_ids(tokens: dict[str, str]) -> tuple[str, str]:
 def timestamp_to_text(value: Any) -> str:
     if not isinstance(value, (int, float)):
         return str(value)
-
     if value < 100_000_000_000:
         return str(value)
-
     dt_utc = datetime.fromtimestamp(value / 1000, tz=timezone.utc)
     dt_jst = dt_utc.astimezone(JST)
     return (
@@ -233,17 +235,10 @@ def timestamp_to_text(value: Any) -> str:
 
 def format_json_with_timestamps(value: Any, key: str | None = None) -> Any:
     if isinstance(value, dict):
-        return {
-            k: format_json_with_timestamps(v, k)
-            for k, v in value.items()
-        }
+        return {k: format_json_with_timestamps(v, k) for k, v in value.items()}
     if isinstance(value, list):
         return [format_json_with_timestamps(v) for v in value]
-    if (
-        key
-        and key.lower().endswith("at")
-        and isinstance(value, (int, float))
-    ):
+    if key and key.lower().endswith("at") and isinstance(value, (int, float)):
         return {
             "epoch_ms": value,
             "formatted": timestamp_to_text(value),
@@ -279,7 +274,6 @@ def print_create_message(payload: list[Any]) -> None:
         return
 
     entity = entity_wrapper.get("entity", {})
-    refers = entity_wrapper.get("refers", {})
 
     if isinstance(entity, dict):
         log("CREATE", f"chatKey    : {entity.get('chatKey')}")
@@ -327,7 +321,6 @@ def print_create_message(payload: list[Any]) -> None:
 # Socket.IO / Engine.IO
 # ---------------------------------------------------------------------------
 def parse_socketio_event(packet: str) -> tuple[str | None, list[Any] | None]:
-    """Socket.IO の 42... パケットを namespace と JSON payload に分解する。"""
     log("PARSE", f"packet={packet!r}")
     if not packet.startswith("42"):
         log("PARSE", "-> Socket.IO イベント(42)ではありません。")
@@ -366,10 +359,7 @@ def send_channel_connect(
     message = (
         '40/desk/channel,'
         + json.dumps(
-            {
-                "jwt": session_token,
-                "channelId": channel_id,
-            },
+            {"jwt": session_token, "channelId": channel_id},
             ensure_ascii=False,
             separators=(",", ":"),
         )
@@ -378,6 +368,22 @@ def send_channel_connect(
     ws.send(message)
 
 
+def send_account_connect(
+    ws: LoggingWS,
+    session_token: str,
+) -> None:
+    message = (
+        '40/desk/account,'
+        + json.dumps(
+            {"jwt": session_token},
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+    )
+    log("SOCKET", "account connect 送信")
+    ws.send(message)
+
+
 def send_refresh(
     ws: LoggingWS,
     session_token: str,
@@ -390,7 +396,7 @@ def send_refresh(
             separators=(",", ":"),
         )
     )
-    log("SOCKET", "refresh を送信します。")
+    log("SOCKET", "refresh を送信します（account WS）。")
     ws.send(message)
 
 
@@ -408,7 +414,6 @@ def send_heartbeat(ws: LoggingWS, manager_id: str) -> None:
 
 
 def send_pong(ws: LoggingWS) -> None:
-    """Engine.IO ping(2) に対する pong(3)。"""
     log("SOCKET", "Engine.IO ping(2) を受信 -> pong(3) を返します。")
     ws.send("3")
 
@@ -418,7 +423,7 @@ def wait_engineio_open(
     session_token: str,
     channel_id: str,
 ) -> None:
-    log("SOCKET", "Engine.IO open 待機開始")
+    log("SOCKET", "Channel Engine.IO open 待機開始")
     deadline = time.monotonic() + REQUEST_TIMEOUT
 
     while time.monotonic() < deadline:
@@ -429,10 +434,10 @@ def wait_engineio_open(
             continue
 
         if packet is None:
-            raise RuntimeError("WebSocketが切断されました。")
+            raise RuntimeError("Channel WebSocketが切断されました。")
 
         if packet.startswith("0"):
-            log("SOCKET", "Engine.IO open パケットを受信しました。")
+            log("SOCKET", "Channel Engine.IO open パケットを受信しました。")
             send_channel_connect(ws, session_token, channel_id)
             log("SOCKET", "desk/channel に接続しました。")
             return
@@ -443,7 +448,144 @@ def wait_engineio_open(
 
         log("SOCKET", f"接続待機中に受信: {packet!r}")
 
-    raise TimeoutError("Engine.IO open パケットを受信できませんでした。")
+    raise TimeoutError("Channel Engine.IO open パケットを受信できませんでした。")
+
+
+def wait_account_engineio_open(
+    ws: LoggingWS,
+    session_token: str,
+) -> None:
+    log("SOCKET", "Account Engine.IO open 待機開始")
+    deadline = time.monotonic() + REQUEST_TIMEOUT
+
+    while time.monotonic() < deadline:
+        try:
+            packet = ws.recv()
+        except websocket.WebSocketTimeoutException:
+            continue
+
+        if packet is None:
+            raise RuntimeError("Account WebSocketが切断されました。")
+
+        if packet.startswith("0"):
+            log("SOCKET", "Account Engine.IO open パケットを受信しました。")
+            send_account_connect(ws, session_token)
+            log("SOCKET", "desk/account に接続しました。")
+            return
+
+        if packet == "2":
+            send_pong(ws)
+            continue
+
+        log("SOCKET", f"Account 接続待機中に受信: {packet!r}")
+
+    raise TimeoutError("Account Engine.IO open パケットを受信できませんでした。")
+
+
+# ---------------------------------------------------------------------------
+# 各 WebSocket の受信ループ
+# ---------------------------------------------------------------------------
+def channel_loop(
+    ws: LoggingWS,
+    session_token: str,
+    channel_id: str,
+    manager_id: str,
+    stop_event: threading.Event,
+) -> None:
+    wait_engineio_open(ws, session_token, channel_id)
+    last_heartbeat = time.monotonic()
+
+    log("SOCKET", "Channel 待機中...")
+
+    while not stop_event.is_set():
+        now = time.monotonic()
+
+        if now - last_heartbeat >= HEARTBEAT_INTERVAL:
+            send_heartbeat(ws, manager_id)
+            last_heartbeat = now
+
+        try:
+            packet = ws.recv()
+        except websocket.WebSocketTimeoutException:
+            continue
+
+        if packet is None:
+            raise RuntimeError("Channel WebSocketが切断されました。")
+
+        if packet == "2":
+            send_pong(ws)
+            continue
+
+        if packet == "3":
+            log("SOCKET", "Channel Engine.IO pong(3) を受信（無視）")
+            continue
+
+        namespace, payload = parse_socketio_event(packet)
+        if namespace is None or payload is None:
+            continue
+
+        if namespace == "/desk/channel" and payload:
+            event_name = payload[0]
+            log("SOCKET", f"event={event_name}")
+            if event_name == "create":
+                print_create_message(payload)
+
+
+def account_loop(
+    ws: LoggingWS,
+    token_holder: dict[str, str],
+    client: ChannelWorksClient,
+    stop_event: threading.Event,
+) -> None:
+    wait_account_engineio_open(ws, token_holder["token"])
+    last_token_check = time.monotonic()
+
+    log("SOCKET", "Account 待機中...")
+
+    while not stop_event.is_set():
+        now = time.monotonic()
+
+        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"]
+            ):
+                log(
+                    "AUTH",
+                    "x-account の変更を検出しました（Account WS に refresh 送信）。",
+                )
+                send_refresh(ws, current_session_token)
+                token_holder["token"] = current_session_token
+
+            last_token_check = now
+
+        try:
+            packet = ws.recv()
+        except websocket.WebSocketTimeoutException:
+            continue
+
+        if packet is None:
+            raise RuntimeError("Account WebSocketが切断されました。")
+
+        if packet == "2":
+            send_pong(ws)
+            continue
+
+        if packet == "3":
+            log("SOCKET", "Account Engine.IO pong(3) を受信（無視）")
+            continue
+
+        namespace, payload = parse_socketio_event(packet)
+        if namespace is None or payload is None:
+            continue
+
+        log(
+            "SOCKET",
+            f"Account event: namespace={namespace}, payload={payload!r}",
+        )
 
 
 # ---------------------------------------------------------------------------
@@ -459,6 +601,7 @@ def run() -> None:
     log("MAIN", f"CHANNEL_API_URL={CHANNEL_API_URL}")
     log("MAIN", f"CHANNEL_LOOKUP_URL={CHANNEL_LOOKUP_URL}")
     log("MAIN", f"WS_URL={WS_URL}")
+    log("MAIN", f"ACCOUNT_WS_URL={ACCOUNT_WS_URL}")
 
     client = ChannelWorksClient(
         room_id=room_id,
@@ -489,8 +632,7 @@ def run() -> None:
         missing = [key for key in required if not tokens.get(key)]
         if missing:
             raise RuntimeError(
-                "必要なトークンが取得できませんでした: "
-                + ", ".join(missing)
+                "必要なトークンが取得できませんでした: " + ", ".join(missing)
             )
         log("AUTH", f"トークン取得 OK（{len(tokens)} 件）")
 
@@ -500,83 +642,80 @@ def run() -> None:
         log("HTTP", f"channel.id = {channel_id}")
 
         last_session_token = tokens["x-account"]
+        token_holder: dict[str, str] = {"token": last_session_token}
 
-        raw_ws = websocket.create_connection(
+        # ---- Channel 用 WebSocket ----
+        channel_raw_ws = websocket.create_connection(
             WS_URL,
             timeout=3.0,
             origin="https://desk.channel.io",
             host="desk-ws.channel.io",
         )
-        log("WS", "WebSocket 接続完了")
-        ws = LoggingWS(raw_ws)
-
-        try:
-            wait_engineio_open(
-                ws,
-                last_session_token,
-                channel_id,
-            )
-
-            last_heartbeat = time.monotonic()
-            last_token_check = time.monotonic()
-
-            log("SOCKET", "待機中...")
-
-            while True:
-                now = time.monotonic()
-
-                if now - last_heartbeat >= HEARTBEAT_INTERVAL:
-                    send_heartbeat(ws, manager_id)
-                    last_heartbeat = now
+        log("WS", "Channel WebSocket 接続完了")
+        channel_ws = LoggingWS(channel_raw_ws, label="CHANNEL-WS")
 
-                if now - last_token_check >= TOKEN_CHECK_INTERVAL:
-                    current_tokens = client.get_tokens()
-                    current_session_token = current_tokens.get("x-account")
-
-                    if (
-                        current_session_token
-                        and current_session_token != last_session_token
-                    ):
-                        log(
-                            "AUTH",
-                            "x-account の変更を検出しました。",
-                        )
-                        send_refresh(ws, current_session_token)
-                        last_session_token = current_session_token
-
-                    last_token_check = now
-
-                try:
-                    packet = ws.recv()
-                except websocket.WebSocketTimeoutException:
-                    continue
-
-                if packet is None:
-                    raise RuntimeError("WebSocketが切断されました。")
-
-                # Engine.IO ping -> pong
-                if packet == "2":
-                    send_pong(ws)
-                    continue
-
-                # Engine.IO pong
-                if packet == "3":
-                    log("SOCKET", "Engine.IO pong(3) を受信（無視）")
-                    continue
+        # ---- Account 用 WebSocket ----
+        account_raw_ws = websocket.create_connection(
+            ACCOUNT_WS_URL,
+            timeout=ACCOUNT_RECV_TIMEOUT,
+            origin="https://desk.channel.io",
+            host="account-ws.channel.io",
+        )
+        log("WS", "Account WebSocket 接続完了")
+        account_ws = LoggingWS(account_raw_ws, label="ACCOUNT-WS")
+
+        stop_event = threading.Event()
+
+        def run_channel() -> None:
+            try:
+                channel_loop(
+                    channel_ws,
+                    token_holder["token"],
+                    channel_id,
+                    manager_id,
+                    stop_event,
+                )
+            except Exception as exc:
+                log("CHANNEL", f"エラーで終了: {exc!r}")
+                stop_event.set()
+
+        def run_account() -> None:
+            try:
+                account_loop(account_ws, token_holder, client, stop_event)
+            except Exception as exc:
+                log("ACCOUNT", f"エラーで終了: {exc!r}")
+                stop_event.set()
+
+        channel_thread = threading.Thread(
+            target=run_channel, name="channel-ws", daemon=True
+        )
+        account_thread = threading.Thread(
+            target=run_account, name="account-ws", daemon=True
+        )
+        channel_thread.start()
+        account_thread.start()
 
-                namespace, payload = parse_socketio_event(packet)
-                if namespace is None or payload is None:
-                    continue
+        log("MAIN", "両 WebSocket スレッド起動。待機中...")
 
-                if namespace == "/desk/channel" and payload:
-                    event_name = payload[0]
-                    log("SOCKET", f"event={event_name}")
+        try:
+            while not stop_event.is_set():
+                stop_event.wait(0.5)
+        except KeyboardInterrupt:
+            log("MAIN", "KeyboardInterrupt を受信しました。")
+            stop_event.set()
 
-                    if event_name == "create":
-                        print_create_message(payload)
+        # スレッド終了を待つ
+        channel_thread.join(timeout=5)
+        account_thread.join(timeout=5)
 
-        finally:
-            ws.close()
+        try:
+            channel_ws.close()
+        except Exception:
+            pass
+        try:
+            account_ws.close()
+        except Exception:
+            pass
 
     finally:
         client.stop_auto_touch()