105 lines
4.3 KiB
Python
105 lines
4.3 KiB
Python
from typing import Optional, List, Dict
|
|
|
|
from app.chain import ChainBase
|
|
from app.core.config import settings
|
|
from app.core.context import Context, MediaInfo, TorrentInfo
|
|
from app.core.meta import MetaBase
|
|
from app.core.metainfo import MetaInfo
|
|
from app.helper.sites import SitesHelper
|
|
from app.log import logger
|
|
from app.utils.string import StringUtils
|
|
|
|
|
|
class SearchChain(ChainBase):
|
|
"""
|
|
站点资源搜索处理链
|
|
"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.siteshelper = SitesHelper()
|
|
|
|
def process(self, meta: MetaBase, mediainfo: MediaInfo,
|
|
keyword: str = None,
|
|
no_exists: Dict[int, List[dict]] = None) -> Optional[List[Context]]:
|
|
"""
|
|
根据媒体信息，执行搜索
|
|
:param meta: 元数据
|
|
:param mediainfo: 媒体信息
|
|
:param keyword: 搜索关键词
|
|
:param no_exists: 缺失的媒体信息
|
|
"""
|
|
logger.info(f'开始搜索资源，关键词：{keyword or mediainfo.title} ...')
|
|
# 未开启的站点不搜索
|
|
indexer_sites = []
|
|
for indexer in self.siteshelper.get_indexers():
|
|
if not settings.INDEXER_SITES \
|
|
or any([s in indexer.get("domain") for s in settings.INDEXER_SITES.split(',')]):
|
|
indexer_sites.append(indexer)
|
|
if not indexer_sites:
|
|
logger.warn('未开启任何有效站点，无法搜索资源')
|
|
return []
|
|
# 补充媒体信息
|
|
if not mediainfo.names:
|
|
mediainfo: MediaInfo = self.recognize_media(meta=MetaInfo(title=mediainfo.get_title_string()),
|
|
mtype=mediainfo.type,
|
|
tmdbid=mediainfo.tmdb_id)
|
|
if not mediainfo:
|
|
logger.error(f'媒体信息识别失败！')
|
|
return []
|
|
# 缺失的媒体信息
|
|
if no_exists:
|
|
# 过滤剧集
|
|
season_episodes = {info.get('season'): info.get('episodes')
|
|
for info in no_exists.get(mediainfo.tmdb_id)}
|
|
else:
|
|
season_episodes = None
|
|
# 执行搜索
|
|
torrents: List[TorrentInfo] = self.search_torrents(
|
|
mediainfo=mediainfo,
|
|
keyword=keyword,
|
|
sites=indexer_sites
|
|
)
|
|
if not torrents:
|
|
logger.warn(f'{keyword or mediainfo.title} 未搜索到资源')
|
|
return []
|
|
# 过滤种子
|
|
result: List[TorrentInfo] = self.filter_torrents(torrent_list=torrents,
|
|
season_episodes=season_episodes)
|
|
if result is not None:
|
|
torrents = result
|
|
if not torrents:
|
|
logger.warn(f'{keyword or mediainfo.title} 没有符合过滤条件的资源')
|
|
return []
|
|
# 过滤不匹配的资源
|
|
_match_torrents = []
|
|
if mediainfo:
|
|
for torrent in torrents:
|
|
# 比对IMDBID
|
|
if torrent.imdbid \
|
|
and mediainfo.imdb_id \
|
|
and torrent.imdbid == mediainfo.imdb_id:
|
|
logger.info(f'{mediainfo.title} 匹配到资源：{torrent.title}')
|
|
_match_torrents.append(torrent)
|
|
continue
|
|
# 识别
|
|
torrent_meta = MetaInfo(torrent.title, torrent.description)
|
|
# 比对标题
|
|
if torrent_meta.get_name() in [mediainfo.title, mediainfo.original_title]:
|
|
logger.info(f'{mediainfo.title} 匹配到资源：{torrent.title}')
|
|
_match_torrents.append(torrent)
|
|
continue
|
|
# 比对别名和译名
|
|
for name in mediainfo.names:
|
|
if StringUtils.clear(name).strip().upper() == \
|
|
StringUtils.clear(torrent_meta.get_name()).strip().upper():
|
|
logger.info(f'{mediainfo.title} 匹配到资源：{torrent.title}')
|
|
_match_torrents.append(torrent)
|
|
break
|
|
else:
|
|
_match_torrents = torrents
|
|
# 组装上下文返回
|
|
return [Context(meta=MetaInfo(torrent.title),
|
|
mediainfo=mediainfo,
|
|
torrentinfo=torrent) for torrent in _match_torrents]
|