nas/Auto_Bangumi
1
0
Fork 0
mirror of https://github.com/EstrellaXD/Auto_Bangumi.git synced 2026-04-14 18:41:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

change: Decoupling of notification module

Browse Source
This commit is contained in:
EstrellaXD
2023-05-31 19:31:02 +08:00
parent 294bd966b5
commit e36d66f9fe
8 changed files with 72 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/module/core/sub_thread.py
View File

@@ -52,6 +52,9 @@ class RenameThread(ProgramStatus):
while not self.stop_event.is_set():
with Renamer() as renamer:
renamed_info = renamer.rename()
with PostNotification() as notifier:
for info in renamed_info:
notifier.send_msg(info)
self.stop_event.wait(settings.program.rename_time)
def rename_start(self):

35
src/module/manager/renamer.py
View File

@@ -3,10 +3,8 @@ import logging
from module.downloader import DownloadClient
from module.parser import TitleParser
# from module.network import PostNotification
from module.models import SubtitleFile, EpisodeFile, Notification
from module.conf import settings
# from module.database import BangumiDatabase
logger = logging.getLogger(__name__)
@@ -47,24 +45,6 @@ class Renamer(DownloadClient):
logger.error(f"[Renamer] Unknown rename method: {method}")
return file_info.media_path
# @staticmethod
# def send_notification(bangumi_name, ep: EpisodeFile):
# with BangumiDatabase() as db:
# poster_path = db.match_poster(bangumi_name)
# poster_link = "https://mikanani.me" + poster_path
# n = Notification(
# official_title=bangumi_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"[Renamer] Notification sent: {ep.title} S{ep.season}E{ep.episode}")
# else:
# logger.warning(f"[Renamer] Notification failed: {ep.title} S{ep.season}E{ep.episode}")
def rename_file(
self,
torrent_name: str,
@@ -87,9 +67,12 @@ class Renamer(DownloadClient):
_hash=_hash, old_path=media_path, new_path=new_path
)
if renamed:
# if settings.notification.enable:
# self.send_notification(bangumi_name, ep)
return ep
n = Notification(
official_title=bangumi_name,
season=ep.season,
episode=ep.episode,
)
return n
else:
logger.warning(f"[Renamer] {media_path} parse failed")
if settings.bangumi_manage.remove_bad_torrent:
@@ -169,9 +152,9 @@ class Renamer(DownloadClient):
}
# Rename single media file
if len(media_list) == 1:
ep_info = self.rename_file(media_path=media_list[0], **kwargs)
if ep_info:
renamed_info.append(ep_info)
notify_info = self.rename_file(media_path=media_list[0], **kwargs)
if notify_info:
renamed_info.append(notify_info)
# Rename subtitle file
if len(subtitle_list) > 0:
self.rename_subtitles(subtitle_list=subtitle_list, **kwargs)

1
src/module/models/bangumi.py
View File

@@ -28,7 +28,6 @@ class Notification(BaseModel):
official_title: str = Field(..., alias="official_title", title="番剧名")
season: int = Field(..., alias="season", title="番剧季度")
episode: int = Field(..., alias="episode", title="番剧集数")
poster_link: str | None = Field(None, alias="poster_link", title="番剧海报链接")
@dataclass

44
src/module/notification/notification.py
View File

@@ -4,16 +4,13 @@ from .plugin import *
from module.models import Notification
from module.conf import settings
from module.database import BangumiDatabase
logger = logging.getLogger(__name__)
type = settings.notification.type
token = settings.notification.token
chat_id = settings.notification.chat_id
def getClient():
def getClient(type=settings.notification.type):
if type.lower() == "telegram":
return TelegramNotification
elif type.lower() == "server-chan":
@@ -25,15 +22,37 @@ def getClient():
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"
def __init__(self):
super().__init__(
token=settings.notification.token,
chat_id=settings.notification.chat_id
)
@staticmethod
def _gen_message(notify: Notification) -> str:
with BangumiDatabase() as db:
poster_path = db.match_poster(notify.official_title)
if poster_path:
poster_link = "https://mikanani.me" + poster_path
text = f"""
番剧名称：{info.official_title}\n
季度： 第{info.season}季\n
更新集数： 第{info.episode}集\n
{poster_link}\n
"""
else:
text = """
番剧名称：{info.official_title}\n
季度： 第{info.season}季\n
更新集数： 第{info.episode}集\n
"""
return text
def send_msg(self, notify: Notification) -> bool:
text = self._gen_message(notify)
try:
return self.post_msg(text)
self.post_msg(text)
logger.debug(f"Send notification: {text}")
except Exception as e:
logger.warning(f"Failed to send notification: {e}")
return False
@@ -44,7 +63,6 @@ if __name__ == "__main__":
official_title="魔法纪录 魔法少女小圆外传",
season=2,
episode=1,
poster_link="https://mikanani.me/images/Bangumi/202107/3788b33f.jpg",
)
with PostNotification() as client:
client.send_msg(info)

9
src/module/notification/plugin/bark.py
View File

@@ -1,7 +1,12 @@
import logging
from module.network import RequestContent
logger = logging.getLogger(__name__)
class BarkNotification(RequestContent):
def __init__(self):
def __init__(self, token, **kwargs):
super().__init__()
self.token = token
self.notification_url = "https://api.day.app/push"
@@ -10,4 +15,4 @@ class BarkNotification(RequestContent):
data = {"title": "AutoBangumi 番剧更新", "body": text, "device_key": self.token}
resp = self.post_data(self.notification_url, data)
logger.debug(f"Bark notification: {resp.status_code}")
return resp.status_code == 200
return resp.status_code == 200

8
src/module/notification/plugin/server_chan.py
View File

@@ -1,10 +1,14 @@
import logging
from module.network import RequestContent
logger = logging.getLogger(__name__)
class ServerChanNotification(RequestContent):
"""Server酱推送"""
def __init__(self):
def __init__(self, token, **kwargs):
super().__init__()
self.notification_url = f"https://sctapi.ftqq.com/{token}.send"
@@ -15,4 +19,4 @@ class ServerChanNotification(RequestContent):
}
resp = self.post_data(self.notification_url, data)
logger.debug(f"ServerChan notification: {resp.status_code}")
return resp.status_code == 200
return resp.status_code == 200

11
src/module/notification/plugin/slack.py
View File

@@ -1,5 +1,12 @@
class BarkNotification(RequestContent):
def __init__(self):
import logging
from module.network import RequestContent
logger = logging.getLogger(__name__)
class SlackNotification(RequestContent):
def __init__(self, token, **kwargs):
super().__init__()
self.token = token
self.notification_url = "https://api.day.app/push"

9
src/module/notification/plugin/telegram.py
View File

@@ -1,7 +1,12 @@
import logging
from module.network.request_contents import RequestContent
logger = logging.getLogger(__name__)
class TelegramNotification(RequestContent):
def __init__(self):
def __init__(self, token, chat_id):
super().__init__()
self.notification_url = f"https://api.telegram.org/bot{token}/sendMessage"
self.chat_id = chat_id
@@ -14,4 +19,4 @@ class TelegramNotification(RequestContent):
}
resp = self.post_data(self.notification_url, data)
logger.debug(f"Telegram notification: {resp.status_code}")
return resp.status_code == 200
return resp.status_code == 200