add webhook schemas

app/chain/__init__.py

@@ -12,7 +12,8 @@ from app.core.event import EventManager
 from app.core.meta import MetaBase
 from app.core.module import ModuleManager
 from app.log import logger
-from app.schemas import TransferInfo, TransferTorrent, ExistMediaInfo, DownloadingTorrent, CommingMessage, Notification
+from app.schemas import TransferInfo, TransferTorrent, ExistMediaInfo, DownloadingTorrent, CommingMessage, Notification, \
+    WebhookEventInfo
 from app.schemas.types import TorrentStatus, MediaType, MediaImageType
 from app.utils.object import ObjectUtils
 from app.utils.singleton import AbstractSingleton, Singleton
@@ -170,7 +171,7 @@ class ChainBase(AbstractSingleton, metaclass=Singleton):
         """
         return self.run_module("message_parser", body=body, form=form, args=args)
 
-    def webhook_parser(self, body: Any, form: Any, args: Any) -> Optional[dict]:
+    def webhook_parser(self, body: Any, form: Any, args: Any) -> Optional[WebhookEventInfo]:
         """
         解析Webhook报文体
         :param body:  请求体

app/chain/webhook.py

@@ -3,8 +3,8 @@ from typing import Any
 
 from app.chain import ChainBase
 from app.schemas import Notification
-from app.utils.http import WebUtils
 from app.schemas.types import EventType, MediaImageType, MediaType, NotificationType
+from app.utils.http import WebUtils
 
 
 class WebhookChain(ChainBase):
@@ -17,7 +17,7 @@ class WebhookChain(ChainBase):
         处理Webhook报文并发送消息
         """
         # 获取主体内容
-        event_info: dict = self.webhook_parser(body=body, form=form, args=args)
+        event_info = self.webhook_parser(body=body, form=form, args=args)
         if not event_info:
             return
         # 广播事件
@@ -42,55 +42,55 @@ class WebhookChain(ChainBase):
             "jellyfin": "https://play-lh.googleusercontent.com/SCsUK3hCCRqkJbmLDctNYCfehLxsS4ggD1ZPHIFrrAN1Tn9yhjmGMPep2D9lMaaa9eQi"
         }
 
-        if not _webhook_actions.get(event_info.get('event')):
+        if not _webhook_actions.get(event_info.event):
             return
 
         # 消息标题
-        if event_info.get('item_type') in ["TV", "SHOW"]:
+        if event_info.item_type in ["TV", "SHOW"]:
-            message_title = f"{_webhook_actions.get(event_info.get('event'))}剧集 {event_info.get('item_name')}"
+            message_title = f"{_webhook_actions.get(event_info.event)}剧集 {event_info.item_name}"
-        elif event_info.get('item_type') == "MOV":
+        elif event_info.item_type == "MOV":
-            message_title = f"{_webhook_actions.get(event_info.get('event'))}电影 {event_info.get('item_name')}"
+            message_title = f"{_webhook_actions.get(event_info.event)}电影 {event_info.item_name}"
-        elif event_info.get('item_type') == "AUD":
+        elif event_info.item_type == "AUD":
-            message_title = f"{_webhook_actions.get(event_info.get('event'))}有声书 {event_info.get('item_name')}"
+            message_title = f"{_webhook_actions.get(event_info.event)}有声书 {event_info.item_name}"
         else:
-            message_title = f"{_webhook_actions.get(event_info.get('event'))}"
+            message_title = f"{_webhook_actions.get(event_info.event)}"
 
         # 消息内容
         message_texts = []
-        if event_info.get('user_name'):
+        if event_info.user_name:
-            message_texts.append(f"用户：{event_info.get('user_name')}")
+            message_texts.append(f"用户：{event_info.user_name}")
-        if event_info.get('device_name'):
+        if event_info.device_name:
-            message_texts.append(f"设备：{event_info.get('client')} {event_info.get('device_name')}")
+            message_texts.append(f"设备：{event_info.client} {event_info.device_name}")
-        if event_info.get('ip'):
+        if event_info.ip:
-            message_texts.append(f"IP地址：{event_info.get('ip')} {WebUtils.get_location(event_info.get('ip'))}")
+            message_texts.append(f"IP地址：{event_info.ip} {WebUtils.get_location(event_info.ip)}")
-        if event_info.get('percentage'):
+        if event_info.percentage:
-            percentage = round(float(event_info.get('percentage')), 2)
+            percentage = round(float(event_info.percentage), 2)
             message_texts.append(f"进度：{percentage}%")
-        if event_info.get('overview'):
+        if event_info.overview:
-            message_texts.append(f"剧情：{event_info.get('overview')}")
+            message_texts.append(f"剧情：{event_info.overview}")
         message_texts.append(f"时间：{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))}")
 
         # 消息内容
         message_content = "\n".join(message_texts)
 
         # 消息图片
-        image_url = event_info.get("image_url")
+        image_url = event_info.image_url
         # 查询剧集图片
-        if event_info.get("tmdb_id") \
+        if event_info.tmdb_id \
-                and event_info.get("season_id"):
+                and event_info.season_id:
-            mtype = MediaType.TV if event_info.get("item_type") == "TV" else MediaType.MOVIE
+            mtype = MediaType.TV if event_info.item_type == "TV" else MediaType.MOVIE
             specific_image = self.obtain_specific_image(
-                mediaid=event_info.get("tmdb_id"),
+                mediaid=event_info.tmdb_id,
                 mtype=mtype,
                 image_type=MediaImageType.Backdrop,
-                season=event_info.get("season_id"),
+                season=event_info.season_id,
-                episode=event_info.get("episode_id")
+                episode=event_info.episode_id
             )
             if specific_image:
                 image_url = specific_image
         # 使用默认图片
         if not image_url:
-            image_url = _webhook_images.get(event_info.get("channel"))
+            image_url = _webhook_images.get(event_info.channel)
 
         # 发送消息
         self.post_message(Notification(mtype=NotificationType.MediaServer,

app/modules/emby/__init__.py

@@ -6,7 +6,7 @@ from app.core.context import MediaInfo
 from app.log import logger
 from app.modules import _ModuleBase
 from app.modules.emby.emby import Emby
-from app.schemas import ExistMediaInfo, RefreshMediaItem
+from app.schemas import ExistMediaInfo, RefreshMediaItem, WebhookEventInfo
 from app.schemas.types import MediaType
 
 
@@ -32,7 +32,7 @@ class EmbyModule(_ModuleBase):
         # Emby认证
         return self.emby.authenticate(name, password)
 
-    def webhook_parser(self, body: Any, form: Any, args: Any) -> Optional[dict]:
+    def webhook_parser(self, body: Any, form: Any, args: Any) -> WebhookEventInfo:
         """
         解析Webhook报文体
         :param body:  请求体

app/modules/emby/emby.py

@@ -5,7 +5,7 @@ from typing import List, Optional, Union, Dict, Generator
 
 from app.core.config import settings
 from app.log import logger
-from app.schemas import RefreshMediaItem
+from app.schemas import RefreshMediaItem, WebhookEventInfo
 from app.schemas.types import MediaType
 from app.utils.http import RequestUtils
 from app.utils.singleton import Singleton
@@ -537,58 +537,58 @@ class Emby(metaclass=Singleton):
             logger.error(f"连接Users/Items出错：" + str(e))
         yield {}
 
-    def get_webhook_message(self, message_str: str) -> dict:
+    def get_webhook_message(self, message_str: str) -> WebhookEventInfo:
         """
         解析Emby Webhook报文
         """
         message = json.loads(message_str)
-        eventItem = {'event': message.get('Event', ''), "channel": "emby"}
+        eventItem = WebhookEventInfo(event=message.get('Event', ''), channel="emby")
         if message.get('Item'):
             if message.get('Item', {}).get('Type') == 'Episode':
-                eventItem['item_type'] = "TV"
+                eventItem.item_type = "TV"
-                eventItem['item_name'] = "%s %s%s %s" % (
+                eventItem.item_name = "%s %s%s %s" % (
                     message.get('Item', {}).get('SeriesName'),
                     "S" + str(message.get('Item', {}).get('ParentIndexNumber')),
                     "E" + str(message.get('Item', {}).get('IndexNumber')),
                     message.get('Item', {}).get('Name'))
-                eventItem['item_id'] = message.get('Item', {}).get('SeriesId')
+                eventItem.item_id = message.get('Item', {}).get('SeriesId')
-                eventItem['season_id'] = message.get('Item', {}).get('ParentIndexNumber')
+                eventItem.season_id = message.get('Item', {}).get('ParentIndexNumber')
-                eventItem['episode_id'] = message.get('Item', {}).get('IndexNumber')
+                eventItem.episode_id = message.get('Item', {}).get('IndexNumber')
             elif message.get('Item', {}).get('Type') == 'Audio':
-                eventItem['item_type'] = "AUD"
+                eventItem.item_type = "AUD"
                 album = message.get('Item', {}).get('Album')
                 file_name = message.get('Item', {}).get('FileName')
-                eventItem['item_name'] = album
+                eventItem.item_name = album
-                eventItem['overview'] = file_name
+                eventItem.overview = file_name
-                eventItem['item_id'] = message.get('Item', {}).get('AlbumId')
+                eventItem.item_id = message.get('Item', {}).get('AlbumId')
             else:
-                eventItem['item_type'] = "MOV"
+                eventItem.item_type = "MOV"
-                eventItem['item_name'] = "%s %s" % (
+                eventItem.item_name = "%s %s" % (
                     message.get('Item', {}).get('Name'), "(" + str(message.get('Item', {}).get('ProductionYear')) + ")")
-                eventItem['item_path'] = message.get('Item', {}).get('Path')
+                eventItem.item_path = message.get('Item', {}).get('Path')
-                eventItem['item_id'] = message.get('Item', {}).get('Id')
+                eventItem.item_id = message.get('Item', {}).get('Id')
 
-            eventItem['tmdb_id'] = message.get('Item', {}).get('ProviderIds', {}).get('Tmdb')
+            eventItem.tmdb_id = message.get('Item', {}).get('ProviderIds', {}).get('Tmdb')
             if message.get('Item', {}).get('Overview') and len(message.get('Item', {}).get('Overview')) > 100:
-                eventItem['overview'] = str(message.get('Item', {}).get('Overview'))[:100] + "..."
+                eventItem.overview = str(message.get('Item', {}).get('Overview'))[:100] + "..."
             else:
-                eventItem['overview'] = message.get('Item', {}).get('Overview')
+                eventItem.overview = message.get('Item', {}).get('Overview')
-            eventItem['percentage'] = message.get('TranscodingInfo', {}).get('CompletionPercentage')
+            eventItem.percentage = message.get('TranscodingInfo', {}).get('CompletionPercentage')
-            if not eventItem['percentage']:
+            if not eventItem.percentage:
                 if message.get('PlaybackInfo', {}).get('PositionTicks'):
-                    eventItem['percentage'] = message.get('PlaybackInfo', {}).get('PositionTicks') / \
+                    eventItem.percentage = message.get('PlaybackInfo', {}).get('PositionTicks') / \
-                                              message.get('Item', {}).get('RunTimeTicks') * 100
+                                           message.get('Item', {}).get('RunTimeTicks') * 100
         if message.get('Session'):
-            eventItem['ip'] = message.get('Session').get('RemoteEndPoint')
+            eventItem.ip = message.get('Session').get('RemoteEndPoint')
-            eventItem['device_name'] = message.get('Session').get('DeviceName')
+            eventItem.device_name = message.get('Session').get('DeviceName')
-            eventItem['client'] = message.get('Session').get('Client')
+            eventItem.client = message.get('Session').get('Client')
         if message.get("User"):
-            eventItem['user_name'] = message.get("User").get('Name')
+            eventItem.user_name = message.get("User").get('Name')
 
         # 获取消息图片
-        if eventItem.get("item_id"):
+        if eventItem.item_id:
             # 根据返回的item_id去调用媒体服务器获取
-            eventItem['image_url'] = self.get_remote_image_by_id(item_id=eventItem.get('item_id'),
+            eventItem.image_url = self.get_remote_image_by_id(item_id=eventItem.item_id,
-                                                                 image_type="Backdrop")
+                                                              image_type="Backdrop")
 
         return eventItem

app/modules/jellyfin/__init__.py

@@ -7,7 +7,7 @@ from app.core.context import MediaInfo
 from app.log import logger
 from app.modules import _ModuleBase
 from app.modules.jellyfin.jellyfin import Jellyfin
-from app.schemas import ExistMediaInfo
+from app.schemas import ExistMediaInfo, WebhookEventInfo
 from app.schemas.types import MediaType
 
 
@@ -33,7 +33,7 @@ class JellyfinModule(_ModuleBase):
         # Jellyfin认证
         return self.jellyfin.authenticate(name, password)
 
-    def webhook_parser(self, body: Any, form: Any, args: Any) -> Optional[dict]:
+    def webhook_parser(self, body: Any, form: Any, args: Any) -> WebhookEventInfo:
         """
         解析Webhook报文体
         :param body:  请求体

app/modules/jellyfin/jellyfin.py

@@ -4,7 +4,7 @@ from typing import List, Union, Optional, Dict, Generator
 
 from app.core.config import settings
 from app.log import logger
-from app.schemas import MediaType
+from app.schemas import MediaType, WebhookEventInfo
 from app.utils.http import RequestUtils
 from app.utils.singleton import Singleton
 from app.utils.string import StringUtils
@@ -365,21 +365,22 @@ class Jellyfin(metaclass=Singleton):
             logger.error(f"连接Library/Refresh出错：" + str(e))
             return False
 
-    def get_webhook_message(self, message: dict) -> dict:
+    def get_webhook_message(self, message: dict) -> WebhookEventInfo:
         """
         解析Jellyfin报文
         """
-        eventItem = {'event': message.get('NotificationType', ''),
+        eventItem = WebhookEventInfo(
-                     'item_name': message.get('Name'),
+            event=message.get('NotificationType', ''),
-                     'user_name': message.get('NotificationUsername'),
+            item_name=message.get('Name'),
-                     "channel": "jellyfin"
+            user_name=message.get('NotificationUsername'),
-                     }
+            channel="jellyfin"
+        )
 
         # 获取消息图片
-        if eventItem.get("item_id"):
+        if eventItem.item_id:
             # 根据返回的item_id去调用媒体服务器获取
-            eventItem['image_url'] = self.get_remote_image_by_id(item_id=eventItem.get('item_id'),
+            eventItem.image_url = self.get_remote_image_by_id(item_id=eventItem.item_id,
-                                                                 image_type="Backdrop")
+                                                              image_type="Backdrop")
 
         return eventItem
 

app/modules/plex/__init__.py

@@ -6,7 +6,7 @@ from app.core.context import MediaInfo
 from app.log import logger
 from app.modules import _ModuleBase
 from app.modules.plex.plex import Plex
-from app.schemas import ExistMediaInfo, RefreshMediaItem
+from app.schemas import ExistMediaInfo, RefreshMediaItem, WebhookEventInfo
 from app.schemas.types import MediaType
 
 
@@ -23,7 +23,7 @@ class PlexModule(_ModuleBase):
     def init_setting(self) -> Tuple[str, Union[str, bool]]:
         return "MEDIASERVER", "plex"
 
-    def webhook_parser(self, body: Any, form: Any, args: Any) -> Optional[dict]:
+    def webhook_parser(self, body: Any, form: Any, args: Any) -> WebhookEventInfo:
         """
         解析Webhook报文体
         :param body:  请求体

app/modules/plex/plex.py

@@ -8,7 +8,7 @@ from plexapi.server import PlexServer
 
 from app.core.config import settings
 from app.log import logger
-from app.schemas import RefreshMediaItem, MediaType
+from app.schemas import RefreshMediaItem, MediaType, WebhookEventInfo
 from app.utils.singleton import Singleton
 
 
@@ -317,7 +317,7 @@ class Plex(metaclass=Singleton):
             logger.error(f"获取媒体库列表出错：{err}")
         yield {}
 
-    def get_webhook_message(self, message_str: str) -> dict:
+    def get_webhook_message(self, message_str: str) -> WebhookEventInfo:
         """
         解析Plex报文
         eventItem  字段的含义
@@ -328,44 +328,44 @@ class Plex(metaclass=Singleton):
         overview   剧情描述
         """
         message = json.loads(message_str)
-        eventItem = {'event': message.get('event', ''), "channel": "plex"}
+        eventItem = WebhookEventInfo(event=message.get('Event', ''), channel="plex")
         if message.get('Metadata'):
             if message.get('Metadata', {}).get('type') == 'episode':
-                eventItem['item_type'] = "TV"
+                eventItem.item_type = "TV"
-                eventItem['item_name'] = "%s %s%s %s" % (
+                eventItem.item_name = "%s %s%s %s" % (
                     message.get('Metadata', {}).get('grandparentTitle'),
                     "S" + str(message.get('Metadata', {}).get('parentIndex')),
                     "E" + str(message.get('Metadata', {}).get('index')),
                     message.get('Metadata', {}).get('title'))
-                eventItem['item_id'] = message.get('Metadata', {}).get('ratingKey')
+                eventItem.item_id = message.get('Metadata', {}).get('ratingKey')
-                eventItem['season_id'] = message.get('Metadata', {}).get('parentIndex')
+                eventItem.season_id = message.get('Metadata', {}).get('parentIndex')
-                eventItem['episode_id'] = message.get('Metadata', {}).get('index')
+                eventItem.episode_id = message.get('Metadata', {}).get('index')
 
                 if message.get('Metadata', {}).get('summary') and len(message.get('Metadata', {}).get('summary')) > 100:
-                    eventItem['overview'] = str(message.get('Metadata', {}).get('summary'))[:100] + "..."
+                    eventItem.overview = str(message.get('Metadata', {}).get('summary'))[:100] + "..."
                 else:
-                    eventItem['overview'] = message.get('Metadata', {}).get('summary')
+                    eventItem.overview = message.get('Metadata', {}).get('summary')
             else:
-                eventItem['item_type'] = "MOV" if message.get('Metadata', {}).get('type') == 'movie' else "SHOW"
+                eventItem.item_type = "MOV" if message.get('Metadata', {}).get('type') == 'movie' else "SHOW"
-                eventItem['item_name'] = "%s %s" % (
+                eventItem.item_name = "%s %s" % (
                     message.get('Metadata', {}).get('title'), "(" + str(message.get('Metadata', {}).get('year')) + ")")
-                eventItem['item_id'] = message.get('Metadata', {}).get('ratingKey')
+                eventItem.item_id = message.get('Metadata', {}).get('ratingKey')
                 if len(message.get('Metadata', {}).get('summary')) > 100:
-                    eventItem['overview'] = str(message.get('Metadata', {}).get('summary'))[:100] + "..."
+                    eventItem.overview = str(message.get('Metadata', {}).get('summary'))[:100] + "..."
                 else:
-                    eventItem['overview'] = message.get('Metadata', {}).get('summary')
+                    eventItem.overview = message.get('Metadata', {}).get('summary')
         if message.get('Player'):
-            eventItem['ip'] = message.get('Player').get('publicAddress')
+            eventItem.ip = message.get('Player').get('publicAddress')
-            eventItem['client'] = message.get('Player').get('title')
+            eventItem.client = message.get('Player').get('title')
             # 这里给个空,防止拼消息的时候出现None
-            eventItem['device_name'] = ' '
+            eventItem.device_name = ' '
         if message.get('Account'):
-            eventItem['user_name'] = message.get("Account").get('title')
+            eventItem.user_name = message.get("Account").get('title')
 
         # 获取消息图片
-        if eventItem.get("item_id"):
+        if eventItem.item_id:
             # 根据返回的item_id去调用媒体服务器获取
-            eventItem['image_url'] = self.get_remote_image_by_id(item_id=eventItem.get('item_id'),
+            eventItem.image_url = self.get_remote_image_by_id(item_id=eventItem.item_id,
-                                                                 image_type="Backdrop")
+                                                              image_type="Backdrop")
 
         return eventItem

app/schemas/mediaserver.py

@@ -109,3 +109,25 @@ class MediaServerSeasonInfo(BaseModel):
     """
     season: Optional[int] = None
     episodes: Optional[List[int]] = []
+
+
+class WebhookEventInfo(BaseModel):
+    """
+    Webhook事件信息
+    """
+    event: Optional[str] = None
+    channel: Optional[str] = None
+    item_type: Optional[str] = None
+    item_name: Optional[str] = None
+    item_id: Optional[str] = None
+    item_path: Optional[str] = None
+    season_id: Optional[str] = None
+    episode_id: Optional[str] = None
+    tmdb_id: Optional[str] = None
+    overview: Optional[str] = None
+    percentage: Optional[float] = None
+    ip: Optional[str] = None
+    device_name: Optional[str] = None
+    client: Optional[str] = None
+    user_name: Optional[str] = None
+    image_url: Optional[str] = None