Files
2025-02-07 19:55:41 +08:00

644 lines
21 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from typing import Any, List, Dict, Tuple, Optional
from cachetools import cached, TTLCache
from app import schemas
from app.core.config import settings
from app.core.event import eventmanager, Event
from app.log import logger
from app.plugins import _PluginBase
from app.schemas import DiscoverSourceEventData
from app.schemas.types import ChainEventType
from app.utils.http import RequestUtils
class TvdbDiscover(_PluginBase):
# 插件名称
plugin_name = "TheTVDB探索"
# 插件描述
plugin_desc = "让探索支持TheTVDB的数据浏览。"
# 插件图标
plugin_icon = "TheTVDB_A.png"
# 插件版本
plugin_version = "1.1"
# 插件作者
plugin_author = "jxxghp"
# 作者主页
author_url = "https://github.com/jxxghp"
# 插件配置项ID前缀
plugin_config_prefix = "tvdbdiscover_"
# 加载顺序
plugin_order = 99
# 可使用的用户级别
auth_level = 1
# 私有属性
_base_api = "https://api4.thetvdb.com/v4"
_enabled = False
_proxy = False
_api_key = None
def init_plugin(self, config: dict = None):
if config:
self._enabled = config.get("enabled")
self._proxy = config.get("proxy")
self._api_key = config.get("api_key")
def get_state(self) -> bool:
return self._enabled
@staticmethod
def get_command() -> List[Dict[str, Any]]:
pass
def get_api(self) -> List[Dict[str, Any]]:
"""
获取插件API
[{
"path": "/xx",
"endpoint": self.xxx,
"methods": ["GET", "POST"],
"summary": "API说明"
}]
"""
return [{
"path": "/tvdb_discover",
"endpoint": self.tvdb_discover,
"methods": ["GET"],
"summary": "TheTVDB探索数据源",
"description": "获取TheTVDB探索数据",
}]
def get_form(self) -> Tuple[List[dict], Dict[str, Any]]:
"""
拼装插件配置页面需要返回两块数据1、页面配置2、数据结构
"""
return [
{
'component': 'VForm',
'content': [
{
'component': 'VRow',
'content': [
{
'component': 'VCol',
'props': {
'cols': 12,
'md': 4
},
'content': [
{
'component': 'VSwitch',
'props': {
'model': 'enabled',
'label': '启用插件',
}
}
]
},
{
'component': 'VCol',
'props': {
'cols': 12,
'md': 4
},
'content': [
{
'component': 'VSwitch',
'props': {
'model': 'proxy',
'label': '使用代理服务器',
}
}
]
}
]
},
{
'component': 'VRow',
'content': [
{
'component': 'VCol',
'props': {
'cols': 12
},
'content': [
{
'component': 'VTextField',
'props': {
'model': 'api_key',
'label': 'API Key'
}
}
]
}
]
}
]
}
], {
"enabled": False,
"proxy": False,
"api_key": "ed2aa66b-7899-4677-92a7-67bc9ce3d93a"
}
def get_page(self) -> List[dict]:
pass
@cached(cache=TTLCache(maxsize=1, ttl=30 * 24 * 3600))
def __get_token(self) -> Optional[str]:
"""
根据APIKEY获取token使用
"""
api_url = f"{self._base_api}/login"
headers = {
"Accept": "application/json",
"Content-Type": "application/json"
}
data = {
"apikey": self._api_key
}
res = RequestUtils(headers=headers).post_res(
api_url,
json=data,
proxies=settings.PROXY if self._proxy else None
)
if not res:
logger.error("获取TheMovieDB token失败")
return None
return res.json().get("data", {}).get("token")
@cached(cache=TTLCache(maxsize=32, ttl=1800))
def __request(self, mtype: str, **kwargs):
"""
请求TheTVDB API
"""
api_url = f"{self._base_api}/{mtype}/filter"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {self.__get_token()}"
}
res = RequestUtils(headers=headers).get_res(
api_url,
params=kwargs,
proxies=settings.PROXY if self._proxy else None
)
if res is None:
raise Exception("无法连接TheTVDB请检查网络连接")
if not res.ok:
raise Exception(f"请求TheTVDB API失败{res.text}")
return res.json().get("data")
def tvdb_discover(self, apikey: str, mtype: str = "series",
company: int = None, contentRating: int = None, country: str = "usa",
genre: int = None, lang: str = "eng", sort: str = "score", sortType: str = "desc",
status: int = None, year: int = None,
page: int = 1, count: int = 30) -> List[schemas.MediaInfo]:
"""
获取TheTVDB探索数据
"""
def __movie_to_media(movie_info: dict) -> schemas.MediaInfo:
"""
电影数据转换为MediaInfo
{
"id": 353554,
"name": "I Am: Celine Dion",
"slug": "i-am-celine-dion",
"image": "/banners/v4/movie/353554/posters/6656173b5167f.jpg",
"nameTranslations": null,
"overviewTranslations": null,
"aliases": null,
"score": 22669,
"runtime": 102,
"status": {
"id": 5,
"name": "Released",
"recordType": "movie",
"keepUpdated": true
},
"lastUpdated": "2024-08-10 10:37:05",
"year": "2024"
}
"""
return schemas.MediaInfo(
type="电影",
title=movie_info.get("name"),
year=movie_info.get("year"),
title_year=f"{movie_info.get('name')} ({movie_info.get('year')})",
mediaid_prefix="tvdb",
media_id=str(movie_info.get("id")),
poster_path=f"https://www.thetvdb.com{movie_info.get('image')}",
vote_average=movie_info.get("score"),
runtime=movie_info.get("runtime"),
overview=movie_info.get("overview")
)
def __series_to_media(series_info: dict) -> schemas.MediaInfo:
"""
电视剧数据转换为MediaInfo
{
"id": 79399,
"name": "Who Wants to Be a Superhero?",
"slug": "who-wants-to-be-a-superhero",
"image": "https://artworks.thetvdb.com/banners/posters/79399-1.jpg",
"nameTranslations": null,
"overviewTranslations": null,
"aliases": null,
"firstAired": "2006-07-27",
"lastAired": "2007-09-06",
"nextAired": "",
"score": 190,
"status": {
"id": 2,
"name": "Ended",
"recordType": "series",
"keepUpdated": false
},
"originalCountry": "usa",
"originalLanguage": "eng",
"defaultSeasonType": 1,
"isOrderRandomized": false,
"lastUpdated": "2022-01-16 03:32:39",
"averageRuntime": 45,
"episodes": null,
"overview": "",
"year": "2006"
}
"""
return schemas.MediaInfo(
type="电视剧",
title=series_info.get("name"),
year=series_info.get("year"),
title_year=f"{series_info.get('name')} ({series_info.get('year')})",
mediaid_prefix="tvdb",
media_id=str(series_info.get("id")),
release_date=series_info.get("firstAired"),
poster_path=series_info.get("image"),
vote_average=series_info.get("score"),
runtime=series_info.get("averageRuntime"),
overview=series_info.get("overview")
)
if apikey != settings.API_TOKEN:
return []
try:
# 计算页码TVDB为固定每页500条
if page * count > 500:
req_page = 500 // count
else:
req_page = page - 1
result = self.__request(
mtype,
company=company,
contentRating=contentRating,
country=country,
genre=genre,
lang=lang,
sort=sort,
sortType=sortType,
status=status,
year=year,
page=req_page
)
except Exception as err:
logger.error(str(err))
return []
if not result:
return []
if mtype == "movies":
results = [__movie_to_media(movie) for movie in result]
else:
results = [__series_to_media(series) for series in result]
return results[(page - 1) * count:page * count]
@staticmethod
def tvdb_filter_ui() -> List[dict]:
"""
TheTVDB过滤参数UI配置
"""
# 国家字典
country_dict = {
"usa": "美国",
"chn": "中国",
"jpn": "日本",
"kor": "韩国",
"ind": "印度",
"fra": "法国",
"ger": "德国",
"ita": "意大利",
"esp": "西班牙",
"uk": "英国",
"aus": "澳大利亚",
"can": "加拿大",
"rus": "俄罗斯",
"bra": "巴西",
"mex": "墨西哥",
"arg": "阿根廷",
"other": "其他"
}
cuntry_ui = [
{
"component": "VChip",
"props": {
"filter": True,
"tile": True,
"value": key
},
"text": value
} for key, value in country_dict.items()
]
# 原始语种字典
lang_dict = {
"eng": "英语",
"chi": "中文",
"jpn": "日语",
"kor": "韩语",
"hin": "印地语",
"fra": "法语",
"deu": "德语",
"ita": "意大利语",
"spa": "西班牙语",
"por": "葡萄牙语",
"rus": "俄语",
"other": "其他"
}
lang_ui = [
{
"component": "VChip",
"props": {
"filter": True,
"tile": True,
"value": key
},
"text": value
} for key, value in lang_dict.items()
]
# 风格字典
genre_dict = {
"1": "Soap",
"2": "Science Fiction",
"3": "Reality",
"4": "News",
"5": "Mini-Series",
"6": "Horror",
"7": "Home and Garden",
"8": "Game Show",
"9": "Food",
"10": "Fantasy",
"11": "Family",
"12": "Drama",
"13": "Documentary",
"14": "Crime",
"15": "Comedy",
"16": "Children",
"17": "Animation",
"18": "Adventure",
"19": "Action",
"21": "Sport",
"22": "Suspense",
"23": "Talk Show",
"24": "Thriller",
"25": "Travel",
"26": "Western",
"27": "Anime",
"28": "Romance",
"29": "Musical",
"30": "Podcast",
"31": "Mystery",
"32": "Indie",
"33": "History",
"34": "War",
"35": "Martial Arts",
"36": "Awards Show"
}
genre_ui = [
{
"component": "VChip",
"props": {
"filter": True,
"tile": True,
"value": key
},
"text": value
} for key, value in genre_dict.items()
]
# 排序字典
sort_dict = {
"score": "评分",
"firstAired": "首播日期",
"name": "名称"
}
sort_ui = [
{
"component": "VChip",
"props": {
"filter": True,
"tile": True,
"value": key
},
"text": value
} for key, value in sort_dict.items()
]
return [
{
"component": "div",
"props": {
"class": "flex justify-start items-center"
},
"content": [
{
"component": "div",
"props": {
"class": "mr-5"
},
"content": [
{
"component": "VLabel",
"text": "类型"
}
]
},
{
"component": "VChipGroup",
"props": {
"model": "mtype"
},
"content": [
{
"component": "VChip",
"props": {
"filter": True,
"tile": True,
"value": "movies"
},
"text": "电影"
},
{
"component": "VChip",
"props": {
"filter": True,
"tile": True,
"value": "series"
},
"text": "电视剧"
}
]
}
]
},
{
"component": "div",
"props": {
"class": "flex justify-start items-center"
},
"content": [
{
"component": "div",
"props": {
"class": "mr-5"
},
"content": [
{
"component": "VLabel",
"text": "风格"
}
]
},
{
"component": "VChipGroup",
"props": {
"model": "genre"
},
"content": genre_ui
}
]
},
{
"component": "div",
"props": {
"class": "flex justify-start items-center"
},
"content": [
{
"component": "div",
"props": {
"class": "mr-5"
},
"content": [
{
"component": "VLabel",
"text": "国家"
}
]
},
{
"component": "VChipGroup",
"props": {
"model": "country"
},
"content": cuntry_ui
}
]
},
{
"component": "div",
"props": {
"class": "flex justify-start items-center"
},
"content": [
{
"component": "div",
"props": {
"class": "mr-5"
},
"content": [
{
"component": "VLabel",
"text": "语言"
}
]
},
{
"component": "VChipGroup",
"props": {
"model": "lang"
},
"content": lang_ui
}
]
},
{
"component": "div",
"props": {
"class": "flex justify-start items-center"
},
"content": [
{
"component": "div",
"props": {
"class": "mr-5"
},
"content": [
{
"component": "VLabel",
"text": "排序"
}
]
},
{
"component": "VChipGroup",
"props": {
"model": "sort"
},
"content": sort_ui
}
]
}
]
@eventmanager.register(ChainEventType.DiscoverSource)
def discover_source(self, event: Event):
"""
监听识别事件使用ChatGPT辅助识别名称
"""
if not self._enabled or not self._api_key:
return
event_data: DiscoverSourceEventData = event.event_data
tvdb_source = schemas.DiscoverMediaSource(
name="TheTVDB",
mediaid_prefix="tvdb",
api_path=f"plugin/TvdbDiscover/tvdb_discover?apikey={settings.API_TOKEN}",
filter_params={
"mtype": "series",
"company": None,
"contentRating": None,
"country": "usa",
"genre": None,
"lang": "eng",
"sort": "score",
"sortType": "desc",
"status": None,
"year": None,
},
filter_ui=self.tvdb_filter_ui()
)
if not event_data.extra_sources:
event_data.extra_sources = [tvdb_source]
else:
event_data.extra_sources.append(tvdb_source)
def stop_service(self):
"""
退出插件
"""
pass