diff --git a/.env.example b/.env.example
new file mode 100644
index 0000000..27220a6
--- /dev/null
+++ b/.env.example
@@ -0,0 +1,20 @@
+# Copy this file to .env and fill it in: cp .env.example .env
+# .env is listed in .gitignore. Never commit a real token.
+# How to get a token and which scopes it needs: README, "Get a token".
+
+# The channel to join: a Twitch login name, without the #
+TWITCH_CHANNEL=your_channel
+
+# The bot account's login name. It must be the account the token belongs to.
+TWITCH_BOT_NAME=your_bot_account
+
+# A user access token of the bot account, with or without "oauth:" in front.
+# Scopes: chat:read chat:edit (main.py and examples/irc_library_echo.py)
+# user:read:chat user:write:chat (examples/eventsub_echo.py)
+TWITCH_OAUTH_TOKEN=
+
+# Optional. Leave these out to use the real Twitch chat (irc.chat.twitch.tv:6697 with TLS).
+# For the practice server (python tools/fake_twitch_server.py) use:
+# TWITCH_IRC_HOST=127.0.0.1
+# TWITCH_IRC_PORT=6667
+# TWITCH_IRC_TLS=false
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..b194699
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,42 @@
+name: CI
+
+on:
+ push:
+ branches: [master]
+ pull_request:
+ workflow_dispatch:
+
+permissions:
+ contents: read
+
+jobs:
+ test:
+ name: Python ${{ matrix.python }}
+ runs-on: ubuntu-latest
+ timeout-minutes: 10
+ strategy:
+ fail-fast: false
+ matrix:
+ python: ["3.10", "3.11", "3.12", "3.13", "3.14"]
+ steps:
+ - uses: actions/checkout@v7
+
+ - uses: actions/setup-python@v7
+ with:
+ python-version: ${{ matrix.python }}
+ cache: pip
+ cache-dependency-path: |
+ requirements-dev.txt
+ examples/requirements.txt
+
+ - name: The bot needs only the standard library
+ run: python -c "import main, twitch_chat, config, twitchSocket; print('imports fine without any packages')"
+
+ - name: Install test tools
+ run: python -m pip install -r requirements-dev.txt
+
+ - name: Lint
+ run: ruff check .
+
+ - name: Tests (against the fake Twitch server, never the real one)
+ run: pytest -v
diff --git a/.gitignore b/.gitignore
index e4824e8..fd844f3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,7 +1,14 @@
-__pycache__
+# your settings and token (copy .env.example)
+.env
-.DS_Store
+# Python
+__pycache__/
+*.py[cod]
+.venv/
+.pytest_cache/
+.ruff_cache/
+# editors and OS
+.vscode/
+.DS_Store
desktop.ini
-
-.vscode
\ No newline at end of file
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..a230fea
--- /dev/null
+++ b/README.md
@@ -0,0 +1,436 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+A Twitch chat bot that repeats every message back into the channel, written with a plain
+socket and nothing outside Python's standard library, so you can see every line that goes
+over the wire. Read [`twitch_chat.py`](twitch_chat.py) from top to bottom and you will know
+how Twitch chat works: connecting, logging in, joining a channel, reading messages,
+answering PING, staying under the rate limit and reconnecting. No Twitch account yet? A
+small fake Twitch server in this repo lets you try everything on your own computer.
+
+- [Quick start](#quick-start): try it with the fake server, then with real Twitch
+- [How Twitch chat works, step by step](#how-twitch-chat-works-step-by-step)
+- [The same bot with a ready-made package](#the-same-bot-with-a-ready-made-package) (issue #1)
+- [Next step: EventSub, the way Twitch recommends now](#next-step-eventsub-the-way-twitch-recommends-now)
+- [Tests](#tests) · [Files](#files) · [Code from 2019](#code-from-2019)
+
+## Quick start
+
+You need Python 3.10 or newer. The bot itself installs nothing.
+
+```bash
+git clone https://github.com/NatLee/twitch-echo-bot-with-python.git
+cd twitch-echo-bot-with-python
+```
+
+### 1. Try it without a Twitch account
+
+Open two terminals in the repo folder. In the first one, start the fake Twitch chat:
+
+```bash
+python tools/fake_twitch_server.py
+```
+
+In the second one, start the bot and point it at the fake server. The fake server takes any
+token that starts with `oauth:`.
+
+```bash
+# macOS / Linux / Git Bash
+TWITCH_IRC_HOST=127.0.0.1 TWITCH_IRC_PORT=6667 TWITCH_IRC_TLS=false \
+TWITCH_CHANNEL=demo TWITCH_BOT_NAME=echo_bot TWITCH_OAUTH_TOKEN=oauth:anything \
+python main.py
+```
+
+```powershell
+# Windows PowerShell
+$env:TWITCH_IRC_HOST="127.0.0.1"; $env:TWITCH_IRC_PORT="6667"; $env:TWITCH_IRC_TLS="false"
+$env:TWITCH_CHANNEL="demo"; $env:TWITCH_BOT_NAME="echo_bot"; $env:TWITCH_OAUTH_TOKEN="oauth:anything"
+python main.py
+```
+
+Now type `alice: hi bot!` in the first terminal. You are alice, a viewer; the bot answers.
+Type `/ping` to see the bot answer a PING, and `/reconnect` to watch it reconnect.
+
+
+
+
+
+Add `--debug` to see every raw IRC line the bot sends (`>`) and receives (`<`). The token
+is never printed. From a second practice run (the final `bye` left out):
+
+```text
+23:17:07 INFO connecting to 127.0.0.1:6667 (plain text)
+23:17:07 DEBUG > CAP REQ :twitch.tv/tags twitch.tv/commands
+23:17:07 DEBUG > PASS ***
+23:17:07 DEBUG > NICK echo_bot
+23:17:07 DEBUG < :tmi.twitch.tv CAP * ACK :twitch.tv/tags twitch.tv/commands
+23:17:07 DEBUG < :tmi.twitch.tv 001 echo_bot :Welcome, GLHF!
+23:17:07 DEBUG > JOIN #demo
+(… 7 lines left out: 002 to 376 and GLOBALUSERSTATE …)
+23:17:08 DEBUG < :echo_bot!echo_bot@echo_bot.tmi.twitch.tv JOIN #demo
+23:17:08 DEBUG < :echo_bot.tmi.twitch.tv 353 echo_bot = #demo :echo_bot
+23:17:08 DEBUG < :echo_bot.tmi.twitch.tv 366 echo_bot #demo :End of /NAMES list
+23:17:08 INFO joined #demo as echo_bot
+23:17:08 DEBUG < @badge-info=;badges=;color=;display-name=echo_bot;emote-sets=0;mod=0;subscriber=0;user-type= :tmi.twitch.tv USERSTATE #demo
+23:17:08 DEBUG < @room-id=1000;slow=0 :tmi.twitch.tv ROOMSTATE #demo
+23:17:08 DEBUG < @badge-info=;badges=;color=#1E90FF;display-name=Alice;emotes=;first-msg=0;flags=;id=00000000-0000-0000-0000-000000000001;mod=0;room-id=1000;subscriber=0;tmi-sent-ts=1793747827973;turbo=0;user-id=2510;user-type= :alice!alice@alice.tmi.twitch.tv PRIVMSG #demo :hi bot!
+#demo Alice: hi bot!
+23:17:08 DEBUG > PRIVMSG #demo :hi bot!
+#demo echo_bot (bot): hi bot!
+23:17:08 DEBUG < @badge-info=;badges=;color=;display-name=echo_bot;emote-sets=0;mod=0;subscriber=0;user-type= :tmi.twitch.tv USERSTATE #demo
+23:17:08 DEBUG < PING :tmi.twitch.tv
+23:17:08 DEBUG > PONG :tmi.twitch.tv
+```
+
+### 2. Get a token
+
+Twitch chat needs a **user access token** for the bot's account. For this bot it needs two
+scopes (permissions): `chat:read` to read chat and `chat:edit` to send messages. If you also
+want to try the [EventSub example](#next-step-eventsub-the-way-twitch-recommends-now), add
+`user:read:chat` and `user:write:chat`.
+
+One way that uses only Twitch's own tools:
+
+1. Log in to the [Twitch Developer Console](https://dev.twitch.tv/console/apps) with the bot
+ account and register an application. Set the OAuth Redirect URL to
+ `http://localhost:3000`. Note the Client ID and create a Client Secret.
+2. Install the [Twitch CLI](https://dev.twitch.tv/docs/cli/) and give it the two values:
+ `twitch configure`.
+3. Ask for a token. A browser window opens; log in as the bot and allow access.
+
+ ```bash
+ twitch token -u -s 'chat:read chat:edit user:read:chat user:write:chat'
+ ```
+
+The command prints a User Access Token. That is what goes in `TWITCH_OAUTH_TOKEN`; the
+`oauth:` in front is optional. User tokens expire, so run the command again when the bot
+reports `Login authentication failed`.
+
+Treat the token like a password. Anyone who has it can chat as your bot.
+
+### 3. Run it on Twitch
+
+Copy the settings template and fill it in:
+
+```bash
+cp .env.example .env
+```
+
+```ini
+TWITCH_CHANNEL=your_channel # the channel to join, without #
+TWITCH_BOT_NAME=your_bot_account # the account the token belongs to
+TWITCH_OAUTH_TOKEN=... # from step 2
+```
+
+`.env` is in `.gitignore`, so git will not pick it up. Settings already set in your shell
+win over the file. Then:
+
+```bash
+python main.py
+```
+
+The bot connects to `irc.chat.twitch.tv:6697` with TLS, joins your channel and repeats
+whatever people say. Stop it with Ctrl+C. To build your own bot, change the line marked
+`<- change this line` in [`main.py`](main.py).
+
+> [!NOTE]
+> Everything in this repo was tested against the fake server and in CI, not against the real
+> Twitch: we had no Twitch account token to test with. The fake server answers with the
+> lines Twitch's documentation shows, but it is not Twitch. If something behaves differently
+> on Twitch, please open an issue.
+
+## How Twitch chat works, step by step
+
+Twitch chat is an [IRC](https://en.wikipedia.org/wiki/IRC) server. IRC is a text protocol
+from 1988: every command is one line of text that ends with `\r\n`. A bot opens a
+connection, sends a few such lines to log in and join a channel, and from then on reads
+lines from Twitch and answers some of them. The steps below match the `STEP` comments in
+[`twitch_chat.py`](twitch_chat.py). Twitch's own description is in
+[IRC Concepts](https://dev.twitch.tv/docs/chat/irc/) (read on 2026-11-04).
+
+### Steps 1 to 4: connect, ask for extras, log in, join
+
+```mermaid
+sequenceDiagram
+ autonumber
+ participant Bot as Your bot (twitch_chat.py)
+ participant Twitch as Twitch chat (irc.chat.twitch.tv)
+ Bot->>Twitch: open a TCP connection to port 6697 and start TLS
+ Bot->>Twitch: CAP REQ :twitch.tv/tags twitch.tv/commands
+ Bot->>Twitch: PASS oauth:YOUR_TOKEN
+ Bot->>Twitch: NICK echo_bot
+ Twitch-->>Bot: :tmi.twitch.tv CAP * ACK :twitch.tv/tags twitch.tv/commands
+ alt the token is right
+ Twitch-->>Bot: :tmi.twitch.tv 001 echo_bot :Welcome, GLHF!
+ Note over Twitch: then 002, 003, 004, 375, 372, 376 and GLOBALUSERSTATE
+ else the token is wrong or expired
+ Twitch-->>Bot: :tmi.twitch.tv NOTICE * :Login authentication failed
+ Note over Twitch: hangs up
+ end
+ Bot->>Twitch: JOIN #35;demo
+ Twitch-->>Bot: JOIN #35;demo, then 353 (who is here)
+ Twitch-->>Bot: 366 echo_bot #35;demo :End of /NAMES list
+ Note over Bot: connect() returns: we are in the channel
+```
+
+1. **Connect.** `socket.create_connection` opens the connection and `ssl` wraps it in TLS,
+ so nobody on the way can read your token. Port 6697 is TLS; 6667 is the same chat
+ without encryption (the 2019 version used that). Twitch also offers
+ `wss://irc-ws.chat.twitch.tv:443` for WebSocket clients such as web pages.
+2. **Ask for extras.** `CAP REQ` asks for Twitch's additions to IRC. `twitch.tv/tags`
+ adds details to each message, such as the viewer's display name and user ID. Twitch only
+ gives tags to bots that also ask for `twitch.tv/commands`.
+3. **Log in.** `PASS` carries the token with `oauth:` in front, then `NICK` gives the
+ bot's login name in lowercase. The order matters: Twitch rejects `NICK` before `PASS`.
+ Success is a burst of numbered lines that starts with `001`.
+4. **Join.** `JOIN #channel`. Twitch confirms with `366 … :End of /NAMES list`. For a
+ channel that does not exist, Twitch says nothing at all, so the bot gives up after 15
+ seconds of silence (`HANDSHAKE_TIMEOUT`) and tells you to check the name.
+
+### Step 5: TCP is a stream, not a list of messages
+
+`recv(4096)` returns whatever bytes have arrived: maybe half a line, maybe five lines.
+Twitch's docs even warn that what the server sends "may contain a single message or
+multiple messages". So the bot keeps a buffer (`LineBuffer`), cuts it at `\r\n` and only
+hands out complete lines.
+It cuts the bytes first and decodes each finished line afterwards, because a character
+like 你 is three bytes in UTF-8 and can be split between two packets.
+
+### Step 6: the four parts of a line
+
+
+
+
+
+
+
+
+`parse_line()` splits a line into **tags** (only there because we asked in step 2),
+**prefix** (who sent it), **command** and **params**. The rule that matters: the first
+`" :"` starts the last parameter, and everything after it belongs to that parameter. That
+is why you must never cut a line with `split(":")`: chat text is full of colons, and every
+link has one.
+
+Tag values cannot contain spaces or `;`, so Twitch writes them as escapes from the
+[IRCv3 message tags](https://ircv3.net/specs/extensions/message-tags) spec: `\s` is a
+space, `\:` is `;`, `\\` is a backslash. `unescape_tag_value()` turns them back.
+
+### Step 7: PING and PONG
+
+```mermaid
+sequenceDiagram
+ participant Twitch as Twitch chat
+ participant Bot as Your bot
+ Twitch-->>Bot: PING :tmi.twitch.tv
+ Bot->>Twitch: PONG :tmi.twitch.tv
+ Note over Twitch,Bot: no PONG? Twitch closes the connection
+```
+
+From time to time (Twitch does not say how often) Twitch checks that the bot is still
+there. The bot answers `PONG` with the same text the `PING` carried. `read_message()` does
+this for you and never returns a PING as a chat message. Only a line whose *command* is `PING` counts; a viewer who types
+"PING :tmi.twitch.tv" in chat is just chatting.
+
+### Step 8: sending, and the rate limit
+
+```mermaid
+sequenceDiagram
+ participant Alice as alice (a viewer)
+ participant Twitch as Twitch chat
+ participant Bot as Your bot
+ Alice->>Twitch: types "hi bot!"
+ Twitch-->>Bot: @…display-name=Alice… :alice!alice@alice.tmi.twitch.tv PRIVMSG #35;demo :hi bot!
+ Note over Bot: read_message() gives user "alice", text "hi bot!"
+ Bot->>Twitch: PRIVMSG #35;demo :hi bot!
+ Twitch-->>Alice: echo_bot: hi bot!
+ Twitch-->>Bot: USERSTATE #35;demo (Twitch never sends your own message back)
+```
+
+`send_message()` sends `PRIVMSG #channel :text`. A line break in the text would end the
+command and start a new one, so the bot turns line breaks into spaces first.
+
+Twitch limits how fast a bot may talk. For a bot that is not the broadcaster, a moderator
+or a VIP, the [rate limits](https://dev.twitch.tv/docs/chat/#rate-limits) are 20 messages
+in any 30 seconds and 1 message per second in a channel. Go over and Twitch ignores the
+bot's messages for an hour. `RateLimiter` remembers recent sends and waits before the next
+one would break a rule. The numbers are in `DEFAULT_RATE_LIMITS`; if your bot is a moderator
+in the channel, Twitch allows 100 per 30 seconds and you can pass your own `rate_limits`.
+Twitch also drops a message that is identical to the one you sent less than 30 seconds
+ago; the bot logs Twitch's `NOTICE` when that happens.
+
+### Step 9: when the connection ends
+
+Connections end. Twitch restarts servers and sends `RECONNECT` first, networks drop, and
+when Twitch closes the connection `recv()` returns empty bytes. In all these cases
+`TwitchChat` raises `ConnectionClosed`. [`main.py`](main.py) catches it, waits 1, 2, 4, 8,
+16, then 30 seconds between tries (`RECONNECT_DELAYS`), connects again and joins the channel
+again. A wrong token or channel name is different: retrying will not help, so `main.py`
+stops and tells you what Twitch said.
+
+## The same bot with a ready-made package
+
+[Issue #1](https://github.com/NatLee/twitch-echo-bot-with-python/issues/1) asked why not use
+an existing IRC package. For a real bot, you should.
+[`examples/irc_library_echo.py`](examples/irc_library_echo.py) is the same echo bot on the
+[`irc`](https://pypi.org/project/irc/) package, about 50 lines of logic:
+
+```bash
+pip install -r examples/requirements.txt
+python -m examples.irc_library_echo
+```
+
+It reads the same `.env`. Compare it with `twitch_chat.py`: the package opens the socket,
+splits lines, parses them, answers PING and reconnects (steps 1, 5, 6, 7 and 9). You still
+write the Twitch parts yourself: the `oauth:` password, `CAP REQ`, `JOIN`, the
+`display-name` tag and the rate limit.
+
+This repo keeps a raw socket on purpose. It is a lesson, and the package hides exactly the
+parts you came to see. The bugs fixed in the 2019 code (next section) were all in those
+hidden parts. Once you have seen them, using a package is the right call. For a bigger bot,
+also look at [TwitchIO](https://pypi.org/project/twitchio/), which speaks EventSub and the
+Twitch API for you.
+
+## Next step: EventSub, the way Twitch recommends now
+
+IRC still works, but Twitch's [chat documentation](https://dev.twitch.tv/docs/chat/) now
+recommends EventSub for reading chat and the Send Chat Message API for sending. Its
+[IRC page](https://dev.twitch.tv/docs/chat/irc/) says IRC "has some limitations versus
+EventSub, and is more complicated to parse". EventSub means Twitch
+pushes events to you as JSON over a WebSocket. Twitch has not announced an end date for IRC
+(checked 2026-11-04).
+
+[`examples/eventsub_echo.py`](examples/eventsub_echo.py) is the echo bot built that way,
+following Twitch's [example chatbot](https://dev.twitch.tv/docs/chat/chatbot-guide/):
+
+```mermaid
+sequenceDiagram
+ autonumber
+ participant Bot as Your bot
+ participant ID as id.twitch.tv
+ participant API as api.twitch.tv/helix
+ participant WS as EventSub WebSocket
+ Bot->>ID: GET /oauth2/validate (who am I, which scopes?)
+ Bot->>API: GET /users?login=demo (channel name to user id)
+ Bot->>WS: connect wss://eventsub.wss.twitch.tv/ws
+ WS-->>Bot: session_welcome (session id)
+ Bot->>API: POST /eventsub/subscriptions, type channel.chat.message (within 10 s)
+ WS-->>Bot: notification: alice said "hi bot!"
+ Bot->>API: POST /chat/messages "hi bot!"
+ WS-->>Bot: notification: echo_bot said "hi bot!" (your own message, skipped)
+```
+
+```bash
+pip install -r examples/requirements.txt
+python -m examples.eventsub_echo
+```
+
+What changes compared with IRC:
+
+| | IRC (`main.py`) | EventSub (`examples/eventsub_echo.py`) |
+|---|---|---|
+| Scopes | `chat:read`, `chat:edit` | `user:read:chat`, `user:write:chat` |
+| Reading | `PRIVMSG` lines on one socket | JSON `notification` messages on a WebSocket |
+| Sending | `PRIVMSG` on the same socket | `POST /helix/chat/messages` |
+| Still alive? | answer `PING` with `PONG` | expect a message every `keepalive_timeout_seconds` |
+| Your own messages | never sent back to you | sent back: skip them or the bot echoes itself forever |
+| Moving servers | `RECONNECT`: connect again | `session_reconnect`: connect to the new URL, the subscription moves along |
+
+The example closes the old WebSocket before it opens the new one after `session_reconnect`.
+Twitch suggests keeping both open until the new one says welcome, so a message sent in
+that moment can be lost.
+
+## Tests
+
+The tests start a fake Twitch server on a free local port and never touch the real Twitch.
+They cover each step above, the 2019 bugs below, the old `twitchSocket` calls, `main.py`
+as a real process, and both examples (the EventSub one against fakes of the Twitch API and
+EventSub WebSocket).
+
+```bash
+pip install -r requirements-dev.txt
+pytest
+ruff check .
+```
+
+No Python on your machine? Docker works too:
+
+```bash
+docker run --rm -v "$PWD:/src:ro" python:3.12-slim sh -c \
+ 'cp -r /src /w && cd /w && pip install -q -r requirements-dev.txt && pytest'
+```
+
+GitHub Actions runs the same on Python 3.10 to 3.14 for every push and pull request.
+
+## Files
+
+| File | What it is |
+|---|---|
+| [`twitch_chat.py`](twitch_chat.py) | the lesson: the IRC client in 9 steps, standard library only |
+| [`main.py`](main.py) | the echo bot: settings, the echo loop, reconnecting |
+| [`config.py`](config.py) | reads settings from the environment and `.env` |
+| [`.env.example`](.env.example) | template for your `.env` |
+| [`tools/fake_twitch_server.py`](tools/fake_twitch_server.py) | the fake Twitch chat, for practice and tests |
+| [`examples/`](examples/) | the same bot on the `irc` package, and on EventSub |
+| [`twitchSocket.py`](twitchSocket.py) | the 2019 names, kept so old scripts keep working |
+| [`tests/`](tests/) | pytest tests |
+| [`docs/images/src/`](docs/images/src/) | HTML sources of the pictures in this README |
+
+## Code from 2019
+
+Scripts written for the 2019 version still work, unchanged:
+
+```python
+from twitchSocket import twitchSocket
+
+ts = twitchSocket(channelName="your_channel", botName="your_bot", botAuth="oauth:...")
+while True:
+ username, message = ts.getMsg()
+ ts.sendMsg(message)
+```
+
+`twitchSocket` now runs on `twitch_chat.TwitchChat` and prints a `DeprecationWarning`
+pointing there. It stays until at least 2027-11-04; [`twitchSocket.py`](twitchSocket.py)
+says what to check before removing it. The new names:
+
+| 2019 | now |
+|---|---|
+| `twitchSocket(channelName=, botName=, botAuth=)` | `TwitchChat(channel=, nick=, token=)`, then `connect()` |
+| `getMsg()` returns `(username, message)` | `read_message()` returns a `ChatMessage` (`.user`, `.text`, `.display_name`, `.tags`) |
+| `sendMsg(message)` | `send_message(text)` |
+
+### What the 2019 version got wrong
+
+All of these were reproduced against the fake server before the rewrite, and each has a
+test now.
+
+- **It crashed within minutes.** A PING made `getMsg()` return `(None, None)`, and
+ `sendMsg(None)` then raised `TypeError`, which the loop did not catch.
+- **Colons cut messages short.** `split(":")` turned `look at this: https://example.com`
+ into `look at this`.
+- **Two lines in one packet lost the second.** One `recv()` was treated as one line. When a
+ PING and a chat line arrived together, the bot reported a viewer called `tmi.twitch.tv`
+ and echoed IRC protocol text.
+- **A character split between packets crashed it.** Decoding each `recv()` on its own
+ failed on half a 你, and the error handler then failed too.
+- **It never noticed a closed connection.** `recv()` returned empty bytes forever and the
+ loop spun at full CPU. A wrong token or a misspelled channel hung the constructor.
+- **It sent the token in plain text** to the old host `irc.twitch.tv:6667`, and the token
+ lived in `main.py`, one `git commit` away from being public.
+- **It answered PONG to chat text** that contained `PING :tmi.twitch.tv`, always with a
+ hard-coded reply.
+- **No rate limit**, so a busy chat could get the bot ignored for an hour.
+
+## License
+
+[MIT](./LICENSE)
diff --git a/config.py b/config.py
new file mode 100644
index 0000000..c68853c
--- /dev/null
+++ b/config.py
@@ -0,0 +1,74 @@
+"""Settings for the bots in this repo, read from environment variables.
+
+Keep the token out of your code: put it in a file called .env next to this one
+(copy .env.example and fill it in). .env is in .gitignore, so git will not pick it up.
+Variables that are already set in your shell win over the file.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Mapping, MutableMapping
+from dataclasses import dataclass
+from pathlib import Path
+
+from twitch_chat import TWITCH_IRC_HOST
+
+ENV_FILE = Path(__file__).with_name(".env")
+
+# name of the environment variable -> what goes in it
+REQUIRED = {
+ "TWITCH_CHANNEL": "the channel to join, e.g. your own login name",
+ "TWITCH_BOT_NAME": "the login name of the bot account",
+ "TWITCH_OAUTH_TOKEN": "a user access token of the bot account (see README: Get a token)",
+}
+FALSE_WORDS = {"0", "false", "no", "off"}
+
+
+class ConfigError(Exception):
+ """A required setting is missing or has a bad value."""
+
+
+@dataclass(frozen=True)
+class Settings:
+ channel: str
+ bot_name: str
+ token: str
+ host: str = TWITCH_IRC_HOST
+ port: int | None = None # None: the usual port for TLS or plain text
+ use_tls: bool = True
+
+ @classmethod
+ def from_env(cls, environ: Mapping[str, str] = os.environ) -> Settings:
+ missing = [name for name in REQUIRED if not environ.get(name, "").strip()]
+ if missing:
+ lines = [f" {name}: {REQUIRED[name]}" for name in missing]
+ raise ConfigError("missing settings (put them in .env, see .env.example):\n" + "\n".join(lines))
+ port = environ.get("TWITCH_IRC_PORT", "").strip()
+ if port and not port.isdigit():
+ raise ConfigError(f"TWITCH_IRC_PORT must be a number, got {port!r}")
+ return cls(
+ channel=environ["TWITCH_CHANNEL"].strip(),
+ bot_name=environ["TWITCH_BOT_NAME"].strip(),
+ token=environ["TWITCH_OAUTH_TOKEN"].strip(),
+ host=environ.get("TWITCH_IRC_HOST", "").strip() or TWITCH_IRC_HOST,
+ port=int(port) if port else None,
+ use_tls=environ.get("TWITCH_IRC_TLS", "true").strip().lower() not in FALSE_WORDS,
+ )
+
+
+def load_env_file(path: Path = ENV_FILE, environ: MutableMapping[str, str] = os.environ) -> None:
+ """Copy KEY=value lines from ``path`` into ``environ``, without overwriting.
+
+ A small stand-in for the python-dotenv package, so the bot needs nothing
+ outside the standard library. Blank lines and lines starting with # are skipped;
+ quotes around a value are removed.
+ """
+ if not path.is_file():
+ return
+ for raw in path.read_text(encoding="utf-8").splitlines():
+ line = raw.strip()
+ if not line or line.startswith("#") or "=" not in line:
+ continue
+ key, _, value = line.partition("=")
+ environ.setdefault(key.strip(), value.strip().strip("'\""))
diff --git a/docs/images/hero-dark.png b/docs/images/hero-dark.png
new file mode 100644
index 0000000..0cebf67
Binary files /dev/null and b/docs/images/hero-dark.png differ
diff --git a/docs/images/hero-light.png b/docs/images/hero-light.png
new file mode 100644
index 0000000..21d5ffb
Binary files /dev/null and b/docs/images/hero-light.png differ
diff --git a/docs/images/message-anatomy-dark.png b/docs/images/message-anatomy-dark.png
new file mode 100644
index 0000000..a325556
Binary files /dev/null and b/docs/images/message-anatomy-dark.png differ
diff --git a/docs/images/message-anatomy-light.png b/docs/images/message-anatomy-light.png
new file mode 100644
index 0000000..2995731
Binary files /dev/null and b/docs/images/message-anatomy-light.png differ
diff --git a/docs/images/src/hero.html b/docs/images/src/hero.html
new file mode 100644
index 0000000..25aa699
--- /dev/null
+++ b/docs/images/src/hero.html
@@ -0,0 +1,157 @@
+
+
+
+
+
+twitch-echo-bot-with-python cover
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
twitch-echo-bot-with-python
+
Learn how Twitch chat works from one small file: a raw socket, nothing else.
Practice run, not the real Twitch. Recorded on 2026-11-04 against tools/fake_twitch_server.py on 127.0.0.1:6667, with a dummy token. The bot runs the same code it runs against Twitch, except for TLS.
+
Terminal 1 · the fake Twitch chat
+
$ python tools/fake_twitch_server.py # chat lines piped in: "alice: hi bot!", "/ping", "Bob: …"
+[server] fake Twitch chat listening on 127.0.0.1:6667 (plain TCP, not the real Twitch)
+[server] point the bot here, for example with these lines in .env:
+ TWITCH_IRC_HOST=127.0.0.1 TWITCH_IRC_PORT=6667 TWITCH_IRC_TLS=false
+ TWITCH_CHANNEL=demo TWITCH_BOT_NAME=any_name TWITCH_OAUTH_TOKEN=oauth:anything
+Type a chat line as name: message (example: alice: hello there)
+Commands: /ping /reconnect /quit
+Lines you type before a bot joins wait until one does.
+[server] bot #1 logged in as echo_bot
+[server] bot #1 joined #demo
+[chat] alice: hi bot!
+[server] echo_bot -> #demo: hi bot!
+[server] PING sent
+[server] bot #1 answered: PONG :tmi.twitch.tv
+[chat] Bob: 冒號也沒問題: https://dev.twitch.tv
+[server] echo_bot -> #demo: 冒號也沒問題: https://dev.twitch.tv
+[server] bot #1 disconnected
+
Terminal 2 · the bot
+
$ TWITCH_IRC_HOST=127.0.0.1 TWITCH_IRC_PORT=6667 TWITCH_IRC_TLS=false python main.py # also set: TWITCH_CHANNEL=demo TWITCH_BOT_NAME=echo_bot TWITCH_OAUTH_TOKEN=oauth:not-a-real-token
+23:17:02 INFO connecting to 127.0.0.1:6667 (plain text)
+23:17:02 INFO joined #demo as echo_bot
+#demo alice: hi bot!
+#demo echo_bot (bot): hi bot!
+#demo Bob: 冒號也沒問題: https://dev.twitch.tv
+#demo echo_bot (bot): 冒號也沒問題: https://dev.twitch.tv
+bye
+
diff --git a/docs/images/terminal-demo.png b/docs/images/terminal-demo.png
new file mode 100644
index 0000000..cc00c21
Binary files /dev/null and b/docs/images/terminal-demo.png differ
diff --git a/examples/eventsub_echo.py b/examples/eventsub_echo.py
new file mode 100644
index 0000000..06c91fb
--- /dev/null
+++ b/examples/eventsub_echo.py
@@ -0,0 +1,225 @@
+"""Next step: the echo bot on the stack Twitch recommends today.
+
+Twitch's docs now say: read chat with EventSub (Twitch pushes events to you over a
+WebSocket) and send with the Send Chat Message API, instead of IRC
+(https://dev.twitch.tv/docs/chat/). This is the same bot built that way. It follows
+Twitch's own example chatbot (https://dev.twitch.tv/docs/chat/chatbot-guide/):
+
+ 1. check the token and learn who we are ........ GET id.twitch.tv/oauth2/validate
+ 2. look up the channel's user id ................ GET /helix/users?login=...
+ 3. open the EventSub WebSocket, read "welcome" .. wss://eventsub.wss.twitch.tv/ws
+ 4. subscribe to chat within 10 seconds .......... POST /helix/eventsub/subscriptions
+ 5. for every chat message, send it back ......... POST /helix/chat/messages
+
+Run it from the repo root with the same .env as main.py. The token needs the scopes
+user:read:chat and user:write:chat (IRC used chat:read and chat:edit instead):
+
+ pip install -r examples/requirements.txt
+ python -m examples.eventsub_echo
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from websockets.exceptions import ConnectionClosed, WebSocketException
+from websockets.sync.client import connect
+
+from config import ConfigError, Settings, load_env_file
+from twitch_chat import DEFAULT_RATE_LIMITS, RateLimiter, reconnect_delay
+
+log = logging.getLogger("eventsub_echo")
+
+TWITCH_ID_URL = "https://id.twitch.tv/oauth2"
+TWITCH_API_URL = "https://api.twitch.tv/helix"
+EVENTSUB_WS_URL = "wss://eventsub.wss.twitch.tv/ws"
+REQUIRED_SCOPES = ("user:read:chat", "user:write:chat")
+WELCOME_TIMEOUT = 10.0 # Twitch closes the WebSocket if we have not subscribed 10 s after "welcome"
+KEEPALIVE_MARGIN = 5.0 # extra seconds on top of Twitch's keepalive_timeout_seconds
+HTTP_TIMEOUT = 15.0
+
+
+class TwitchAPIError(Exception):
+ pass
+
+
+@dataclass(frozen=True)
+class Endpoints:
+ """Where the three Twitch services live. Tests point these at fake servers."""
+
+ id_url: str = TWITCH_ID_URL
+ api_url: str = TWITCH_API_URL
+ eventsub_url: str = EVENTSUB_WS_URL
+
+ @classmethod
+ def from_env(cls, environ: Mapping[str, str] = os.environ) -> Endpoints:
+ return cls(
+ id_url=environ.get("TWITCH_ID_URL") or TWITCH_ID_URL,
+ api_url=environ.get("TWITCH_API_URL") or TWITCH_API_URL,
+ eventsub_url=environ.get("TWITCH_EVENTSUB_URL") or EVENTSUB_WS_URL,
+ )
+
+
+class TwitchAPI:
+ """Just the four HTTP calls this bot needs."""
+
+ def __init__(self, token: str, endpoints: Endpoints) -> None:
+ self.token = token.removeprefix("oauth:") # IRC wants "oauth:", the API does not
+ self.endpoints = endpoints
+ self.client_id = ""
+ self.user_id = ""
+ self.login = ""
+
+ def validate(self) -> list[str]:
+ """Step 1: who does this token belong to, and what may it do? Returns its scopes."""
+ info = self._call("GET", f"{self.endpoints.id_url}/validate", auth=f"OAuth {self.token}")
+ self.client_id, self.user_id, self.login = info["client_id"], info["user_id"], info["login"]
+ return info.get("scopes", [])
+
+ def user_id_of(self, login: str) -> str:
+ """Step 2: EventSub and the API use numeric user ids, not names."""
+ query = urllib.parse.urlencode({"login": login.lstrip("#").lower()})
+ users = self._call("GET", f"{self.endpoints.api_url}/users?{query}")["data"]
+ if not users:
+ raise TwitchAPIError(f"no Twitch user called {login!r}")
+ return users[0]["id"]
+
+ def subscribe_to_chat(self, session_id: str, broadcaster_id: str) -> None:
+ """Step 4: ask Twitch to push this channel's chat messages to our WebSocket."""
+ self._call("POST", f"{self.endpoints.api_url}/eventsub/subscriptions", expect=202, body={
+ "type": "channel.chat.message",
+ "version": "1",
+ "condition": {"broadcaster_user_id": broadcaster_id, "user_id": self.user_id},
+ "transport": {"method": "websocket", "session_id": session_id},
+ })
+
+ def send_chat_message(self, broadcaster_id: str, text: str) -> dict[str, Any]:
+ """Step 5: say something. Twitch answers whether it really went out (is_sent)."""
+ return self._call("POST", f"{self.endpoints.api_url}/chat/messages", body={
+ "broadcaster_id": broadcaster_id,
+ "sender_id": self.user_id,
+ "message": text,
+ })["data"][0]
+
+ def _call(self, method: str, url: str, *, expect: int = 200, body: dict | None = None, auth: str | None = None) -> dict:
+ headers = {"Authorization": auth or f"Bearer {self.token}", "Client-Id": self.client_id}
+ data = None
+ if body is not None:
+ headers["Content-Type"] = "application/json"
+ data = json.dumps(body).encode()
+ request = urllib.request.Request(url, data=data, method=method, headers=headers)
+ try:
+ with urllib.request.urlopen(request, timeout=HTTP_TIMEOUT) as response:
+ status, payload = response.status, response.read()
+ except urllib.error.HTTPError as error:
+ status, payload = error.code, error.read()
+ if status != expect:
+ raise TwitchAPIError(f"{method} {url} -> HTTP {status}: {payload.decode(errors='replace')}")
+ return json.loads(payload or b"{}")
+
+
+class EchoBot:
+ def __init__(self, settings: Settings, endpoints: Endpoints) -> None:
+ self.settings = settings
+ self.endpoints = endpoints
+ self.api = TwitchAPI(settings.token, endpoints)
+ self.limiter = RateLimiter(DEFAULT_RATE_LIMITS)
+ self.broadcaster_id = ""
+
+ def prepare(self) -> None:
+ """Steps 1 and 2, done once."""
+ scopes = self.api.validate()
+ missing = [scope for scope in REQUIRED_SCOPES if scope not in scopes]
+ if missing:
+ raise TwitchAPIError(f"the token is missing the scopes {missing}; get a new one (README: Get a token)")
+ if self.api.login != self.settings.bot_name.lower():
+ log.warning("the token belongs to %s, not TWITCH_BOT_NAME=%s", self.api.login, self.settings.bot_name)
+ self.broadcaster_id = self.api.user_id_of(self.settings.channel)
+
+ def listen(self) -> None:
+ """Steps 3 to 5. Returns when Twitch closes the connection."""
+ url, subscribed = self.endpoints.eventsub_url, False
+ try:
+ while True:
+ with connect(url) as websocket:
+ welcome = json.loads(websocket.recv(timeout=WELCOME_TIMEOUT))
+ session = welcome["payload"]["session"]
+ if not subscribed: # after a session_reconnect, Twitch moves our subscription over
+ self.api.subscribe_to_chat(session["id"], self.broadcaster_id)
+ subscribed = True
+ print(f"listening to #{self.settings.channel} as {self.api.login}", flush=True)
+ url = self._read_events(websocket, session["keepalive_timeout_seconds"] + KEEPALIVE_MARGIN)
+ # Simplification: Twitch suggests keeping the old socket open until the new one
+ # says "welcome", so that no event is lost in between. We close it first.
+ except ConnectionClosed as closed:
+ log.warning("Twitch closed the EventSub connection (%s)", closed)
+
+ def _read_events(self, websocket: Any, timeout: float) -> str:
+ """Handle messages until Twitch tells us to move to another server (returns its URL)."""
+ while True:
+ # EventSub's version of PING: if nothing (not even a keepalive) arrives in time,
+ # the connection is dead and recv() raises TimeoutError.
+ message = json.loads(websocket.recv(timeout=timeout))
+ kind = message["metadata"]["message_type"]
+ if kind == "notification":
+ self.on_chat_message(message["payload"]["event"])
+ elif kind == "session_reconnect":
+ return message["payload"]["session"]["reconnect_url"]
+ elif kind == "revocation":
+ raise TwitchAPIError(f"Twitch cancelled the subscription: {message['payload']['subscription']['status']}")
+ # "session_keepalive" needs no answer: it only proves the connection is alive.
+
+ def on_chat_message(self, event: dict[str, Any]) -> None:
+ if event["chatter_user_id"] == self.api.user_id:
+ return # unlike IRC, EventSub also delivers the bot's own messages; echoing them would never end
+ text = event["message"]["text"]
+ print(f"#{event['broadcaster_user_login']} {event['chatter_user_name']}: {text}", flush=True)
+ self.limiter.wait()
+ result = self.api.send_chat_message(self.broadcaster_id, text)
+ if not result["is_sent"]:
+ log.warning("Twitch did not send it: %s", result.get("drop_reason"))
+ return
+ print(f"#{event['broadcaster_user_login']} {self.api.login} (bot): {text}", flush=True)
+
+
+def run(settings: Settings, endpoints: Endpoints, *, sleep: Callable[[float], None] = time.sleep) -> None:
+ bot = EchoBot(settings, endpoints)
+ bot.prepare()
+ failures = 0
+ while True:
+ try:
+ bot.listen() # returns after a session that worked and then ended
+ failures = 0
+ except (OSError, WebSocketException) as error: # timeouts, refused connections, bad handshakes
+ log.warning("could not stay connected (%r)", error)
+ failures += 1
+ delay = reconnect_delay(failures)
+ log.info("reconnecting in %s s", delay)
+ sleep(delay)
+
+
+def main() -> int:
+ logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-7s %(message)s", datefmt="%H:%M:%S")
+ load_env_file()
+ try:
+ run(Settings.from_env(), Endpoints.from_env())
+ except (ConfigError, TwitchAPIError) as error:
+ print(f"error: {error}", file=sys.stderr)
+ return 1
+ except KeyboardInterrupt:
+ print("bye", file=sys.stderr)
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/examples/irc_library_echo.py b/examples/irc_library_echo.py
new file mode 100644
index 0000000..413ee03
--- /dev/null
+++ b/examples/irc_library_echo.py
@@ -0,0 +1,85 @@
+"""The same echo bot, written with a ready-made IRC package.
+
+Issue #1 asked: why not use an existing IRC package? Fair question. Here is the bot
+on top of `irc` (https://pypi.org/project/irc/), the package Twitch's own old Python
+sample used. Put it next to main.py + twitch_chat.py and compare:
+
+* the package opens the socket, splits lines, parses them, answers PING and
+ reconnects with back-off (STEP 1, 5, 6, 7 and 9 in twitch_chat.py);
+* we still do the Twitch-specific parts: the oauth: password, CAP REQ, JOIN,
+ reading the display-name tag and the rate limit (STEP 2, 3, 4 and 8).
+
+Run it from the repo root, with the same .env as main.py:
+
+ pip install -r examples/requirements.txt
+ python -m examples.irc_library_echo
+"""
+
+from __future__ import annotations
+
+import functools
+import logging
+import ssl
+import sys
+
+import irc.bot
+import irc.client
+import irc.connection
+from jaraco.stream import buffer
+
+from config import ConfigError, Settings, load_env_file
+from twitch_chat import DEFAULT_CAPABILITIES, DEFAULT_RATE_LIMITS, RateLimiter, default_port, irc_password
+
+# Chat text is not always valid UTF-8; the package's documented way to cope with that.
+irc.client.ServerConnection.buffer_class = buffer.LenientDecodingLineBuffer
+
+
+class EchoBot(irc.bot.SingleServerIRCBot):
+ def __init__(self, settings: Settings) -> None:
+ self.channel = "#" + settings.channel.lstrip("#").lower()
+ self.limiter = RateLimiter(DEFAULT_RATE_LIMITS)
+ server = irc.bot.ServerSpec(settings.host, settings.port or default_port(settings.use_tls), irc_password(settings.token))
+ super().__init__([server], settings.bot_name.lower(), settings.bot_name, connect_factory=connect_factory(settings))
+
+ def on_welcome(self, connection: irc.client.ServerConnection, event: irc.client.Event) -> None:
+ """Logged in (Twitch sent 001): ask for tags, then join."""
+ connection.cap("REQ", ":" + " ".join(DEFAULT_CAPABILITIES))
+ connection.join(self.channel)
+
+ def on_join(self, connection: irc.client.ServerConnection, event: irc.client.Event) -> None:
+ print(f"joined {event.target} as {connection.get_nickname()}", flush=True)
+
+ def on_pubmsg(self, connection: irc.client.ServerConnection, event: irc.client.Event) -> None:
+ """A chat message. event.arguments[0] is the text, already parsed for us."""
+ text = event.arguments[0]
+ tags = {tag["key"]: tag["value"] for tag in event.tags or []}
+ print(f"{event.target} {tags.get('display-name') or event.source.nick}: {text}", flush=True)
+ self.limiter.wait()
+ connection.privmsg(event.target, text)
+ print(f"{event.target} {connection.get_nickname()} (bot): {text}", flush=True)
+
+
+def connect_factory(settings: Settings) -> irc.connection.Factory:
+ if not settings.use_tls:
+ return irc.connection.Factory()
+ context = ssl.create_default_context()
+ return irc.connection.Factory(wrapper=functools.partial(context.wrap_socket, server_hostname=settings.host))
+
+
+def main() -> int:
+ logging.basicConfig(level=logging.INFO)
+ load_env_file()
+ try:
+ bot = EchoBot(Settings.from_env())
+ except ConfigError as error:
+ print(f"error: {error}", file=sys.stderr)
+ return 2
+ try:
+ bot.start() # runs forever; the package reconnects on its own
+ except KeyboardInterrupt:
+ bot.disconnect("bye")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/examples/requirements.txt b/examples/requirements.txt
new file mode 100644
index 0000000..86e1e97
--- /dev/null
+++ b/examples/requirements.txt
@@ -0,0 +1,3 @@
+# Only the examples need these; main.py does not.
+irc>=20.5 # examples/irc_library_echo.py
+websockets>=13.0 # examples/eventsub_echo.py (uses the sync client)
diff --git a/main.py b/main.py
index bd3a609..aeb627e 100644
--- a/main.py
+++ b/main.py
@@ -1,14 +1,81 @@
-from twitchSocket import twitchSocket
+"""Twitch echo bot: repeats every chat message back into the channel.
-ts = twitchSocket(channelName='',
- botName ='',
- botAuth =' None:
+ """The bot's logic: read a message, send the same text back."""
+ for message in chat.messages():
+ print(f"#{message.channel} {message.display_name}: {message.text}", flush=True)
+ reply = message.text # <- change this line to make your own bot
+ chat.send_message(reply)
+ print(f"#{chat.channel} {chat.nick} (bot): {reply}", flush=True)
+
+
+def run(settings: Settings, *, sleep: Callable[[float], None] = time.sleep) -> None:
+ """Connect, echo, and reconnect whenever the connection ends."""
+ failures = 0
+ while True:
+ chat = TwitchChat(
+ channel=settings.channel,
+ nick=settings.bot_name,
+ token=settings.token,
+ host=settings.host,
+ port=settings.port,
+ use_tls=settings.use_tls,
+ )
+ try:
+ with chat:
+ chat.connect()
+ failures = 0
+ echo(chat)
+ except OSError as error: # includes ConnectionClosed, refused connections, network trouble
+ delay = reconnect_delay(failures)
+ failures += 1
+ log.warning("connection lost (%s); reconnecting in %s s", error, delay)
+ sleep(delay)
+
+
+def main(argv: list[str] | None = None) -> int:
+ parser = argparse.ArgumentParser(description="Twitch echo bot")
+ parser.add_argument("--debug", action="store_true", help="print every IRC line sent and received")
+ args = parser.parse_args(argv)
+ logging.basicConfig(
+ level=logging.DEBUG if args.debug else logging.INFO,
+ format="%(asctime)s %(levelname)-7s %(message)s",
+ datefmt="%H:%M:%S",
+ )
+ load_env_file()
try:
- username, message = ts.getMsg()
- # Your logic here.
- ts.sendMsg(message)
+ run(Settings.from_env())
+ except ConfigError as error:
+ print(f"error: {error}", file=sys.stderr)
+ return 2
+ except (AuthenticationError, JoinError) as error:
+ # Retrying will not help here: fix the token or the channel name first.
+ print(f"error: {error}", file=sys.stderr)
+ return 1
+ except KeyboardInterrupt:
+ print("bye", file=sys.stderr)
+ return 0
+
- except(KeyboardInterrupt, EOFError, SystemExit):
- break
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..04859d5
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,15 @@
+# Tool settings only. This repo is a few scripts to read and run, not a package to install.
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["."]
+addopts = "-ra"
+timeout = 60 # a socket test that hangs fails instead of blocking CI
+
+[tool.ruff]
+line-length = 130
+target-version = "py310"
+
+[tool.ruff.lint]
+select = ["E", "F", "W", "I", "N", "UP", "B"]
+ignore = ["N818"] # ConnectionClosed reads better than ConnectionClosedError (websockets uses the same name)
diff --git a/readme.md b/readme.md
deleted file mode 100644
index c64fe2b..0000000
--- a/readme.md
+++ /dev/null
@@ -1,44 +0,0 @@
-Twitch Echo Bot with Python
-========
-[](./LICENSE) [](https://www.python.org/)
-
-Fast and easy way to use.
-
-## Features
-
-- Join the twitch channel.
-- Send message.
-- Get new message.
-
-## Usage
-
-```bash
-git clone https://github.com/NatLee/Twitch-Echo-Bot-with-Python.git
-```
-
-## Example (in main.py)
-
-```python
-
-from twitchSocket import twitchSocket
-
-ts = twitchSocket(channelName='',
- botName ='',
- botAuth ='=8.3
+pytest-timeout>=2.3
+trustme>=1.2
+ruff>=0.15
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..1d4731b
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,83 @@
+"""Shared test helpers: a fake Twitch server per test, and a way to connect to it.
+
+Nothing here talks to the real Twitch. Servers listen on 127.0.0.1 with port 0,
+so the operating system picks a free port for every test.
+"""
+
+from __future__ import annotations
+
+import threading
+from collections.abc import Callable, Iterator
+from typing import Any
+
+import pytest
+
+from tools.fake_twitch_server import FakeTwitchServer
+from twitch_chat import TwitchChat
+
+TOKEN = "oauth:test-token-not-real"
+CHANNEL = "demo"
+BOT = "echo_bot"
+
+
+@pytest.fixture
+def server() -> Iterator[FakeTwitchServer]:
+ with FakeTwitchServer(channels=[CHANNEL], accepted_token=TOKEN) as fake:
+ yield fake
+
+
+@pytest.fixture
+def chat_options(server: FakeTwitchServer) -> dict[str, Any]:
+ """TwitchChat arguments that point at the fake server (plain TCP, short timeouts)."""
+ return {
+ "channel": CHANNEL,
+ "nick": BOT,
+ "token": TOKEN,
+ "host": server.host,
+ "port": server.port,
+ "use_tls": False,
+ "handshake_timeout": 2.0,
+ }
+
+
+@pytest.fixture
+def connect(chat_options: dict[str, Any]) -> Iterator[Callable[..., TwitchChat]]:
+ """connect(**overrides) -> a connected TwitchChat; closed after the test."""
+ opened: list[TwitchChat] = []
+
+ def _connect(**overrides: Any) -> TwitchChat:
+ chat = TwitchChat(**{**chat_options, **overrides})
+ opened.append(chat)
+ chat.connect()
+ return chat
+
+ yield _connect
+ for chat in opened:
+ chat.close()
+
+
+class Background:
+ """Run a blocking call in a thread and collect its result or exception."""
+
+ def __init__(self, target: Callable[[], Any]) -> None:
+ self.result: Any = None
+ self.error: BaseException | None = None
+ self._thread = threading.Thread(target=self._run, args=(target,), daemon=True)
+ self._thread.start()
+
+ def _run(self, target: Callable[[], Any]) -> None:
+ try:
+ self.result = target()
+ except BaseException as error: # noqa: B036 - hand every failure back to the test
+ self.error = error
+
+ def join(self, timeout: float = 5.0) -> Any:
+ self._thread.join(timeout)
+ assert not self._thread.is_alive(), "the call is still running"
+ if self.error is not None:
+ raise self.error
+ return self.result
+
+
+class StopBot(Exception):
+ """Raised by a fake sleep() to end a bot's reconnect loop inside a test."""
diff --git a/tests/test_examples.py b/tests/test_examples.py
new file mode 100644
index 0000000..df13e5f
--- /dev/null
+++ b/tests/test_examples.py
@@ -0,0 +1,203 @@
+"""The two examples: the `irc` package version, and the EventSub version.
+
+The EventSub test uses small fakes of Twitch's HTTP API and EventSub WebSocket that
+answer with the shapes from Twitch's docs. Like everything here, no real Twitch.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+
+from config import Settings
+from tests.conftest import BOT, CHANNEL, TOKEN, Background, StopBot
+
+REPO = Path(__file__).resolve().parents[1]
+
+
+# ----- examples/irc_library_echo.py ---------------------------------------------------------
+
+
+def test_irc_library_example_echoes(server):
+ pytest.importorskip("irc")
+ env = {**os.environ, "PYTHONUNBUFFERED": "1", "TWITCH_CHANNEL": CHANNEL, "TWITCH_BOT_NAME": BOT,
+ "TWITCH_OAUTH_TOKEN": TOKEN, "TWITCH_IRC_HOST": server.host, "TWITCH_IRC_PORT": str(server.port),
+ "TWITCH_IRC_TLS": "false"}
+ with subprocess.Popen([sys.executable, "-m", "examples.irc_library_echo"], cwd=REPO, env=env,
+ stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) as bot:
+ try:
+ server.wait_for_join(timeout=15)
+ server.say("alice", "hello from the irc package: yes", display_name="Alice")
+ server.wait_for(lambda line: line == "PRIVMSG #demo :hello from the irc package: yes")
+ server.ping("still-there")
+ server.wait_for(lambda line: line.startswith("PONG") and "still-there" in line)
+ finally:
+ bot.terminate()
+ output = bot.communicate(timeout=10)[0]
+
+ sent = server.sent_lines()
+ assert sent.index(f"PASS {TOKEN}") < sent.index(f"NICK {BOT}")
+ assert "CAP REQ :twitch.tv/tags twitch.tv/commands" in sent
+ assert "#demo Alice: hello from the irc package: yes" in output
+
+
+# ----- examples/eventsub_echo.py -------------------------------------------------------------
+
+BOT_ID, CHANNEL_ID, VIEWER_ID = "100", "200", "300"
+API_TOKEN = TOKEN.removeprefix("oauth:")
+
+
+class FakeTwitchAPI(ThreadingHTTPServer):
+ """id.twitch.tv/oauth2/validate and the three Helix calls the example makes."""
+
+ def __init__(self, scopes: tuple[str, ...] = ("chat:read", "chat:edit", "user:read:chat", "user:write:chat")) -> None:
+ super().__init__(("127.0.0.1", 0), FakeAPIHandler)
+ self.scopes = list(scopes) # what the fake token is allowed to do
+ self.requests: list[tuple[str, str, dict, dict]] = [] # method, path, headers, body
+ self.subscribed = threading.Event()
+ self.chat_sent = threading.Event()
+
+ @property
+ def url(self) -> str:
+ return f"http://127.0.0.1:{self.server_address[1]}"
+
+ def bodies(self, path: str) -> list[dict]:
+ return [body for _, request_path, _, body in self.requests if request_path == path]
+
+
+class FakeAPIHandler(BaseHTTPRequestHandler):
+ server: FakeTwitchAPI
+
+ def log_message(self, *args: object) -> None: # keep test output quiet
+ pass
+
+ def do_GET(self) -> None: # noqa: N802
+ url = urlparse(self.path)
+ self.server.requests.append(("GET", url.path, dict(self.headers), {}))
+ if url.path == "/oauth2/validate" and self.headers["Authorization"] == f"OAuth {API_TOKEN}":
+ self._reply(200, {"client_id": "fake-client", "login": BOT, "user_id": BOT_ID,
+ "scopes": self.server.scopes, "expires_in": 3600})
+ elif url.path == "/helix/users" and parse_qs(url.query) == {"login": [CHANNEL]}:
+ self._reply(200, {"data": [{"id": CHANNEL_ID, "login": CHANNEL, "display_name": "Demo"}]})
+ else:
+ self._reply(401, {"status": 401, "message": "invalid request"})
+
+ def do_POST(self) -> None: # noqa: N802
+ body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+ self.server.requests.append(("POST", self.path, dict(self.headers), body))
+ if self.headers["Authorization"] != f"Bearer {API_TOKEN}" or self.headers["Client-Id"] != "fake-client":
+ self._reply(401, {"status": 401, "message": "bad token or client id"})
+ elif self.path == "/helix/eventsub/subscriptions":
+ self.server.subscribed.set()
+ self._reply(202, {"data": [{"id": "sub-1", "status": "enabled", **body}]})
+ elif self.path == "/helix/chat/messages":
+ self.server.chat_sent.set()
+ self._reply(200, {"data": [{"message_id": "msg-1", "is_sent": True}]})
+ else:
+ self._reply(404, {"status": 404})
+
+ def _reply(self, status: int, payload: dict) -> None:
+ data = json.dumps(payload).encode()
+ self.send_response(status)
+ self.send_header("Content-Type", "application/json")
+ self.send_header("Content-Length", str(len(data)))
+ self.end_headers()
+ self.wfile.write(data)
+
+
+def eventsub(kind: str, payload: dict) -> str:
+ return json.dumps({"metadata": {"message_id": f"id-{time.monotonic_ns()}", "message_type": kind,
+ "message_timestamp": "2026-11-04T00:00:00Z"}, "payload": payload})
+
+
+def welcome(session_id: str) -> str:
+ return eventsub("session_welcome", {"session": {"id": session_id, "status": "connected",
+ "keepalive_timeout_seconds": 10, "reconnect_url": None}})
+
+
+def chat_notification(chatter_id: str, name: str, text: str) -> str:
+ event = {"broadcaster_user_id": CHANNEL_ID, "broadcaster_user_login": CHANNEL, "broadcaster_user_name": "Demo",
+ "chatter_user_id": chatter_id, "chatter_user_login": name.lower(), "chatter_user_name": name,
+ "message_id": f"m-{time.monotonic_ns()}", "message": {"text": text, "fragments": []}}
+ return eventsub("notification", {"subscription": {"type": "channel.chat.message", "version": "1"}, "event": event})
+
+
+def test_eventsub_example_subscribes_echoes_and_skips_its_own_messages(capsys):
+ websockets_server = pytest.importorskip("websockets.sync.server")
+ from examples import eventsub_echo
+
+ api = FakeTwitchAPI()
+ threading.Thread(target=api.serve_forever, daemon=True).start()
+
+ def session(websocket) -> None:
+ if websocket.request.path == "/ws":
+ websocket.send(welcome("session-1"))
+ assert api.subscribed.wait(5)
+ websocket.send(eventsub("session_keepalive", {}))
+ websocket.send(chat_notification(VIEWER_ID, "Alice", "hello: eventsub"))
+ websocket.send(eventsub("session_reconnect", {"session": {"id": "session-1", "status": "reconnecting",
+ "reconnect_url": f"{ws_url}/moved"}}))
+ time.sleep(0.5) # Twitch would close the old socket a bit later
+ else: # the reconnect URL: the subscription comes along, no new POST expected
+ websocket.send(welcome("session-2"))
+ websocket.send(chat_notification(BOT_ID, BOT, "hello: eventsub")) # the bot's own echo
+ websocket.send(chat_notification(VIEWER_ID, "Bob", "after the move"))
+ deadline = time.monotonic() + 5
+ while len(api.bodies("/helix/chat/messages")) < 2 and time.monotonic() < deadline:
+ time.sleep(0.05)
+ time.sleep(0.3) # room for a (wrong) third send
+ # returning closes the connection, which ends bot.listen()
+
+ sleeps: list[float] = []
+
+ def fake_sleep(seconds: float) -> None:
+ sleeps.append(seconds)
+ raise StopBot
+
+ with websockets_server.serve(session, "127.0.0.1", 0) as ws_server:
+ ws_url = f"ws://127.0.0.1:{ws_server.socket.getsockname()[1]}"
+ threading.Thread(target=ws_server.serve_forever, daemon=True).start()
+ endpoints = eventsub_echo.Endpoints(id_url=f"{api.url}/oauth2", api_url=f"{api.url}/helix", eventsub_url=f"{ws_url}/ws")
+ settings = Settings(channel=CHANNEL, bot_name=BOT, token=TOKEN)
+ bot = Background(lambda: eventsub_echo.run(settings, endpoints, sleep=fake_sleep))
+ with pytest.raises(StopBot):
+ bot.join(timeout=20)
+ ws_server.shutdown()
+ api.shutdown()
+
+ assert api.bodies("/helix/eventsub/subscriptions") == [{
+ "type": "channel.chat.message",
+ "version": "1",
+ "condition": {"broadcaster_user_id": CHANNEL_ID, "user_id": BOT_ID},
+ "transport": {"method": "websocket", "session_id": "session-1"},
+ }]
+ assert api.bodies("/helix/chat/messages") == [
+ {"broadcaster_id": CHANNEL_ID, "sender_id": BOT_ID, "message": "hello: eventsub"},
+ {"broadcaster_id": CHANNEL_ID, "sender_id": BOT_ID, "message": "after the move"},
+ ]
+ assert sleeps == [1]
+ out = capsys.readouterr().out
+ assert "#demo Alice: hello: eventsub\n#demo echo_bot (bot): hello: eventsub\n" in out
+
+
+def test_eventsub_example_refuses_a_token_without_chat_scopes():
+ pytest.importorskip("websockets")
+ from examples import eventsub_echo
+
+ api = FakeTwitchAPI(scopes=("chat:read", "chat:edit")) # an IRC-only token
+ threading.Thread(target=api.serve_forever, daemon=True).start()
+ try:
+ endpoints = eventsub_echo.Endpoints(id_url=f"{api.url}/oauth2", api_url=f"{api.url}/helix", eventsub_url="ws://unused")
+ with pytest.raises(eventsub_echo.TwitchAPIError, match="user:read:chat"):
+ eventsub_echo.run(Settings(channel=CHANNEL, bot_name=BOT, token=TOKEN), endpoints)
+ finally:
+ api.shutdown()
diff --git a/tests/test_legacy.py b/tests/test_legacy.py
new file mode 100644
index 0000000..c760b8e
--- /dev/null
+++ b/tests/test_legacy.py
@@ -0,0 +1,62 @@
+"""The 2019 names in twitchSocket.py must keep working.
+
+Delete this file together with twitchSocket.py (not before 2027-11-04, see that file).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from tests.conftest import CHANNEL, TOKEN
+from twitchSocket import twitchSocket
+
+
+@pytest.fixture
+def legacy(server):
+ """The 2019 call, plus the options that point it at the fake server instead of Twitch."""
+
+ def _open(*args, **kwargs):
+ with pytest.warns(DeprecationWarning, match="twitch_chat.TwitchChat"):
+ return twitchSocket(*args, **kwargs, host=server.host, port=server.port, use_tls=False)
+
+ return _open
+
+
+def test_old_keyword_call_joins_in_the_constructor(server, legacy):
+ ts = legacy(channelName=CHANNEL, botName="Echo_Bot", botAuth=TOKEN)
+
+ server.wait_for_join()
+ assert server.sent_lines()[-1] == f"JOIN #{CHANNEL}"
+ ts.close()
+
+
+def test_old_positional_call(server, legacy):
+ legacy(CHANNEL, "echo_bot", TOKEN).close()
+
+
+def test_getmsg_and_sendmsg_print_like_before(server, legacy, capsys):
+ ts = legacy(channelName=CHANNEL, botName="Echo_Bot", botAuth=TOKEN)
+ server.say("alice", "hi: there")
+
+ assert ts.getMsg() == ("alice", "hi: there")
+ ts.sendMsg("hello")
+
+ server.wait_for(lambda line: line == "PRIVMSG #demo :hello")
+ assert capsys.readouterr().out == "alice: hi: there\nEcho_Bot: hello\n"
+ ts.close()
+
+
+def test_the_2019_main_loop_survives_a_ping(server, legacy):
+ # Regression: in 2019, getMsg() returned (None, None) for a PING and the next line,
+ # ts.sendMsg(message), raised TypeError, which main.py did not catch.
+ ts = legacy(channelName=CHANNEL, botName="echo_bot", botAuth=TOKEN)
+ server.ping()
+ server.say("alice", "still here?")
+
+ username, message = ts.getMsg() # the body of the 2019 main.py loop
+ ts.sendMsg(message)
+
+ assert (username, message) == ("alice", "still here?")
+ server.wait_for(lambda line: line == "PRIVMSG #demo :still here?")
+ assert "PONG :tmi.twitch.tv" in server.sent_lines()
+ ts.close()
diff --git a/tests/test_main.py b/tests/test_main.py
new file mode 100644
index 0000000..36b033d
--- /dev/null
+++ b/tests/test_main.py
@@ -0,0 +1,144 @@
+"""config.py and main.py: settings, the echo loop, reconnecting, and the real command line."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+import main
+from config import REQUIRED, ConfigError, Settings, load_env_file
+from tests.conftest import BOT, CHANNEL, TOKEN, Background, StopBot
+from twitch_chat import TWITCH_IRC_HOST
+
+REPO = Path(__file__).resolve().parents[1]
+GOOD_ENV = {"TWITCH_CHANNEL": CHANNEL, "TWITCH_BOT_NAME": BOT, "TWITCH_OAUTH_TOKEN": TOKEN}
+
+
+def fake_server_env(server) -> dict[str, str]:
+ return {**GOOD_ENV, "TWITCH_IRC_HOST": server.host, "TWITCH_IRC_PORT": str(server.port), "TWITCH_IRC_TLS": "false"}
+
+
+# ----- config.py ----------------------------------------------------------------------
+
+
+def test_missing_settings_are_all_named():
+ with pytest.raises(ConfigError) as error:
+ Settings.from_env({"TWITCH_CHANNEL": " "})
+
+ for name in REQUIRED:
+ assert name in str(error.value)
+
+
+def test_settings_default_to_real_twitch_over_tls():
+ settings = Settings.from_env(GOOD_ENV)
+
+ assert (settings.host, settings.port, settings.use_tls) == (TWITCH_IRC_HOST, None, True)
+
+
+@pytest.mark.parametrize("word", ["false", "0", "No", "OFF"])
+def test_settings_for_the_practice_server(word):
+ settings = Settings.from_env({**GOOD_ENV, "TWITCH_IRC_HOST": "127.0.0.1", "TWITCH_IRC_PORT": "6667", "TWITCH_IRC_TLS": word})
+
+ assert (settings.host, settings.port, settings.use_tls) == ("127.0.0.1", 6667, False)
+
+
+def test_bad_port_is_a_config_error():
+ with pytest.raises(ConfigError, match="TWITCH_IRC_PORT"):
+ Settings.from_env({**GOOD_ENV, "TWITCH_IRC_PORT": "six"})
+
+
+def test_env_file_is_read_without_overwriting(tmp_path):
+ env_file = tmp_path / ".env"
+ env_file.write_text(
+ "# a comment\n\nTWITCH_CHANNEL = demo\nTWITCH_BOT_NAME='echo_bot'\n"
+ "TWITCH_OAUTH_TOKEN=\"oauth:from-file\"\nnot a setting\n",
+ encoding="utf-8",
+ )
+ environ = {"TWITCH_OAUTH_TOKEN": "oauth:from-shell"}
+
+ load_env_file(env_file, environ)
+
+ assert environ == {"TWITCH_CHANNEL": "demo", "TWITCH_BOT_NAME": "echo_bot", "TWITCH_OAUTH_TOKEN": "oauth:from-shell"}
+
+
+def test_missing_env_file_is_fine(tmp_path):
+ environ: dict[str, str] = {}
+ load_env_file(tmp_path / "nope.env", environ)
+ assert environ == {}
+
+
+def test_env_example_lists_every_setting():
+ example = (REPO / ".env.example").read_text(encoding="utf-8")
+
+ for name in [*REQUIRED, "TWITCH_IRC_HOST", "TWITCH_IRC_PORT", "TWITCH_IRC_TLS"]:
+ assert name in example
+
+
+# ----- main.py ------------------------------------------------------------------------
+
+
+def test_run_echoes_and_reconnects(server, capsys):
+ sleeps: list[float] = []
+
+ def fake_sleep(seconds: float) -> None:
+ sleeps.append(seconds)
+ if len(sleeps) == 2:
+ raise StopBot
+
+ bot = Background(lambda: main.run(Settings.from_env(fake_server_env(server)), sleep=fake_sleep))
+
+ server.wait_for_join()
+ server.say("alice", "hello: bot", display_name="Alice")
+ server.wait_for(lambda line: line == "PRIVMSG #demo :hello: bot")
+
+ server.request_reconnect() # Twitch's "please come back" -> the bot reconnects after 1 s
+ server.wait_for_join(count=2)
+ server.say("bob", "again")
+ server.wait_for(lambda line: line == "PRIVMSG #demo :again")
+
+ server.disconnect_all() # second loss: the fake sleep ends the test
+ with pytest.raises(StopBot):
+ bot.join()
+ assert sleeps == [1, 1] # the failure count starts over after each good connection
+ out = capsys.readouterr().out
+ assert "#demo Alice: hello: bot\n#demo echo_bot (bot): hello: bot\n" in out
+
+
+def test_main_stops_on_a_wrong_token(server, monkeypatch, capsys):
+ monkeypatch.setattr(main, "load_env_file", lambda: None)
+ for name, value in {**fake_server_env(server), "TWITCH_OAUTH_TOKEN": "oauth:wrong"}.items():
+ monkeypatch.setenv(name, value)
+
+ assert main.main([]) == 1
+ assert "Login authentication failed" in capsys.readouterr().err
+
+
+def test_main_explains_missing_settings(monkeypatch, capsys):
+ monkeypatch.setattr(main, "load_env_file", lambda: None)
+ for name in REQUIRED:
+ monkeypatch.delenv(name, raising=False)
+
+ assert main.main([]) == 2
+ assert ".env.example" in capsys.readouterr().err
+
+
+def test_python_main_py_end_to_end(server):
+ """The command from the README, as a real process."""
+ env = {**os.environ, **fake_server_env(server), "PYTHONUNBUFFERED": "1"}
+ with subprocess.Popen([sys.executable, "main.py", "--debug"], cwd=REPO, env=env,
+ stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) as bot:
+ try:
+ server.wait_for_join(timeout=10)
+ server.say("alice", "end to end")
+ server.wait_for(lambda line: line == "PRIVMSG #demo :end to end")
+ finally:
+ bot.terminate()
+ output = bot.communicate(timeout=10)[0]
+
+ assert "> PASS ***" in output
+ assert TOKEN.removeprefix("oauth:") not in output
+ assert "#demo alice: end to end" in output
diff --git a/tests/test_parsing.py b/tests/test_parsing.py
new file mode 100644
index 0000000..90c89d6
--- /dev/null
+++ b/tests/test_parsing.py
@@ -0,0 +1,123 @@
+"""STEP 5 and STEP 6: turning bytes into lines, and lines into their parts."""
+
+import pytest
+
+from twitch_chat import ChatMessage, LineBuffer, parse_line, unescape_tag_value
+
+# Modelled on the PRIVMSG example in https://dev.twitch.tv/docs/chat/irc/ (checked 2026-11-04)
+TWITCH_DOCS_EXAMPLE = (
+ "@badge-info=;badges=turbo/1;color=#0D4200;display-name=ronni;emotes=25:0-4,12-16/1902:6-10;"
+ "id=b34ccfc7-4977-403a-8a94-33c6bac34fb8;mod=0;room-id=1337;subscriber=0;tmi-sent-ts=1507246572675;"
+ "turbo=1;user-id=1337;user-type=global_mod :ronni!ronni@ronni.tmi.twitch.tv PRIVMSG #ronni :Kappa Keepo Kappa"
+)
+
+
+def test_twitch_docs_example_splits_into_tags_prefix_command_params():
+ message = parse_line(TWITCH_DOCS_EXAMPLE)
+
+ assert message.command == "PRIVMSG"
+ assert message.prefix == "ronni!ronni@ronni.tmi.twitch.tv"
+ assert message.nick == "ronni"
+ assert message.params == ["#ronni", "Kappa Keepo Kappa"]
+ assert message.tags["display-name"] == "ronni"
+ assert message.tags["emotes"] == "25:0-4,12-16/1902:6-10"
+ assert message.tags["badge-info"] == ""
+ assert len(message.tags) == 13
+
+
+def test_chat_text_with_colons_stays_whole():
+ # Regression: the 2019 code used split(":") and kept only "look at this".
+ message = parse_line(":alice!alice@alice.tmi.twitch.tv PRIVMSG #demo :look at this: https://example.com :)")
+
+ assert message.trailing == "look at this: https://example.com :)"
+
+
+def test_text_that_starts_with_a_colon():
+ assert parse_line(":a!a@a.tmi.twitch.tv PRIVMSG #demo ::)").trailing == ":)"
+
+
+def test_ping_has_no_prefix():
+ message = parse_line("PING :tmi.twitch.tv")
+
+ assert (message.prefix, message.command, message.params) == (None, "PING", ["tmi.twitch.tv"])
+
+
+def test_numeric_reply():
+ message = parse_line(":tmi.twitch.tv 001 echo_bot :Welcome, GLHF!")
+
+ assert message.command == "001"
+ assert message.params == ["echo_bot", "Welcome, GLHF!"]
+
+
+def test_line_without_trailing_parameter():
+ assert parse_line(":echo_bot!echo_bot@echo_bot.tmi.twitch.tv JOIN #demo").params == ["#demo"]
+
+
+def test_empty_trailing_parameter_is_kept():
+ assert parse_line("PRIVMSG #demo :").params == ["#demo", ""]
+
+
+def test_cap_nak():
+ assert parse_line(":tmi.twitch.tv CAP * NAK :twitch.tv/foo").params == ["*", "NAK", "twitch.tv/foo"]
+
+
+def test_tag_without_value():
+ assert parse_line("@flag;key=1 PING :x").tags == {"flag": "", "key": "1"}
+
+
+@pytest.mark.parametrize(
+ ("escaped", "plain"),
+ [
+ (r"Hello\sthere", "Hello there"),
+ (r"a\:b", "a;b"),
+ (r"back\\slash", "back\\slash"),
+ (r"line\rbreak\n", "line\rbreak\n"),
+ (r"\b", "b"), # unknown escape: drop the backslash
+ ("test\\", "test"), # lone backslash at the end: drop it
+ ("", ""),
+ ],
+)
+def test_tag_values_are_unescaped_like_ircv3_says(escaped, plain):
+ assert unescape_tag_value(escaped) == plain
+
+
+def test_display_name_comes_from_tags_and_falls_back_to_login():
+ tagged = ChatMessage.from_irc(parse_line(r"@display-name=Alice\sW :alice!alice@alice.tmi.twitch.tv PRIVMSG #demo :hi"))
+ untagged = ChatMessage.from_irc(parse_line(":alice!alice@alice.tmi.twitch.tv PRIVMSG #demo :hi"))
+
+ assert (tagged.channel, tagged.user, tagged.text, tagged.display_name) == ("demo", "alice", "hi", "Alice W")
+ assert untagged.display_name == "alice"
+
+
+# ----- LineBuffer -------------------------------------------------------------------
+
+
+def test_two_lines_in_one_chunk():
+ # Regression: the 2019 code treated one recv() as one line and lost the second.
+ assert LineBuffer().feed(b"PING :a\r\nPING :b\r\n") == ["PING :a", "PING :b"]
+
+
+def test_half_a_line_waits_for_the_rest():
+ buffer = LineBuffer()
+
+ assert buffer.feed(b"PRIVMSG #demo :hel") == []
+ assert buffer.feed(b"lo\r\nPING") == ["PRIVMSG #demo :hello"]
+ assert buffer.feed(b" :x\r\n") == ["PING :x"]
+
+
+def test_character_cut_between_two_chunks():
+ # Regression: decoding each recv() on its own crashed on a split UTF-8 character.
+ data = "PRIVMSG #demo :你好\r\n".encode()
+ cut = data.index("你".encode()) + 1
+ buffer = LineBuffer()
+
+ assert buffer.feed(data[:cut]) == []
+ assert buffer.feed(data[cut:]) == ["PRIVMSG #demo :你好"]
+
+
+def test_bare_newline_and_blank_lines():
+ assert LineBuffer().feed(b"PING :a\n\r\n\r\nPING :b\n") == ["PING :a", "PING :b"]
+
+
+def test_invalid_utf8_does_not_crash():
+ assert LineBuffer().feed(b"PRIVMSG #demo :\xff\r\n") == ["PRIVMSG #demo :�"]
diff --git a/tests/test_rate_limit.py b/tests/test_rate_limit.py
new file mode 100644
index 0000000..40444f5
--- /dev/null
+++ b/tests/test_rate_limit.py
@@ -0,0 +1,67 @@
+"""STEP 8: RateLimiter, with a fake clock so the tests run instantly."""
+
+from twitch_chat import DEFAULT_RATE_LIMITS, RateLimiter
+
+
+class FakeClock:
+ def __init__(self) -> None:
+ self.now = 0.0
+ self.slept: list[float] = []
+
+ def __call__(self) -> float:
+ return self.now
+
+ def sleep(self, seconds: float) -> None:
+ self.slept.append(seconds)
+ self.now += seconds
+
+
+def limiter(rules, clock):
+ return RateLimiter(rules, clock=clock, sleep=clock.sleep)
+
+
+def test_first_messages_go_out_at_once():
+ clock = FakeClock()
+ limit = limiter([(3, 10.0)], clock)
+
+ assert [limit.wait() for _ in range(3)] == [0.0, 0.0, 0.0]
+
+
+def test_one_per_second():
+ clock = FakeClock()
+ limit = limiter([(1, 1.0)], clock)
+
+ limit.wait()
+ clock.now += 0.25
+
+ assert limit.wait() == 0.75
+
+
+def test_twitch_defaults_twenty_one_messages_take_thirty_seconds():
+ # DEFAULT_RATE_LIMITS = 20 per 30 s and 1 per s (Twitch's numbers for a normal bot).
+ # Sends 1-20 go out one second apart (t = 0..19). Send 21 must wait until
+ # send 1 is 30 s old, so it leaves at t = 30 after an 11 s wait.
+ clock = FakeClock()
+ limit = limiter(DEFAULT_RATE_LIMITS, clock)
+
+ for _ in range(21):
+ limit.wait()
+
+ assert clock.slept == [1.0] * 19 + [11.0]
+ assert clock.now == 30.0
+
+
+def test_messages_spread_out_never_wait():
+ clock = FakeClock()
+ limit = limiter(DEFAULT_RATE_LIMITS, clock)
+
+ for _ in range(50):
+ assert limit.wait() == 0.0
+ clock.now += 2.0 # slower than 20 per 30 s
+
+
+def test_no_rules_means_no_limit():
+ clock = FakeClock()
+ limit = limiter([], clock)
+
+ assert sum(limit.wait() for _ in range(100)) == 0.0
diff --git a/tests/test_twitch_chat.py b/tests/test_twitch_chat.py
new file mode 100644
index 0000000..1e17e8f
--- /dev/null
+++ b/tests/test_twitch_chat.py
@@ -0,0 +1,247 @@
+"""TwitchChat against the fake Twitch server: the handshake, reading, PING, sending, endings.
+
+Tests marked "Regression" failed with the 2019 twitchSocket.py; the reproduction
+script and its output are kept in the revamp notes.
+"""
+
+from __future__ import annotations
+
+import logging
+import ssl
+import time
+
+import pytest
+import trustme
+
+from tests.conftest import BOT, CHANNEL, TOKEN, Background
+from tools.fake_twitch_server import FakeTwitchServer
+from twitch_chat import (
+ DEFAULT_RATE_LIMITS,
+ TWITCH_IRC_HOST,
+ TWITCH_IRC_PLAIN_PORT,
+ TWITCH_IRC_TLS_PORT,
+ AuthenticationError,
+ ConnectionClosed,
+ JoinError,
+ TwitchChat,
+)
+
+ALICE = ":alice!alice@alice.tmi.twitch.tv PRIVMSG #demo :"
+
+
+# ----- STEP 1 to 4: connecting --------------------------------------------------------
+
+
+def test_defaults_are_twitchs_tls_endpoint():
+ chat = TwitchChat(channel="demo", nick="bot", token="t")
+
+ assert (chat.host, chat.port, chat.use_tls) == (TWITCH_IRC_HOST, TWITCH_IRC_TLS_PORT, True)
+ assert TwitchChat(channel="demo", nick="bot", token="t", use_tls=False).port == TWITCH_IRC_PLAIN_PORT
+
+
+def test_handshake_sends_cap_pass_nick_join_in_order(server, connect):
+ connect()
+
+ assert server.sent_lines() == [
+ "CAP REQ :twitch.tv/tags twitch.tv/commands",
+ f"PASS {TOKEN}",
+ f"NICK {BOT}",
+ f"JOIN #{CHANNEL}",
+ ]
+
+
+def test_names_and_token_are_normalised(server, connect):
+ connect(channel="#Demo", nick="Echo_Bot", token=TOKEN.removeprefix("oauth:"))
+
+ assert server.sent_lines()[1:] == [f"PASS {TOKEN}", "NICK echo_bot", "JOIN #demo"]
+
+
+def test_wrong_token_raises_authentication_error_with_twitchs_words(connect):
+ # Regression: the 2019 constructor kept looping after Twitch hung up.
+ with pytest.raises(AuthenticationError, match="Login authentication failed"):
+ connect(token="oauth:wrong")
+
+
+def test_unknown_channel_times_out_with_a_hint(connect):
+ # Regression: Twitch ignores a JOIN for a channel that does not exist; the 2019 code waited forever.
+ started = time.monotonic()
+
+ with pytest.raises(JoinError, match="spelled right"):
+ connect(channel="no_such_channel", handshake_timeout=0.5)
+ assert time.monotonic() - started < 3
+
+
+def test_suspended_channel_notice_becomes_join_error():
+ with FakeTwitchServer(channels=[CHANNEL], suspended_channels=["gone"], accepted_token=TOKEN) as fake:
+ chat = TwitchChat(channel="gone", nick=BOT, token=TOKEN, host=fake.host, port=fake.port, use_tls=False)
+ with chat, pytest.raises(JoinError, match="has been suspended"):
+ chat.connect()
+
+
+def test_tls_connection_checks_the_certificate():
+ ca = trustme.CA()
+ server_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+ ca.issue_cert("127.0.0.1").configure_cert(server_context)
+ trusting = ssl.create_default_context()
+ ca.configure_trust(trusting)
+
+ with FakeTwitchServer(channels=[CHANNEL], accepted_token=TOKEN, ssl_context=server_context) as tls_server:
+ options = dict(channel=CHANNEL, nick=BOT, token=TOKEN, host="127.0.0.1", port=tls_server.port, handshake_timeout=2)
+ with TwitchChat(**options, ssl_context=trusting) as chat:
+ chat.connect()
+ tls_server.say("alice", "over TLS")
+ assert chat.read_message().text == "over TLS"
+
+ # The default context does not know our test CA, so it must refuse the server.
+ with TwitchChat(**options) as untrusted, pytest.raises(ssl.SSLCertVerificationError):
+ untrusted.connect()
+
+
+def test_token_never_reaches_the_debug_log(connect, caplog):
+ caplog.set_level(logging.DEBUG, logger="twitch_chat")
+
+ connect()
+
+ assert "> PASS ***" in caplog.text
+ assert TOKEN.removeprefix("oauth:") not in caplog.text
+ assert "< :tmi.twitch.tv 001 echo_bot :Welcome, GLHF!" in caplog.text
+
+
+# ----- STEP 5 to 7: reading ------------------------------------------------------------
+
+
+def test_message_with_colons_and_tags(server, connect):
+ # Regression: the 2019 code returned only "look at this".
+ chat = connect()
+ server.say("alice", "look at this: https://example.com", display_name="Alice")
+
+ message = chat.read_message()
+
+ assert (message.channel, message.user, message.display_name) == ("demo", "alice", "Alice")
+ assert message.text == "look at this: https://example.com"
+ assert message.tags["user-id"]
+
+
+def test_two_messages_in_one_packet(server, connect):
+ # Regression: the 2019 code returned the first and silently dropped the second.
+ chat = connect()
+ server.send_raw(f"{ALICE}first\r\n{ALICE}second\r\n".encode())
+
+ assert [chat.read_message().text, chat.read_message().text] == ["first", "second"]
+
+
+def test_line_and_character_split_across_packets(server, connect):
+ # Regression: the 2019 code crashed (UnicodeDecodeError, then UnboundLocalError).
+ chat = connect()
+ data = f"{ALICE}你好，世界\r\n".encode()
+ cut = data.index("你".encode()) + 1 # inside the 3 bytes of 你
+ reader = Background(chat.read_message) # already waiting in recv() when the first half arrives
+ time.sleep(0.1)
+ server.send_raw(data[:cut])
+ time.sleep(0.1)
+ server.send_raw(data[cut:])
+
+ assert reader.join().text == "你好，世界"
+
+
+def test_ping_gets_the_same_text_back_and_is_not_a_chat_message(server, connect):
+ # Regression: the 2019 code returned (None, None) for PING, and main.py crashed on it.
+ chat = connect()
+ server.ping("custom-ping-text")
+ server.say("alice", "after the ping")
+
+ assert chat.read_message().text == "after the ping"
+ assert server.wait_for(lambda line: line.startswith("PONG")) == "PONG :custom-ping-text"
+
+
+def test_ping_and_chat_in_the_same_packet(server, connect):
+ # Regression: the 2019 code reported user "tmi.twitch.tv\r\n" and echoed IRC protocol text.
+ chat = connect()
+ server.send_raw(f"PING :tmi.twitch.tv\r\n{ALICE}hello\r\n".encode())
+
+ message = chat.read_message()
+
+ assert (message.user, message.text) == ("alice", "hello")
+ server.wait_for(lambda line: line == "PONG :tmi.twitch.tv")
+
+
+def test_chat_text_that_says_ping_is_just_text(server, connect):
+ # Regression: the 2019 code answered PONG to any packet containing "PING :tmi.twitch.tv".
+ chat = connect()
+ server.say("alice", "PING :tmi.twitch.tv")
+
+ assert chat.read_message().text == "PING :tmi.twitch.tv"
+ assert not [line for line in server.sent_lines() if line.startswith("PONG")]
+
+
+def test_notices_are_logged(server, connect, caplog):
+ chat = connect()
+ server.send_line_to_all("@msg-id=msg_duplicate :tmi.twitch.tv NOTICE #demo :Your message is identical to the one you sent.")
+ server.say("alice", "next")
+
+ assert chat.read_message().text == "next"
+ assert "Twitch notice: Your message is identical" in caplog.text
+
+
+# ----- STEP 8: sending ------------------------------------------------------------------
+
+
+def test_send_message(server, connect):
+ connect().send_message("hello: chat")
+
+ assert server.wait_for(lambda line: line.startswith("PRIVMSG")) == "PRIVMSG #demo :hello: chat"
+
+
+def test_line_breaks_cannot_smuggle_in_another_command(server, connect):
+ chat = connect()
+ chat.send_message("one\r\nJOIN #someone_else\ntwo")
+ chat.send_message("done")
+
+ server.wait_for(lambda line: line == "PRIVMSG #demo :done")
+ assert "PRIVMSG #demo :one JOIN #someone_else two" in server.sent_lines()
+ assert "JOIN #someone_else" not in server.sent_lines()
+
+
+def test_sending_waits_for_the_rate_limit(server, connect):
+ assert DEFAULT_RATE_LIMITS == ((20, 30.0), (1, 1.0)) # Twitch's numbers for a normal bot
+ chat = connect(rate_limits=[(1, 0.3)])
+ started = time.monotonic()
+
+ chat.send_message("a")
+ chat.send_message("b")
+
+ assert time.monotonic() - started >= 0.29
+ server.wait_for(lambda line: line == "PRIVMSG #demo :b")
+
+
+# ----- STEP 9: endings -------------------------------------------------------------------
+
+
+def test_server_hanging_up_raises_connection_closed(server, connect):
+ # Regression: the 2019 code returned (None, None) in a busy loop forever.
+ chat = connect()
+ server.disconnect_all()
+
+ with pytest.raises(ConnectionClosed, match="closed the connection"):
+ chat.read_message()
+
+
+def test_reconnect_request_raises_connection_closed_and_connect_starts_over(server, connect):
+ chat = connect()
+ server.request_reconnect()
+
+ with pytest.raises(ConnectionClosed, match="reconnect"):
+ chat.read_message()
+
+ chat.connect()
+ server.wait_for(lambda line: line == f"JOIN #{CHANNEL}", count=2)
+ server.say("alice", "welcome back")
+ assert chat.read_message().text == "welcome back"
+
+
+def test_using_a_closed_chat_says_so(connect):
+ chat = connect()
+ chat.close()
+
+ with pytest.raises(ConnectionClosed, match="not connected"):
+ chat.send_message("hi")
diff --git a/tools/fake_twitch_server.py b/tools/fake_twitch_server.py
new file mode 100644
index 0000000..6881521
--- /dev/null
+++ b/tools/fake_twitch_server.py
@@ -0,0 +1,480 @@
+"""A pretend Twitch chat server you can run on your own computer.
+
+It speaks just enough of Twitch's IRC dialect for this repo: CAP REQ, PASS, NICK,
+JOIN, PRIVMSG, PING/PONG and RECONNECT, with the same reply lines that Twitch's
+docs show (https://dev.twitch.tv/docs/chat/irc/). Two uses:
+
+* Practice without a Twitch account. In one terminal:
+
+ python tools/fake_twitch_server.py
+
+ In another one, point the bot at it (the server prints the exact command):
+
+ TWITCH_IRC_HOST=127.0.0.1 TWITCH_IRC_PORT=6667 TWITCH_IRC_TLS=false python main.py
+
+ Then type ``alice: hello`` in the server window to chat as a viewer called alice.
+ ``/ping`` sends a PING, ``/reconnect`` asks the bot to reconnect, ``/quit`` stops.
+
+* Tests. ``FakeTwitchServer`` is a context manager; tests start it on a free port,
+ make it say things (``say``, ``ping``, ``send_raw``) and check what the bot sent
+ (``wait_for``, ``sent_lines``).
+
+This is not Twitch. It does not check tokens against anything real, has no rate
+limits and knows only the channels you give it. It is deliberately written without
+importing the bot's own code, so a parser bug in the bot cannot hide behind the same
+bug in the server.
+"""
+
+from __future__ import annotations
+
+import argparse
+import itertools
+import socket
+import ssl
+import sys
+import threading
+import time
+from collections.abc import Callable, Iterable
+from dataclasses import dataclass, field
+
+SERVER_NAME = "tmi.twitch.tv"
+DEFAULT_HOST = "127.0.0.1"
+DEFAULT_PORT = 6667 # Twitch's plain-text IRC port; the practice server has no TLS
+DEFAULT_CHANNELS = ("demo",)
+ACCEPT_POLL = 0.1 # seconds between checks whether stop() was called
+KNOWN_CAPABILITIES = ("twitch.tv/tags", "twitch.tv/commands", "twitch.tv/membership")
+
+# Exact texts from Twitch's docs, so the bot sees what it would see on Twitch.
+AUTH_FAILED = "Login authentication failed"
+CHANNEL_SUSPENDED = "This channel does not exist or has been suspended."
+BAD_AUTH_ORDER = "Improperly formatted auth"
+WELCOME_LINES = (
+ ("001", "Welcome, GLHF!"),
+ ("002", "Your host is tmi.twitch.tv"),
+ ("003", "This server is rather new"),
+ ("004", "-"),
+ ("375", "-"),
+ ("372", "You are in a maze of twisty passages, all alike."),
+ ("376", ">"),
+)
+
+_TAG_ESCAPES = {";": r"\:", " ": r"\s", "\\": "\\\\", "\r": r"\r", "\n": r"\n"}
+
+
+def escape_tag_value(value: str) -> str:
+ """Escape a tag value the way IRCv3 says (https://ircv3.net/specs/extensions/message-tags)."""
+ return "".join(_TAG_ESCAPES.get(char, char) for char in value)
+
+
+def format_tags(tags: dict[str, str]) -> str:
+ return "@" + ";".join(f"{key}={escape_tag_value(value)}" for key, value in tags.items())
+
+
+@dataclass
+class _Client:
+ """One connected bot."""
+
+ number: int
+ sock: socket.socket
+ caps: set[str] = field(default_factory=set)
+ password: str | None = None
+ nick: str | None = None
+ channels: set[str] = field(default_factory=set)
+ lock: threading.Lock = field(default_factory=threading.Lock)
+
+ def send(self, data: bytes) -> None:
+ with self.lock:
+ try:
+ self.sock.sendall(data)
+ except OSError:
+ pass # the bot already hung up; nothing to deliver
+
+ def send_line(self, line: str) -> None:
+ self.send((line + "\r\n").encode("utf-8"))
+
+ def close(self) -> None:
+ try:
+ self.sock.shutdown(socket.SHUT_RDWR)
+ except OSError:
+ pass
+ self.sock.close()
+
+
+class FakeTwitchServer:
+ """A threaded IRC server that behaves like Twitch chat for the parts this repo uses."""
+
+ def __init__(
+ self,
+ host: str = DEFAULT_HOST,
+ port: int = 0,
+ *,
+ channels: Iterable[str] = DEFAULT_CHANNELS,
+ suspended_channels: Iterable[str] = (),
+ accepted_token: str | None = None,
+ ssl_context: ssl.SSLContext | None = None,
+ on_event: Callable[[str], None] | None = None,
+ ) -> None:
+ names = [name.lstrip("#").lower() for name in channels]
+ self.channels = set(names)
+ self.default_channel = names[0] # where say() and the console send chat by default
+ self.suspended_channels = {name.lstrip("#").lower() for name in suspended_channels}
+ self.accepted_token = accepted_token # None: any "oauth:..." password logs in
+ self.ssl_context = ssl_context
+ self.on_event = on_event or (lambda text: None)
+ self._listener = socket.create_server((host, port))
+ self.address = self._listener.getsockname()[:2]
+ self._clients: list[_Client] = []
+ self._sent: list[tuple[int, str]] = [] # (client number, line the bot sent)
+ self._joins: dict[str, int] = {} # channel -> how many successful JOINs so far
+ self._changed = threading.Condition()
+ self._numbers = itertools.count(1)
+ self._message_ids = itertools.count(1)
+ self._running = False
+ self._threads: list[threading.Thread] = []
+
+ # ----- start / stop ---------------------------------------------------------
+
+ def start(self) -> FakeTwitchServer:
+ self._running = True
+ self._spawn(self._accept_loop)
+ return self
+
+ def stop(self) -> None:
+ self._running = False
+ self._listener.close()
+ self.disconnect_all()
+ for thread in self._threads:
+ thread.join(timeout=2)
+
+ def __enter__(self) -> FakeTwitchServer:
+ return self.start()
+
+ def __exit__(self, *exc_info: object) -> None:
+ self.stop()
+
+ @property
+ def host(self) -> str:
+ return self.address[0]
+
+ @property
+ def port(self) -> int:
+ return self.address[1]
+
+ # ----- things a test (or the console) can make the server do ---------------
+
+ def say(self, user: str, text: str, *, channel: str | None = None, display_name: str | None = None) -> None:
+ """Send a chat message from viewer ``user`` to every bot in ``channel``."""
+ channel = (channel or self.default_channel).lower()
+ login = user.lower()
+ tags = {
+ "badge-info": "",
+ "badges": "",
+ "color": "#1E90FF",
+ "display-name": display_name or user,
+ "emotes": "",
+ "first-msg": "0",
+ "flags": "",
+ "id": f"00000000-0000-0000-0000-{next(self._message_ids):012d}",
+ "mod": "0",
+ "room-id": "1000",
+ "subscriber": "0",
+ "tmi-sent-ts": str(int(time.time() * 1000)),
+ "turbo": "0",
+ "user-id": str(2000 + sum(map(ord, login))),
+ "user-type": "",
+ }
+ line = f":{login}!{login}@{login}.{SERVER_NAME} PRIVMSG #{channel} :{text}"
+ for client in self._clients_in(channel):
+ client.send_line(self._with_tags(client, tags, line))
+
+ def ping(self, text: str = SERVER_NAME) -> None:
+ self.send_line_to_all(f"PING :{text}")
+
+ def request_reconnect(self) -> None:
+ """Twitch sends RECONNECT before maintenance. Only bots with twitch.tv/commands get it."""
+ for client in self._snapshot_clients():
+ if "twitch.tv/commands" in client.caps:
+ client.send_line(f":{SERVER_NAME} RECONNECT")
+
+ def send_line_to_all(self, line: str) -> None:
+ for client in self._snapshot_clients():
+ client.send_line(line)
+
+ def send_raw(self, data: bytes) -> None:
+ """Send exact bytes, e.g. two lines in one packet or half a line."""
+ for client in self._snapshot_clients():
+ client.send(data)
+
+ def disconnect_all(self) -> None:
+ for client in self._snapshot_clients():
+ client.close()
+
+ # ----- things a test can check ----------------------------------------------
+
+ def sent_lines(self) -> list[str]:
+ """Every line the bots sent, oldest first."""
+ with self._changed:
+ return [line for _, line in self._sent]
+
+ def wait_for(self, predicate: Callable[[str], bool], timeout: float | None = 5.0, count: int = 1) -> str:
+ """Wait until bots have sent ``count`` lines that match ``predicate``; return the last of them.
+ ``timeout=None`` waits forever."""
+ def matches() -> list[str]:
+ return [line for _, line in self._sent if predicate(line)]
+
+ with self._changed:
+ if not self._changed.wait_for(lambda: len(matches()) >= count, timeout):
+ sent = [line for _, line in self._sent]
+ raise TimeoutError(f"wanted {count} matching line(s) within {timeout} s; bots sent {sent}")
+ return matches()[count - 1]
+
+ def wait_for_join(self, channel: str | None = None, timeout: float | None = 5.0, count: int = 1) -> None:
+ """Wait until a bot is in ``channel`` and there have been ``count`` JOINs to it so far.
+ ``timeout=None`` waits forever."""
+ wanted = (channel or self.default_channel).lower()
+ with self._changed:
+ joined = self._changed.wait_for(lambda: self._clients_in(wanted) and self._joins.get(wanted, 0) >= count, timeout)
+ if not joined:
+ raise TimeoutError(f"no bot joined #{wanted} {count} time(s) within {timeout} s")
+
+ # ----- the server itself ----------------------------------------------------
+
+ def _spawn(self, target: Callable[..., None], *args: object) -> None:
+ thread = threading.Thread(target=target, args=args, daemon=True)
+ thread.start()
+ self._threads.append(thread)
+
+ def _accept_loop(self) -> None:
+ # On Linux, closing a socket does not wake up accept(); a short timeout lets stop() end this loop.
+ self._listener.settimeout(ACCEPT_POLL)
+ while self._running:
+ try:
+ sock, _ = self._listener.accept()
+ except TimeoutError:
+ continue
+ except OSError:
+ return # listener closed by stop()
+ sock.settimeout(None)
+ self._spawn(self._serve, sock)
+
+ def _serve(self, sock: socket.socket) -> None:
+ if self.ssl_context is not None:
+ try:
+ sock = self.ssl_context.wrap_socket(sock, server_side=True)
+ except (OSError, ssl.SSLError):
+ sock.close()
+ return
+ client = _Client(next(self._numbers), sock)
+ with self._changed:
+ self._clients.append(client)
+ try:
+ self._read_lines(client)
+ finally:
+ with self._changed:
+ self._clients.remove(client)
+ self._changed.notify_all()
+ client.close()
+ self.on_event(f"bot #{client.number} disconnected")
+
+ def _read_lines(self, client: _Client) -> None:
+ buffer = b""
+ while True:
+ try:
+ data = client.sock.recv(4096)
+ except OSError:
+ return
+ if not data:
+ return
+ buffer += data
+ *lines, buffer = buffer.split(b"\n")
+ for raw in lines:
+ line = raw.rstrip(b"\r").decode("utf-8", errors="replace")
+ with self._changed:
+ self._sent.append((client.number, line))
+ self._changed.notify_all()
+ if not self._handle(client, line):
+ return
+
+ def _handle(self, client: _Client, line: str) -> bool:
+ """React to one line from a bot. Returns False when the connection should end."""
+ command, _, rest = line.partition(" ")
+ handler = getattr(self, f"_on_{command.lower()}", self._on_unknown)
+ return handler(client, rest, command)
+
+ def _on_cap(self, client: _Client, rest: str, command: str) -> bool:
+ wanted = rest.removeprefix("REQ").strip().lstrip(":").split()
+ if all(cap in KNOWN_CAPABILITIES for cap in wanted):
+ client.caps.update(wanted)
+ client.send_line(f":{SERVER_NAME} CAP * ACK :{' '.join(wanted)}")
+ else:
+ client.send_line(f":{SERVER_NAME} CAP * NAK :{' '.join(wanted)}")
+ return True
+
+ def _on_pass(self, client: _Client, rest: str, command: str) -> bool:
+ client.password = rest.strip()
+ return True
+
+ def _on_nick(self, client: _Client, rest: str, command: str) -> bool:
+ if client.password is None:
+ return self._refuse(client, BAD_AUTH_ORDER)
+ token_ok = client.password.startswith("oauth:") and client.password != "oauth:"
+ if self.accepted_token is not None:
+ token_ok = client.password == self.accepted_token
+ if not token_ok:
+ return self._refuse(client, AUTH_FAILED)
+ client.nick = rest.strip().lower()
+ for number, text in WELCOME_LINES:
+ client.send_line(f":{SERVER_NAME} {number} {client.nick} :{text}")
+ if "twitch.tv/commands" in client.caps:
+ tags = {"badge-info": "", "badges": "", "color": "", "display-name": client.nick,
+ "emote-sets": "0", "user-id": "1", "user-type": ""}
+ client.send_line(self._with_tags(client, tags, f":{SERVER_NAME} GLOBALUSERSTATE"))
+ self.on_event(f"bot #{client.number} logged in as {client.nick}")
+ return True
+
+ def _on_user(self, client: _Client, rest: str, command: str) -> bool:
+ return True # IRC libraries send USER; Twitch does not need it
+
+ def _on_join(self, client: _Client, rest: str, command: str) -> bool:
+ nick = client.nick
+ for name in rest.strip().split(","):
+ channel = name.strip().lstrip("#").lower()
+ if channel in self.suspended_channels:
+ client.send_line(self._with_tags(client, {"msg-id": "msg_channel_suspended"},
+ f":{SERVER_NAME} NOTICE #{channel} :{CHANNEL_SUSPENDED}"))
+ continue
+ if channel not in self.channels:
+ continue # Twitch quietly drops JOINs for channels that do not exist
+ with self._changed: # count the bot in before it hears 366, so chat can reach it right away
+ client.channels.add(channel)
+ self._joins[channel] = self._joins.get(channel, 0) + 1
+ self._changed.notify_all()
+ self.on_event(f"bot #{client.number} joined #{channel}")
+ client.send_line(f":{nick}!{nick}@{nick}.{SERVER_NAME} JOIN #{channel}")
+ client.send_line(f":{nick}.{SERVER_NAME} 353 {nick} = #{channel} :{nick}")
+ client.send_line(f":{nick}.{SERVER_NAME} 366 {nick} #{channel} :End of /NAMES list")
+ if "twitch.tv/commands" in client.caps:
+ client.send_line(self._with_tags(client, self._userstate(nick), f":{SERVER_NAME} USERSTATE #{channel}"))
+ roomstate = {"room-id": "1000", "slow": "0"}
+ client.send_line(self._with_tags(client, roomstate, f":{SERVER_NAME} ROOMSTATE #{channel}"))
+ return True
+
+ def _on_part(self, client: _Client, rest: str, command: str) -> bool:
+ channel = rest.strip().lstrip("#").lower()
+ client.channels.discard(channel)
+ client.send_line(f":{client.nick}!{client.nick}@{client.nick}.{SERVER_NAME} PART #{channel}")
+ return True
+
+ def _on_privmsg(self, client: _Client, rest: str, command: str) -> bool:
+ target, _, text = rest.partition(" :")
+ self.on_event(f"{client.nick} -> {target}: {text}")
+ if "twitch.tv/commands" in client.caps:
+ # Twitch does not echo the bot's own PRIVMSG back; it sends USERSTATE instead.
+ client.send_line(self._with_tags(client, self._userstate(client.nick), f":{SERVER_NAME} USERSTATE {target}"))
+ return True
+
+ def _on_pong(self, client: _Client, rest: str, command: str) -> bool:
+ self.on_event(f"bot #{client.number} answered: PONG {rest}")
+ return True
+
+ def _on_quit(self, client: _Client, rest: str, command: str) -> bool:
+ return False
+
+ def _on_unknown(self, client: _Client, rest: str, command: str) -> bool:
+ client.send_line(f":{SERVER_NAME} 421 {client.nick or '*'} {command} :Unknown command")
+ return True
+
+ def _refuse(self, client: _Client, notice: str) -> bool:
+ client.send_line(f":{SERVER_NAME} NOTICE * :{notice}")
+ self.on_event(f"bot #{client.number} refused: {notice}")
+ return False # Twitch closes the connection after a failed login
+
+ def _userstate(self, nick: str | None) -> dict[str, str]:
+ return {"badge-info": "", "badges": "", "color": "", "display-name": nick or "",
+ "emote-sets": "0", "mod": "0", "subscriber": "0", "user-type": ""}
+
+ @staticmethod
+ def _with_tags(client: _Client, tags: dict[str, str], line: str) -> str:
+ return f"{format_tags(tags)} {line}" if "twitch.tv/tags" in client.caps else line
+
+ def _snapshot_clients(self) -> list[_Client]:
+ with self._changed:
+ return list(self._clients)
+
+ def _clients_in(self, channel: str) -> list[_Client]:
+ return [client for client in self._snapshot_clients() if channel in client.channels]
+
+
+# ----- console mode -------------------------------------------------------------
+
+CONSOLE_HELP = """\
+Type a chat line as name: message (example: alice: hello there)
+Commands: /ping /reconnect /quit
+Lines you type before a bot joins wait until one does."""
+
+
+def run_console(server: FakeTwitchServer, lines: Iterable[str], delay: float) -> None:
+ """Read lines from the keyboard (or a pipe) and play them into the chat."""
+ channel = server.default_channel
+ commands = {
+ "/ping": ("PING sent", server.ping),
+ "/reconnect": ("RECONNECT sent", server.request_reconnect),
+ }
+ for raw in lines:
+ line = raw.strip()
+ if not line:
+ continue
+ if line == "/quit":
+ return
+ server.wait_for_join(channel, timeout=None)
+ if line in commands:
+ note, action = commands[line]
+ print(f"[server] {note}", flush=True)
+ action()
+ else:
+ user, separator, text = (part.strip() for part in line.partition(":"))
+ if not separator or not user:
+ print("[server] write it as name: message", flush=True)
+ continue
+ print(f"[chat] {user}: {text}", flush=True)
+ server.say(user, text, channel=channel)
+ time.sleep(delay)
+
+
+def main(argv: list[str] | None = None) -> None:
+ parser = argparse.ArgumentParser(description="Pretend Twitch chat server for practice (plain TCP, no TLS).")
+ parser.add_argument("--host", default=DEFAULT_HOST, help=f"address to listen on (default {DEFAULT_HOST})")
+ parser.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"port to listen on (default {DEFAULT_PORT})")
+ parser.add_argument("--channel", action="append",
+ help=f"channel that exists (default {DEFAULT_CHANNELS[0]}); repeat for more")
+ parser.add_argument("--token", help="only accept this PASS value (default: any oauth:... value)")
+ parser.add_argument("--delay", type=float, default=0.3, help="seconds to wait after each typed line (default 0.3)")
+ args = parser.parse_args(argv)
+
+ server = FakeTwitchServer(
+ args.host, args.port,
+ channels=args.channel or DEFAULT_CHANNELS,
+ accepted_token=args.token,
+ on_event=lambda text: print(f"[server] {text}", flush=True),
+ )
+ with server:
+ host, port = server.address
+ channel = server.default_channel
+ print(f"[server] fake Twitch chat listening on {host}:{port} (plain TCP, not the real Twitch)", flush=True)
+ print("[server] point the bot here, for example with these lines in .env:", flush=True)
+ print(f" TWITCH_IRC_HOST={host} TWITCH_IRC_PORT={port} TWITCH_IRC_TLS=false", flush=True)
+ print(f" TWITCH_CHANNEL={channel} TWITCH_BOT_NAME=any_name "
+ f"TWITCH_OAUTH_TOKEN={args.token or 'oauth:anything'}", flush=True)
+ print(CONSOLE_HELP, flush=True)
+ try:
+ run_console(server, sys.stdin, args.delay)
+ if not sys.stdin.isatty():
+ # input came from a pipe: keep serving until Ctrl+C so the bot can finish
+ threading.Event().wait()
+ except KeyboardInterrupt:
+ pass
+ print("[server] stopped", flush=True)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/twitchSocket.py b/twitchSocket.py
index 80286ec..c6c5cbc 100644
--- a/twitchSocket.py
+++ b/twitchSocket.py
@@ -1,64 +1,57 @@
-import socket
-
-
-class twitchSocket(object):
- def __init__(self, channelName:str, botName:str, botAuth:str):
- self.__HOST = 'irc.twitch.tv'
- self.__PORT = 6667
- self.__NICK = botName
- self.__PASS = botAuth
- self.__CHANNEL = channelName
-
- self.__s = self.__setSocket()
-
- while True:
- if 'End of /NAMES list' in self.__getPacket():
- break
-
- def __setSocket(self) -> socket:
- s = socket.socket()
- s.connect((self.__HOST, self.__PORT))
- s.send(bytes('PASS ' + self.__PASS + '\r\n', 'UTF-8'))
- s.send(bytes('NICK ' + self.__NICK + '\r\n', 'UTF-8'))
- s.send(bytes('JOIN #' + self.__CHANNEL + '\r\n', 'UTF-8'))
-
- return s
-
- def sendMsg(self, message:str):
- self.__s.send(bytes('PRIVMSG #' + self.__CHANNEL + ' :' + message + '\r\n', 'UTF-8'))
- print(self.__NICK + ': ' + message)
-
- def __sendPong(self):
- self.__s.send(bytes('PONG :tmi.twitch.tv\r\n', 'UTF-8'))
-
- def __getPacket(self) -> str:
- return self.__s.recv(1024).rstrip().decode('UTF-8')
-
- def __getDataFromPacket(self) -> str:
- # rawData Sample:
- # :usr!usr@usr.tmi.twitch.tv PRIVMSG #channel :string
- rawData = self.__getPacket()
- if 'PING :tmi.twitch.tv' in rawData:
- self.__sendPong()
- # Filter the twitch send <3
- # Avoid twitch ping you without any message.
- if len(rawData.split(':')) < 3:
- return None
- return rawData
-
- def getMsg(self) -> tuple:
- username = None
- message = None
- try:
- rawData = self.__getDataFromPacket()
- # get message and username with extracting raw data.
- if rawData is not None:
- username = rawData.split(':')[1].split('!')[0]
- message = rawData.split(':')[2].strip()
- print(username + ': '+ message)
- except Exception as e:
- print(rawData)
- print(e)
-
- return username, message
-
+"""The 2019 names, kept so scripts written for the old version keep working.
+
+ from twitchSocket import twitchSocket
+
+ ts = twitchSocket(channelName="...", botName="...", botAuth="oauth:...")
+ username, message = ts.getMsg()
+ ts.sendMsg(message)
+
+New code should use ``twitch_chat.TwitchChat`` (PEP 8 names, see the README).
+This file is the only place that knows the old names; nothing else imports it.
+
+What changed underneath (all bug fixes, same calls):
+* it connects to irc.chat.twitch.tv:6697 with TLS instead of irc.twitch.tv:6667 in plain text;
+* getMsg() waits for the next real chat message. It used to return (None, None) for
+ PINGs and other server lines, and main.py then crashed on sendMsg(None);
+* messages that contain ":" come through whole, and the other 2019 bugs listed in the
+ README ("What the 2019 version got wrong") are gone too.
+
+When can this file go? Not before 2027-11-04, one year after the new names were
+announced in the README. Before deleting it, search GitHub for "from twitchSocket import"
+to see whether anyone still uses it, then remove this file and its test
+(tests/test_legacy.py) and the "Code from 2019" section of the README together.
+"""
+
+from __future__ import annotations
+
+import warnings
+from typing import Any
+
+from twitch_chat import TwitchChat
+
+
+class twitchSocket(TwitchChat): # noqa: N801 - the 2019 name, on purpose
+ def __init__(self, channelName: str, botName: str, botAuth: str, **options: Any) -> None: # noqa: N803
+ """Connects and joins right away, like the 2019 version did.
+
+ ``options`` go to TwitchChat (host, port, use_tls, ...); old code passes none.
+ """
+ warnings.warn(
+ "twitchSocket is the 2019 name; use twitch_chat.TwitchChat (see README)",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+ super().__init__(channel=channelName, nick=botName, token=botAuth, **options)
+ self._bot_name = botName # printed as given, like before
+ self.connect()
+
+ def getMsg(self) -> tuple[str, str]: # noqa: N802
+ """Wait for the next chat message; print and return (username, message)."""
+ message = self.read_message()
+ print(message.user + ": " + message.text)
+ return message.user, message.text
+
+ def sendMsg(self, message: str) -> None: # noqa: N802
+ """Send ``message`` to the channel and print it."""
+ self.send_message(message)
+ print(self._bot_name + ": " + message)
diff --git a/twitch_chat.py b/twitch_chat.py
new file mode 100644
index 0000000..5416622
--- /dev/null
+++ b/twitch_chat.py
@@ -0,0 +1,424 @@
+"""Talk to Twitch chat with nothing but a socket.
+
+Twitch chat is an IRC server: you open a connection, send short text commands,
+and Twitch sends text lines back. This file does that by hand, so you can see
+every step. Read it from top to bottom; each STEP matches a section of the README.
+
+ STEP 1 open a connection (TLS) ............ TwitchChat._open_socket
+ STEP 2 ask for Twitch's extra features .... TwitchChat._request_capabilities
+ STEP 3 log in with PASS and NICK .......... TwitchChat._log_in
+ STEP 4 join a channel ..................... TwitchChat._join
+ STEP 5 cut the byte stream into lines ..... LineBuffer
+ STEP 6 split a line into its parts ........ parse_line
+ STEP 7 answer PING with PONG .............. TwitchChat._next_message
+ STEP 8 send a message, not too fast ....... TwitchChat.send_message, RateLimiter
+ STEP 9 notice when the connection ends .... ConnectionClosed
+
+Everything Twitch-specific below comes from Twitch's docs, checked on 2026-11-04:
+https://dev.twitch.tv/docs/chat/irc/ and https://dev.twitch.tv/docs/chat/#rate-limits
+
+Only the Python standard library is used.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+import socket
+import ssl
+import time
+from collections import deque
+from collections.abc import Callable, Iterable, Iterator
+from dataclasses import dataclass, field
+
+log = logging.getLogger("twitch_chat")
+
+# ----- Where Twitch chat lives -------------------------------------------------
+TWITCH_IRC_HOST = "irc.chat.twitch.tv"
+TWITCH_IRC_TLS_PORT = 6697 # encrypted, use this one
+TWITCH_IRC_PLAIN_PORT = 6667 # not encrypted: your token travels in clear text
+
+# Extra features we ask for in STEP 2. "tags" adds details such as display-name
+# to each message; Twitch only gives tags to bots that also ask for "commands".
+DEFAULT_CAPABILITIES = ("twitch.tv/tags", "twitch.tv/commands")
+
+# How long to wait for Twitch to answer while logging in and joining (seconds).
+HANDSHAKE_TIMEOUT = 15.0
+
+# Twitch's limits for a bot that is not the broadcaster, a moderator or a VIP:
+# at most 20 messages in any 30 seconds, and 1 message per second in a channel.
+# Go over and Twitch ignores the bot's messages for an hour. Each pair is
+# (messages, seconds).
+DEFAULT_RATE_LIMITS = ((20, 30.0), (1, 1.0))
+
+# STEP 9: seconds to wait before reconnecting after the 1st, 2nd, ... failure in a row.
+RECONNECT_DELAYS = (1, 2, 4, 8, 16, 30)
+
+RECV_SIZE = 4096 # bytes to read per recv() call; any size works, see STEP 5
+ENCODING = "utf-8"
+
+
+# ----- What can go wrong --------------------------------------------------------
+class TwitchChatError(Exception):
+ """Base class for the errors below."""
+
+
+class AuthenticationError(TwitchChatError):
+ """Twitch refused the login (STEP 3). Usually a wrong or expired token."""
+
+
+class JoinError(TwitchChatError):
+ """Twitch did not let us into the channel (STEP 4)."""
+
+
+class ConnectionClosed(TwitchChatError, ConnectionError):
+ """The connection ended, or Twitch asked us to reconnect (STEP 9). Connect again."""
+
+
+def reconnect_delay(failures: int) -> int:
+ """STEP 9: wait longer after each failure in a row, so a broken network is not hammered."""
+ return RECONNECT_DELAYS[min(failures, len(RECONNECT_DELAYS) - 1)]
+
+
+def irc_password(token: str) -> str:
+ """The IRC password is the token with "oauth:" in front. Accept it with or without."""
+ return token if token.startswith("oauth:") else "oauth:" + token
+
+
+def default_port(use_tls: bool) -> int:
+ return TWITCH_IRC_TLS_PORT if use_tls else TWITCH_IRC_PLAIN_PORT
+
+
+# ----- The client ------------------------------------------------------------------
+class TwitchChat:
+ """One bot account in one Twitch channel.
+
+ >>> chat = TwitchChat(channel="your_channel", nick="your_bot", token="oauth:...")
+ >>> chat.connect()
+ >>> for message in chat.messages():
+ ... chat.send_message(message.text)
+ """
+
+ def __init__(
+ self,
+ channel: str,
+ nick: str,
+ token: str,
+ *,
+ host: str = TWITCH_IRC_HOST,
+ port: int | None = None,
+ use_tls: bool = True,
+ capabilities: Iterable[str] = DEFAULT_CAPABILITIES,
+ rate_limits: Iterable[tuple[int, float]] = DEFAULT_RATE_LIMITS,
+ handshake_timeout: float = HANDSHAKE_TIMEOUT,
+ ssl_context: ssl.SSLContext | None = None,
+ ) -> None:
+ # Twitch wants lowercase names. Channels are written "#name" in IRC,
+ # so we accept both "name" and "#name" and store the bare name.
+ self.channel = channel.strip().lstrip("#").lower()
+ self.nick = nick.strip().lower()
+ self._password = irc_password(token)
+ self.host = host
+ self.port = port or default_port(use_tls)
+ self.use_tls = use_tls
+ self.capabilities = tuple(capabilities)
+ self.handshake_timeout = handshake_timeout
+ self._ssl_context = ssl_context
+ self._limiter = RateLimiter(rate_limits)
+ self._sock: socket.socket | None = None
+ self._buffer = LineBuffer()
+ self._lines: deque[str] = deque()
+
+ # STEP 1 to 4 ------------------------------------------------------------------
+ def connect(self) -> None:
+ """Open the connection, log in and join the channel. Returns once we are in.
+ Calling it again after ConnectionClosed starts over with a fresh connection."""
+ self.close()
+ self._buffer = LineBuffer() # leftovers from an old connection mean nothing now
+ self._lines.clear()
+ self._sock = self._open_socket() # STEP 1
+ try:
+ self._request_capabilities() # STEP 2
+ self._log_in() # STEP 3
+ self._join() # STEP 4
+ except BaseException:
+ self.close() # don't leave a half-open connection behind
+ raise
+ self._sock.settimeout(None) # from now on, wait as long as chat is quiet
+ log.info("joined #%s as %s", self.channel, self.nick)
+
+ def _open_socket(self) -> socket.socket:
+ """STEP 1: a TCP connection, wrapped in TLS so nobody can read the token."""
+ log.info("connecting to %s:%s (%s)", self.host, self.port, "TLS" if self.use_tls else "plain text")
+ sock = socket.create_connection((self.host, self.port), timeout=self.handshake_timeout)
+ if not self.use_tls:
+ return sock
+ context = self._ssl_context or ssl.create_default_context()
+ return context.wrap_socket(sock, server_hostname=self.host)
+
+ def _request_capabilities(self) -> None:
+ """STEP 2: CAP REQ asks for Twitch's extras. Twitch answers "CAP * ACK" (yes)
+ or "CAP * NAK" (no); we read that answer in STEP 3 together with the login reply.
+ """
+ if self.capabilities:
+ self._send_line("CAP REQ :" + " ".join(self.capabilities))
+
+ def _log_in(self) -> None:
+ """STEP 3: PASS (the token) must come before NICK (the bot's login name).
+
+ Success is a burst of numbered lines starting with "001". A wrong token gets
+ "NOTICE * :Login authentication failed" and Twitch hangs up.
+ """
+ self._send_line("PASS " + self._password, secret=True)
+ self._send_line("NICK " + self.nick)
+ self._wait_for("001", error=AuthenticationError, waiting_for="the login reply")
+
+ def _join(self) -> None:
+ """STEP 4: JOIN #channel. Twitch answers with JOIN, a member list (353)
+ and "366 ... :End of /NAMES list". For a channel that does not exist Twitch
+ answers nothing at all, which is why we wait with a timeout.
+ """
+ self._send_line(f"JOIN #{self.channel}")
+ self._wait_for(
+ "366",
+ error=JoinError,
+ waiting_for=f"the JOIN reply (is #{self.channel} spelled right? Twitch ignores unknown channels)",
+ )
+
+ def _wait_for(self, command: str, *, error: type[TwitchChatError], waiting_for: str) -> IrcMessage:
+ """Read lines until Twitch sends ``command``. During login and join, a NOTICE
+ means Twitch said no, so it becomes ``error`` with Twitch's own words."""
+ try:
+ while True:
+ message = self._next_message()
+ if message.command == command:
+ return message
+ if message.command == "NOTICE":
+ raise error(f"Twitch said: {message.trailing}")
+ if message.command == "CAP" and "NAK" in message.params:
+ log.warning("Twitch refused capabilities: %s", message.trailing)
+ except TimeoutError:
+ raise error(f"no answer from Twitch within {self.handshake_timeout:g} s while waiting for {waiting_for}") from None
+
+ # STEP 5 to 7: reading ----------------------------------------------------------------
+ def messages(self) -> Iterator[ChatMessage]:
+ """Chat messages, one after another, forever (until the connection ends)."""
+ while True:
+ yield self.read_message()
+
+ def read_message(self) -> ChatMessage:
+ """Wait for the next chat message. PING and other server lines are handled here."""
+ while True:
+ message = self._next_message()
+ if message.command == "PRIVMSG":
+ return ChatMessage.from_irc(message)
+ if message.command == "NOTICE":
+ # e.g. "Your message was not sent because you are sending messages too quickly."
+ log.warning("Twitch notice: %s", message.trailing)
+
+ def _next_message(self) -> IrcMessage:
+ """Read one line (STEP 5), parse it (STEP 6) and handle the lines that are about
+ the connection itself (STEP 7 and STEP 9)."""
+ line = self._next_line()
+ log.debug("< %s", line)
+ message = parse_line(line)
+ if message.command == "PING":
+ # STEP 7: Twitch checks that we are still here. Answer with the same text
+ # it sent, or Twitch closes the connection.
+ self._send_line("PONG :" + message.trailing)
+ elif message.command == "RECONNECT":
+ # STEP 9: Twitch is about to restart this server; we should come back.
+ raise ConnectionClosed("Twitch asked us to reconnect (server maintenance)")
+ return message
+
+ def _next_line(self) -> str:
+ """STEP 5: one complete line. recv() returns whatever bytes have arrived:
+ maybe half a line, maybe five lines. LineBuffer keeps the leftovers."""
+ while not self._lines:
+ data = self._require_socket().recv(RECV_SIZE)
+ if not data:
+ # STEP 9: recv() returns b"" only when the other side closed the connection.
+ raise ConnectionClosed("Twitch closed the connection")
+ self._lines.extend(self._buffer.feed(data))
+ return self._lines.popleft()
+
+ # STEP 8: sending ----------------------------------------------------------------------
+ def send_message(self, text: str) -> None:
+ """Say ``text`` in the channel. Waits first if we would go over the rate limit."""
+ # One IRC line is one command, so a line break would end the message and
+ # start a new (unwanted) command. Twitch shows messages on one line anyway.
+ text = " ".join(text.splitlines())
+ waited = self._limiter.wait()
+ if waited:
+ log.info("waited %.1f s to stay under Twitch's rate limit", waited)
+ self._send_line(f"PRIVMSG #{self.channel} :{text}")
+
+ def _send_line(self, line: str, *, secret: bool = False) -> None:
+ """Every command goes out through here: text + CRLF, encoded as UTF-8."""
+ if "\r" in line or "\n" in line:
+ raise ValueError("an IRC line cannot contain CR or LF")
+ log.debug("> %s", line.split(" ", 1)[0] + " ***" if secret else line)
+ self._require_socket().sendall((line + "\r\n").encode(ENCODING)) # sendall: send() may send only part
+
+ # Housekeeping ---------------------------------------------------------------------------
+ def close(self) -> None:
+ if self._sock is not None:
+ self._sock.close()
+ self._sock = None
+
+ def __enter__(self) -> TwitchChat:
+ return self
+
+ def __exit__(self, *exc_info: object) -> None:
+ self.close()
+
+ def _require_socket(self) -> socket.socket:
+ if self._sock is None:
+ raise ConnectionClosed("not connected; call connect() first")
+ return self._sock
+
+
+# ----- STEP 5: from a byte stream to lines --------------------------------------------
+class LineBuffer:
+ """TCP delivers a stream of bytes, not messages. Twitch ends every line with
+ CRLF ("\\r\\n"), so we collect bytes and hand out only complete lines.
+
+ We split the *bytes* and decode each finished line. Decoding chunk by chunk
+ would break when a character such as "你" (3 bytes in UTF-8) is cut in two.
+ """
+
+ def __init__(self) -> None:
+ self._pending = b""
+
+ def feed(self, data: bytes) -> list[str]:
+ *lines, self._pending = (self._pending + data).split(b"\n")
+ # rstrip("\r") also accepts a bare "\n"; errors="replace" keeps one bad byte from killing the bot
+ return [line.rstrip(b"\r").decode(ENCODING, errors="replace") for line in lines if line.strip()]
+
+
+# ----- STEP 6: the parts of one line --------------------------------------------------
+@dataclass(frozen=True)
+class IrcMessage:
+ """One IRC line, split into its four parts. See the picture in the README.
+
+ @badge-info=;display-name=Alice;... :alice!alice@alice.tmi.twitch.tv PRIVMSG #demo :hi: there
+ |------------- tags --------------| |------------ prefix ------------| |command| |params----|
+ """
+
+ tags: dict[str, str]
+ prefix: str | None # who sent it: "nick!user@host", or a server name
+ command: str # PRIVMSG, PING, JOIN, NOTICE, or a number such as 001
+ params: list[str] = field(default_factory=list)
+
+ @property
+ def trailing(self) -> str:
+ """The last parameter, which is the only one that may contain spaces."""
+ return self.params[-1] if self.params else ""
+
+ @property
+ def nick(self) -> str | None:
+ """The sender's login name: the part of the prefix before "!"."""
+ return self.prefix.split("!", 1)[0] if self.prefix else None
+
+
+def parse_line(line: str) -> IrcMessage:
+ """Split one IRC line (without its CRLF) into tags, prefix, command and params.
+
+ Never use line.split(":") for this: the chat text itself may contain ":".
+ The rule is: the first " :" starts the last parameter, and everything after it,
+ colons and spaces included, belongs to that parameter.
+ """
+ rest = line
+ tags: dict[str, str] = {}
+ if rest.startswith("@"): # IRCv3 tags: "@key=value;key2=value2 "
+ raw_tags, _, rest = rest[1:].partition(" ")
+ tags = parse_tags(raw_tags)
+ prefix = None
+ if rest.startswith(":"): # prefix: ":nick!user@host "
+ prefix, _, rest = rest[1:].partition(" ")
+ middle, has_trailing, trailing = rest.partition(" :")
+ command, *params = middle.split()
+ if has_trailing:
+ params.append(trailing)
+ return IrcMessage(tags=tags, prefix=prefix, command=command.upper(), params=params)
+
+
+def parse_tags(raw: str) -> dict[str, str]:
+ """"badges=;display-name=Alice" -> {"badges": "", "display-name": "Alice"}"""
+ tags = {}
+ for item in filter(None, raw.split(";")):
+ key, _, value = item.partition("=")
+ tags[key] = unescape_tag_value(value)
+ return tags
+
+
+# Tag values cannot contain ";" or spaces, so IRCv3 writes them as escapes.
+# https://ircv3.net/specs/extensions/message-tags#escaping-values
+_TAG_UNESCAPES = {":": ";", "s": " ", "\\": "\\", "r": "\r", "n": "\n"}
+
+
+def unescape_tag_value(value: str) -> str:
+ r""""Hello\sthere" -> "Hello there". An unknown escape "\x" becomes "x", and a
+ lone "\" at the very end is dropped, as the spec says."""
+ return re.sub(r"\\(.?)", lambda match: _TAG_UNESCAPES.get(match[1], match[1]), value)
+
+
+@dataclass(frozen=True)
+class ChatMessage:
+ """A chat message, the part of a PRIVMSG line a bot usually cares about."""
+
+ channel: str # without "#"
+ user: str # login name, lowercase
+ text: str
+ tags: dict[str, str] = field(default_factory=dict)
+
+ @property
+ def display_name(self) -> str:
+ """How the name looks in chat (may have capitals or CJK); the login if not set."""
+ return self.tags.get("display-name") or self.user
+
+ @classmethod
+ def from_irc(cls, message: IrcMessage) -> ChatMessage:
+ # PRIVMSG params are [#channel, text]
+ return cls(
+ channel=message.params[0].lstrip("#"),
+ user=message.nick or "",
+ text=message.trailing,
+ tags=message.tags,
+ )
+
+
+# ----- STEP 8: staying under the rate limit ---------------------------------------------
+class RateLimiter:
+ """Remembers when we sent recently, and waits when one more message would break
+ a rule such as "20 messages in any 30 seconds".
+
+ ``rules`` is a list of (messages, seconds) pairs; all of them must hold.
+ ``clock`` and ``sleep`` can be swapped out in tests.
+ """
+
+ def __init__(
+ self,
+ rules: Iterable[tuple[int, float]],
+ *,
+ clock: Callable[[], float] = time.monotonic,
+ sleep: Callable[[float], None] = time.sleep,
+ ) -> None:
+ self.rules = tuple(rules)
+ self._clock = clock
+ self._sleep = sleep
+ self._sent: deque[float] = deque(maxlen=max((limit for limit, _ in self.rules), default=0))
+
+ def wait(self) -> float:
+ """Block until sending is allowed, record the send, and return the seconds waited."""
+ now = self._clock()
+ # For "N messages per T seconds": the N-th most recent send must be at least T seconds ago.
+ ready_at = max(
+ (self._sent[-limit] + period for limit, period in self.rules if len(self._sent) >= limit),
+ default=now,
+ )
+ delay = max(0.0, ready_at - now)
+ if delay:
+ self._sleep(delay)
+ self._sent.append(now + delay)
+ return delay
+ 
+ 
-brightgreen){kind=icon}
+
+