Update get_msg.py

get_msg.py

@@ -1,9 +1,51 @@
+"""
+ChannelWorks メッセージ受信ライブラリ
+
+使い方:
+    from channel_auth import ChannelWorksClient
+    from channel_messenger import ChannelMessenger
+
+    client = ChannelWorksClient(
+        room_id="...",
+        email="you@example.com",
+        password="your-password",
+        log_level=ChannelWorksClient.LOG_ERROR,
+    )
+    client.login()
+    client.start_auto_touch()
+
+    def on_push(wrapper):
+        entity = wrapper.get("entity") or {}
+        print("受信:", entity.get("plainText"))
+
+    messenger = ChannelMessenger(
+        client=client,
+        channel_slug="...",                # channel_id を自動取得したい場合
+        # manager_id="...", channel_id="...",  # 直接指定してもよい
+        group_path="/groups/グループ番号",
+        log_level=ChannelMessenger.LOG_ERROR,
+        on_push=on_push,
+    )
+    messenger.start()
+    try:
+        messenger.wait()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        messenger.stop()
+        client.stop_auto_touch()
+
+log_level:
+    0 = 何も出力しない
+    1 = エラー・警告のみ (デフォルト)
+    2 = デバッグログも出力 (送受信パケット等をすべて出力)
+"""
+
 import json
-import os
 import threading
 import time
 from datetime import datetime, timezone, timedelta
-from typing import Any
+from typing import Any, Callable, Dict, Optional, Tuple
 
 import requests
 import websocket
@@ -11,11 +53,14 @@ import websocket
 from channel_auth import ChannelWorksClient
 
 
-ROOM_ID = os.getenv("CW_ROOM_ID")
-CHANNEL_SLUG = os.getenv("CW_CHANNEL_SLUG")
+# ---------------------------------------------------------------------------
+# 定数
+# ---------------------------------------------------------------------------
+DEFAULT_GROUP_PATH = "/groups/574628"
+
+CHANNEL_API_URL_TEMPLATE = "https://api.channel.works/desk/channels/{room_id}"
+CHANNEL_LOOKUP_URL_TEMPLATE = "https://api.channel.works/desk/channels/{slug}"
 
-CHANNEL_API_URL = f"https://api.channel.works/desk/channels/{ROOM_ID}"
-CHANNEL_LOOKUP_URL = f"https://api.channel.works/desk/channels/{CHANNEL_SLUG}"
 WS_URL = (
     "wss://desk-ws.channel.io/socket.io/"
     "?platform=web&EIO=4&transport=websocket"
@@ -24,73 +69,121 @@ ACCOUNT_WS_URL = (
     "wss://account-ws.channel.io/socket.io/"
     "?platform=web&EIO=4&transport=websocket"
 )
-GROUP_PATH = os.getenv("CW_GROUP_ID", "/groups/574628")
-
-# heartbeat 応答を待つタイムアウト
-HEARTBEAT_RESPONSE_TIMEOUT = 5.0
-TOKEN_CHECK_INTERVAL = 1.0
-REQUEST_TIMEOUT = 30.0
-# heartbeat を送る間隔
-HEARTBEAT_INTERVAL = 30.0
-
-# recv 用のタイムアウト
-CHANNEL_RECV_TIMEOUT = 3.0
-ACCOUNT_RECV_TIMEOUT = 1.0
 
 JST = timezone(timedelta(hours=9))
 
-    
+
 # ---------------------------------------------------------------------------
-# ログ用ヘルパ
+# モジュールレベルユーティリティ（ログに依存しない純粋関数）
 # ---------------------------------------------------------------------------
-def log(tag: str, message: str) -> None:
-    now = datetime.now(JST).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
-    print(f"[{now}] [{tag}] {message}", flush=True)
+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 (
+        f"{dt_jst:%Y-%m-%d %H:%M:%S.%f} JST "
+        f"(UTC {dt_utc:%Y-%m-%d %H:%M:%S.%f})"
+    )
 
 
-def log_block(tag: str, title: str, body: str) -> None:
-    now = datetime.now(JST).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
-    print(f"\n[{now}] [{tag}] ===== {title} =====", flush=True)
-    print(body, flush=True)
-    print(f"[{now}] [{tag}] ===== /{title} =====", flush=True)
+def _format_json_with_timestamps(value: Any, key: Optional[str] = None) -> Any:
+    if isinstance(value, dict):
+        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)):
+        return {"epoch_ms": value, "formatted": _timestamp_to_text(value)}
+    return value
+
+
+def _get_nested(data: Any, *path: str) -> Any:
+    current = data
+    for key in path:
+        if not isinstance(current, dict) or key not in current:
+            return None
+        current = current[key]
+    return current
+
+
+def _find_first_key(data: Any, key: str) -> Any:
+    if isinstance(data, dict):
+        if key in data:
+            return data[key]
+        for value in data.values():
+            result = _find_first_key(value, key)
+            if result is not None:
+                return result
+    elif isinstance(data, list):
+        for value in data:
+            result = _find_first_key(value, key)
+            if result is not None:
+                return result
+    return None
+
+
+def _parse_socketio_event(packet: str) -> Tuple[Optional[str], Optional[list]]:
+    if not packet.startswith("42"):
+        return None, None
+    payload_text = packet[2:]
+    if payload_text.startswith("/"):
+        try:
+            namespace, payload_text = payload_text.split(",", 1)
+        except ValueError:
+            return None, None
+    else:
+        namespace = "/"
+    try:
+        payload = json.loads(payload_text)
+    except json.JSONDecodeError:
+        return namespace, None
+    if not isinstance(payload, list):
+        return namespace, None
+    return namespace, payload
 
 
 # ---------------------------------------------------------------------------
-# WebSocket ラッパ
+# WebSocket ラッパ（送受信をデバッグログするだけの薄いラッパ）
 # ---------------------------------------------------------------------------
-class LoggingWS:
-    def __init__(self, ws: websocket.WebSocket, label: str = "WS") -> None:
+class _LoggingWS:
+    def __init__(self, ws: websocket.WebSocket, label: str, messenger: "ChannelMessenger") -> None:
         self._ws = ws
         self._label = label
+        self._messenger = messenger
         self._send_count = 0
         self._recv_count = 0
 
-    def send(self, payload) -> None:
+    def send(self, payload: Any) -> None:
         self._send_count += 1
         text = (
             payload.decode("utf-8", errors="replace")
             if isinstance(payload, (bytes, bytearray))
             else payload
         )
-        log_block(
+        self._messenger._log_block_debug(
             f"{self._label}-SEND",
             f"#{self._send_count} ({len(text)} bytes)",
             text,
         )
         self._ws.send(payload)
 
-    def recv(self):
+    def recv(self) -> Optional[str]:
         packet = self._ws.recv()
         self._recv_count += 1
         if packet is None:
-            log(f"{self._label}-RECV", f"#{self._recv_count} (None / 切断)")
+            self._messenger._log_debug(
+                f"{self._label}-RECV",
+                f"#{self._recv_count} (None / 切断)",
+            )
             return None
         text = (
             packet.decode("utf-8", errors="replace")
             if isinstance(packet, (bytes, bytearray))
             else packet
         )
-        log_block(
+        self._messenger._log_block_debug(
             f"{self._label}-RECV",
             f"#{self._recv_count} ({len(text)} chars)",
             text,
@@ -104,683 +197,578 @@ class LoggingWS:
         try:
             self._ws.close()
         finally:
-            log(self._label, "WebSocket をクローズしました。")
+            self._messenger._log_debug(self._label, "WebSocket をクローズしました。")
 
     def __getattr__(self, name: str) -> Any:
         return getattr(self._ws, name)
 
 
 # ---------------------------------------------------------------------------
-# 環境変数
-# ---------------------------------------------------------------------------
-def require_env(name: str) -> str:
-    value = os.getenv(name)
-    if not value:
-        raise RuntimeError(f"環境変数 {name} が設定されていません。")
-    return value
-
-
-# ---------------------------------------------------------------------------
-# HTTP
-# ---------------------------------------------------------------------------
-def get_json(session: requests.Session, url: str, tokens) -> Any:
-    log("HTTP", f"GET {url}")
-    response = session.get(
-        url,
-        headers={
-            "Accept": "application/json",
-            "Accept-Language": "ja",
-            "Referer": "https://channel.works/",
-            "Origin": "https://channel.works",
-            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
-                          "AppleWebKit/537.36 (KHTML, like Gecko) "
-                          "Chrome/152.0.0.0 Safari/537.36",
-            "x-account": tokens["x-account"],
-        },
-        timeout=REQUEST_TIMEOUT,
-    )
-    log("HTTP", f"-> status={response.status_code} ({url})")
-    response.raise_for_status()
-    try:
-        data = response.json()
-    except ValueError as exc:
-        raise RuntimeError(
-            f"JSONを取得できませんでした: {url}\n"
-            f"status={response.status_code}\n"
-            f"body={response.text[:500]}"
-        ) from exc
-    log_block("HTTP-BODY", url, json.dumps(data, ensure_ascii=False, indent=2))
-    return data
-
-
-def get_nested(data: Any, *path: str) -> Any:
-    current = data
-    for key in path:
-        if not isinstance(current, dict) or key not in current:
-            return None
-        current = current[key]
-    return current
-
-
-def find_first_key(data: Any, key: str) -> Any:
-    if isinstance(data, dict):
-        if key in data:
-            return data[key]
-        for value in data.values():
-            result = find_first_key(value, key)
-            if result is not None:
-                return result
-    elif isinstance(data, list):
-        for value in data:
-            result = find_first_key(value, key)
-            if result is not None:
-                return result
-    return None
-
-
-def fetch_ids(tokens: dict[str, str]) -> tuple[str, str]:
-    log("HTTP", "fetch_ids 開始")
-    session = requests.Session()
-    session.cookies.update(
-        {
-            "ch-session-1": tokens["ch-session-1"],
-            "ch-veil-id": tokens["ch-veil-id"],
-            "x-account": tokens["x-account"],
-            "x-account-refresh": tokens["x-account-refresh"],
-        }
-    )
-    room_json = get_json(session, CHANNEL_API_URL, tokens)
-    manager_id = get_nested(room_json, "manager", "id")
-    if manager_id is None:
-        manager_id = find_first_key(room_json, "manager")
-        if isinstance(manager_id, dict):
-            manager_id = manager_id.get("id")
-
-    channel_json = get_json(session, CHANNEL_LOOKUP_URL, tokens)
-    channel_id = get_nested(channel_json, "channel", "id")
-    if channel_id is None:
-        channel_id = find_first_key(channel_json, "id")
-
-    if manager_id is None:
-        raise RuntimeError(
-            f"{CHANNEL_API_URL} のJSONから manager.id を取得できませんでした。"
-        )
-    if channel_id is None:
-        raise RuntimeError(
-            f"{CHANNEL_LOOKUP_URL} のJSONから channel.id を取得できませんでした。"
-        )
-    log("HTTP", f"manager.id = {manager_id}")
-    log("HTTP", f"channel.id = {channel_id}")
-    return str(manager_id), str(channel_id)
-
-
-# ---------------------------------------------------------------------------
-# 表示整形
+# 本体
 # ---------------------------------------------------------------------------
-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 (
-        f"{dt_jst:%Y-%m-%d %H:%M:%S.%f} JST "
-        f"(UTC {dt_utc:%Y-%m-%d %H:%M:%S.%f})"
-    )
+class ChannelMessenger:
+    # ログレベル
+    LOG_NONE = 0
+    LOG_ERROR = 1
+    LOG_DEBUG = 2
+
+    def __init__(
+        self,
+        client: ChannelWorksClient,
+        channel_slug: Optional[str] = None,
+        manager_id: Optional[str] = None,
+        channel_id: Optional[str] = None,
+        group_path: str = DEFAULT_GROUP_PATH,
+        log_level: int = LOG_ERROR,
+        heartbeat_interval: float = 30.0,
+        heartbeat_response_timeout: float = 5.0,
+        token_check_interval: float = 1.0,
+        request_timeout: float = 30.0,
+        channel_recv_timeout: float = 3.0,
+        account_recv_timeout: float = 1.0,
+        on_push: Optional[Callable[[Dict[str, Any]], None]] = None,
+    ) -> None:
+        if client is None:
+            raise ValueError("client (ChannelWorksClient) は必須です。")
+
+        self.client = client
+        self.channel_slug = channel_slug
+        self.manager_id = manager_id
+        self.channel_id = channel_id
+        self.group_path = group_path
+        self.log_level = log_level
+
+        self.heartbeat_interval = heartbeat_interval
+        self.heartbeat_response_timeout = heartbeat_response_timeout
+        self.token_check_interval = token_check_interval
+        self.request_timeout = request_timeout
+        self.channel_recv_timeout = channel_recv_timeout
+        self.account_recv_timeout = account_recv_timeout
+
+        self.on_push = on_push
+
+        self._lock = threading.RLock()
+        self._stop_event = threading.Event()
+        self._channel_thread: Optional[threading.Thread] = None
+        self._account_thread: Optional[threading.Thread] = None
+        # x-account の最新値。channel/account 両ワーカーがこの dict を参照する。
+        self._token_holder: Dict[str, Optional[str]] = {"token": None}
+
+    # ------------------------------------------------------------------
+    # ログ
+    # ------------------------------------------------------------------
+    @staticmethod
+    def _now_str() -> str:
+        return datetime.now(JST).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
+
+    def _log_debug(self, tag: str, msg: str) -> None:
+        if self.log_level >= self.LOG_DEBUG:
+            print(f"[{self._now_str()}] [{tag}] {msg}", flush=True)
+
+    def _log_error(self, tag: str, msg: str) -> None:
+        if self.log_level >= self.LOG_ERROR:
+            print(f"[{self._now_str()}] [{tag}] {msg}", flush=True)
+
+    def _log_block_debug(self, tag: str, title: str, body: str) -> None:
+        if self.log_level < self.LOG_DEBUG:
+            return
+        ts = self._now_str()
+        print(f"\n[{ts}] [{tag}] ===== {title} =====", flush=True)
+        print(body, flush=True)
+        print(f"[{ts}] [{tag}] ===== /{title} =====", flush=True)
+
+    def set_log_level(self, level: int) -> None:
+        self.log_level = level
+
+    # ------------------------------------------------------------------
+    # 公開 API
+    # ------------------------------------------------------------------
+    def start(self) -> None:
+        """channel / account の各 WebSocket をバックグラウンドで起動する。"""
+        with self._lock:
+            if self._is_running():
+                self._log_debug("MAIN", "ChannelMessenger は既に起動中です。")
+                return
+
+            tokens = self.client.get_tokens() or {}
+            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(
+                    "client から必要なトークンが取得できません。先に login() を実行してください: "
+                    + ", ".join(missing)
+                )
 
+            if self.manager_id is None or self.channel_id is None:
+                self._resolve_ids(tokens)
 
-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()}
-    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)):
-        return {"epoch_ms": value, "formatted": timestamp_to_text(value)}
-    return value
-
+            self._token_holder["token"] = tokens["x-account"]
+            self._stop_event.clear()
 
-def _print_message_entity(tag: str, entity: dict[str, Any]) -> None:
-    """メッセージ entity の中身を整形してログ出力する共通処理。"""
-    for key in (
-        "chatKey", "id", "mainKey", "channelId", "chatType", "chatId",
-        "hubId", "personType", "personId", "language", "plainText",
-        "writingType",
-    ):
-        log(tag, f"{key:<11}: {entity.get(key)}")
-    for key in ("createdAt", "updatedAt"):
-        if key in entity:
-            log(tag, f"{key:<11}: {timestamp_to_text(entity[key])}")
-    blocks = entity.get("blocks")
-    if blocks:
-        log(tag, "blocks:")
-        for index, block in enumerate(blocks, 1):
-            if isinstance(block, dict):
-                log(
-                    tag,
-                    f"  [{index}] type={block.get('type')} "
-                    f"value={block.get('value')!r}",
-                )
-            else:
-                log(tag, f"  [{index}] {block!r}")
-
-
-def print_push_message(payload: list[Any]) -> None:
-    """push イベントの payload をパースして表示する。
-
-    想定フォーマット:
-      42/desk/channel,["push",{"event":"push","entity":{...},
-                               "type":"message","refers":{...}}]
-    すなわち payload = ["push", {"event":..., "entity":..., "type":..., "refers":...}]
-    """
-    if len(payload) < 2:
-        log_block("PUSH", "payload", json.dumps(payload, ensure_ascii=False, indent=2))
-        return
-
-    event_name = payload[0]
-    wrapper = payload[1]
-    log("PUSH", f"受信イベント: {event_name}")
-
-    if not isinstance(wrapper, dict):
-        log_block(
-            "PUSH",
-            "wrapper",
-            json.dumps(
-                format_json_with_timestamps(wrapper),
-                ensure_ascii=False,
-                indent=2,
-            ),
-        )
-        return
-
-    if "type" in wrapper:
-        log("PUSH", f"type       : {wrapper.get('type')}")
-    if "event" in wrapper:
-        log("PUSH", f"event      : {wrapper.get('event')}")
-
-    entity = wrapper.get("entity", {})
-    if isinstance(entity, dict):
-        _print_message_entity("PUSH", entity)
-
-    refers = wrapper.get("refers")
-    if isinstance(refers, dict):
-        manager = refers.get("manager")
-        if isinstance(manager, dict):
-            log(
-                "PUSH",
-                "refers.manager: "
-                f"id={manager.get('id')} name={manager.get('name')!r}",
+            self._channel_thread = threading.Thread(
+                target=self._channel_worker,
+                args=(tokens["x-account"], self.channel_id, self.manager_id),
+                name="channel-ws",
+                daemon=True,
             )
-        channel = refers.get("channel")
-        if isinstance(channel, dict):
-            log(
-                "PUSH",
-                "refers.channel: "
-                f"id={channel.get('id')} name={channel.get('name')!r}",
+            self._account_thread = threading.Thread(
+                target=self._account_worker,
+                name="account-ws",
+                daemon=True,
             )
-        group = refers.get("group")
-        if isinstance(group, dict):
-            log(
-                "PUSH",
-                "refers.group  : "
-                f"id={group.get('id')} title={group.get('title')!r}",
+            self._channel_thread.start()
+            self._account_thread.start()
+            self._log_debug(
+                "MAIN",
+                f"起動: manager_id={self.manager_id} channel_id={self.channel_id}",
             )
 
-    log_block(
-        "PUSH",
-        "詳細JSON",
-        json.dumps(
-            format_json_with_timestamps(payload),
-            ensure_ascii=False,
-            indent=2,
-        ),
-    )
-
-
-# ---------------------------------------------------------------------------
-# Socket.IO / Engine.IO
-# ---------------------------------------------------------------------------
-def parse_socketio_event(packet: str) -> tuple[str | None, list[Any] | None]:
-    log("PARSE", f"packet={packet!r}")
-    if not packet.startswith("42"):
-        log("PARSE", "-> Socket.IO イベント(42)ではありません。")
-        return None, None
-
-    payload_text = packet[2:]
-    if payload_text.startswith("/"):
+    def wait(self, poll_interval: float = 0.5) -> None:
+        """両ワーカーが終了するまでブロックする。"""
         try:
-            namespace, payload_text = payload_text.split(",", 1)
-        except ValueError:
-            return None, None
-    else:
-        namespace = "/"
+            while self._is_running():
+                time.sleep(poll_interval)
+        except KeyboardInterrupt:
+            self._log_debug("MAIN", "KeyboardInterrupt を受信")
+            raise
+
+    def stop(self, join_timeout: float = 5.0) -> None:
+        """ワーカーを停止する。"""
+        self._stop_event.set()
+        for thread in (self._channel_thread, self._account_thread):
+            if thread and thread.is_alive():
+                thread.join(timeout=join_timeout)
+        self._channel_thread = None
+        self._account_thread = None
+        self._log_debug("MAIN", "ChannelMessenger を停止しました。")
+
+    def is_running(self) -> bool:
+        return self._is_running()
 
-    try:
-        payload = json.loads(payload_text)
-    except json.JSONDecodeError:
-        log("PARSE", f"-> JSON解析失敗: {payload_text!r}")
-        return namespace, None
+    def close(self) -> None:
+        self.stop()
 
-    if not isinstance(payload, list):
-        return namespace, None
-    return namespace, payload
+    def __enter__(self) -> "ChannelMessenger":
+        return self
 
+    def __exit__(self, exc_type, exc, tb) -> None:
+        self.close()
 
-def send_channel_connect(ws: LoggingWS, session_token: str, channel_id: str) -> None:
-    message = (
-        '40/desk/channel,'
-        + json.dumps(
-            {"jwt": session_token, "channelId": channel_id},
-            ensure_ascii=False,
-            separators=(",", ":"),
+    # ------------------------------------------------------------------
+    # 内部: 状態
+    # ------------------------------------------------------------------
+    def _is_running(self) -> bool:
+        return any(
+            t is not None and t.is_alive()
+            for t in (self._channel_thread, self._account_thread)
         )
-    )
-    log("SOCKET", f"channel connect 送信: channelId={channel_id}")
-    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=(",", ":"),
+    # ------------------------------------------------------------------
+    # 内部: HTTP / ID 解決
+    # ------------------------------------------------------------------
+    def _resolve_ids(self, tokens: Dict[str, str]) -> None:
+        room_id = getattr(self.client, "room_id", None)
+        if not room_id:
+            raise RuntimeError("client.room_id が取得できません。")
+
+        session = requests.Session()
+        session.cookies.update(
+            {
+                "ch-session-1": tokens.get("ch-session-1") or "",
+                "ch-veil-id": tokens.get("ch-veil-id") or "",
+                "x-account": tokens.get("x-account") or "",
+                "x-account-refresh": tokens.get("x-account-refresh") or "",
+            }
         )
-    )
-    log("SOCKET", "account connect 送信")
-    ws.send(message)
 
+        room_url = CHANNEL_API_URL_TEMPLATE.format(room_id=room_id)
+
+        if self.manager_id is None:
+            room_json = self._http_get_json(session, room_url, tokens)
+            manager_id = _get_nested(room_json, "manager", "id")
+            if manager_id is None:
+                manager = _find_first_key(room_json, "manager")
+                if isinstance(manager, dict):
+                    manager_id = manager.get("id")
+            if manager_id is None:
+                raise RuntimeError(
+                    f"{room_url} の JSON から manager.id を取得できませんでした。"
+                )
+            self.manager_id = str(manager_id)
 
-def send_refresh(ws: LoggingWS, session_token: str) -> None:
-    """account WS 用の refresh（channelId は送らない）"""
-    message = (
-        '42/desk/account,0'
-        + json.dumps(
-            ["refresh", {"jwt": session_token}],
-            ensure_ascii=False,
-            separators=(",", ":"),
+        if self.channel_id is None:
+            if not self.channel_slug:
+                raise RuntimeError(
+                    "channel_id を自動解決するには channel_slug を指定してください。"
+                )
+            lookup_url = CHANNEL_LOOKUP_URL_TEMPLATE.format(slug=self.channel_slug)
+            channel_json = self._http_get_json(session, lookup_url, tokens)
+            channel_id = _get_nested(channel_json, "channel", "id")
+            if channel_id is None:
+                channel_id = _find_first_key(channel_json, "id")
+            if channel_id is None:
+                raise RuntimeError(
+                    f"{lookup_url} の JSON から channel.id を取得できませんでした。"
+                )
+            self.channel_id = str(channel_id)
+
+    def _http_get_json(
+        self, session: requests.Session, url: str, tokens: Dict[str, str]
+    ) -> Any:
+        self._log_debug("HTTP", f"GET {url}")
+        response = session.get(
+            url,
+            headers={
+                "Accept": "application/json",
+                "Accept-Language": "ja",
+                "Referer": "https://channel.works/",
+                "Origin": "https://channel.works",
+                "User-Agent": (
+                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                    "AppleWebKit/537.36 (KHTML, like Gecko) "
+                    "Chrome/152.0.0.0 Safari/537.36"
+                ),
+                "x-account": tokens["x-account"],
+            },
+            timeout=self.request_timeout,
         )
-    )
-    log("SOCKET", "refresh を送信します（account WS）。")
-    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=(",", ":"),
+        self._log_debug("HTTP", f"-> status={response.status_code} ({url})")
+        response.raise_for_status()
+        try:
+            data = response.json()
+        except ValueError as exc:
+            raise RuntimeError(
+                f"JSON を取得できませんでした: {url}\n"
+                f"status={response.status_code}\n"
+                f"body={response.text[:500]}"
+            ) from exc
+        self._log_block_debug(
+            "HTTP-BODY",
+            url,
+            json.dumps(data, ensure_ascii=False),
         )
-    )
-    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,'
-        + json.dumps(
-            ["heartbeat", manager_id],
-            ensure_ascii=False,
-            separators=(",", ":"),
+        return data
+
+    # ------------------------------------------------------------------
+    # 内部: WebSocket 接続
+    # ------------------------------------------------------------------
+    def _open_channel_ws(self, session_token: str, channel_id: str) -> _LoggingWS:
+        raw = websocket.create_connection(
+            WS_URL,
+            timeout=self.request_timeout,
+            origin="https://desk.channel.io",
+            host="desk-ws.channel.io",
         )
-    )
-    log("SOCKET", f"heartbeat 送信: manager_id={manager_id}")
-    ws.send(message)
-
-
-def send_pong(ws: LoggingWS) -> None:
-    log("SOCKET", "Engine.IO ping(2) を受信 -> pong(3) を返します。")
-    ws.send("3")
-
-
-# ---------------------------------------------------------------------------
-# 各 WS のセットアップ
-# ---------------------------------------------------------------------------
-def open_channel_ws(session_token: str, channel_id: str) -> LoggingWS:
-    raw = websocket.create_connection(
-        WS_URL,
-        timeout=REQUEST_TIMEOUT,
-        origin="https://desk.channel.io",
-        host="desk-ws.channel.io",
-    )
-    raw.settimeout(CHANNEL_RECV_TIMEOUT)
-    log("WS", "Channel WebSocket 接続完了")
-    ws = LoggingWS(raw, label="CHANNEL-WS")
-
-    # Engine.IO open を待ってから connect を送る
-    deadline = time.monotonic() + REQUEST_TIMEOUT
-    connect_sent = False
-    while time.monotonic() < deadline:
-        try:
-            packet = ws.recv()
-        except websocket.WebSocketTimeoutException:
-            continue
-        if packet is None:
-            raise RuntimeError("Channel WebSocket が切断されました。")
-        if packet.startswith("0"):
-            log("SOCKET", "Channel Engine.IO open パケットを受信しました。")
-            send_channel_connect(ws, session_token, channel_id)
-            connect_sent = True
-            continue
-        if packet == "2":
-            send_pong(ws)
-            continue
-        # connect 応答待ち: 40/desk/channel,{"sid":"..."}
-        if connect_sent and packet.startswith("40/desk/channel,"):
-            log("SOCKET", "Channel connect 応答 (sid) を受信しました。")
-            send_join_group(ws, GROUP_PATH)
-            return ws
-        log("SOCKET", f"Channel 接続待機中に受信: {packet!r}")
-    raise TimeoutError("Channel connect 応答を受信できませんでした。")
-
-
-def open_account_ws(session_token: str) -> LoggingWS:
-    raw = websocket.create_connection(
-        ACCOUNT_WS_URL,
-        timeout=REQUEST_TIMEOUT,
-        origin="https://desk.channel.io",
-        host="account-ws.channel.io",
-    )
-    raw.settimeout(ACCOUNT_RECV_TIMEOUT)
-    log("WS", "Account WebSocket 接続完了")
-    ws = LoggingWS(raw, label="ACCOUNT-WS")
-
-    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)
-            return ws
-        if packet == "2":
-            send_pong(ws)
-            continue
-        log("SOCKET", f"Account 接続待機中に受信: {packet!r}")
-    raise TimeoutError("Account Engine.IO open パケットを受信できませんでした。")
-
-
-# ---------------------------------------------------------------------------
-# 各スレッド
-# ---------------------------------------------------------------------------
-def channel_worker(
-    session_token: str,
-    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)
-    except Exception as exc:
-        log("CHANNEL", f"接続失敗: {exc!r}")
-        return
-
-    current_token = session_token
-    last_token_check = time.monotonic()
-
-    ready_received = False
-    # フェーズ: "waiting"（30秒待機中） / "first_sent" / "second_sent"
-    heartbeat_phase = "waiting"
-    phase_started_at = 0.0
-
-    log("SOCKET", "Channel 待機中...")
-    try:
-        while not stop_event.is_set():
-            now = time.monotonic()
-
-            if ready_received:
-                if heartbeat_phase == "waiting":
-                    # 30 秒待ってから 1 回目を送る
-                    if now - phase_started_at >= HEARTBEAT_INTERVAL:
-                        send_heartbeat(ws, manager_id)
-                        heartbeat_phase = "first_sent"
-                        phase_started_at = now
-                elif heartbeat_phase == "first_sent":
-                    # 1 回目の応答が 5 秒以内に来なければ再送
-                    if now - phase_started_at >= HEARTBEAT_RESPONSE_TIMEOUT:
-                        log(
-                            "SOCKET",
-                            f"1回目 heartbeat の応答が "
-                            f"{HEARTBEAT_RESPONSE_TIMEOUT}s 以内に来ません。再送します。",
-                        )
-                        send_heartbeat(ws, manager_id)
-                        phase_started_at = now
-                elif heartbeat_phase == "second_sent":
-                    # 2 回目の応答が 5 秒以内に来なければ再送
-                    if now - phase_started_at >= HEARTBEAT_RESPONSE_TIMEOUT:
-                        log(
-                            "SOCKET",
-                            f"2回目 heartbeat の応答が "
-                            f"{HEARTBEAT_RESPONSE_TIMEOUT}s 以内に来ません。再送します。",
-                        )
-                        send_heartbeat(ws, manager_id)
-                        phase_started_at = 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
+        raw.settimeout(self.channel_recv_timeout)
+        self._log_debug("WS", "Channel WebSocket 接続完了")
+        ws = _LoggingWS(raw, label="CHANNEL-WS", messenger=self)
 
+        deadline = time.monotonic() + self.request_timeout
+        connect_sent = False
+        while time.monotonic() < deadline:
             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) 受信（無視）")
+            if packet.startswith("0"):
+                self._log_debug("SOCKET", "Channel Engine.IO open 受信")
+                self._send_channel_connect(ws, session_token, channel_id)
+                connect_sent = True
                 continue
-
-            namespace, payload = parse_socketio_event(packet)
-            if namespace is None or payload is None:
+            if packet == "2":
+                self._send_pong(ws)
                 continue
+            if connect_sent and packet.startswith("40/desk/channel,"):
+                self._log_debug("SOCKET", "Channel connect 応答 (sid) 受信")
+                self._send_join_group(ws, self.group_path)
+                return ws
+            self._log_debug("SOCKET", f"Channel 接続待機中の受信: {packet!r}")
+        raise TimeoutError("Channel connect 応答を受信できませんでした。")
+
+    def _open_account_ws(self, session_token: str) -> _LoggingWS:
+        raw = websocket.create_connection(
+            ACCOUNT_WS_URL,
+            timeout=self.request_timeout,
+            origin="https://desk.channel.io",
+            host="account-ws.channel.io",
+        )
+        raw.settimeout(self.account_recv_timeout)
+        self._log_debug("WS", "Account WebSocket 接続完了")
+        ws = _LoggingWS(raw, label="ACCOUNT-WS", messenger=self)
 
-            if namespace == "/desk/channel" and payload:
-                event_name = payload[0]
-                log("SOCKET", f"event={event_name}")
-
-                if event_name == "ready":
-                    # ready 受信でサイクル開始（まず 30 秒待機）
-                    ready_received = True
-                    heartbeat_phase = "waiting"
-                    phase_started_at = time.monotonic()
-
-                elif event_name == "heartbeat":
-                    if heartbeat_phase == "first_sent":
-                        # 1 回目の応答 → すぐ 2 回目を送る
-                        log("SOCKET", "1回目の応答を受信 → 2回目を送信します。")
-                        send_heartbeat(ws, manager_id)
-                        heartbeat_phase = "second_sent"
-                        phase_started_at = time.monotonic()
-                    elif heartbeat_phase == "second_sent":
-                        # 2 回目の応答 → 次のサイクルへ（30 秒待機）
-                        log("SOCKET", "2回目の応答を受信 → 次のサイクルまで待機します。")
-                        heartbeat_phase = "waiting"
-                        phase_started_at = time.monotonic()
-
-                elif event_name == "create":
-                    # create はパースせず生パケットをそのままログ出力
-                    log_block("CREATE", "raw packet", packet)
-
-                elif event_name == "push":
-                    # push をパースしてメッセージを表示
-                    print_push_message(payload)
-
-                elif event_name == "expired":
-                    log("CHANNEL", "expired を受信しました。接続を終了します。")
-                    break
-    except Exception as exc:
-        log("CHANNEL", f"エラーで終了: {exc!r}")
-    finally:
-        ws.close()
-
-
-def account_worker(
-    token_holder: dict[str, str],
-    client: ChannelWorksClient,
-    stop_event: threading.Event,
-) -> None:
-    try:
-        ws = open_account_ws(token_holder["token"])
-    except Exception as exc:
-        log("ACCOUNT", f"接続失敗: {exc!r}")
-        return
-
-    last_token_check = time.monotonic()
-    log("SOCKET", "Account 待機中...")
-    try:
-        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 変更検出 -> refresh 送信")
-                    send_refresh(ws, current_session_token)
-                    token_holder["token"] = current_session_token
-                last_token_check = now
-
+        deadline = time.monotonic() + self.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"):
+                self._log_debug("SOCKET", "Account Engine.IO open 受信")
+                self._send_account_connect(ws, session_token)
+                return ws
             if packet == "2":
-                send_pong(ws)
+                self._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: ns={namespace}, payload={payload!r}")
-    except Exception as exc:
-        log("ACCOUNT", f"エラーで終了: {exc!r}")
-    finally:
-        ws.close()
-
-def send_join_group(ws: LoggingWS, group_path: str) -> None:
-    message = (
-        '42/desk/channel,'
-        + json.dumps(
+            self._log_debug("SOCKET", f"Account 接続待機中の受信: {packet!r}")
+        raise TimeoutError("Account Engine.IO open パケットを受信できませんでした。")
+
+    # ------------------------------------------------------------------
+    # 内部: Socket.IO パケット送信
+    # ------------------------------------------------------------------
+    def _send_join_group(self, ws: _LoggingWS, group_path: str) -> None:
+        message = '42/desk/channel,' + json.dumps(
             ["join", group_path],
             ensure_ascii=False,
             separators=(",", ":"),
         )
-    )
-    log("SOCKET", f"join 送信: {group_path}")
-    ws.send(message)
-# ---------------------------------------------------------------------------
-# メイン
-# ---------------------------------------------------------------------------
-def run() -> None:
-    log("MAIN", "起動")
-    room_id = require_env("CW_ROOM_ID") if os.getenv("CW_ROOM_ID") else ROOM_ID
-    email = require_env("CW_EMAIL")
-    password = require_env("CW_PASSWORD")
+        self._log_debug("SOCKET", f"join 送信: {group_path}")
+        ws.send(message)
 
-    log("MAIN", f"room_id={room_id}")
-    log("MAIN", f"WS_URL={WS_URL}")
-    log("MAIN", f"ACCOUNT_WS_URL={ACCOUNT_WS_URL}")
+    def _send_channel_connect(
+        self, ws: _LoggingWS, session_token: str, channel_id: str
+    ) -> None:
+        message = '40/desk/channel,' + json.dumps(
+            {"jwt": session_token, "channelId": channel_id},
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+        self._log_debug("SOCKET", f"channel connect 送信: channelId={channel_id}")
+        ws.send(message)
 
-    client = ChannelWorksClient(
-        room_id=room_id,
-        email=email,
-        password=password,
-        log_level=ChannelWorksClient.LOG_ERROR,
-        touch_interval=600,
-        timeout=REQUEST_TIMEOUT,
-    )
+    def _send_account_connect(self, ws: _LoggingWS, session_token: str) -> None:
+        message = '40/desk/account,' + json.dumps(
+            {"jwt": session_token},
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+        self._log_debug("SOCKET", "account connect 送信")
+        ws.send(message)
 
-    try:
-        client.login()
-        client.start_auto_touch()
-        tokens = client.get_tokens()
-        if not tokens:
-            raise RuntimeError("get_tokens() が空でした。")
-        required = (
-            "ch-session-1",
-            "ch-veil-id",
-            "x-account",
-            "x-account-refresh",
+    def _send_refresh(self, ws: _LoggingWS, session_token: str) -> None:
+        message = '42/desk/account,0' + json.dumps(
+            ["refresh", {"jwt": session_token}],
+            ensure_ascii=False,
+            separators=(",", ":"),
         )
-        missing = [k for k in required if not tokens.get(k)]
-        if missing:
-            raise RuntimeError(
-                "必要なトークンが取得できません: " + ", ".join(missing)
-            )
+        self._log_debug("SOCKET", "refresh 送信（account WS）")
+        ws.send(message)
 
-        manager_id, channel_id = fetch_ids(tokens)
-        token_holder = {"token": tokens["x-account"]}
-
-        stop_event = threading.Event()
-
-        # channel 側を先に起動
-        channel_thread = threading.Thread(
-            target=channel_worker,
-            args=(
-                token_holder["token"],
-                channel_id,
-                manager_id,
-                stop_event,
-                token_holder,
-            ),
-            name="channel-ws",
-            daemon=True,
+    def _send_channel_refresh(
+        self, ws: _LoggingWS, session_token: str, channel_id: str
+    ) -> None:
+        message = '42/desk/channel,3' + json.dumps(
+            ["refresh", {"jwt": session_token, "channelId": channel_id}],
+            ensure_ascii=False,
+            separators=(",", ":"),
         )
-        account_thread = threading.Thread(
-            target=account_worker,
-            args=(token_holder, client, stop_event),
-            name="account-ws",
-            daemon=True,
+        self._log_debug(
+            "SOCKET", f"refresh 送信（channel WS, channelId={channel_id}）"
         )
-        channel_thread.start()
-        account_thread.start()
+        ws.send(message)
 
-        log("MAIN", "両スレッド���動。待機中...")
+    def _send_heartbeat(self, ws: _LoggingWS, manager_id: str) -> None:
+        message = '42/desk/channel,' + json.dumps(
+            ["heartbeat", manager_id],
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+        self._log_debug("SOCKET", f"heartbeat 送信: manager_id={manager_id}")
+        ws.send(message)
+
+    def _send_pong(self, ws: _LoggingWS) -> None:
+        self._log_debug("SOCKET", "Engine.IO ping(2) 受信 -> pong(3) 送信")
+        ws.send("3")
+
+    # ------------------------------------------------------------------
+    # 内部: push 処理
+    # ------------------------------------------------------------------
+    def _handle_push(self, payload: list) -> None:
+        if len(payload) < 2:
+            return
+        wrapper = payload[1]
+        if not isinstance(wrapper, dict):
+            return
+
+        if self.log_level >= self.LOG_DEBUG:
+            self._log_block_debug(
+                "PUSH",
+                f"type={wrapper.get('type')} event={wrapper.get('event')}",
+                json.dumps(
+                    _format_json_with_timestamps(wrapper),
+                    ensure_ascii=False,
+                ),
+            )
+
+        if self.on_push is not None:
+            try:
+                self.on_push(wrapper)
+            except Exception as exc:
+                self._log_error("PUSH", f"on_push コールバックで例外: {exc!r}")
+
+    # ------------------------------------------------------------------
+    # 内部: channel ワーカー
+    # ------------------------------------------------------------------
+    def _channel_worker(
+        self, session_token: str, channel_id: str, manager_id: str
+    ) -> None:
         try:
-            while channel_thread.is_alive() or account_thread.is_alive():
-                time.sleep(0.5)
-        except KeyboardInterrupt:
-            log("MAIN", "KeyboardInterrupt 受信")
+            ws = self._open_channel_ws(session_token, channel_id)
+        except Exception as exc:
+            self._log_error("CHANNEL", f"接続失敗: {exc!r}")
+            return
+
+        current_token = session_token
+        last_token_check = time.monotonic()
+
+        ready_received = False
+        heartbeat_phase = "waiting"  # "waiting" / "first_sent" / "second_sent"
+        phase_started_at = 0.0
+
+        try:
+            while not self._stop_event.is_set():
+                now = time.monotonic()
+
+                if ready_received:
+                    if heartbeat_phase == "waiting":
+                        if now - phase_started_at >= self.heartbeat_interval:
+                            self._send_heartbeat(ws, manager_id)
+                            heartbeat_phase = "first_sent"
+                            phase_started_at = now
+                    elif heartbeat_phase == "first_sent":
+                        if now - phase_started_at >= self.heartbeat_response_timeout:
+                            self._log_debug(
+                                "SOCKET",
+                                f"1回目 heartbeat 応答なし ({self.heartbeat_response_timeout}s) → 再送",
+                            )
+                            self._send_heartbeat(ws, manager_id)
+                            phase_started_at = now
+                    elif heartbeat_phase == "second_sent":
+                        if now - phase_started_at >= self.heartbeat_response_timeout:
+                            self._log_debug(
+                                "SOCKET",
+                                f"2回目 heartbeat 応答なし ({self.heartbeat_response_timeout}s) → 再送",
+                            )
+                            self._send_heartbeat(ws, manager_id)
+                            phase_started_at = now
+
+                if now - last_token_check >= self.token_check_interval:
+                    new_token = self._token_holder.get("token")
+                    if new_token and new_token != current_token:
+                        self._log_debug(
+                            "AUTH", "x-account 変更検出 → Channel refresh 送信"
+                        )
+                        self._send_channel_refresh(ws, new_token, channel_id)
+                        current_token = new_token
+                    last_token_check = now
+
+                try:
+                    packet = ws.recv()
+                except websocket.WebSocketTimeoutException:
+                    continue
+
+                if packet is None:
+                    raise RuntimeError("Channel WebSocket が切断されました。")
+                if packet == "2":
+                    self._send_pong(ws)
+                    continue
+                if packet == "3":
+                    self._log_debug("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]
+                    self._log_debug("SOCKET", f"event={event_name}")
+
+                    if event_name == "ready":
+                        ready_received = True
+                        heartbeat_phase = "waiting"
+                        phase_started_at = time.monotonic()
+                    elif event_name == "heartbeat":
+                        if heartbeat_phase == "first_sent":
+                            self._log_debug(
+                                "SOCKET", "1回目の応答受信 → 2回目を送信"
+                            )
+                            self._send_heartbeat(ws, manager_id)
+                            heartbeat_phase = "second_sent"
+                            phase_started_at = time.monotonic()
+                        elif heartbeat_phase == "second_sent":
+                            self._log_debug(
+                                "SOCKET", "2回目の応答受信 → 次サイクル待機"
+                            )
+                            heartbeat_phase = "waiting"
+                            phase_started_at = time.monotonic()
+                    elif event_name == "create":
+                        self._log_block_debug("CREATE", "raw packet", packet)
+                    elif event_name == "push":
+                        self._handle_push(payload)
+                    elif event_name == "expired":
+                        self._log_error(
+                            "CHANNEL", "expired を受信 → 接続を終了します。"
+                        )
+                        break
+        except Exception as exc:
+            self._log_error("CHANNEL", f"エラーで終了: {exc!r}")
         finally:
-            stop_event.set()
-            channel_thread.join(timeout=5)
-            account_thread.join(timeout=5)
-    finally:
-        client.stop_auto_touch()
-        log("AUTH", "auto_touch 停止")
+            ws.close()
+
+    # ------------------------------------------------------------------
+    # 内部: account ワーカー
+    # ------------------------------------------------------------------
+    def _account_worker(self) -> None:
+        try:
+            ws = self._open_account_ws(self._token_holder["token"] or "")
+        except Exception as exc:
+            self._log_error("ACCOUNT", f"接続失敗: {exc!r}")
+            return
 
+        last_token_check = time.monotonic()
+        try:
+            while not self._stop_event.is_set():
+                now = time.monotonic()
+                if now - last_token_check >= self.token_check_interval:
+                    current_tokens = self.client.get_tokens() or {}
+                    current_session_token = current_tokens.get("x-account")
+                    if (
+                        current_session_token
+                        and current_session_token != self._token_holder["token"]
+                    ):
+                        self._log_debug(
+                            "AUTH", "x-account 変更検出 → refresh 送信（account WS）"
+                        )
+                        self._send_refresh(ws, current_session_token)
+                        self._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":
+                    self._send_pong(ws)
+                    continue
+                if packet == "3":
+                    self._log_debug("SOCKET", "Account Engine.IO pong(3) 受信")
+                    continue
+
+                namespace, payload = _parse_socketio_event(packet)
+                if namespace is None or payload is None:
+                    continue
+                self._log_debug(
+                    "SOCKET", f"Account event: ns={namespace}, payload={payload!r}"
+                )
+        except Exception as exc:
+            self._log_error("ACCOUNT", f"エラーで終了: {exc!r}")
+        finally:
+            ws.close()
 
-if __name__ == "__main__":
-    run()