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

Merge branch '3.0-dev' into 3.1-dev

Browse Source 
# Conflicts:
#	backend/src/module/checker/checker.py
#	backend/src/module/database/bangumi.py
#	backend/src/module/notification/notification.py
#	backend/src/module/notification/plugin/bark.py
#	backend/src/module/parser/analyser/tmdb_parser.py
This commit is contained in:
EstrellaXD
2023-06-13 22:02:48 +08:00
parent c7c71da7bd 2a51490169
commit de832888ac
16 changed files with 177 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
backend/src/module/checker/checker.py
View File

@@ -1,7 +1,9 @@
import os.path
from module.conf import settings
from module.downloader import DownloadClient
from module.network import RequestContent
from module.conf import settings
from module.models import Config
class Checker:
@@ -45,8 +47,7 @@ class Checker:
@staticmethod
def check_first_run() -> bool:
token_exist = False if settings.rss_parser.token in ["", "token"] else True
if token_exist:
return False
else:
if settings.dict() == Config().dict():
return True
else:
return False

2
backend/src/module/core/sub_thread.py
View File

@@ -19,6 +19,8 @@ class RSSThread(ProgramStatus):
)
def rss_loop(self):
with DownloadClient() as client:
client.init_downloader()
while not self.stop_event.is_set():
# Analyse RSS
with BangumiDatabase() as db:

92
backend/src/module/database/bangumi.py
View File

@@ -1,8 +1,8 @@
import logging
from module.ab_decorator import locked
from module.database.connector import DataConnector
from module.models import BangumiData
from module.ab_decorator import locked
logger = logging.getLogger(__name__)
@@ -115,54 +115,27 @@ class BangumiDatabase(DataConnector):
def search_id(self, _id: int) -> BangumiData | None:
condition = {"id": _id}
value = self._search_data(table_name=self.__table_name, condition=condition)
# self._cursor.execute(
# """
# SELECT * FROM bangumi WHERE id = :id
# """,
# {"id": _id},
# )
# values = self._cursor.fetchone()
if value is None:
dict_data = self._search_data(table_name=self.__table_name, condition=condition)
if dict_data is None:
return None
keys = [x[0] for x in self._cursor.description]
dict_data = dict(zip(keys, value))
return self.__db_to_data(dict_data)
def search_official_title(self, official_title: str) -> BangumiData | None:
value = self._search_data(
dict_data = self._search_data(
table_name=self.__table_name, condition={"official_title": official_title}
)
# self._cursor.execute(
# """
# SELECT * FROM bangumi WHERE official_title = :official_title
# """,
# {"official_title": official_title},
# )
# values = self._cursor.fetchone()
if value is None:
if dict_data is None:
return None
keys = [x[0] for x in self._cursor.description]
dict_data = dict(zip(keys, value))
return self.__db_to_data(dict_data)
def match_poster(self, bangumi_name: str) -> str:
condition = f"INSTR({bangumi_name}, official_title) > 0"
condition = {"_custom_condition": "INSTR(:bangumi_name, official_title) > 0"}
keys = ["official_title", "poster_link"]
data = self._search_data(
table_name=self.__table_name,
keys=keys,
condition=condition,
)
# self._cursor.execute(
# """
# SELECT official_title, poster_link
# FROM bangumi
# WHERE INSTR(:bangumi_name, official_title) > 0
# """,
# {"bangumi_name": bangumi_name},
# )
# data = self._cursor.fetchone()
if not data:
return ""
official_title, poster_link = data
@@ -170,33 +143,26 @@ class BangumiDatabase(DataConnector):
return ""
return poster_link
@locked
def match_list(self, torrent_list: list, rss_link: str) -> list:
# Match title_raw in database
keys = ["title_raw", "rss_link", "poster_link"]
data = self._search_datas(
match_datas = self._search_datas(
table_name=self.__table_name,
keys=keys,
)
# self._cursor.execute(
# """
# SELECT title_raw, rss_link, poster_link FROM bangumi
# """
# )
# data = self._cursor.fetchall()
if not data:
if not match_datas:
return torrent_list
# Match title
i = 0
while i < len(torrent_list):
torrent = torrent_list[i]
for title_raw, rss_set, poster_link in data:
if title_raw in torrent.name:
if rss_link not in rss_set:
rss_set += "," + rss_link
self.update_rss(title_raw, rss_set)
if not poster_link:
self.update_poster(title_raw, torrent.poster_link)
for match_data in match_datas:
if match_data.get("title_raw") in torrent.name:
if rss_link not in match_data.get("rss_link"):
match_data["rss_link"] += f",{rss_link}"
self.update_rss(match_data.get("title_raw"), match_data.get("rss_link"))
if not match_data.get("poster_link"):
self.update_poster(match_data.get("title_raw"), torrent.poster_link)
torrent_list.pop(i)
break
else:
@@ -205,27 +171,20 @@ class BangumiDatabase(DataConnector):
def not_complete(self) -> list[BangumiData]:
# Find eps_complete = False
condition = "eps_complete = 0"
self._search_datas(
condition = {"eps_collect": 0}
dict_data = self._search_datas(
table_name=self.__table_name,
condition=condition,
)
self._cursor.execute(
"""
SELECT * FROM bangumi WHERE eps_collect = 0
"""
)
return self.__fetch_data()
return [self.__db_to_data(x) for x in dict_data]
def not_added(self) -> list[BangumiData]:
self._cursor.execute(
"""
SELECT * FROM bangumi
WHERE added = 0 OR rule_name IS NULL OR save_path IS NULL
"""
condition = {"added": 0, "rule_name": None, "save_path": None}
dict_data = self._search_datas(
table_name=self.__table_name,
condition=condition,
)
return self.__fetch_data()
return [self.__db_to_data(x) for x in dict_data]
def gen_id(self) -> int:
self._cursor.execute(
@@ -255,3 +214,8 @@ class BangumiDatabase(DataConnector):
if self.__check_exist(data):
return True
return False
if __name__ == '__main__':
with BangumiDatabase() as db:
print(db.not_added())
print(db.not_complete())

40
backend/src/module/database/connector.py
View File

@@ -93,23 +93,16 @@ class DataConnector:
)
self._conn.commit()
def _delete_all(self, table_name: str):
self._cursor.execute(f"DELETE FROM {table_name}")
self._conn.commit()
def _search_data(self, table_name: str, keys: list[str] | None, condition: str) -> dict:
if keys is None:
self._cursor.execute(f"SELECT * FROM {table_name} WHERE {condition}")
else:
self._cursor.execute(
f"SELECT {', '.join(keys)} FROM {table_name} WHERE {condition}"
)
return dict(zip(keys, self._cursor.fetchone()))
def _delete(self, table_name: str, condition: dict):
condition_sql = " AND ".join([f"{key} = :{key}" for key in condition.keys()])
self._cursor.execute(f"DELETE FROM {table_name} WHERE {condition_sql}", condition)
self._conn.commit()
def _search_datas(self, table_name: str, keys: list[str] | None, condition: str = None) -> list[dict]:
def _search(self, table_name: str, keys: list[str] | None = None, condition: dict = None):
if keys is None:
select_sql = "*"
else:
@@ -117,9 +110,24 @@ class DataConnector:
if condition is None:
self._cursor.execute(f"SELECT {select_sql} FROM {table_name}")
else:
self._cursor.execute(
f"SELECT {select_sql} FROM {table_name} WHERE {condition}"
custom_condition = condition.pop("_custom_condition", None)
condition_sql = " AND ".join([f"{key} = :{key}" for key in condition.keys()]) + (
f" AND {custom_condition}" if custom_condition else ""
)
self._cursor.execute(
f"SELECT {select_sql} FROM {table_name} WHERE {condition_sql}", condition
)
def _search_data(self, table_name: str, keys: list[str] | None = None, condition: dict = None) -> dict:
if keys is None:
keys = self.__get_table_columns(table_name)
self._search(table_name, keys, condition)
return dict(zip(keys, self._cursor.fetchone()))
def _search_datas(self, table_name: str, keys: list[str] | None = None, condition: dict = None) -> list[dict]:
if keys is None:
keys = self.__get_table_columns(table_name)
self._search(table_name, keys, condition)
return [dict(zip(keys, row)) for row in self._cursor.fetchall()]
def _table_exists(self, table_name: str) -> bool:
@@ -129,6 +137,10 @@ class DataConnector:
)
return len(self._cursor.fetchall()) == 1
def __get_table_columns(self, table_name: str) -> list[str]:
self._cursor.execute(f"PRAGMA table_info({table_name})")
return [column_info[1] for column_info in self._cursor.fetchall()]
@staticmethod
def __python_to_sqlite_type(value) -> str:
if isinstance(value, int):

5
backend/src/module/manager/collector.py
View File

@@ -57,8 +57,3 @@ def eps_complete():
bd.update_list(datas)
if __name__ == "__main__":
from module.conf import setup_logger
setup_logger()
eps_complete()

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

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

35
backend/src/module/notification/notification.py
View File

@@ -1,15 +1,11 @@
import logging
from .plugin import *
from module.models import Notification
from module.conf import settings
from module.database import BangumiDatabase
from module.models import Notification
from .plugin import (
BarkNotification,
ServerChanNotification,
TelegramNotification,
WecomNotification,
)
logger = logging.getLogger(__name__)
@@ -31,28 +27,30 @@ class PostNotification:
def __init__(self):
Notifier = getClient(settings.notification.type)
self.notifier = Notifier(
token=settings.notification.token, chat_id=settings.notification.chat_id
token=settings.notification.token,
chat_id=settings.notification.chat_id
)
@staticmethod
def _gen_message(notify: Notification) -> str:
def _get_poster(notify: Notification):
with BangumiDatabase() as db:
poster_path = db.match_poster(notify.official_title)
if poster_path:
poster_link = "https://mikanani.me" + poster_path
text = f"""
番剧名称：{notify.official_title}\n季度：第{notify.season}季\n更新集数：第{notify.episode}集\n{poster_link}\n
"""
# text = f"""
# 番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n{poster_link}\n
# """
else:
text = f"""
番剧名称：{notify.official_title}\n季度：第{notify.season}季\n更新集数：第{notify.episode}集\n
"""
return text
poster_link = "https://mikanani.me"
# text = f"""
# 番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n
# """
notify.poster_path = poster_link
def send_msg(self, notify: Notification) -> bool:
text = self._gen_message(notify)
self._get_poster(notify)
try:
self.notifier.post_msg(text)
self.notifier.post_msg(notify)
logger.debug(f"Send notification: {notify.official_title}")
except Exception as e:
logger.warning(f"Failed to send notification: {e}")
@@ -65,7 +63,6 @@ class PostNotification:
def __exit__(self, exc_type, exc_val, exc_tb):
self.notifier.__exit__(exc_type, exc_val, exc_tb)
if __name__ == "__main__":
info = Notification(
official_title="魔法纪录 魔法少女小圆外传",

14
backend/src/module/notification/plugin/bark.py
View File

@@ -1,6 +1,8 @@
import logging
from module.network import RequestContent
from module.models import Notification
logger = logging.getLogger(__name__)
@@ -11,8 +13,16 @@ class BarkNotification(RequestContent):
self.token = token
self.notification_url = "https://api.day.app/push"
def post_msg(self, text) -> bool:
data = {"title": "AutoBangumi 番剧更新", "body": text, "device_key": self.token}
@staticmethod
def gen_message(notify: Notification) -> str:
text = f"""
番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n{notify.poster_path}\n
"""
return text
def post_msg(self, notify: Notification) -> bool:
text = self.gen_message(notify)
data = {"title": notify.official_title, "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

13
backend/src/module/notification/plugin/server_chan.py
View File

@@ -1,6 +1,7 @@
import logging
from module.network import RequestContent
from module.models import Notification
logger = logging.getLogger(__name__)
@@ -12,9 +13,17 @@ class ServerChanNotification(RequestContent):
super().__init__()
self.notification_url = f"https://sctapi.ftqq.com/{token}.send"
def post_msg(self, text: str) -> bool:
@staticmethod
def gen_message(notify: Notification) -> str:
text = f"""
番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n{notify.poster_path}\n
"""
return text
def post_msg(self, notify: Notification) -> bool:
text = self.gen_message(notify)
data = {
"title": "AutoBangumi 番剧更新",
"title": notify.official_title,
"desp": text,
}
resp = self.post_data(self.notification_url, data)

13
backend/src/module/notification/plugin/slack.py
View File

@@ -1,6 +1,7 @@
import logging
from module.network import RequestContent
from module.models import Notification
logger = logging.getLogger(__name__)
@@ -11,8 +12,16 @@ class SlackNotification(RequestContent):
self.token = token
self.notification_url = "https://api.day.app/push"
def post_msg(self, text) -> bool:
data = {"title": "AutoBangumi 番剧更新", "body": text, "device_key": self.token}
@staticmethod
def gen_message(notify: Notification) -> str:
text = f"""
番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n{notify.poster_path}\n
"""
return text
def post_msg(self, notify: Notification) -> bool:
text = self.gen_message(notify)
data = {"title": notify.official_title, "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

13
backend/src/module/notification/plugin/telegram.py
View File

@@ -1,6 +1,7 @@
import logging
from module.network.request_contents import RequestContent
from module.network import RequestContent
from module.models import Notification
logger = logging.getLogger(__name__)
@@ -11,7 +12,15 @@ class TelegramNotification(RequestContent):
self.notification_url = f"https://api.telegram.org/bot{token}/sendMessage"
self.chat_id = chat_id
def post_msg(self, text: str) -> bool:
@staticmethod
def gen_message(notify: Notification) -> str:
text = f"""
番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n{notify.poster_path}\n
"""
return text
def post_msg(self, notify: Notification) -> bool:
text = self.gen_message(notify)
data = {
"chat_id": self.chat_id,
"text": text,

28
backend/src/module/notification/plugin/wecom.py
View File

@@ -1,6 +1,7 @@
import logging
from module.network import RequestContent
from module.models import Notification
logger = logging.getLogger(__name__)
@@ -14,21 +15,26 @@ class WecomNotification(RequestContent):
self.notification_url = f"{chat_id}"
self.token = token
def post_msg(self, text: str) -> bool:
@staticmethod
def gen_message(notify: Notification) -> str:
text = f"""
番剧名称：{notify.official_title}\n季度： 第{notify.season}季\n更新集数： 第{notify.episode}集\n{notify.poster_path}\n
"""
return text
def post_msg(self, notify: Notification) -> bool:
##Change message format to match Wecom push better
info = text.split("：")
print(info)
title = "【番剧更新】" + info[1].split("\n")[0].strip()
msg = info[2].split("\n")[0].strip()+" "+info[3].split("\n")[0].strip()
picurl = info[3].split("\n")[1].strip()
title = "【番剧更新】" + notify.official_title
msg = self.gen_message(notify)
picurl = notify.poster_path
#Default pic to avoid blank in message. Resolution:1068*455
if picurl == "":
if picurl == "https://mikanani.me":
picurl = "https://article.biliimg.com/bfs/article/d8bcd0408bf32594fd82f27de7d2c685829d1b2e.png"
data = {
"key":self.token,
"type": "news",
"title": title,
"msg": msg,
"key":self.token,
"type": "news",
"title": title,
"msg": msg,
"picurl":picurl
}
resp = self.post_data(self.notification_url, data)

5
backend/src/module/parser/analyser/raw_parser.py
View File

@@ -179,3 +179,8 @@ def raw_parser(raw: str) -> Episode | None:
return Episode(
name_en, name_zh, name_jp, season, sr, episode, sub, group, dpi, source
)
if __name__ == '__main__':
title = "[动漫国字幕组&LoliHouse] THE MARGINAL SERVICE - 08 [WebRip 1080p HEVC-10bit AAC][简繁内封字幕]"
print(raw_parser(title))

46
backend/src/module/parser/analyser/tmdb_parser.py
View File

@@ -2,8 +2,8 @@ import re
import time
from dataclasses import dataclass
from module.conf import TMDB_API
from module.network import RequestContent
from module.conf import TMDB_API
@dataclass
@@ -16,15 +16,16 @@ class TMDBInfo:
year: str
LANGUAGE = {"zh": "zh-CN", "jp": "ja-JP", "en": "en-US"}
LANGUAGE = {
"zh": "zh-CN",
"jp": "ja-JP",
"en": "en-US"
}
def search_url(e):
return f"https://api.themoviedb.org/3/search/tv?api_key={TMDB_API}&page=1&query={e}&include_adult=false"
def info_url(e, key):
return f"https://api.themoviedb.org/3/tv/{e}?api_key={TMDB_API}&language={LANGUAGE[key]}"
search_url = lambda e: \
f"https://api.themoviedb.org/3/search/tv?api_key={TMDB_API}&page=1&query={e}&include_adult=false"
info_url = lambda e, key: \
f"https://api.themoviedb.org/3/tv/{e}?api_key={TMDB_API}&language={LANGUAGE[key]}"
def is_animation(tv_id, language) -> bool:
@@ -38,10 +39,11 @@ def is_animation(tv_id, language) -> bool:
def get_season(seasons: list) -> int:
for season in seasons:
ss = sorted(seasons, key=lambda e: e.get("air_date"), reverse=True)
for season in ss:
if re.search(r"第 \d 季", season.get("season")) is not None:
date = season.get("air_date").split("-")
[year, _, _] = date
[year, _ , _] = date
now_year = time.localtime().tm_year
if int(year) <= now_year:
return int(re.findall(r"\d", season.get("season"))[0])
@@ -62,30 +64,16 @@ def tmdb_parser(title, language) -> TMDBInfo | None:
break
url_info = info_url(id, language)
info_content = req.get_json(url_info)
season = [
{
"season": s.get("name"),
"air_date": s.get("air_date"),
"poster_path": s.get("poster_path"),
}
for s in info_content.get("seasons")
]
season = [{"season": s.get("name"), "air_date": s.get("air_date"), "poster_path": s.get("poster_path")} for s in info_content.get("seasons")]
last_season = get_season(season)
original_title = info_content.get("original_name")
official_title = info_content.get("name")
year_number = info_content.get("first_air_date").split("-")[0]
return TMDBInfo(
id,
official_title,
original_title,
season,
last_season,
str(year_number),
)
return TMDBInfo(id, official_title, original_title, season, last_season, str(year_number))
else:
return None
if __name__ == '__main__':
title = "鬼灭之刃"
print(tmdb_parser(title, "zh"))
title = "海盗战记"
print(tmdb_parser(title, "zh").last_season)

8
backend/src/module/parser/analyser/torrent_parser.py
View File

@@ -13,13 +13,13 @@ RULES = [
r"(.*) - (\d{1,4}(?!\d|p)|\d{1,4}\.\d{1,2}(?!\d|p))(?:v\d{1,2})?(?: )?(?:END)?(.*)",
r"(.*)[\[\ E](\d{1,4}|\d{1,4}\.\d{1,2})(?:v\d{1,2})?(?: )?(?:END)?[\]\ ](.*)",
r"(.*)\[(?:第)?(\d*\.*\d*)[话集話](?:END)?\](.*)",
r"(.*)第(\d*\.*\d*)[话話集](?:END)?(.*)",
r"(.*)第?(\d*\.*\d*)[话話集](?:END)?(.*)",
r"(.*)(?:S\d{2})?EP?(\d+)(.*)",
]
SUBTITLE_LANG = {
"zh-tw": ["TC", "CHT", "cht", "繁", "zh-tw"],
"zh": ["SC", "CHS", "chs", "简", "zh"],
"zh-tw": ["tc", "cht", "繁", "zh-tw"],
"zh": ["sc", "chs", "简", "zh"],
}
@@ -54,7 +54,7 @@ def get_season_and_title(season_and_title) -> tuple[str, int]:
def get_subtitle_lang(subtitle_name: str) -> str:
for key, value in SUBTITLE_LANG.items():
for v in value:
if v in subtitle_name:
if v in subtitle_name.lower():
return key

11
backend/src/module/parser/title_parser.py
View File

@@ -49,7 +49,16 @@ class TitleParser:
"jp": episode.title_jp,
}
title_raw = episode.title_en if episode.title_en else episode.title_zh
official_title = titles[language] if titles[language] else titles["zh"]
if titles[language]:
official_title = titles[language]
elif titles["zh"]:
official_title = titles["zh"]
elif titles["en"]:
official_title = titles["en"]
elif titles["jp"]:
official_title = titles["jp"]
else:
official_title = title_raw
_season = episode.season
data = BangumiData(
official_title=official_title,