style: restrict line-length

libs/slack_api/api.py

@@ -1,59 +1,57 @@
 import json
-from quart import Quart, jsonify,request
+from quart import Quart, jsonify, request
 from slack_sdk.web.async_client import AsyncWebClient
 from .slackevent import SlackEvent
-from typing import Callable, Dict, Any
-from pkg.platform.types import events as platform_events, message as platform_message
-
-class SlackClient():
-    
-      def __init__(self,bot_token:str,signing_secret:str):
-
-            self.bot_token = bot_token
-            self.signing_secret = signing_secret
-            self.app = Quart(__name__)
-            self.client = AsyncWebClient(self.bot_token)
-            self.app.add_url_rule('/callback/command', 'handle_callback', self.handle_callback_request, methods=['GET', 'POST'])
-            self._message_handlers = {
-            "example":[],
-            }
-            self.bot_user_id = None # 避免机器人回复自己的消息
-
-      async def handle_callback_request(self):
-            try:
-                  body = await request.get_data()
-                  data = json.loads(body)
-                  if 'type' in data:
-                        if data['type'] == 'url_verification':
-                              return data['challenge']
-                  
-                  bot_user_id = data.get("event",{}).get("bot_id","")
-
-                  if self.bot_user_id and bot_user_id == self.bot_user_id:
-                        return jsonify({'status': 'ok'})
-                  
-                  
-                  # 处理私信
-                  if data and data.get("event", {}).get("channel_type") in ["im"]:
-                        event = SlackEvent.from_payload(data)
-                        await self._handle_message(event)
-                        return jsonify({'status': 'ok'})
-                  
-                  #处理群聊
-                  if data.get("event",{}).get("type") == 'app_mention':
-                             data.setdefault("event", {})["channel_type"] = "channel"
-                             event = SlackEvent.from_payload(data)
-                             await self._handle_message(event)
-                             return jsonify({'status':'ok'})
-
-                  return jsonify({'status': 'ok'})
-            
-            except Exception as e:
-                 raise(e)
-            
+from typing import Callable
+from pkg.platform.types import events as platform_events
 
 
-      async def _handle_message(self, event: SlackEvent):
+class SlackClient:
+    def __init__(self, bot_token: str, signing_secret: str):
+        self.bot_token = bot_token
+        self.signing_secret = signing_secret
+        self.app = Quart(__name__)
+        self.client = AsyncWebClient(self.bot_token)
+        self.app.add_url_rule(
+            '/callback/command', 'handle_callback', self.handle_callback_request, methods=['GET', 'POST']
+        )
+        self._message_handlers = {
+            'example': [],
+        }
+        self.bot_user_id = None  # 避免机器人回复自己的消息
+
+    async def handle_callback_request(self):
+        try:
+            body = await request.get_data()
+            data = json.loads(body)
+            if 'type' in data:
+                if data['type'] == 'url_verification':
+                    return data['challenge']
+
+            bot_user_id = data.get('event', {}).get('bot_id', '')
+
+            if self.bot_user_id and bot_user_id == self.bot_user_id:
+                return jsonify({'status': 'ok'})
+
+            # 处理私信
+            if data and data.get('event', {}).get('channel_type') in ['im']:
+                event = SlackEvent.from_payload(data)
+                await self._handle_message(event)
+                return jsonify({'status': 'ok'})
+
+            # 处理群聊
+            if data.get('event', {}).get('type') == 'app_mention':
+                data.setdefault('event', {})['channel_type'] = 'channel'
+                event = SlackEvent.from_payload(data)
+                await self._handle_message(event)
+                return jsonify({'status': 'ok'})
+
+            return jsonify({'status': 'ok'})
+
+        except Exception as e:
+            raise (e)
+
+    async def _handle_message(self, event: SlackEvent):
         """
         处理消息事件。
         """
@@ -62,50 +60,38 @@ class SlackClient():
             for handler in self._message_handlers[msg_type]:
                 await handler(event)
 
-      def on_message(self, msg_type: str):
+    def on_message(self, msg_type: str):
         """注册消息类型处理器"""
+
         def decorator(func: Callable[[platform_events.Event], None]):
             if msg_type not in self._message_handlers:
                 self._message_handlers[msg_type] = []
             self._message_handlers[msg_type].append(func)
             return func
+
         return decorator
 
-      async def send_message_to_channel(self,text:str,channel_id:str):
-            try:
-                  response = await self.client.chat_postMessage(
-                        channel=channel_id,
-                        text=text
-                  )
-                  if self.bot_user_id is None and response.get("ok"):
-                        self.bot_user_id = response["message"]["bot_id"]
-                  return 
-            except Exception as e:
-                  raise e
+    async def send_message_to_channel(self, text: str, channel_id: str):
+        try:
+            response = await self.client.chat_postMessage(channel=channel_id, text=text)
+            if self.bot_user_id is None and response.get('ok'):
+                self.bot_user_id = response['message']['bot_id']
+            return
+        except Exception as e:
+            raise e
 
-      async def send_message_to_one(self,text:str,user_id:str):
-            try:
-                  response = await self.client.chat_postMessage(
-                        channel = '@'+user_id,
-                        text= text
-                  )
-                  if self.bot_user_id is None and response.get("ok"):
-                        self.bot_user_id = response["message"]["bot_id"]
-                  
-                  return 
-            except Exception as e:
-                  raise e
-
-      async def run_task(self, host: str, port: int, *args, **kwargs):
-            """
-            启动 Quart 应用。
-            """
-            await self.app.run_task(host=host, port=port, *args, **kwargs)
-
-
-
-
-
-            
+    async def send_message_to_one(self, text: str, user_id: str):
+        try:
+            response = await self.client.chat_postMessage(channel='@' + user_id, text=text)
+            if self.bot_user_id is None and response.get('ok'):
+                self.bot_user_id = response['message']['bot_id']
 
+            return
+        except Exception as e:
+            raise e
 
+    async def run_task(self, host: str, port: int, *args, **kwargs):
+        """
+        启动 Quart 应用。
+        """
+        await self.app.run_task(host=host, port=port, *args, **kwargs)

libs/slack_api/slackevent.py

@@ -1,86 +1,82 @@
 from typing import Dict, Any, Optional
 
+
 class SlackEvent(dict):
     @staticmethod
-    def from_payload(payload: Dict[str, Any]) -> Optional["SlackEvent"]:
+    def from_payload(payload: Dict[str, Any]) -> Optional['SlackEvent']:
         try:
             event = SlackEvent(payload)
             return event
         except KeyError:
             return None
-    
+
     @property
     def text(self) -> str:
-        
-        if self.get("event", {}).get("channel_type") == "im":
-                blocks = self.get("event", {}).get("blocks", [])
-                if not blocks:
-                    return ""
+        if self.get('event', {}).get('channel_type') == 'im':
+            blocks = self.get('event', {}).get('blocks', [])
+            if not blocks:
+                return ''
 
-                elements = blocks[0].get("elements", [])
-                if not elements:
-                    return "" 
+            elements = blocks[0].get('elements', [])
+            if not elements:
+                return ''
 
-                elements = elements[0].get("elements", [])
-                text = ""
+            elements = elements[0].get('elements', [])
+            text = ''
 
-                for el in elements:
-                    if el.get("type") == "text":
-                        text += el.get("text", "")
-                    elif el.get("type") == "link":
-                        text += el.get("url", "")
+            for el in elements:
+                if el.get('type') == 'text':
+                    text += el.get('text', '')
+                elif el.get('type') == 'link':
+                    text += el.get('url', '')
 
-                return text 
+            return text
 
-                
-        if self.get("event",{}).get("channel_type") == 'channel': 
-            message_text = ""
-            for block in self.get("event", {}).get("blocks", []):
-                if block.get("type") == "rich_text":
-                    for element in block.get("elements", []):
-                        if element.get("type") == "rich_text_section":
+        if self.get('event', {}).get('channel_type') == 'channel':
+            message_text = ''
+            for block in self.get('event', {}).get('blocks', []):
+                if block.get('type') == 'rich_text':
+                    for element in block.get('elements', []):
+                        if element.get('type') == 'rich_text_section':
                             parts = []
-                            for el in element.get("elements", []):
-                                if el.get("type") == "text":
-                                    parts.append(el["text"])
-                                elif el.get("type") == "link":
-                                    parts.append(el["url"])
-                            message_text = "".join(parts)
-                            
+                            for el in element.get('elements', []):
+                                if el.get('type') == 'text':
+                                    parts.append(el['text'])
+                                elif el.get('type') == 'link':
+                                    parts.append(el['url'])
+                            message_text = ''.join(parts)
+
             return message_text
 
-
-    
     @property
     def user_id(self) -> Optional[str]:
-        return self.get("event", {}).get("user","")
-    
+        return self.get('event', {}).get('user', '')
+
     @property
     def channel_id(self) -> Optional[str]:
-        return self.get("event", {}).get("channel","")
-    
+        return self.get('event', {}).get('channel', '')
+
     @property
     def type(self) -> str:
-        """ message对应私聊，app_mention对应频道at """
-        return self.get("event", {}).get("channel_type", "")
-    
+        """message对应私聊，app_mention对应频道at"""
+        return self.get('event', {}).get('channel_type', '')
+
     @property
     def message_id(self) -> str:
-        return self.get("event_id","")
-    
+        return self.get('event_id', '')
+
     @property
     def pic_url(self) -> str:
         """提取 Slack 事件中的图片 URL"""
-        files = self.get("event", {}).get("files", [])
+        files = self.get('event', {}).get('files', [])
         if files:
-            return files[0].get("url_private", "")
+            return files[0].get('url_private', '')
         return None
-        
-    
+
     @property
     def sender_name(self) -> str:
-        return self.get("event", {}).get("user","")
-    
+        return self.get('event', {}).get('user', '')
+
     def __getattr__(self, key: str) -> Optional[Any]:
         return self.get(key)
 
@@ -88,4 +84,4 @@ class SlackEvent(dict):
         self[key] = value
 
     def __repr__(self) -> str:
-        return f"<SlackEvent {super().__repr__()}>"
+        return f'<SlackEvent {super().__repr__()}>'