""" ChannelWorks クライアントライブラリ 使い方: from channel_works import ChannelWorksClient client = ChannelWorksClient( email="you@example.com", password="your-password", log_level=ChannelWorksClient.LOG_DEBUG, ) client.login() # sign_in + boot client.start_auto_touch() # 10分ごとに自動で touch を叩いてトークン更新 tokens = client.get_tokens() # 最新トークンを取得 ... client.stop_auto_touch() log_level: 0 = 何も出力しない 1 = エラー・警告のみ (デフォルト) 2 = デバッグログも出力 """ import json import threading from collections import OrderedDict from urllib.parse import urlencode import requests SIGN_IN_URI = "https://api.channel.works/desk/account/sign-in" BOOT_URI = "https://api.channel.io/front/v8/elastic/plugins/b823a684-f97e-4e4e-bf12-48b70c1939ae/boot" TOUCH_URI = "https://api.channel.works/desk/account/touch" BOOT_URL_TEMPLATE = "https://channel.works/{room_id}/user-chats" BOOT_AVATAR_URL = "https://cf.channel.io/avatar/emoji/bee.f7dc8e.png" DEFAULT_TOUCH_INTERVAL = 600 # 10 分 # Set-Cookie から抜き出す対象トークン _TOKEN_NAMES = ("x-account", "x-account-refresh", "ch-session-1", "ch-veil-id") class ChannelWorksClient: # ログレベル LOG_NONE = 0 LOG_ERROR = 1 LOG_DEBUG = 2 def __init__( self, room_id, email=None, password=None, log_level=LOG_ERROR, touch_interval=DEFAULT_TOUCH_INTERVAL, timeout=30, ): if not room_id: raise ValueError("room_id は必須です。") self.room_id = room_id self.email = email self.password = password self.log_level = log_level self.touch_interval = touch_interval self.timeout = timeout self._lock = threading.RLock() # 最新トークン（この値は常に上書きされる。固定値にはならない） self._x_account = None self._x_account_refresh = None self._ch_session_1 = None self._ch_veil_id = None # boot に必要なアカウント情報 self._account_name = None self._account_email = None self._account_id = None self._member_hash = None # 自動 touch 用スレッド self._touch_thread = None self._touch_stop = threading.Event() # ------------------------------------------------------------------ # ログ # ------------------------------------------------------------------ def _log_debug(self, msg): if self.log_level >= self.LOG_DEBUG: print(f"[DEBUG] {msg}") def _log_error(self, msg): if self.log_level >= self.LOG_ERROR: print(f"[ERROR] {msg}") def set_log_level(self, level): self.log_level = level # ------------------------------------------------------------------ # 内部ユーティリティ # ------------------------------------------------------------------ @staticmethod def _common_headers(): return { "accept": "application/json", "accept-language": "ja", "content-type": "application/json", "priority": "u=1, i", "sec-ch-ua": "\"Chromium\";v=\"152\", \"Not?A_Brand\";v=\"24\", \"Google Chrome\";v=\"152\"", "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": "\"Windows\"", "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-site", "Referer": "https://channel.works/", } @staticmethod def _get_set_cookie_headers(resp): try: headers = resp.raw.headers.getlist("Set-Cookie") if headers: return headers except Exception: pass raw = resp.headers.get("Set-Cookie") return [raw] if raw else [] @staticmethod def _extract_new_tokens(set_cookie_headers): new_tokens = {} for set_cookie in set_cookie_headers: first_part = set_cookie.split(";", 1)[0].strip() if "=" not in first_part: continue name, value = first_part.split("=", 1) name = name.strip() if name in _TOKEN_NAMES and value: new_tokens[name] = value return new_tokens def _build_cookie_header(self): cookies = OrderedDict() if self._ch_session_1: cookies["ch-session-1"] = self._ch_session_1 if self._ch_veil_id: cookies["ch-veil-id"] = self._ch_veil_id if self._x_account: cookies["x-account"] = self._x_account if self._x_account_refresh: cookies["x-account-refresh"] = self._x_account_refresh cookies["x-account-refresh-bound"] = "false" return "; ".join(f"{k}={v}" for k, v in cookies.items()) def _apply_new_tokens(self, new_tokens): """Set-Cookie から取得した新トークンを保持値に反映する。""" with self._lock: if "x-account" in new_tokens: self._x_account = new_tokens["x-account"] if "x-account-refresh" in new_tokens: self._x_account_refresh = new_tokens["x-account-refresh"] if "ch-session-1" in new_tokens: self._ch_session_1 = new_tokens["ch-session-1"] if "ch-veil-id" in new_tokens: self._ch_veil_id = new_tokens["ch-veil-id"] # ------------------------------------------------------------------ # 認証 # ------------------------------------------------------------------ def sign_in(self): if not self.email or not self.password: raise ValueError("email と password を指定してください。") headers = self._common_headers() headers["x-supports-refresh"] = "true" body = json.dumps({"email": self.email, "password": self.password}).encode("utf-8") self._log_debug(f"sign_in POST {SIGN_IN_URI}") resp = requests.post(SIGN_IN_URI, headers=headers, data=body, timeout=self.timeout) if not (200 <= resp.status_code < 300): self._log_error(f"sign_in 失敗 status={resp.status_code} body={resp.text}") return resp set_cookies = self._get_set_cookie_headers(resp) new_tokens = self._extract_new_tokens(set_cookies) self._apply_new_tokens(new_tokens) self._log_debug(f"sign_in 取得トークン: {sorted(new_tokens.keys())}") if not self._x_account or not self._x_account_refresh: self._log_error("sign_in 応答から x-account / x-account-refresh を取得できませんでした。") try: data = resp.json() except ValueError: self._log_error("sign_in 応答の JSON パースに失敗しました。") return resp account_info = data["account"] with self._lock: self._account_name = account_info["name"] self._account_email = account_info["email"] self._account_id = account_info["id"] self._member_hash = data["memberHash"] self._log_debug("sign_in 成功") return resp def boot(self): if not self._account_id or not self._member_hash: self._log_error("boot 前に sign_in を実行してください。") return None profile = json.dumps( { "name": self._account_name, "email": self._account_email, "avatarUrl": BOOT_AVATAR_URL, }, ensure_ascii=False, separators=(",", ":"), ) boot_url_value = BOOT_URL_TEMPLATE.format(room_id=self.room_id) body = urlencode({ "url": boot_url_value, # ← 変更 "profile": profile, "memberId": self._account_id, "memberHash": self._member_hash, }).encode("utf-8") headers = { "accept": "application/json", "accept-language": "ja", "content-type": "application/x-www-form-urlencoded", "priority": "u=1, i", "sec-ch-ua": "\"Chromium\";v=\"152\", \"Not?A_Brand\";v=\"24\", \"Google Chrome\";v=\"152\"", "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": "\"Windows\"", "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "cross-site", "sec-fetch-storage-access": "none", "Referer": "https://channel.works/", "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " # User-Agentは必須 "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/152.0.0.0 Safari/537.36" ), } self._log_debug(f"boot POST {BOOT_URI} memberId={self._account_id}") resp = requests.post(BOOT_URI, headers=headers, data=body, timeout=self.timeout) if not (200 <= resp.status_code < 300): self._log_error(f"boot 失敗 status={resp.status_code} body={resp.text}") return resp # boot 応答で ch-session-1 / ch-veil-id が降ってくる場合に備えて反映 self._apply_new_tokens(self._extract_new_tokens(self._get_set_cookie_headers(resp))) try: data = resp.json() except ValueError: self._log_error("boot 応答の JSON パースに失敗しました。") return resp with self._lock: self._ch_session_1 = data.get("sessionJWT", self._ch_session_1) self._ch_veil_id = data.get("veilId", self._ch_veil_id) self._log_debug("boot 成功") return resp def login(self): """sign_in → boot をまとめて実行する。""" self.sign_in() self.boot() # ------------------------------------------------------------------ # touch # ------------------------------------------------------------------ def touch(self): """最新トークンを使って touch を 1 回叩き、返ってきた新トークンで保持値を更新する。""" with self._lock: if not self._x_account: self._log_error("touch 実行不可: x-account 未取得") return None account = self._x_account cookie_header = self._build_cookie_header() headers = self._common_headers() headers["x-account"] = account headers["Cookie"] = cookie_header body = b'{"refreshOnly":true,"skipSessionExtension":false}' self._log_debug(f"touch POST {TOUCH_URI}") try: resp = requests.post(TOUCH_URI, headers=headers, data=body, timeout=self.timeout) except Exception as e: self._log_error(f"touch 通信例外: {e!r}") return None if not (200 <= resp.status_code < 300): self._log_error(f"touch 失敗 status={resp.status_code} body={resp.text}") return resp new_tokens = self._extract_new_tokens(self._get_set_cookie_headers(resp)) self._apply_new_tokens(new_tokens) self._log_debug(f"touch 成功 更新トークン: {sorted(new_tokens.keys()) or '(更新なし)'}") return resp # ------------------------------------------------------------------ # トークン取得 # ------------------------------------------------------------------ def get_tokens(self): """常に最新のトークンを辞書で返す。""" with self._lock: return { "x-account": self._x_account, "x-account-refresh": self._x_account_refresh, "ch-session-1": self._ch_session_1, "ch-veil-id": self._ch_veil_id, } # ------------------------------------------------------------------ # 自動 touch # ------------------------------------------------------------------ def start_auto_touch(self): """バックグラウンドで touch_interval 秒ごとに touch を実行する。""" with self._lock: if self._touch_thread and self._touch_thread.is_alive(): self._log_debug("auto-touch は既に起動中") return self._touch_stop.clear() self._touch_thread = threading.Thread( target=self._touch_loop, name="ChannelWorksAutoTouch", daemon=True ) self._touch_thread.start() self._log_debug(f"auto-touch 開始 (interval={self.touch_interval}s)") def stop_auto_touch(self): """自動 touch を停止する。""" self._touch_stop.set() thread = self._touch_thread if thread and thread.is_alive(): thread.join(timeout=5) self._touch_thread = None self._log_debug("auto-touch 停止") def _touch_loop(self): while not self._touch_stop.wait(self.touch_interval): try: self.touch() except Exception as e: self._log_error(f"auto-touch 例外: {e!r}") # ------------------------------------------------------------------ # コンテキストマネージャ # ------------------------------------------------------------------ def close(self): self.stop_auto_touch() def __enter__(self): return self def __exit__(self, exc_type, exc, tb): self.close()