Add twitch

poetry.lock

@@ -1,5 +1,17 @@
 # This file is automatically @generated by Poetry and should not be changed by hand.
 
+[[package]]
+name = "aiofiles"
+version = "24.1.0"
+description = "File support for asyncio."
+category = "main"
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "aiofiles-24.1.0-py3-none-any.whl", hash = "sha256:b4ec55f4195e3eb5d7abd1bf7e061763e864dd4954231fb8539a0ef8bb8260e5"},
+    {file = "aiofiles-24.1.0.tar.gz", hash = "sha256:22a075c9e5a3810f0c2e48f3008c94d68c65d763b9b03857924c99e57355166c"},
+]
+
 [[package]]
 name = "aiohappyeyeballs"
 version = "2.3.5"
@@ -700,4 +712,4 @@ multidict = ">=4.0"
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.11"
-content-hash = "e53f2a74aafd8e01839640d9dee230641a332427025d48ce5e9906377e604274"
+content-hash = "ecdeecc583d261bc5c6fbf0c07c2d31e7b11941f2a63667d280b14a9104dae03"

pyproject.toml

@@ -12,6 +12,7 @@ discord-py = "^2.4.0"
 twitchapi = "^4.2.1"
 pydantic = "^2.8.2"
 pydantic-settings = "^2.4.0"
+aiofiles = "^24.1.0"
 
 
 [build-system]

src/config.py

@@ -19,5 +19,7 @@ class Config(BaseSettings):
     TWITCH_CLIENT_SECRET: str
     TWITCH_CHANNEL_ID: str
 
+    SECRETS_FILE_PATH: str
+
 
 config = Config()  # type: ignore

src/main.py

@@ -1,5 +1,11 @@
-from services.discord import start_discord_sevice
+from asyncio import gather
 
+from services.discord import start_discord_sevice
+from services.twitch import start_twitch_service
 
 async def main():
-    await start_discord_sevice()
+    print("Starting services...")
+    await gather(
+        start_discord_sevice(),
+        start_twitch_service()
+    )

src/services/twitch.py

@@ -0,0 +1,71 @@
+from asyncio import Lock
+import json
+
+from twitchAPI.twitch import Twitch
+from twitchAPI.type import AuthScope
+
+import aiofiles
+
+from config import config
+
+
+class State:
+    pass
+
+
+class TokenStorage:
+    lock = Lock()
+
+    @staticmethod
+    async def save(acceess_token: str, refresh_token: str):
+        data = json.dumps({"access_token": acceess_token, "refresh_token": refresh_token})
+
+        async with TokenStorage.lock:
+            async with aiofiles.open(config.SECRETS_FILE_PATH, "w") as f:
+                await f.write(data)
+
+    @staticmethod
+    async def get() -> tuple[str, str]:
+        async with TokenStorage.lock:
+            async with aiofiles.open(config.SECRETS_FILE_PATH, "r") as f:
+                data_str = await f.read()
+
+        data = json.loads(data_str)
+        return data["access_token"], data["refresh_token"]
+
+
+class TwitchService:
+    SCOPES = [AuthScope.CHANNEL_BOT]
+
+    def __init__(self, twitch: Twitch):
+        self.twitch = twitch
+
+        self.state: State | None = None
+
+    @classmethod
+    async def authorize(cls):
+        twitch = Twitch(
+            config.TWITCH_CLIENT_ID,
+            config.TWITCH_CLIENT_SECRET
+        )
+
+        twitch.user_auth_refresh_callback = TokenStorage.save
+
+        token, refresh_token = await TokenStorage.get()
+        await twitch.set_user_authentication(token, cls.SCOPES, refresh_token)
+
+        return twitch
+
+    @classmethod
+    async def run(cls):
+        pass
+
+    @classmethod
+    async def start(cls):
+        twith = await cls.authorize()
+
+        await cls(twith).run()
+
+
+async def start_twitch_service():
+    await TwitchService.start()