Init

.gitignore

@@ -0,0 +1,13 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+
+.env
+tokens.json

.python-version

@@ -0,0 +1 @@
+3.13

pyproject.toml

@@ -0,0 +1,11 @@
+[project]
+name = "twitch-chat-bot"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "aiofiles>=24.1.0",
+    "pydantic-ai>=0.1.3",
+    "twitchapi>=4.4.0",
+]

src/auth.py

@@ -0,0 +1,71 @@
+import os
+import json
+import aiofiles
+
+from twitchAPI.twitch import Twitch
+from twitchAPI.oauth import UserAuthenticator
+from twitchAPI.type import AuthScope
+
+
+APP_ID = os.environ["TWITCH_APP_ID"]
+APP_SECRET = os.environ["TWITCH_APP_SECRET"]
+USER_SCOPE = [AuthScope.CHAT_READ, AuthScope.CHAT_EDIT]
+
+
+class TokenManager:
+    FILENAME = "tokens.json"
+
+    @classmethod
+    async def load(cls) -> tuple[str, str] | None:
+        try:
+            async with aiofiles.open(cls.FILENAME, "r") as f:
+                data = await f.read()
+
+                json_data = json.loads(data)
+
+            return json_data["auth_token"], json_data["refresh_token"]
+        except Exception as e:
+            print(e)
+
+        return None
+
+    @classmethod
+    async def save(cls, auth_token: str, refresh_token: str):
+        async with aiofiles.open(cls.FILENAME, "w") as f:
+            await f.write(
+                json.dumps({
+                    "auth_token": auth_token,
+                    "refresh_token": refresh_token
+                })
+            )
+
+
+async def get_auth_token(client: Twitch) -> tuple[str, str]:
+    auth = UserAuthenticator(client, USER_SCOPE)
+
+    token_data = await auth.authenticate()
+    if token_data is None:
+        raise RuntimeError("Authorization failed!")
+
+    return token_data
+
+
+async def get_client() -> Twitch:
+    client = Twitch(APP_ID, APP_SECRET)
+
+    saved_token = await TokenManager.load()
+
+    if saved_token:
+        token, refresh_token = saved_token
+    else:
+        token, refresh_token = await get_auth_token(client)
+        await TokenManager.save(token, refresh_token)
+
+    await client.set_user_authentication(
+        token,
+        scope=USER_SCOPE,
+        refresh_token=refresh_token,
+        validate=True,
+    )
+
+    return client

src/chatbot.py

@@ -0,0 +1,53 @@
+from asyncio import sleep
+
+from twitchAPI.type import ChatEvent
+from twitchAPI.chat import Chat, EventData, ChatMessage
+
+from auth import get_client, Twitch
+from handlers import HANDLERS
+
+
+class ChatBot:
+    TARGET_CHANNELS = [
+        "kurbezz",
+        "kamsyll"
+    ]
+
+    def __init__(self, client: Twitch):
+        self.client = client
+
+    @classmethod
+    async def on_ready(cls, ready_event: EventData):
+        print("[system]: Ready!")
+
+        for channel in cls.TARGET_CHANNELS:
+            print(f"[system]: Subscribe to {channel}...")
+            await ready_event.chat.join_room(channel)
+            print(f"[system]: Subscribed to {channel}!")
+
+    @classmethod
+    async def on_message(cls, msg: ChatMessage):
+        print(f"[{msg.user.name}]: {msg.text}")
+
+        for handler in HANDLERS:
+            await handler(msg)
+
+    @classmethod
+    async def run(cls):
+        client = await get_client()
+
+        chat = await Chat(client)
+
+        chat.register_event(ChatEvent.READY, cls.on_ready)
+        chat.register_event(ChatEvent.MESSAGE, cls.on_message)
+
+        chat.start()
+
+        try:
+            while True:
+                await sleep(1)
+        except KeyboardInterrupt:
+            print("[system]: Shutting down...")
+        finally:
+            chat.stop()
+            await client.close()

src/handlers/__init__.py

@@ -0,0 +1,18 @@
+from typing import Callable, Awaitable
+
+from twitchAPI.chat import ChatMessage
+
+from .goida import on_goida_handler
+from .lasqexx import on_lasqexx_message
+from .greetings import on_greetings
+from .farewells import on_farewells
+from .gemini import on_gemini_handler
+
+
+HANDLERS: list[Callable[[ChatMessage], Awaitable[bool]]] = [
+    on_goida_handler,
+    on_lasqexx_message,
+    on_greetings,
+    on_farewells,
+    on_gemini_handler,
+]

src/handlers/farewells.py

@@ -0,0 +1,15 @@
+from twitchAPI.chat import ChatMessage
+
+
+TRIGGER_AND_RESPONSE: list[tuple[str, str]] = [
+    ("пока", "Пока")
+]
+
+
+async def on_farewells(msg: ChatMessage) -> bool:
+    for trigger, response in TRIGGER_AND_RESPONSE:
+        if trigger in msg.text.lower():
+            await msg.reply(response)
+            return True
+
+    return False

src/handlers/gemini.py

@@ -0,0 +1,48 @@
+import os
+
+from twitchAPI.chat import ChatMessage
+
+from pydantic_ai import Agent
+from pydantic_ai.models.gemini import GeminiModel
+from pydantic_ai.providers.google_gla import GoogleGLAProvider
+
+
+GEMINI_API_KEY = os.environ["GEMINI_API_KEY"]
+
+SYSTEM_PROMPT = """
+Ты чат бот на сервисе Twitch.
+К тебе могут обратиться с помощью имен: Булат, курбез, kurbezz.
+
+### Правила:
+1. Отвечай на запросы пользователей
+2. Не признавайся, что ты чат бот или нейросеть. Отвечай, что ты человек.
+3. Не используй запрещенные слова на сервисе Twitch.
+4. Используй для ответа чистый текст или смайлы. (Не markdown или html)
+5. Отвечай коротко и ясно.
+"""
+
+
+model = GeminiModel(
+    'gemini-2.0-flash',
+    provider=GoogleGLAProvider(api_key=GEMINI_API_KEY)
+)
+
+agent = Agent(
+    model=model,
+    retries=5,
+    instrument=True,
+    system_prompt=SYSTEM_PROMPT
+)
+
+
+async def on_gemini_handler(msg: ChatMessage) -> bool:
+    if not msg.text.startswith("!gemini "):
+        return False
+
+    prompt = msg.text[8:]
+
+    result = await agent.run(prompt)
+
+    await msg.reply(result.output)
+
+    return True

src/handlers/goida.py

@@ -0,0 +1,10 @@
+from twitchAPI.chat import ChatMessage
+
+
+async def on_goida_handler(message: ChatMessage) -> bool:
+    if "гойда" not in message.text.lower():
+        return False
+
+    await message.reply("ГООООООООООООООООООООООООООООООООООООЙДА!")
+
+    return True

src/handlers/greetings.py

@@ -0,0 +1,16 @@
+from twitchAPI.chat import ChatMessage
+
+
+TRIGGER_AND_RESPONSE: list[tuple[str, str]] = [
+    # ("ку", "Ку"),
+    ("привет", "Привет")
+]
+
+
+async def on_greetings(msg: ChatMessage) -> bool:
+    for trigger, response in TRIGGER_AND_RESPONSE:
+        if trigger in msg.text.lower():
+            await msg.reply(response)
+            return True
+
+    return False

src/handlers/lasqexx.py

@@ -0,0 +1,20 @@
+from twitchAPI.chat import ChatMessage
+
+
+TRIGGER_AND_RESPONSE: list[tuple[str, str]] = [
+    ("здароу", "Здароу, давай иди уже"),
+    ("сосал?", "А ты? Иди уже"),
+    ("лан я пошёл", "да да, иди уже")
+]
+
+
+async def on_lasqexx_message(msg: ChatMessage):
+    if 'lasqexx' != msg.user.name:
+        return False
+
+    for trigger, response in TRIGGER_AND_RESPONSE:
+        if trigger in msg.text.lower():
+            await msg.reply(response)
+            return True
+
+    return False

src/main.py

@@ -0,0 +1,7 @@
+from asyncio import run
+
+from chatbot import ChatBot
+
+
+if __name__ == "__main__":
+    run(ChatBot.run())