nas/Auto_Bangumi
1
0
Fork 0
mirror of https://github.com/EstrellaXD/Auto_Bangumi.git synced 2026-05-02 06:23:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

Complete notification.py

Browse Source
This commit is contained in:
EstrellaXD
2023-03-15 20:39:14 +08:00
parent ad1d431549
commit 0a819d32bd
10 changed files with 93 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/module/api.py
View File

@@ -20,6 +20,7 @@ app.mount("/assets", StaticFiles(directory="templates/assets"), name="assets")
templates = Jinja2Templates(directory="templates")
# HTML Response
@app.get("/", response_class=HTMLResponse)
def index(request: Request):
context = {"request": request}

2
src/module/app.py
View File

@@ -71,7 +71,7 @@ def main_process(bangumi_data, download_client: DownloadClient):
if settings.bangumi_manage.enable:
rename.run()
times += 1
time.sleep(settings.program.sleep_time/settings.program.times)
time.sleep(settings.program.sleep_time / settings.program.times)
def run():

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

@@ -26,3 +26,22 @@ class Episode:
group: str
resolution: str
source: str
@dataclass
class SeasonInfo(dict):
official_title: str
title_raw: str
season: int
season_raw: str
group: str
filter: list | None
offset: int | None
dpi: str
source: str
subtitle: str
added: bool
eps_collect: bool

40
src/module/network/notification.py
View File

@@ -1,7 +1,6 @@
import logging
import requests
from .request_contents import RequestContent
from module.conf import settings
@@ -12,11 +11,23 @@ class PostNotification:
def __init__(self):
self.token = settings.notification_token
self.notification_url = lambda message: f"https://api.pushbullet.com/v2/{self.token}/{message}"
self.client = self.getClient()
def ifttt_post(self, message):
url = self.notification_url(message)
response = requests.get(url)
return response.status_code == 200
@staticmethod
def getClient():
if settings.notification.type.lower() == "telegram":
return TelegramNotification()
elif settings.notification.type.lower() == "serverchan":
return ServerChanNotification()
else:
return None
def send_msg(self, title: str, desp: str) -> bool:
if not settings.notification.enable:
return False
if self.client is None:
return False
return self.client.send_msg(title, desp)
class TelegramNotification:
@@ -35,7 +46,6 @@ class TelegramNotification:
class ServerChanNotification:
"""Server酱推送"""
def __init__(self):
self.token = settings.notification.token
self.notification_url = f"https://sctapi.ftqq.com/{self.token}.send"
@@ -47,16 +57,6 @@ class ServerChanNotification:
"title": title,
"desp": desp,
}
try:
resp = requests.post(self.notification_url, json=data, timeout=3)
resp.raise_for_status()
except requests.RequestException as e:
logging.error("[ServerChanNotification] send fail, error: %s" % e)
return False
return True
if __name__ == '__main__':
name = "勇者、辞职不干了"
notification = ServerChanNotification()
notification.send_msg(f"《{name[:10]}》缓存成功", f"[Auto Bangumi]《{name}》缓存成功")
with RequestContent() as req:
resp = req.post_data(self.notification_url, data)
return resp.status_code == 200

10
src/module/network/request_contents.py
View File

@@ -1,12 +1,12 @@
from dataclasses import dataclass
import re
from dataclasses import dataclass
from bs4 import BeautifulSoup
from .request_url import RequestURL
from module.conf import settings
import re
FILTER = "|".join(settings.rss_parser.filter)
@@ -41,3 +41,9 @@ class RequestContent(RequestURL):
# API JSON
def get_json(self, _url) -> dict:
return self.get_url(_url).json()
def post_json(self, _url, data: dict) -> dict:
return self.post_url(_url, data).json()
def post_data(self, _url, data: dict) -> dict:
return self.post_url(_url, data)

24
src/module/network/request_url.py
View File

@@ -18,8 +18,8 @@ class RequestURL:
}
def get_url(self, url):
times = 0
while times < 5:
try_time = 0
while try_time < 5:
try:
req = self.session.get(url=url, headers=self.header)
req.raise_for_status()
@@ -29,7 +29,25 @@ class RequestURL:
logger.debug(e)
logger.warning("ERROR with Connection.Please check DNS/Connection settings")
time.sleep(5)
times += 1
try_time += 1
except Exception as e:
logger.debug(f"URL: {url}")
logger.debug(e)
break
def post_url(self, url: str, data: dict):
try_time = 0
while try_time < 5:
try:
req = self.session.post(url=url, headers=self.header, data=data)
req.raise_for_status()
return req
except requests.RequestException as e:
logger.debug(f"URL: {url}")
logger.debug(e)
logger.warning("ERROR with Connection.Please check DNS/Connection settings")
time.sleep(5)
try_time += 1
except Exception as e:
logger.debug(f"URL: {url}")
logger.debug(e)

42
src/module/parser/fuzz_match.py
View File

@@ -1,42 +0,0 @@
from thefuzz import fuzz
import logging
from module.utils import json_config
from module.conf import settings
logger = logging.getLogger(__name__)
class FuzzMatch:
def __init__(self):
try:
anidb_data = json_config.get(settings.anidb_url)
json_config.save(settings.anidb_path, anidb_data)
except Exception as e:
logger.debug(e)
logger.info(f"Fail to get anidb data, reading local data")
anidb_data = json_config.load(settings.anidb_path)
self.match_data = anidb_data
@staticmethod
def match(title_raw, info: dict):
compare_value = []
for tag in ["main", "en", "ja", "zh-Hans", "zh-Hant"]:
if info[tag] is not None:
a = fuzz.token_sort_ratio(title_raw.lower(), info[tag].lower())
compare_value.append(a)
for compare in info["other"]:
a = fuzz.token_sort_ratio(title_raw.lower(), compare.lower())
compare_value.append(a)
return max(compare_value)
def find_max_name(self, title_raw):
max_value = 0
max_info = None
for info in self.match_data:
a = self.match(title_raw, info)
if a > max_value:
max_value = a
max_info = info
return max_value, max_info["main"]
# logger.debug(max(value))

17
src/module/parser/openai.py Normal file
View File

@@ -0,0 +1,17 @@
from module.network import RequestContent
from module.conf import settings
API_URL = "https://openai.estrella.cloud"
class OpenAIParser:
def __init__(self):
pass
def parse(self, title: str) -> dict:
prompt = {
"prompt": "This is a test prompt",
}
with RequestContent() as req:
req.get_json(API_URL, params=prompt)

7
src/module/parser/title_parser.py
View File

@@ -3,6 +3,7 @@ import logging
from .analyser import RawParser, DownloadParser, TMDBMatcher
from module.conf import settings
from module.models import SeasonInfo
logger = logging.getLogger(__name__)
LANGUAGE = settings.rss_parser.language
@@ -38,7 +39,7 @@ class TitleParser:
official_title = official_title if official_title else title
return official_title, tmdb_season
def return_dict(self, _raw: str):
def return_dict(self, _raw: str) -> dict:
try:
episode = self.raw_parser(_raw)
title_search = episode.title_zh if episode.title_zh else episode.title_en
@@ -59,8 +60,10 @@ class TitleParser:
"subtitle": episode.sub,
"added": False,
"eps_collect": True if episode.episode > 1 else False,
"offset": 0,
"filter": settings.rss_parser.filter
}
logger.debug(f"RAW:{_raw} >> {episode.title_en}")
return data
except Exception as e:
logger.debug(e)
logger.debug(e)