"""Twitch chat collector (IRC 匿名接続、 2026-09-20)。 python scripts/twitch.py [--dry] [--duration SEC] # 単発 ad-hoc (dry = DB に書かず表示だけ) daemon.py からは `check_twitch_live()` (Helix、 .env に TWITCH_CLIENT_ID/SECRET がある時だけ) と `collect_twitch()` を使う。 Helix が無ければ LIVE 検知なしで IRC に常時 JOIN する (匿名 JOIN は認証不要、 オフライン中もチャットは流れるのでそのまま集める)。 受信した本文は YouTube と同じ経路 (mask → normalize → SqliteAppender) に流す。 投稿者名 / channel は保存しない。 Twitch emote は `emotes=` tag の位置情報で本文から除く (= YouTube の sticker 除去に相当)。 """ from __future__ import annotations import json import pathlib import random import socket import ssl import sys import threading import time import urllib.parse import urllib.request from import_youtube_live import SqliteAppender, mask, normalize ROOT = pathlib.Path(__file__).resolve().parent.parent IRC_HOST, IRC_PORT = "irc.chat.twitch.tv", 6697 UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Gecko/20100101 Firefox/120.0" def load_env() -> dict[str, str]: """`.env` (KEY=VALUE、 # コメント) を読む。 python-dotenv は使わない。""" env: dict[str, str] = {} p = ROOT / ".env" if p.exists(): for line in p.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) env[k.strip()] = v.strip().strip('"').strip("'") return env # ---------- Helix (LIVE 検知、 optional) ---------- _token: dict[str, object] = {"value": None, "expires": 0.0} _token_lock = threading.Lock() def helix_configured() -> bool: e = load_env() return bool(e.get("TWITCH_CLIENT_ID") and e.get("TWITCH_CLIENT_SECRET")) def _app_token() -> str | None: e = load_env() cid, sec = e.get("TWITCH_CLIENT_ID"), e.get("TWITCH_CLIENT_SECRET") if not (cid and sec): return None with _token_lock: if _token["value"] and time.time() < float(_token["expires"]) - 60: return str(_token["value"]) data = urllib.parse.urlencode( {"client_id": cid, "client_secret": sec, "grant_type": "client_credentials"} ).encode() req = urllib.request.Request("https://id.twitch.tv/oauth2/token", data=data, headers={"User-Agent": UA}) with urllib.request.urlopen(req, timeout=30) as r: j = json.load(r) _token["value"] = j["access_token"] _token["expires"] = time.time() + float(j.get("expires_in", 3600)) return str(_token["value"]) def check_twitch_live(login: str) -> bool | None: """LIVE なら True、 オフラインなら False。 Helix 未設定なら None (= 不明)。""" tok = _app_token() if tok is None: return None e = load_env() req = urllib.request.Request( "https://api.twitch.tv/helix/streams?" + urllib.parse.urlencode({"user_login": login}), headers={"Client-Id": e["TWITCH_CLIENT_ID"], "Authorization": f"Bearer {tok}", "User-Agent": UA}, ) with urllib.request.urlopen(req, timeout=30) as r: j = json.load(r) return bool(j.get("data")) # ---------- IRC ---------- def _parse_tags(raw: str) -> dict[str, str]: tags: dict[str, str] = {} for kv in raw.split(";"): if "=" in kv: k, v = kv.split("=", 1) tags[k] = v return tags def _strip_emotes(text: str, emotes_tag: str) -> str: """emotes=25:0-4,12-16/1902:6-10 の位置 (code point 単位) を本文から除く。""" if not emotes_tag: return text spans: list[tuple[int, int]] = [] for part in emotes_tag.split("/"): if ":" not in part: continue _, ranges = part.split(":", 1) for r in ranges.split(","): if "-" in r: a, b = r.split("-", 1) try: spans.append((int(a), int(b))) except ValueError: pass if not spans: return text chars = list(text) for a, b in spans: for i in range(a, min(b, len(chars) - 1) + 1): chars[i] = " " return "".join(chars) class IrcClient: """匿名 (justinfan) で 1 channel に JOIN して PRIVMSG を yield する。 PING に応答、 切断時は例外。""" def __init__(self, login: str) -> None: self.login = login.lower().lstrip("#") self.sock: ssl.SSLSocket | None = None self.buf = b"" def connect(self) -> None: raw = socket.create_connection((IRC_HOST, IRC_PORT), timeout=30) self.sock = ssl.create_default_context().wrap_socket(raw, server_hostname=IRC_HOST) self.sock.settimeout(30) # 無音 30 秒ごとに heartbeat を返し、 呼び出し側が stop / deadline を見られるようにする self.idle = 0.0 nick = f"justinfan{random.randint(10000, 99999)}" self._send("CAP REQ :twitch.tv/tags twitch.tv/commands") self._send(f"PASS SCHMOOPIIE") self._send(f"NICK {nick}") self._send(f"JOIN #{self.login}") def _send(self, line: str) -> None: assert self.sock is not None self.sock.sendall((line + "\r\n").encode("utf-8")) def close(self) -> None: if self.sock is not None: try: self.sock.close() except OSError: pass self.sock = None def messages(self): """(ts_ms, body) を yield。 body は emote 除去済み・未 mask。""" assert self.sock is not None while True: try: chunk = self.sock.recv(65536) except (socket.timeout, TimeoutError): self.idle += 30 if self.idle > 360: # Twitch は約 5 分ごとに PING を送る。 それより長く無音なら死んだとみなす raise ConnectionError("idle") yield None, None continue self.idle = 0.0 if not chunk: raise ConnectionError("closed") self.buf += chunk while b"\r\n" in self.buf: line, self.buf = self.buf.split(b"\r\n", 1) text = line.decode("utf-8", errors="replace") if text.startswith("PING"): self._send("PONG" + text[4:]) continue tags: dict[str, str] = {} if text.startswith("@"): tag_raw, text = text[1:].split(" ", 1) tags = _parse_tags(tag_raw) if " PRIVMSG " not in text: if "RECONNECT" in text.split(" ")[:2]: raise ConnectionError("reconnect requested") continue _, _, rest = text.partition(" PRIVMSG ") _, _, body = rest.partition(" :") # /me は ACTION ... で来る if body.startswith("ACTION ") and body.endswith(""): body = body[8:-1] body = _strip_emotes(body, tags.get("emotes", "")) ts = int(tags.get("tmi-sent-ts") or int(time.time() * 1000)) yield ts, body def collect_twitch( login: str, out_dir: pathlib.Path, stop_event: threading.Event | None = None, log_tag: str = "", genre: str | None = None, still_live=None, dry: bool = False, duration_sec: int = 0, ) -> None: """IRC に JOIN して集める。 `still_live()` が False を返したら (Helix で配信終了を検知) 離脱。 Helix が無い時は still_live=None で常駐 (= 切断されたら再接続)。 duration_sec > 0 で時間上限。""" prefix = f"[{log_tag}] " if log_tag else "" db_path = out_dir / "comments.sqlite" writer = None if dry else SqliteAppender.shared(db_path) count = 0 deadline = time.time() + duration_sec if duration_sec > 0 else None last_live_check = time.time() stop_reason = "live_ended" print(f"{prefix}[start] twitch #{login} dry={dry} helix={still_live is not None}", flush=True) try: while True: if stop_event is not None and stop_event.is_set(): stop_reason = "stopped" break if deadline is not None and time.time() >= deadline: stop_reason = "duration_cap" break client = IrcClient(login) try: client.connect() for ts, raw in client.messages(): body = normalize(mask(raw)) if raw is not None else "" if body: if dry: print(f"{prefix} {body}", flush=True) elif writer is not None: writer.write(ts, body, genre) count += 1 if count % 100 == 0 and not dry: print(f"{prefix} seen={count} touched={writer.touched} rejected={writer.rejected}") if stop_event is not None and stop_event.is_set(): break if deadline is not None and time.time() >= deadline: break if still_live is not None and time.time() - last_live_check >= 60: last_live_check = time.time() if still_live() is False: stop_reason = "live_ended" client.close() return except (ConnectionError, OSError, TimeoutError) as e: print(f"{prefix}[irc] {e!r} → reconnect in 10s") client.close() for _ in range(10): if stop_event is not None and stop_event.is_set(): break time.sleep(1) continue finally: client.close() if stop_event is not None and stop_event.is_set(): stop_reason = "stopped" elif deadline is not None and time.time() >= deadline: stop_reason = "duration_cap" break finally: if writer is not None: writer.close() print(f"{prefix}[done] reason={stop_reason} seen={count}") def main() -> int: import argparse ap = argparse.ArgumentParser() ap.add_argument("login") ap.add_argument("--dry", action="store_true") ap.add_argument("--duration", type=int, default=0) ap.add_argument("--out-dir", default=str(ROOT / "data")) ap.add_argument("--genre", default=None) a = ap.parse_args() if helix_configured(): print(f"helix live: {check_twitch_live(a.login)}") collect_twitch(a.login, pathlib.Path(a.out_dir), dry=a.dry, duration_sec=a.duration, genre=a.genre) return 0 if __name__ == "__main__": sys.exit(main())