新增新版本消息推送

src/module/database/bangumi.py

@@ -168,25 +168,19 @@ class BangumiDatabase(DataConnector):
         dict_data = dict(zip(keys, values))
         return self.__db_to_data(dict_data)
 
-    def match_official_title(self, title: str) -> bool:
+    def match_poster(self, torrent_name: str) -> tuple[str, str]:
+        # Find title_raw which in torrent_name
         self._cursor.execute(
             """
-            SELECT official_title FROM bangumi 
-            WHERE official_title = :official_title
-            """,
-            {"official_title": title},
-        )
-        return self._cursor.fetchone() is not None
-
-    def match_title_raw(self, title: str, rss_link: str) -> bool:
-        self._cursor.execute(
+            SELECT title_raw, poster_link, official_title FROM bangumi
             """
-            SELECT title_raw, rss_link FROM bangumi 
-            WHERE INSTR (:title_raw, title_raw) > 0 AND INSTR (:rss_link, rss_link) > 0
-            """,
-            {"title_raw": title, "rss_link": rss_link},
         )
-        return self._cursor.fetchone() is not None
+        data = self._cursor.fetchall()
+        if not data:
+            return "", ""
+        for title_raw, poster_link, official_title in data:
+            if title_raw in torrent_name:
+                return poster_link, official_title
 
     def match_list(self, title_dict: dict, rss_link: str) -> dict:
         # Match title_raw in database
@@ -209,28 +203,6 @@ class BangumiDatabase(DataConnector):
                     break
         return title_dict
 
-    def not_exist_titles(self, titles: list[str], rss_link) -> list[str]:
-        # Select all title in titles that title_raw in database not in title
-        self._cursor.execute(
-            """
-            SELECT title_raw, rss_link FROM bangumi
-            """
-        )
-        data = self._cursor.fetchall()
-        if not data:
-            return titles
-        # Match title
-        for title_raw, rss_set in data:
-            rss_set = rss_set.split(",")
-            for title in titles:
-                if rss_link in rss_set:
-                    if title_raw in title:
-                        titles.remove(title)
-                elif rss_link not in rss_set:
-                    rss_set.append(rss_link)
-                    self.update_rss(title_raw, rss_set)
-        return titles
-
     def not_complete(self) -> list[BangumiData]:
         # Find eps_complete = False
         self._cursor.execute(
@@ -258,3 +230,9 @@ class BangumiDatabase(DataConnector):
         if data is None:
             return 1
         return data[0] + 1
+
+
+if __name__ == '__main__':
+    title = "[Lilith-Raws] Boku no Kokoro no Yabai Yatsu - 01 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4].mp4"
+    with BangumiDatabase() as db:
+        print(db.match_poster(title))

src/module/manager/renamer.py

@@ -7,8 +7,9 @@ from module.downloader import DownloadClient
 
 from module.parser import TitleParser
 from module.network import PostNotification
-from module.models import SubtitleFile, EpisodeFile
+from module.models import SubtitleFile, EpisodeFile, Notification
 from module.conf import settings, PLATFORM
+from module.database import BangumiDatabase
 
 if PLATFORM == "Windows":
     import ntpath as path
@@ -22,7 +23,6 @@ class Renamer(DownloadClient):
     def __init__(self):
         super().__init__()
         self._renamer = TitleParser()
-        self._notification = PostNotification()
 
     @staticmethod
     def print_result(torrent_count, rename_count):
@@ -65,8 +65,25 @@ class Renamer(DownloadClient):
         elif method == "subtitle_advance":
             return f"{bangumi_name} S{season}E{episode}.{file_info.language}{file_info.suffix}"
 
-    def send_notification(self, torrent_name, ep: EpisodeFile, bangumi_name: str):
-        pass
+    @staticmethod
+    def send_notification(torrent_name, ep: EpisodeFile):
+        with BangumiDatabase() as db:
+            poster_path, official_name = db.match_poster(torrent_name)
+        poster_link = settings.rss_parser.custom_url + poster_path
+        if "://" not in poster_link:
+            poster_link = "https://" + poster_link
+        n = Notification(
+            title=official_name,
+            season=ep.season,
+            episode=ep.episode,
+            poster_link=poster_link,
+        )
+        with PostNotification() as notificator:
+            status = notificator.send_msg(n)
+        if status:
+            logger.info(f"Notification sent: {ep.title} S{ep.season}E{ep.episode}")
+        else:
+            logger.warning(f"Notification failed: {ep.title} S{ep.season}E{ep.episode}")
 
     def rename_file(
         self,
@@ -88,8 +105,7 @@ class Renamer(DownloadClient):
                 renamed = self.rename_torrent_file(_hash=_hash, old_path=media_path, new_path=new_path)
                 if renamed:
                     if settings.notification.enable:
-                        self.send_notification(torrent_name, ep, bangumi_name)
-                        pass
+                        self.send_notification(torrent_name, ep)
                     return
         logger.warning(f"{media_path} parse failed")
         if settings.bangumi_manage.remove_bad_torrent:

src/module/network/notification.py

@@ -7,87 +7,83 @@ from module.conf import settings
 
 logger = logging.getLogger(__name__)
 
+type = settings.notification.type
+token = settings.notification.token
+chat_id = settings.notification.chat_id
 
-class PostNotification:
+
+class TelegramNotification(RequestContent):
     def __init__(self):
-        self.type: str = settings.notification.type
-        self.token = settings.notification.token
-        self.chat_id = settings.notification.chat_id
-        self.client = self.getClient()
-
-    def getClient(self):
-        if self.type.lower() == "telegram":
-            return TelegramNotification(self.token, self.chat_id)
-        elif self.type.lower() == "server-chan":
-            return ServerChanNotification(self.token)
-        elif self.type.lower() == "bark":
-            return BarkNotification(self.token)
-        else:
-            return None
-
-    def send_msg(self, info: Notification) -> bool:
-        text = f"番剧名称：{info.official_title}\n" \
-               f"季度： 第{info.season}季\n" \
-               f"更新集数： 第{info.episode}集\n" \
-               f"{info.poster_link}\n"
-        if self.client is None:
-            return False
-        return self.client.send_msg(text)
-
-
-class TelegramNotification:
-    def __init__(self, token, chat_id):
+        super().__init__()
         self.notification_url = f"https://api.telegram.org/bot{token}/sendMessage"
         self.chat_id = chat_id
 
-    def send_msg(self, text: str) -> bool:
+    def post_msg(self, text: str) -> bool:
         data = {
             "chat_id": self.chat_id,
             "text": text,
             "disable_notification": True,
         }
-        with RequestContent() as req:
-            resp = req.post_data(self.notification_url, data)
-            logger.debug(f"Telegram notification: {resp.status_code}")
+        resp = self.post_data(self.notification_url, data)
+        logger.debug(f"Telegram notification: {resp.status_code}")
         return resp.status_code == 200
 
 
-class ServerChanNotification:
+class ServerChanNotification(RequestContent):
     """Server酱推送"""
-
-    def __init__(self, token):
+    def __init__(self):
+        super().__init__()
         self.notification_url = f"https://sctapi.ftqq.com/{token}.send"
 
-    def send_msg(self, text: str) -> bool:
+    def post_msg(self, text: str) -> bool:
         data = {
             "title": "AutoBangumi 番剧更新",
             "desp": text,
         }
-        with RequestContent() as req:
-            resp = req.post_data(self.notification_url, data)
-            logger.debug(f"ServerChan notification: {resp.status_code}")
+        resp = self.post_data(self.notification_url, data)
+        logger.debug(f"ServerChan notification: {resp.status_code}")
         return resp.status_code == 200
 
 
-class BarkNotification:
-    def __init__(self, token):
+class BarkNotification(RequestContent):
+    def __init__(self):
+        super().__init__()
         self.token = token
         self.notification_url = "https://api.day.app/push"
 
-    def send_msg(self, text) -> bool:
+    def post_msg(self, text) -> bool:
         data = {"title": "AutoBangumi 番剧更新", "body": text, "device_key": self.token}
-        with RequestContent() as req:
-            resp = req.post_data(self.notification_url, data)
-            logger.debug(f"Bark notification: {resp.status_code}")
+        resp = self.post_data(self.notification_url, data)
+        logger.debug(f"Bark notification: {resp.status_code}")
         return resp.status_code == 200
 
 
+def getClient():
+    if type.lower() == "telegram":
+        return TelegramNotification
+    elif type.lower() == "server-chan":
+        return ServerChanNotification
+    elif type.lower() == "bark":
+        return BarkNotification
+    else:
+        return None
+
+
+class PostNotification(getClient()):
+    def send_msg(self, info: Notification) -> bool:
+        text = f"番剧名称：{info.official_title}\n" \
+               f"季度： 第{info.season}季\n" \
+               f"更新集数： 第{info.episode}集\n" \
+               f"{info.poster_link}\n"
+        return self.post_msg(text)
+
+
 if __name__ == '__main__':
-    notification = PostNotification()
     info = Notification(
         official_title="魔法纪录 魔法少女小圆外传",
         season=2,
         episode=1,
         poster_link="https://mikanani.me/images/Bangumi/202107/3788b33f.jpg",
     )
-    notification.send_msg(info)
+    with PostNotification() as client:
+        client.send_msg(info)