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 Data Model

Browse Source
This commit is contained in:
EstrellaXD
2023-04-26 18:14:17 +08:00
parent b2fc5c2bba
commit 7dc6d8d8a5
15 changed files with 127 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/module/manager/eps_complete.py
View File

@@ -6,6 +6,7 @@ from module.conf import settings
from module.network import RequestContent
from module.core.download_client import DownloadClient
from module.models import BangumiData
logger = logging.getLogger(__name__)
SEARCH_KEY = ["group", "title_raw", "season_raw", "subtitle", "source", "dpi"]
@@ -17,51 +18,55 @@ class FullSeasonGet:
pass
@staticmethod
def init_eps_complete_search_str(data: dict):
test = [data.get(key).strip() for key in SEARCH_KEY if data.get(key) is not None]
def init_eps_complete_search_str(data: BangumiData):
test = []
for key in SEARCH_KEY:
data_dict = data.dict()
if data_dict[key] is not None:
test.append(data_dict[key])
search_str_pre = "+".join(test)
search_str = re.sub(r"[\W_ ]", "+", search_str_pre)
return search_str
def get_season_torrents(self, data: dict):
def get_season_torrents(self, data: BangumiData):
keyword = self.init_eps_complete_search_str(data)
with RequestContent() as req:
torrents = req.get_torrents(f"{CUSTOM_URL}/RSS/Search?searchstr={keyword}")
return torrents
@staticmethod
def collect_season_torrents(data: dict, torrents):
def collect_season_torrents(data: BangumiData, torrents):
downloads = []
for torrent in torrents:
download_info = {
"url": torrent.torrent_link,
"save_path": os.path.join(
settings.downloader.path,
data["official_title"],
f"Season {data['season']}")
data.official_title,
f"Season {data.season}")
}
downloads.append(download_info)
return downloads
def download_eps(self, data, download_client: DownloadClient):
logger.info(f"Start collecting {data['official_title']} Season {data['season']}...")
def download_eps(self, data: BangumiData, download_client: DownloadClient):
logger.info(f"Start collecting {data.official_title} Season {data.season}...")
torrents = self.get_season_torrents(data)
downloads = self.collect_season_torrents(data, torrents)
for download in downloads:
download_client.add_torrent(download)
logger.info("Completed!")
data["eps_collect"] = False
data.eps_collect = False
def eps_complete(self, bangumi_info, download_client: DownloadClient):
def eps_complete(self, bangumi_info: list[BangumiData], download_client: DownloadClient):
for data in bangumi_info:
if data["eps_collect"]:
if data.eps_collect:
self.download_eps(data, download_client)
def download_collection(self, data, link, download_client: DownloadClient):
def download_collection(self, data: BangumiData, link, download_client: DownloadClient):
with RequestContent() as req:
torrents = req.get_torrents(link)
downloads = self.collect_season_torrents(data, torrents)
logger.info(f"Starting download {data.get('official_title')}")
logger.info(f"Starting download {data.official_title} Season {data.season}...")
for download in downloads:
download_client.add_torrent(download)
logger.info("Completed!")