# main_janken.py import json import os import random import string import threading import time import traceback from datetime import datetime import requests from channel_auth import ChannelWorksClient from channel_msg import ChannelMessenger # --------------------------------------------------------------------------- # 設定 # --------------------------------------------------------------------------- CW_EMAIL = os.environ["CW_EMAIL"] CW_PASSWORD = os.environ["CW_PASSWORD"] ROOM_ID = "241135" TARGET_CHAT_ID = "574798" GROUP_PATH = f"/groups/{TARGET_CHAT_ID}" # じゃんけんの手 HANDS = ["✊", "✌", "✋"] # 手の強さ: ✊=0, ✌=1, ✋=2 (グー<チョキ<パー<グー) HAND_INDEX = {h: i for i, h in enumerate(HANDS)} TRIGGER_WORDS = [ "じゃんけん", "ジャンケン", "じゃんけんぽん", "ジャンケンポン", "じゃんけんぽい", "ジャンケンポイ", "じゃんけんほい", "ジャンケンホイ", ] CHANNEL_SEND_URL = ( f"https://api.channel.works/desk/channels/{ROOM_ID}" f"/groups/{TARGET_CHAT_ID}/messages" ) LOG_LEVEL = ChannelWorksClient.LOG_ERROR # --------------------------------------------------------------------------- # ロガー # --------------------------------------------------------------------------- def log(tag: str, msg: str) -> None: ts = datetime.now().strftime("%m-%d %H:%M:%S.%f")[:-3] print(f"[{ts}] [{tag}] {msg}", flush=True) def _make_request_id() -> str: suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=4)) return f"desk-web-{int(time.time() * 1000)}{suffix}" # --------------------------------------------------------------------------- # じゃんけん判定 # --------------------------------------------------------------------------- def detect_user_hand(text: str) -> str | None: """メッセージ本文からユーザーが出した手(絵文字)を探す。""" for h in HANDS: if h in text: return h return None def has_trigger_word(text: str) -> bool: """じゃんけんを促すキーワードが含まれているか。""" return any(w in text for w in TRIGGER_WORDS) def judge(my_hand: str, user_hand: str) -> str: """ 自分の手と相手の手から勝敗メッセージを返す。 ✊(0) < ✌(1) < ✋(2) < ✊(0) の三すくみ。 (my - user) % 3 == 1 なら自分(ボット)の勝ち (my - user) % 3 == 2 なら相手の勝ち 0 ならあいこ """ diff = (HAND_INDEX[my_hand] - HAND_INDEX[user_hand]) % 3 if diff == 0: return "あいこ" if diff == 2: return "私の勝ち！" return "あなたの勝ち！" def make_reply(user_hand: str) -> str: """ランダムな手を出して勝敗メッセージを生成する。""" my_hand = random.choice(HANDS) result = judge(my_hand, user_hand) return f"{my_hand} {result}" # --------------------------------------------------------------------------- # Channel 送信 # --------------------------------------------------------------------------- def send_channel_message(client: ChannelWorksClient, text: str) -> requests.Response: tokens = client.get_tokens() or {} x_account = tokens.get("x-account") or "" x_account_refresh = tokens.get("x-account-refresh") or "" cookie_parts = [] if x_account: cookie_parts.append(f"x-account={x_account}") if x_account_refresh: cookie_parts.append(f"x-account-refresh={x_account_refresh}") cookie_parts.append("x-account-refresh-bound=false") cookie_header = "; ".join(cookie_parts) headers = { "accept": "application/json", "accept-language": "ja", "content-type": "application/json", "x-account": x_account, "Cookie": cookie_header, "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" ), } body = { "requestId": _make_request_id(), "blocks": [{"type": "text", "value": text}], "buttons": None, "form": None, "webPage": None, "files": None, "customPayload": None, } log("SEND-REQ", f"POST {CHANNEL_SEND_URL} text={text!r}") resp = requests.post( CHANNEL_SEND_URL, headers=headers, data=json.dumps(body, ensure_ascii=False).encode("utf-8"), timeout=30, ) log("SEND-RES", f"status={resp.status_code}") return resp # --------------------------------------------------------------------------- # メイン # --------------------------------------------------------------------------- def main() -> None: log("BOOT", f"room_id={ROOM_ID} / target_chat_id={TARGET_CHAT_ID}") log("AUTH", "ChannelWorksClient 初期化") client = ChannelWorksClient( room_id=ROOM_ID, email=CW_EMAIL, password=CW_PASSWORD, log_level=LOG_LEVEL, ) client.login() log("AUTH", "login() 完了") client.start_auto_touch() log("AUTH", "auto_touch 開始") my_account_id = str(getattr(client, "_account_id", "") or "") log("AUTH", f"my_account_id={my_account_id}") seen_ids: set[str] = set() seen_lock = threading.Lock() # 重複・連投防止用 def on_push(wrapper: dict) -> None: if wrapper.get("event") != "create": return entity = wrapper.get("entity") or {} chat_id = str(entity.get("chatId") or "") person_id = str(entity.get("personId") or "") msg_id = entity.get("id") or "" plain = entity.get("plainText") or "" if chat_id != TARGET_CHAT_ID: return if person_id and my_account_id and person_id == my_account_id: return if msg_id: with seen_lock: if msg_id in seen_ids: return seen_ids.add(msg_id) if len(seen_ids) > 2000: seen_ids.clear() if not plain.strip(): return # 「じゃんけん」系ワード + 手の絵文字が含まれているか判定 if not has_trigger_word(plain): return user_hand = detect_user_hand(plain) if user_hand is None: log("PUSH-SKIP", "手の絵文字なし") return reply = make_reply(user_hand) log("REPLY", f"user={user_hand} -> {reply!r}") try: resp = send_channel_message(client, reply) if not 200 <= resp.status_code < 300: log("SEND-NG", f"status={resp.status_code} body={resp.text[:300]}") except Exception as exc: log("SEND-ERR", f"{exc!r}") traceback.print_exc() log("WS", "ChannelMessenger 初期化") messenger = ChannelMessenger( client=client, channel_id=ROOM_ID, group_path=GROUP_PATH, log_level=LOG_LEVEL, on_push=on_push, ) log("WS", "start() 呼び出し") try: messenger.start() except Exception as exc: log("WS-ERR", f"start() 失敗: {exc!r}") traceback.print_exc() return log("WS", f"start() 完了 running={messenger.is_running()}") def watchdog() -> None: while True: time.sleep(30) if not messenger.is_running(): log("WATCH", "ChannelMessenger が停止しています！") break threading.Thread(target=watchdog, daemon=True).start() try: messenger.wait() except KeyboardInterrupt: log("WS", "KeyboardInterrupt") finally: messenger.stop() client.stop_auto_touch() log("WS", "終了") if __name__ == "__main__": main()