Spaces:

izuemon
/

any-env-code

Running

Update v2_gpt.py

0383d66 verified 24 days ago

7.92 kB

# 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()

	# 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()