codfrm/MoviePilot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat：人物搜索API

Browse Source
This commit is contained in:
jxxghp 2024-04-26 17:47:45 +08:00
parent 2f71e401be
commit f20b1bcfe9
8 changed files with 118 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
app/api/endpoints/media.py
View File

@ -63,15 +63,16 @@ def recognize_file2(path: str,
return recognize_file(path)
@router.get("/search", summary="搜索媒体信息", response_model=List[schemas.MediaInfo])
@router.get("/search", summary="搜索媒体/人物信息", response_model=List[schemas.MediaInfo])
def search_by_title(title: str,
type: str = "media",
page: int = 1,
count: int = 8,
_: schemas.TokenPayload = Depends(verify_token)) -> Any:
"""
模糊搜索媒体信息列表
模糊搜索媒体/人物信息列表 media：媒体信息，person：人物信息
"""
_, medias = MediaChain().search(title=title)
_, medias = MediaChain().search(title=title, stype=type)
if medias:
return [media.to_dict() for media in medias[(page - 1) * count: page * count]]
return []

9
app/chain/__init__.py
View File

@ -19,7 +19,7 @@ from app.db.message_oper import MessageOper
from app.helper.message import MessageHelper
from app.log import logger
from app.schemas import TransferInfo, TransferTorrent, ExistMediaInfo, DownloadingTorrent, CommingMessage, Notification, \
WebhookEventInfo, TmdbEpisode
WebhookEventInfo, TmdbEpisode, TmdbPerson, DoubanPerson
from app.schemas.types import TorrentStatus, MediaType, MediaImageType, EventType
from app.utils.object import ObjectUtils
@ -260,6 +260,13 @@ class ChainBase(metaclass=ABCMeta):
"""
return self.run_module("search_medias", meta=meta)
def search_persons(self, name: str) -> Optional[List[TmdbPerson, DoubanPerson]]:
"""
搜索人物信息
:param name: 人物名称
"""
return self.run_module("search_persons", name=name)
def search_torrents(self, site: CommentedMap,
keywords: List[str],
mtype: MediaType = None,

67
app/chain/media.py
View File

@ -2,7 +2,7 @@ import copy
import time
from pathlib import Path
from threading import Lock
from typing import Optional, List, Tuple
from typing import Optional, List, Tuple, Union
from app.chain import ChainBase
from app.core.context import Context, MediaInfo
@ -10,6 +10,7 @@ from app.core.event import eventmanager, Event
from app.core.meta import MetaBase
from app.core.metainfo import MetaInfo, MetaInfoPath
from app.log import logger
from app.schemas import TmdbPerson, DoubanPerson
from app.schemas.types import EventType, MediaType
from app.utils.singleton import Singleton
from app.utils.string import StringUtils
@ -156,36 +157,48 @@ class MediaChain(ChainBase, metaclass=Singleton):
# 返回上下文
return Context(meta_info=file_meta, media_info=mediainfo)
def search(self, title: str) -> Tuple[MetaBase, List[MediaInfo]]:
def search(self, title: str,
stype: str = "media") -> Tuple[Optional[MetaBase], List[Union[MediaInfo, TmdbPerson, DoubanPerson]]]:
"""
搜索媒体信息
搜索媒体/人物信息
:param title: 搜索内容
:param stype: 搜索类型 media：媒体信息，person：人物信息
:return: 识别元数据，媒体信息列表
"""
# 提取要素
mtype, key_word, season_num, episode_num, year, content = StringUtils.get_keyword(title)
# 识别
meta = MetaInfo(content)
if not meta.name:
meta.cn_name = content
# 合并信息
if mtype:
meta.type = mtype
if season_num:
meta.begin_season = season_num
if episode_num:
meta.begin_episode = episode_num
if year:
meta.year = year
# 开始搜索
logger.info(f"开始搜索媒体信息：{meta.name}")
medias: Optional[List[MediaInfo]] = self.search_medias(meta=meta)
if not medias:
logger.warn(f"{meta.name} 没有找到对应的媒体信息！")
return meta, []
logger.info(f"{content} 搜索到 {len(medias)} 条相关媒体信息")
# 识别的元数据，媒体信息列表
return meta, medias
if stype == "media":
# 提取要素
mtype, key_word, season_num, episode_num, year, content = StringUtils.get_keyword(title)
# 识别
meta = MetaInfo(content)
if not meta.name:
meta.cn_name = content
# 合并信息
if mtype:
meta.type = mtype
if season_num:
meta.begin_season = season_num
if episode_num:
meta.begin_episode = episode_num
if year:
meta.year = year
# 开始搜索
logger.info(f"开始搜索媒体信息：{meta.name}")
medias: Optional[List[MediaInfo]] = self.search_medias(meta=meta)
if not medias:
logger.warn(f"{meta.name} 没有找到对应的媒体信息！")
return meta, []
logger.info(f"{content} 搜索到 {len(medias)} 条相关媒体信息")
# 识别的元数据，媒体信息列表
return meta, medias
else:
# 搜索人物信息
logger.info(f"开始搜索人物信息：{title}")
persons: Optional[List[Union[TmdbPerson, DoubanPerson]]] = self.search_persons(name=title)
if not persons:
logger.warn(f"{title} 没有找到对应的人物信息！")
return None, []
logger.info(f"{title} 搜索到 {len(persons)} 条相关人物信息")
return None, persons
def get_tmdbinfo_by_doubanid(self, doubanid: str, mtype: MediaType = None) -> Optional[dict]:
"""

14
app/modules/douban/__init__.py
View File

@ -13,6 +13,7 @@ from app.modules import _ModuleBase
from app.modules.douban.apiv2 import DoubanApi
from app.modules.douban.douban_cache import DoubanCache
from app.modules.douban.scraper import DoubanScraper
from app.schemas import DoubanPerson
from app.schemas.types import MediaType
from app.utils.common import retry
from app.utils.http import RequestUtils
@ -563,6 +564,19 @@ class DoubanModule(_ModuleBase):
media.season = meta.begin_season
return ret_medias
def search_persons(self, name: str) -> Optional[List[DoubanPerson]]:
"""
搜索人物信息
"""
if settings.RECOGNIZE_SOURCE != "douban":
return None
if not name:
return []
result = self.doubanapi.person_search(name)
if not result:
return []
return [DoubanPerson(**item) for item in result.get("items")]
@retry(Exception, 5, 3, 3, logger=logger)
def match_doubaninfo(self, name: str, imdbid: str = None,
mtype: MediaType = None, year: str = None, season: int = None) -> dict:

11
app/modules/douban/apiv2.py
View File

@ -137,6 +137,9 @@ class DoubanApi(metaclass=Singleton):
# doulist
"doulist": "/doulist/",
"doulist_items": "/doulist/%s/items",
# personlist
"person_search": "/person/search",
}
_user_agents = [
@ -274,6 +277,14 @@ class DoubanApi(metaclass=Singleton):
return self.__invoke(self._urls["group_search"], q=keyword,
start=start, count=count, _ts=ts)
def person_search(self, keyword: str, start: int = 0, count: int = 20,
ts=datetime.strftime(datetime.now(), '%Y%m%d')):
"""
人物搜索
"""
return self.__invoke(self._urls["person_search"], q=keyword,
start=start, count=count, _ts=ts)
def movie_showing(self, start: int = 0, count: int = 20,
ts=datetime.strftime(datetime.now(), '%Y%m%d')):
"""

14
app/modules/themoviedb/__init__.py
View File

@ -13,6 +13,7 @@ from app.modules.themoviedb.category import CategoryHelper
from app.modules.themoviedb.scraper import TmdbScraper
from app.modules.themoviedb.tmdb_cache import TmdbCache
from app.modules.themoviedb.tmdbapi import TmdbApi
from app.schemas import TmdbPerson
from app.schemas.types import MediaType, MediaImageType
from app.utils.http import RequestUtils
from app.utils.system import SystemUtils
@ -261,6 +262,19 @@ class TheMovieDbModule(_ModuleBase):
return medias
return []
def search_persons(self, name: str) -> Optional[List[TmdbPerson]]:
"""
搜索人物信息
"""
if settings.RECOGNIZE_SOURCE != "themoviedb":
return None
if not name:
return []
results = self.tmdb.search_persons(name)
if results:
return [TmdbPerson(**person) for person in results]
return []
def scrape_metadata(self, path: Path, mediainfo: MediaInfo, transfer_type: str,
metainfo: MetaBase = None, force_nfo: bool = False, force_img: bool = False) -> None:
"""

14
app/modules/themoviedb/tmdbapi.py
View File

@ -95,6 +95,20 @@ class TmdbApi:
ret_infos.append(tv)
return ret_infos
def search_persons(self, name: str) -> List[dict]:
"""
查询模糊匹配的所有人物TMDB信息
"""
if not name:
return []
ret_infos = []
persons = self.person.search(query=name) or []
for person in persons:
if name in person.get("name"):
person['media_type'] = MediaType.PERSON
ret_infos.append(person)
return ret_infos
@staticmethod
def __compare_names(file_name: str, tmdb_names: list) -> bool:
"""

13
app/modules/themoviedb/tmdbv3api/objs/person.py
View File

@ -136,3 +136,16 @@ class Person(TMDb):
params="page=%s" % page,
key="results"
)
def search(self, query, page=1):
"""
Search for people.
:param query: str
:param page: int
:return:
"""
return self._request_obj(
self._urls["search_people"],
params="query=%s&page=%s" % (query, page),
key="results"
)