개발종료

app/damination.py

@@ -2,14 +2,14 @@ import asyncio
 import json
 import re
 import requests
-from aiohttp import web
 from websockets import connect
 import logging.handlers
 import configparser
 from pydantic import BaseModel
 from datetime import datetime
 
-from aiohttp import ClientSession
+import aiohttp
+
 
 class StatusEndpointFilter(logging.Filter):
     def filter(self, record):
@@ -17,23 +17,11 @@ class StatusEndpointFilter(logging.Filter):
             return False
         return True
 
-
 # Config Parser
 config = configparser.ConfigParser()
 config.read("config/settings.ini")
 
-
-if config.has_option("widget", "ids"):
-    WIDGET_IDS = config.get("widget", "ids").split(",")
-    if not WIDGET_IDS[0]:
-        WIDGET_IDS = []
-else:
-    WIDGET_IDS = []
-
-
 MAX_RETRIES = config.getint("config", "MAX_RETRIES")
-SERVER_IP = config.get("server", "ip")
-SERVER_PORT = config.get("server", "port")
 
 # Create logger
 logger = logging.getLogger()
@@ -41,11 +29,10 @@ logger.setLevel(logging.DEBUG)
 
 # Create file handler
 log_filename = datetime.now().strftime('logs/%Y-%m-%d_%H:%M:%S_damination') + '.log'
-file_handler = logging.handlers.RotatingFileHandler(log_filename, maxBytes=100000, backupCount=3)
+file_handler = logging.handlers.RotatingFileHandler(log_filename, maxBytes=10000000, backupCount=3)
 file_handler.setLevel(logging.INFO)
 file_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
 file_handler.setFormatter(file_formatter)
-
 file_handler.addFilter(StatusEndpointFilter())
 
 # Create stream handler
@@ -63,6 +50,7 @@ class Roulette(BaseModel):
     name : str
     conf : dict
 
+
 class Content(BaseModel):
     account : str
     name : str
@@ -70,6 +58,7 @@ class Content(BaseModel):
     count : int | None
     amount : int | None
     roulette : Roulette | None
+    streamer : str
 
 class Donate(BaseModel):
     code : str
@@ -78,29 +67,88 @@ class Donate(BaseModel):
 
 class DaminationAPI():
 
+
     def __init__(self):
+        self.WIDGET_IDS = []
         self.queue = asyncio.Queue()
         self.fetch_tasks = []
-        self.stop_event = asyncio.Event()
         self.subscribers = {}
+        self.session = aiohttp.ClientSession()
+        self.connections = {}
+        self.processing_tasks = []
 
 
-    def add_subscriber(self, widget_id: str, url: str) -> bool:
+    async def initialize_widgets(self):
+        if config.has_option("widget", "ids"):
+            widgets = config.get("widget", "ids").split(",")
+
+            if widgets[0]:
+                for widget in widgets:
+                    await self.add_widget_and_start_fetching_notifications(widget)
+                logger.info(f"WIDGET ID 초기화 : {widgets} [initialize_widgets]")
+
+            else:
+                logger.error(f"WIDGET 초기화 실패 - 등록된 WIDGET ID 없음 [initialize_widgets]")
+
+
+    async def process_notification(self):
+        donate = await self.pop_notification_from_queue()
+        if len(donate):
+            try:
+                for widget_id, value in donate.items():
+                   
+                    logger.debug(f"Raw Data : widget_id: {widget_id}, value: {value} [process_notification]")
+                    await self.send_notifications_to_subscribers(widget_id, value)
+
+                    logger.debug(f"QUEUE 추출: {widget_id, value} [process_notification]")
+
+            except Exception as e:
+                logger.error(f"Raw Data : {donate}, {e} [process_notification]")
+
+
+        else:
+            logger.error("QUEUE 추출 실패 - 저장된 QUEUE 없음 [process_notification]")
+
+
+    def get_subscribers(self, widget_id: str) -> list:
+        """
+        주어진 widget_id에 해당하는 구독자 목록을 반환합니다.
+        widget_id: 구독자 목록을 얻을 위젯 ID
+        """
+
+        if widget_id in self.subscribers:
+            subscribers = self.subscribers[widget_id]
+            logger.info(f"구독자 반환({widget_id}) : {subscribers} [get_subscribers]")
+            return subscribers
+        
+        else:
+            logger.error(f"구독자 없음({widget_id}) [get_subscribers]")
+            return []
+
+
+    def add_subscriber(self, widget_id: str, url: str) -> str:
         if widget_id not in self.subscribers:
             self.subscribers[widget_id] = []
 
         if url not in self.subscribers[widget_id]:
             self.subscribers[widget_id].append(url)
-            return True
+            logger.info(f"구독자 추가 widget_id={widget_id}, url={url} [add_subscriber]")
+            return widget_id
+        
         else:
-            return False
+            logger.error(f"이미 추가된 구독자 widget_id={widget_id}, url={url} [add_subscriber]")
+            return ''
 
-    def remove_subscriber(self, widget_id: str, url: str) -> bool:
+
+    def remove_subscriber(self, widget_id: str, url: str) -> str:
         if widget_id in self.subscribers and url in self.subscribers[widget_id]:
             self.subscribers[widget_id].remove(url)
-            return True
+            logger.info(f"구독자 제거: widget_id={widget_id}, url={url} [remove_subscriber]")
+            return widget_id
+        
         else:
-            return False
+            logger.error(f"구독자를 찾을 수 없음 widget_id={widget_id}, url={url} [remove_subscriber]")
+            return ''
         
 
     def get_widget_ids(self) -> list:
@@ -108,22 +156,35 @@ class DaminationAPI():
         현재 설정된 모든 위젯 ID 목록을 반환합니다.
         """
 
-        logger.debug(f"get widget {WIDGET_IDS}")
+        if self.WIDGET_IDS:
-        return WIDGET_IDS
+            logger.info(f"WIDGET ID 반환 {self.WIDGET_IDS} [get_widget_ids]")
+            return self.WIDGET_IDS
+        
+        else:
+            logger.error(f"WIDGET ID 반환 실패 - WIDGET ID 없음 [get_widget_ids]")
+            return []
     
-    def add_widget_id(self, widget_id:str):
+
+    def add_widget_id(self, widget_id:str) -> str:
         """
         주어진 widget_id를 위젯 ID 목록에 추가합니다.
         widget_id: 추가할 위젯 ID
         """
 
         try:
-            WIDGET_IDS.append(widget_id)
+            if widget_id not in self.WIDGET_IDS:
-            logger.debug(f"add widget {widget_id}")
+                self.WIDGET_IDS.append(widget_id)
+                logger.info(f"WIDGET ID 추가 {widget_id} [add_widget_id]")
+                return widget_id
+            
+            else:
+                logger.error(f"이미 추가된 WIDGET ID {widget_id} [add_widget_id]")
+                return ''
 
         except Exception as e:
             logger.error(e)
 
+
     def remove_widget_id(self, widget_id: str) -> str:
         """
         주어진 widget_id를 위젯 ID 목록에서 제거합니다. 제거한 widget_id를 반환하며,
@@ -132,19 +193,20 @@ class DaminationAPI():
         """
 
         try:
-            if widget_id in WIDGET_IDS:
+            if widget_id in self.WIDGET_IDS:
-                widget = WIDGET_IDS.pop(WIDGET_IDS.index(widget_id))
+                widget = self.WIDGET_IDS.pop(self.WIDGET_IDS.index(widget_id))
-                logger.debug(f"delete widget {widget}")
+                logger.info(f"WIDGET ID 제거 {widget} [remove_widget_id]")
                 return widget
             
             else:
+                logger.error(f"WIDGET ID 제거 실패 - WIDGET ID를 찾을 수 없음 [remove_widget_id]")
                 return ''
 
         except Exception as e:
             logger.error(e)
 
 
-    async def fetch_alerts(self, queue, widget_id: str):
+    async def fetch_notifications(self, queue, widget_id: str):
         """
         주어진 widget_id에 대한 알림을 가져와 queue에 저장합니다.
         연결이 끊어지면 최대 MAX_RETRIES 횟수만큼 재시도합니다.
@@ -156,43 +218,43 @@ class DaminationAPI():
 
         while retries < MAX_RETRIES:
             try:
-                logger.info(f"retries: {retries}")
                 payload = self.get_websocket_payload(widget_id)
                 wss_url = f"wss://toon.at:8071/{payload}"
 
                 async with connect(wss_url, ping_interval=12) as websocket:
                     token = json.loads(await websocket.recv())['token'] 
                     logger.debug(f"token : {token}")
-                    logger.info(f"WebSocket connection established for widget_id: {widget_id}")
+                    logger.info(f"웹 소켓 연결 : {widget_id} [fetch_notifications]")
                     while True:
-                        if self.stop_event.is_set():  # 추가: 중단 이벤트가 설정된 경우, 작업을 중단
-                            break
 
                         message = await websocket.recv()
                         logger.debug(f"message: {message}")
                         alert_data = json.loads(message)
+                        alert_data["content"]["streamer"] = token["twitch"]["account"]
+                        logger.info(f"alert_data : {alert_data}")
 
                         try: 
                             reformed_data = Donate(**alert_data)
 
                         except Exception as e:
-                            logger.error(f"Error: {e}, Raw data: {alert_data}")
+                            logger.error(f"Error: {e}, Raw data: {alert_data} [fetch_notifications]")
-
                             pass
 
                         await queue.put({widget_id: json.loads(reformed_data.json().encode('utf-8').decode('unicode_escape'))})
 
-            except asyncio.CancelledError:  # 추가: 취소 예외 처리
+            except asyncio.CancelledError:
-                logger.info(f"Widget {widget_id} fetching task was cancelled.")
+                logger.info(f"웹 소켓 연결 취소 : {widget_id} [fetch_notifications]")
                 break
 
             except Exception as e:
-                logger.error(f"Error: {e}")
                 retries += 1
+                logger.error(f"웹 소켓 연결 실패({widget_id}) : {e}, 재시도 : {retries} [fetch_notifications]")
                 await asyncio.sleep(2 ** retries)
 
-        self.remove_widget_id(widget_id)
+        if retries > 2:
-        logger.info(f"Connection failed for widget_id: {widget_id}")
+            self.remove_widget_id(widget_id)
+            logger.error(f"연결 실패 : {widget_id} [fetch_notifications]")
+
 
     @staticmethod
     def get_websocket_payload(widget_id:str) -> str:
@@ -203,6 +265,7 @@ class DaminationAPI():
 
         res = requests.get(f"https://toon.at/widget/alertbox/{widget_id}")
         match = re.search(r"payload(.+)", res.text)
+
         if not match:
             raise ValueError("Payload not found")
 
@@ -210,88 +273,103 @@ class DaminationAPI():
         return payload
 
 
-    async def start_fetching_alerts(self):
+    async def add_widget_and_start_fetching_notifications(self, widget_id: str) -> str:
-        """
-        설정된 모든 위젯 ID에 대해 알림을 가져오는 작업을 시작합니다.
-        """
-
-        self.stop_event.clear()  # 추가: 중단 이벤트를 초기화합니다.
-
-        for widget_id in WIDGET_IDS:
-            task = asyncio.create_task(self.fetch_alerts(self.queue, widget_id))
-            self.fetch_tasks.append(task)
-
-        await asyncio.gather(*self.fetch_tasks)
-
-
-    async def stop_fetching_alerts(self):
-        """
-        설정된 모든 위젯 ID에 대한 알림 가져오기 작업을 중단합니다.
-        중단 이벤트를 설정하고 모든 fetch_tasks를 기다립니다.
-        """
-
-        self.stop_event.set()
-        await asyncio.gather(*self.fetch_tasks)
-    
-    async def add_widget_id_and_start_fetching(self, widget_id: str):
         """
         주어진 widget_id를 위젯 ID 목록에 추가하고 알림을 가져오는 작업을 시작합니다.
         widget_id: 추가할 위젯 ID
         """
         
-        if widget_id not in WIDGET_IDS:
+        if widget_id not in self.WIDGET_IDS:
             self.add_widget_id(widget_id)
-            task = asyncio.create_task(self.fetch_alerts(self.queue, widget_id))
+            task = asyncio.create_task(self.fetch_notifications(self.queue, widget_id))
             self.fetch_tasks.append(task)
+            logger.info(f"투네이션 알림 수신 시작 : {widget_id} [add_widget_and_start_fetching_notifications]")
+            return widget_id
+        
         else:
-            logger.warning(f"Widget ID {widget_id} already exists in the list.")
+            logger.error(f"이미 투네이션 알림 수신 중 : {widget_id} [add_widget_and_start_fetching_notifications]")
+            return ''
 
 
-    async def remove_widget_id_and_stop_fetching(self, widget_id: str):
+    async def remove_widget_and_stop_fetching_notifications(self, widget_id: str) -> str:
         """
         주어진 widget_id를 위젯 ID 목록에서 제거하고 알림을 가져오는 작업을 중단합니다.
         widget_id: 제거할 위젯 ID
         """
-        if widget_id in WIDGET_IDS:
+
-            index = WIDGET_IDS.index(widget_id)
+        if widget_id in self.WIDGET_IDS:
+            index = self.WIDGET_IDS.index(widget_id)
             self.remove_widget_id(widget_id)
             fetch_task = self.fetch_tasks.pop(index)
             fetch_task.cancel()
             await fetch_task
+            logger.info(f"투네이션 알림 수신 중단 : {widget_id} [remove_widget_and_stop_fetching_notifications]")
+            return widget_id
+        
         else:
-            logger.warning(f"Widget ID {widget_id} not found.")
+            logger.error(f"수신중인 투네이션 알림을 찾을 수 없음 {widget_id} [remove_widget_and_stop_fetching_notifications]")
+            return ''
 
 
-    async def pop_from_queue(self):
+    async def pop_notification_from_queue(self) -> dict:
         """
         queue에서 가장 먼저 저장된 항목을 꺼내어 반환합니다. 
         queue가 비어있는 경우 None을 반환합니다.
         """
+
         if not self.queue.empty():
             item = await self.queue.get()
             return item
-        else:
-            return None
-
-
-    async def notify_subscribers(self, widget_id: str, donate):
-        if widget_id not in self.subscribers:
-            return
         
-        for subscriber_url in self.subscribers[widget_id]:
+        else:
-            async with ClientSession() as session:
+            return {}
-                await session.post(subscriber_url, json=donate)
+
-            
+
-            logger.info(f"send to subscriber({widget_id}: {donate})")
+    async def get_connection(self, url: str):
+        if url not in self.connections:
+            self.connections[url] = aiohttp.ClientSession()
+
+        return self.connections[url]
+
+
+    async def send_notifications_to_subscribers(self, widget_id: str, donate):
+        """
+        주어진 widget_id에 해당하는 모든 구독자들에게 알림을 전송합니다.
+        widget_id: 알림을 보낼 구독자들의 위젯 ID
+        donate: 전송할 알림 데이터
+        """
+
+        subscribers = self.get_subscribers(widget_id)
+        tasks = [asyncio.create_task(self.send_notification_webhook(await self.get_connection(subscriber_url), subscriber_url, donate)) for subscriber_url in subscribers]
+        await asyncio.gather(*tasks)
+
+
+    async def send_notification_webhook(self, session, subscriber_url, donate):
+        payload = {
+            "text": f"**{donate['content']['name']}**({donate['content']['account']}) to **{donate['content']['streamer']}**\n{donate}",
+            "data": donate
+        }
+
+        async with session.post(subscriber_url, json=payload) as response:
+            if response.status == 200:
+                logger.info(f"알림 전송 {subscriber_url}, {payload} [send_notification_webhook]")
+
+            else:
+                logger.error(f"알림 전송 실패 {subscriber_url}, 응답: {response.status} [send_notification_webhook]")
 
 
     async def run(self):
-        await self.start_fetching_alerts()
+        await self.initialize_widgets()
-
+        
         while True:
             if not self.queue.empty():
-                donate = await self.pop_from_queue()
+                task = asyncio.create_task(self.process_notification())
-                for widget_id, value in donate.items():
+                self.processing_tasks.append(task)
-                    await self.notify_subscribers(widget_id, value)
+                await task
-            await asyncio.sleep(2)
 
+            await asyncio.sleep(0.1)
+
+
+    async def close(self):
+        for session in self.connections.values():
+            await session.close()

app/main.py

@@ -1,4 +1,3 @@
-
 import asyncio
 from aiohttp import web
 from damination import DaminationAPI
@@ -9,72 +8,103 @@ toonat = DaminationAPI()
 app = web.Application()
 
 
-@routes.get('/subscribe/{widget_id}/{url}')
+@routes.get('/get_subscriber/{widget_id}')
+async def get_widgets(request):
+    """
+    주어진 widget_id로 연결된 구독 목록을 반환합니다.
+    사용 예: http://localhost/get_subscriber?widget_id=YOUR_WIDGET_ID
+    """
+    widget_id = request.match_info['widget_id']
+    subscribers = toonat.get_subscribers(widget_id)
+    if subscribers:
+        return web.json_response(subscribers)
+    else:
+        return web.Response(text=f"no subscribers")
+
+
+@routes.get('/subscribe/{widget_id}')
 async def subscribe(request):
+    """
+    주어진 widget_id와 url로 구독을 추가합니다.
+    사용 예: http://localhost/subscribe/YOUR_WIDGET_ID?url=YOUR_WEBHOOK_URL
+    """
     widget_id = request.match_info['widget_id']
-    url = request.match_info['url']
+    url = request.query['url']
     success = toonat.add_subscriber(widget_id, url)
-    if success:
-        return web.Response(text=f"Subscribed to widget ID {widget_id} with URL {url}.")
-    else:
-        return web.Response(text=f"Already subscribed to widget ID {widget_id} with URL {url}.")
 
-@routes.get('/unsubscribe/{widget_id}/{url}')
+    if success:
+        return web.Response(text=f"Subscribed to widget ID {success} with URL {url}")
+    else:
+        return web.Response(text=f"Already subscribed to widget ID {widget_id} with URL {url}")
+
+
+@routes.get('/unsubscribe/{widget_id}')
 async def unsubscribe(request):
+    """
+    주어진 widget_id와 url로 구독을 취소합니다.
+    사용 예: http://localhost/unusbscribe/YOUR_WIDGET_ID?=url=YOUR_WEBHOOK_URL
+    """
     widget_id = request.match_info['widget_id']
-    url = request.match_info['url']
+    url = request.query['url']
     success = toonat.remove_subscriber(widget_id, url)
-    if success:
-        return web.Response(text=f"Unsubscribed from widget ID {widget_id} with URL {url}.")
-    else:
-        return web.Response(text=f"Not subscribed to widget ID {widget_id} with URL {url}.")
 
-@routes.get('/init_widget_id')
+    if success:
-async def init_widget_id(request):
+        return web.Response(text=f"Unsubscribed from widget ID {success} with URL {url}")
-    if not toonat.WIDGET_IDS:
-        widget_id = request.query.get("widget_id")
-        if widget_id:
-            toonat.WIDGET_IDS.append(widget_id)
-            return web.Response(text=f"Widget ID {widget_id} added.")
-        else:
-            return web.Response(text="Invalid widget ID. Please try again.")
     else:
-        return web.Response(text="Widget ID already exists.")
+        return web.Response(text=f"Not subscribed to widget ID {widget_id} with URL {url}")
 
 
 @routes.get('/get_widgets')
 async def get_widgets(request):
     """
-    현재 설정된 모든 위젯 ID 목록을 반환하는 웹 API 엔드포인트입니다.
+    현재 설정된 모든 위젯 ID 목록을 반환합니다.
+    사용 예: http://localhost/get_widgets
     """
     widget_ids = toonat.get_widget_ids()
-    return web.json_response(widget_ids)
+
+    if widget_ids:
+        return web.json_response(widget_ids)
+    else:
+        return web.Response(text=f"widget ids empty")
 
 
-@routes.get('/add_widget/{widget_id}')
+@routes.get('/add_widget')
 async def add_widget(request):
     """
-    주어진 widget_id를 위젯 ID 목록에 추가하고 알림을 가져오는 작업을 시작하는 웹 API 엔드포인트입니다.
+    주어진 widget_id를 위젯 ID 목록에 추가하고 알림을 가져오는 작업을 시작합니다.
     widget_id: 추가할 위젯 ID
+    사용 예: http://localhost/add_widget?=widget_id=YOUR_WIDGET_ID
     """
-    widget_id = request.match_info['widget_id']
+    widget_id = request.query['widget_id']
-    await toonat.add_widget_id_and_start_fetching(widget_id)
+    success = await toonat.add_widget_and_start_fetching_notifications(widget_id)
-    return web.Response(text=f"Widget ID {widget_id} added and fetching started.")
 
-@routes.get('/remove_widget/{widget_id}')
+    if success:
+        return web.Response(text=f"Widget ID {success} added and fetching started")
+    else:
+        return web.Response(text=f"Already {widget_id} in the list")
+
+
+@routes.get('/remove_widget')
 async def remove_widget(request):
     """
-    주어진 widget_id를 위젯 ID 목록에서 제거하고 알림을 가져오는 작업을 중단하는 웹 API 엔드포인트입니다.
+    주어진 widget_id를 위젯 ID 목록에서 제거하고 알림을 가져오는 작업을 중단합니다.
     widget_id: 제거할 위젯 ID
+    사용 예: http://localhost/remove_widget?=YOUR_WIDGET_ID
     """
-    widget_id = request.match_info['widget_id']
+    widget_id = request.query['widget_id']
-    await toonat.remove_widget_id_and_stop_fetching(widget_id)
+    success = await toonat.remove_widget_and_stop_fetching_notifications(widget_id)
-    return web.Response(text=f"Widget ID {widget_id} removed and fetching stopped.")
+
+    if success:
+        return web.Response(text=f"Widget ID {success} removed and fetching stopped")
+    else:
+        return web.Response(text=f"Not in widget list {widget_id}")
+
 
 @routes.get('/status')
 async def status(request):
     """
     서버 상태를 확인하는 웹 API 엔드포인트입니다. 이 엔드포인트는 서버가 정상적으로 작동 중임을 확인하는 데 사용됩니다.
+    DOCKER Health Check에 사용됩니다.
     """
     return web.HTTPOk()
 
@@ -98,15 +128,13 @@ async def main():
     # 웹 서버를 실행합니다.
     web_server_task = asyncio.create_task(web_server(app))
 
-    # 두 작업을 병렬로 실행하지 않고 순서대로 처리합니다.
     await api_task
     await web_server_task
 
 
-
 if __name__ == "__main__":
     if not toonat.WIDGET_IDS:
-        print("No widget ID found in settings.ini.")
+        print("settings.ini 파일에서 위젯 ID를 찾을 수 없습니다")
-        print("Please enter the widget ID via web endpoint (http://localhost/init_widget_id?widget_id=YOUR_WIDGET_ID)")
+        print("웹 엔드포인트(http://localhost/add_widget/YOUR_WIDGET_ID)를 통해 위젯 ID를 입력해주세요")
 
     asyncio.run(main())

data/config/settings.ini

@@ -3,7 +3,3 @@ MAX_RETRIES=3
 
 [widget]
 ids=
-
-[server]
-ip=192.168.1.210
-port=8113

docker-compose.yml

@@ -1,6 +1,8 @@
 version: '3'
 services:
-  toonat-test:
+  damination:
+    build:
+      context: .
     image: damination:1.0
     container_name: damination
     ports:
@@ -8,5 +10,4 @@ services:
     volumes:
       - ./data/config:/app/config
       - ./data/logs:/app/logs
-    restart: unless-stopped
+    restart: unless-stopped
-    platform: linux/x86_64

dockerfile

@@ -11,7 +11,7 @@ RUN apk --no-cache add tzdata && \
 COPY ./requirements.txt /
 RUN pip install --no-cache-dir -r requirements.txt
 
-COPY . /app
+COPY ./app /app
 WORKDIR /app
 
 EXPOSE 80
@@ -20,4 +20,4 @@ HEALTHCHECK --interval=5s --timeout=3s --start-period=10s --retries=3 \
     CMD curl --fail http://localhost/status || exit 1
 
 
-CMD ["python", "damination.py"]
+CMD ["python", "main.py"]