mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-02-03 02:25:32 +08:00
906 lines
37 KiB
Python
906 lines
37 KiB
Python
import json
|
||
from pathlib import Path
|
||
from typing import List, Optional, Dict, Tuple, Generator, Any, Union
|
||
from urllib.parse import quote_plus
|
||
|
||
from plexapi import media
|
||
from plexapi.myplex import MyPlexAccount
|
||
from plexapi.server import PlexServer
|
||
from requests import Response, Session
|
||
|
||
from app import schemas
|
||
from app.core.cache import cached
|
||
from app.log import logger
|
||
from app.schemas import MediaType
|
||
from app.utils.http import RequestUtils
|
||
from app.utils.url import UrlUtils
|
||
from app.schemas import MediaServerItem
|
||
|
||
|
||
class Plex:
|
||
_plex = None
|
||
_session = None
|
||
_sync_libraries: List[str] = []
|
||
|
||
def __init__(self, host: Optional[str] = None, token: Optional[str] = None, play_host: Optional[str] = None,
|
||
sync_libraries: list = None, **kwargs):
|
||
if not host or not token:
|
||
logger.error("Plex服务器配置不完整！")
|
||
return
|
||
self._host = host
|
||
if self._host:
|
||
self._host = UrlUtils.standardize_base_url(self._host)
|
||
self._playhost = play_host
|
||
if self._playhost:
|
||
self._playhost = UrlUtils.standardize_base_url(self._playhost)
|
||
self._token = token
|
||
if self._host and self._token:
|
||
try:
|
||
self._plex = PlexServer(self._host, self._token)
|
||
self._libraries = self._plex.library.sections()
|
||
except Exception as e:
|
||
self._plex = None
|
||
logger.error(f"Plex服务器连接失败：{str(e)}")
|
||
self._session = self.__adapt_plex_session()
|
||
self._sync_libraries = sync_libraries or []
|
||
|
||
def is_inactive(self) -> bool:
|
||
"""
|
||
判断是否需要重连
|
||
"""
|
||
if not self._host or not self._token:
|
||
return False
|
||
return True if not self._plex else False
|
||
|
||
def reconnect(self):
|
||
"""
|
||
重连
|
||
"""
|
||
try:
|
||
self._plex = PlexServer(self._host, self._token)
|
||
self._libraries = self._plex.library.sections()
|
||
except Exception as e:
|
||
self._plex = None
|
||
logger.error(f"Plex服务器连接失败：{str(e)}")
|
||
|
||
def authenticate(self, username: str, password: str) -> Optional[Tuple[str, str]]:
|
||
"""
|
||
用户认证
|
||
:param username: 用户名
|
||
:param password: 密码
|
||
:return: 认证成功返回 (token, 用户名)，否则返回 None
|
||
"""
|
||
if not username or not password:
|
||
return None
|
||
try:
|
||
account = MyPlexAccount(username=username, password=password, remember=False)
|
||
if account:
|
||
plex = PlexServer(self._host, account.authToken)
|
||
if not plex:
|
||
return None
|
||
return account.authToken, account.username
|
||
except Exception as e:
|
||
# 处理认证失败或网络错误等情况
|
||
logger.error(f"Authentication failed: {e}")
|
||
return None
|
||
|
||
@cached(maxsize=32, ttl=86400)
|
||
def __get_library_images(self, library_key: str, mtype: int) -> Optional[List[str]]:
|
||
"""
|
||
获取媒体服务器最近添加的媒体的图片列表
|
||
param: library_key
|
||
param: type type的含义: 1 电影 2 剧集 详见 plexapi/utils.py中SEARCHTYPES的定义
|
||
"""
|
||
if not self._plex:
|
||
return None
|
||
# 返回结果
|
||
poster_urls = {}
|
||
# 页码计数
|
||
container_start = 0
|
||
# 需要的总条数/每页的条数
|
||
total_size = 4
|
||
# 如果总数不足,接续获取下一页
|
||
while len(poster_urls) < total_size:
|
||
items = self._plex.fetchItems(f"/hubs/home/recentlyAdded?type={mtype}§ionID={library_key}",
|
||
container_start=container_start,
|
||
container_size=8,
|
||
maxresults=8)
|
||
for item in items:
|
||
if item.type == "episode":
|
||
# 如果是剧集的单集,则去找上级的图片
|
||
if item.parentThumb is not None:
|
||
poster_urls[item.parentThumb] = None
|
||
else:
|
||
# 否则就用自己的图片
|
||
if item.thumb is not None:
|
||
poster_urls[item.thumb] = None
|
||
if len(poster_urls) == total_size:
|
||
break
|
||
if len(items) < total_size:
|
||
break
|
||
container_start += total_size
|
||
return [f"{self._host.rstrip('/') + url}?X-Plex-Token={self._token}" for url in
|
||
list(poster_urls.keys())[:total_size]]
|
||
|
||
def get_librarys(self, hidden: Optional[bool] = False) -> List[schemas.MediaServerLibrary]:
|
||
"""
|
||
获取媒体服务器所有媒体库列表
|
||
"""
|
||
if not self._plex:
|
||
return []
|
||
try:
|
||
self._libraries = self._plex.library.sections()
|
||
except Exception as err:
|
||
logger.error(f"获取媒体服务器所有媒体库列表出错：{str(err)}")
|
||
return []
|
||
libraries = []
|
||
for library in self._libraries:
|
||
if hidden and self._sync_libraries and "all" not in self._sync_libraries \
|
||
and str(library.key) not in self._sync_libraries:
|
||
continue
|
||
if library.type == "movie":
|
||
library_type = MediaType.MOVIE.value
|
||
image_list = self.__get_library_images(library.key, 1)
|
||
elif library.type == "show":
|
||
library_type = MediaType.TV.value
|
||
image_list = self.__get_library_images(library.key, 2)
|
||
else:
|
||
continue
|
||
libraries.append(
|
||
schemas.MediaServerLibrary(
|
||
id=library.key,
|
||
name=library.title,
|
||
path=library.locations,
|
||
type=library_type,
|
||
image_list=image_list,
|
||
link=f"{self._playhost or self._host}web/index.html#!/media/{self._plex.machineIdentifier}"
|
||
f"/com.plexapp.plugins.library?source={library.key}&X-Plex-Token={self._token}",
|
||
server_type='plex'
|
||
)
|
||
)
|
||
return libraries
|
||
|
||
def get_medias_count(self) -> schemas.Statistic:
|
||
"""
|
||
获得电影、电视剧、动漫媒体数量
|
||
:return: movie_count tv_count episode_count
|
||
"""
|
||
if not self._plex:
|
||
return schemas.Statistic()
|
||
sections = self._plex.library.sections()
|
||
movie_count = tv_count = episode_count = 0
|
||
# 媒体库白名单
|
||
allow_library = [str(lib.id) for lib in self.get_librarys(hidden=True)]
|
||
for sec in sections:
|
||
if str(sec.key) not in allow_library:
|
||
continue
|
||
if sec.type == "movie":
|
||
movie_count += sec.totalSize
|
||
if sec.type == "show":
|
||
tv_count += sec.totalSize
|
||
episode_count += sec.totalViewSize(libtype="episode")
|
||
return schemas.Statistic(
|
||
movie_count=movie_count,
|
||
tv_count=tv_count,
|
||
episode_count=episode_count
|
||
)
|
||
|
||
def get_movies(self,
|
||
title: str,
|
||
original_title: Optional[str] = None,
|
||
year: Optional[str] = None,
|
||
tmdb_id: Optional[int] = None) -> Optional[List[schemas.MediaServerItem]]:
|
||
"""
|
||
根据标题和年份，检查电影是否在Plex中存在，存在则返回列表
|
||
:param title: 标题
|
||
:param original_title: 原产地标题
|
||
:param year: 年份，为空则不过滤
|
||
:param tmdb_id: TMDB ID
|
||
:return: 含title、year属性的字典列表
|
||
"""
|
||
if not self._plex:
|
||
return None
|
||
ret_movies = []
|
||
if year:
|
||
movies = self._plex.library.search(title=title,
|
||
year=year,
|
||
libtype="movie")
|
||
# 根据原标题再查一遍
|
||
if original_title and str(original_title) != str(title):
|
||
movies.extend(self._plex.library.search(title=original_title,
|
||
year=year,
|
||
libtype="movie"))
|
||
else:
|
||
movies = self._plex.library.search(title=title,
|
||
libtype="movie")
|
||
if original_title and str(original_title) != str(title):
|
||
movies.extend(self._plex.library.search(title=original_title,
|
||
libtype="movie"))
|
||
for item in set(movies):
|
||
ids = self.__get_ids(item.guids)
|
||
if tmdb_id and ids['tmdb_id']:
|
||
if str(ids['tmdb_id']) != str(tmdb_id):
|
||
continue
|
||
path = None
|
||
if item.locations:
|
||
path = item.locations[0]
|
||
ret_movies.append(
|
||
schemas.MediaServerItem(
|
||
server="plex",
|
||
library=item.librarySectionID,
|
||
item_id=item.key,
|
||
item_type=item.type,
|
||
title=item.title,
|
||
original_title=item.originalTitle,
|
||
year=item.year,
|
||
tmdbid=ids['tmdb_id'],
|
||
imdbid=ids['imdb_id'],
|
||
tvdbid=ids['tvdb_id'],
|
||
path=path,
|
||
)
|
||
)
|
||
return ret_movies
|
||
|
||
def get_tv_episodes(self,
|
||
item_id: Optional[str] = None,
|
||
title: Optional[str] = None,
|
||
original_title: Optional[str] = None,
|
||
year: Optional[str] = None,
|
||
tmdb_id: Optional[int] = None,
|
||
season: Optional[int] = None) -> Tuple[Optional[str], Optional[Dict[int, list]]]:
|
||
"""
|
||
根据标题、年份、季查询电视剧所有集信息
|
||
:param item_id: 媒体ID
|
||
:param title: 标题
|
||
:param original_title: 原产地标题
|
||
:param year: 年份，可以为空，为空时不按年份过滤
|
||
:param tmdb_id: TMDB ID
|
||
:param season: 季号，数字
|
||
:return: 所有集的列表
|
||
"""
|
||
if not self._plex:
|
||
return None, {}
|
||
if item_id:
|
||
videos = self.__fetch_item(item_id)
|
||
else:
|
||
# 兼容年份为空的场景
|
||
kwargs = {"year": year} if year else {}
|
||
# 根据标题和年份模糊搜索，该结果不够准确
|
||
videos = self._plex.library.search(title=title,
|
||
libtype="show",
|
||
**kwargs)
|
||
if (not videos
|
||
and original_title
|
||
and str(original_title) != str(title)):
|
||
videos = self._plex.library.search(title=original_title,
|
||
libtype="show",
|
||
**kwargs)
|
||
|
||
if not videos:
|
||
return None, {}
|
||
if isinstance(videos, list):
|
||
videos = videos[0]
|
||
video_tmdbid = self.__get_ids(videos.guids).get('tmdb_id')
|
||
if tmdb_id and video_tmdbid:
|
||
if str(video_tmdbid) != str(tmdb_id):
|
||
return None, {}
|
||
episodes = videos.episodes()
|
||
season_episodes = {}
|
||
for episode in episodes:
|
||
if season is not None and episode.seasonNumber != int(season):
|
||
continue
|
||
if episode.seasonNumber not in season_episodes:
|
||
season_episodes[episode.seasonNumber] = []
|
||
season_episodes[episode.seasonNumber].append(episode.index)
|
||
return videos.key, season_episodes
|
||
|
||
def get_remote_image_by_id(self,
|
||
item_id: str,
|
||
image_type: str,
|
||
depth: Optional[int] = 0,
|
||
plex_url: Optional[bool] = True) -> Optional[str]:
|
||
"""
|
||
根据ItemId从Plex查询图片地址
|
||
:param item_id: 在Plex中的ID
|
||
:param image_type: 图片的类型，Poster或者Backdrop等
|
||
:param depth: 当前递归深度，默认为0
|
||
:param plex_url: 是否返回Plex的URL，默认为True（仅在配置了外网地址和Token时有效）
|
||
:return: 图片对应在plex服务器或TMDB中的URL
|
||
"""
|
||
if not self._plex or depth > 2 or not item_id:
|
||
return None
|
||
try:
|
||
image_url = None
|
||
ekey = item_id
|
||
item = self._plex.fetchItem(ekey=ekey)
|
||
if not item:
|
||
return None
|
||
# 如果配置了外网播放地址以及Token，则默认从Plex媒体服务器获取图片，否则返回有外网地址的图片资源
|
||
# Plex外网播放地址这个框里目前可以填两种地址
|
||
# 1. Plex的官方转发地址https://app.plex.tv, 2. 自己处理的端口转发地址
|
||
# 如果使用的是1的官方转发地址,那么就不能走这个逻辑，因为官方转发地址无法获取到图片
|
||
if (self._playhost and "app.plex.tv" not in self._playhost
|
||
and self._token and plex_url):
|
||
query = {"X-Plex-Token": self._token}
|
||
if image_type == "Poster":
|
||
if item.thumb:
|
||
image_url = UrlUtils.combine_url(host=self._playhost, path=item.thumb, query=query)
|
||
else:
|
||
# 默认使用art也就是Backdrop进行处理
|
||
if item.art:
|
||
image_url = UrlUtils.combine_url(host=self._playhost, path=item.art, query=query)
|
||
# 这里对episode进行特殊处理，实际上episode的Backdrop是Poster
|
||
# 也有个别情况，比如机智的凡人小子episode就是Poster，因此这里把episode的优先级降低，默认还是取art
|
||
if not image_url and item.TYPE == "episode" and item.thumb:
|
||
image_url = UrlUtils.combine_url(host=self._playhost, path=item.thumb, query=query)
|
||
else:
|
||
if image_type == "Poster":
|
||
images = self._plex.fetchItems(ekey=f"{ekey}/posters",
|
||
cls=media.Poster)
|
||
else:
|
||
# 默认使用art也就是Backdrop进行处理
|
||
images = self._plex.fetchItems(ekey=f"{ekey}/arts",
|
||
cls=media.Art)
|
||
# 这里对episode进行特殊处理，实际上episode的Backdrop是Poster
|
||
# 也有个别情况，比如机智的凡人小子episode就是Poster，因此这里把episode的优先级降低，默认还是取art
|
||
if not images and item.TYPE == "episode":
|
||
images = self._plex.fetchItems(ekey=f"{ekey}/posters",
|
||
cls=media.Poster)
|
||
for image in images:
|
||
if hasattr(image, "key") and image.key.startswith("http"):
|
||
image_url = image.key
|
||
break
|
||
# 如果最后还是找不到，则递归父级进行查找
|
||
if not image_url and hasattr(item, "parentKey"):
|
||
return self.get_remote_image_by_id(item_id=item.parentKey,
|
||
image_type=image_type,
|
||
depth=depth + 1)
|
||
return image_url
|
||
except Exception as e:
|
||
logger.error(f"获取封面出错：" + str(e))
|
||
return None
|
||
|
||
def refresh_root_library(self) -> bool:
|
||
"""
|
||
通知Plex刷新整个媒体库
|
||
"""
|
||
if not self._plex:
|
||
return False
|
||
return self._plex.library.update()
|
||
|
||
def refresh_library_by_items(self, items: List[schemas.RefreshMediaItem]) -> Optional[bool]:
|
||
"""
|
||
按路径刷新媒体库 item: target_path
|
||
"""
|
||
if not self._plex:
|
||
return False
|
||
result_dict = {}
|
||
for item in items:
|
||
file_path = item.target_path
|
||
lib_key, path = self.__find_librarie(file_path, self._libraries)
|
||
# 如果存在同一剧集的多集,key(path)相同会合并
|
||
if path:
|
||
result_dict[path.as_posix()] = lib_key
|
||
else:
|
||
result_dict[""] = lib_key
|
||
if "" in result_dict:
|
||
# 如果有匹配失败的,刷新整个库
|
||
self._plex.library.update()
|
||
else:
|
||
# 否则一个一个刷新
|
||
for path, lib_key in result_dict.items():
|
||
logger.info(f"刷新媒体库：{lib_key} - {path}")
|
||
self._plex.query(f'/library/sections/{lib_key}/refresh?path={quote_plus(Path(path).parent.as_posix())}')
|
||
return None
|
||
return None
|
||
|
||
@staticmethod
|
||
def __find_librarie(path: Path, libraries: List[Any]) -> Tuple[str, Optional[Path]]:
|
||
"""
|
||
判断这个path属于哪个媒体库
|
||
多个媒体库配置的目录不应有重复和嵌套,
|
||
"""
|
||
|
||
def is_subpath(_path: Path, _parent: Path) -> bool:
|
||
"""
|
||
判断_path是否是_parent的子目录下
|
||
"""
|
||
_path = _path.resolve()
|
||
_parent = _parent.resolve()
|
||
return _path.parts[:len(_parent.parts)] == _parent.parts
|
||
|
||
if path is None:
|
||
return "", None
|
||
|
||
try:
|
||
for lib in libraries:
|
||
if hasattr(lib, "locations") and lib.locations:
|
||
for location in lib.locations:
|
||
if is_subpath(path, Path(location)):
|
||
return lib.key, path
|
||
except Exception as err:
|
||
logger.error(f"查找媒体库出错：{str(err)}")
|
||
return "", None
|
||
|
||
def get_iteminfo(self, itemid: str) -> Optional[schemas.MediaServerItem]:
|
||
"""
|
||
获取单个项目详情
|
||
"""
|
||
if not self._plex:
|
||
return None
|
||
try:
|
||
item = self.__fetch_item(itemid)
|
||
return self.__build_media_server_item(item)
|
||
except Exception as err:
|
||
logger.error(f"获取项目详情出错：{str(err)}")
|
||
return None
|
||
|
||
@staticmethod
|
||
def __get_ids(guids: List[Any]) -> dict:
|
||
def parse_tmdb_id(value: str) -> tuple[bool, int]:
|
||
"""尝试将TMDB ID字符串转换为整数。如果成功，返回(True, int)，失败则返回(False, None)。"""
|
||
try:
|
||
int_value = int(value)
|
||
return True, int_value
|
||
except ValueError:
|
||
return False, None
|
||
|
||
guid_mapping = {
|
||
"imdb://": "imdb_id",
|
||
"tmdb://": "tmdb_id",
|
||
"tvdb://": "tvdb_id"
|
||
}
|
||
ids = {varname: None for varname in guid_mapping.values()}
|
||
for guid in guids:
|
||
guid_id = guid['id'] if isinstance(guid, dict) else guid.id
|
||
for prefix, varname in guid_mapping.items():
|
||
if guid_id.startswith(prefix):
|
||
clean_id = guid_id[len(prefix):]
|
||
if varname == "tmdb_id":
|
||
# tmdb_id为int，Plex可能存在脏数据，特别处理tmdb_id
|
||
success, parsed_id = parse_tmdb_id(clean_id)
|
||
if success:
|
||
ids[varname] = parsed_id
|
||
else:
|
||
ids[varname] = clean_id
|
||
break
|
||
|
||
return ids
|
||
|
||
def __fetch_item(self, item_id: Union[int, str]):
|
||
"""
|
||
根据给定的item_id获取媒体项
|
||
:param item_id: 媒体项的ID，可以是整数或字符串，如果是字符串且表示为数字，将会被转换为整数
|
||
"""
|
||
if isinstance(item_id, str) and item_id.isdigit():
|
||
item_id = int(item_id)
|
||
return self._plex.fetchItem(item_id)
|
||
|
||
def __build_media_server_item(self, item) -> Optional[schemas.MediaServerItem]:
|
||
"""
|
||
构造MediaServerItem
|
||
:param item: Plex媒体项目
|
||
:return: MediaServerItem
|
||
"""
|
||
if not item:
|
||
return None
|
||
ids = self.__get_ids(item.guids)
|
||
path = item.locations[0] if item.locations else None
|
||
playback_position = getattr(item, "viewOffset", None) or 0
|
||
duration = getattr(item, "duration", None) or 0
|
||
percentage = (playback_position / duration * 100) if duration > 0 else None
|
||
played = getattr(item, "isPlayed", None) or False
|
||
play_count = getattr(item, "viewCount", None) or 0
|
||
last_played_date = getattr(item, "lastViewedAt", None)
|
||
|
||
user_state = schemas.MediaServerItemUserState(
|
||
played=played,
|
||
resume=playback_position > 0,
|
||
last_played_date=last_played_date.isoformat() if last_played_date and hasattr(last_played_date,
|
||
'isoformat') else None,
|
||
play_count=play_count,
|
||
percentage=percentage,
|
||
)
|
||
|
||
return schemas.MediaServerItem(
|
||
server="plex",
|
||
library=item.librarySectionID,
|
||
item_id=item.key,
|
||
item_type=item.type,
|
||
title=item.title,
|
||
original_title=item.originalTitle,
|
||
year=item.year,
|
||
tmdbid=ids.get("tmdb_id"),
|
||
imdbid=ids.get("imdb_id"),
|
||
tvdbid=ids.get("tvdb_id"),
|
||
path=path,
|
||
user_state=user_state,
|
||
)
|
||
|
||
def get_items(self, parent: Union[str, int], start_index: Optional[int] = 0, limit: Optional[int] = -1) \
|
||
-> Generator[MediaServerItem | None, Any, None]:
|
||
"""
|
||
获取媒体服务器项目列表，支持分页和不分页逻辑，默认不分页获取所有数据
|
||
|
||
:param parent: 媒体库ID，用于标识要获取的媒体库
|
||
:param start_index: 起始索引，用于分页获取数据。默认为 0，即从第一个项目开始获取
|
||
:param limit: 每次请求的最大项目数，用于分页。如果为 None 或 -1，则表示一次性获取所有数据，默认为 -1
|
||
|
||
:return: 返回一个生成器对象，用于逐步获取媒体服务器中的项目
|
||
"""
|
||
if not parent or not self._plex:
|
||
return None
|
||
try:
|
||
section = self._plex.library.sectionByID(int(parent))
|
||
if section:
|
||
if limit is None or limit == -1:
|
||
items = section.all()
|
||
else:
|
||
items = section.all(container_start=start_index, container_size=limit, maxresults=limit)
|
||
for item in items:
|
||
try:
|
||
if not item:
|
||
continue
|
||
yield self.__build_media_server_item(item)
|
||
except Exception as e:
|
||
logger.error(f"处理媒体项目时出错：{str(e)}, 跳过此项目")
|
||
continue
|
||
except Exception as err:
|
||
logger.error(f"获取媒体库列表出错：{str(err)}")
|
||
return None
|
||
|
||
def get_webhook_message(self, form: any) -> Optional[schemas.WebhookEventInfo]:
|
||
"""
|
||
解析Plex报文
|
||
eventItem 字段的含义
|
||
event 事件类型
|
||
item_type 媒体类型 TV,MOV
|
||
item_name TV:琅琊榜 S1E6 剖心明志 虎口脱险
|
||
MOV:猪猪侠大冒险(2001)
|
||
overview 剧情描述
|
||
{
|
||
"event": "media.scrobble",
|
||
"user": false,
|
||
"owner": true,
|
||
"Account": {
|
||
"id": 31646104,
|
||
"thumb": "https://plex.tv/users/xx",
|
||
"title": "播放"
|
||
},
|
||
"Server": {
|
||
"title": "Media-Server",
|
||
"uuid": "xxxx"
|
||
},
|
||
"Player": {
|
||
"local": false,
|
||
"publicAddress": "xx.xx.xx.xx",
|
||
"title": "MagicBook",
|
||
"uuid": "wu0uoa1ujfq90t0c5p9f7fw0"
|
||
},
|
||
"Metadata": {
|
||
"librarySectionType": "show",
|
||
"ratingKey": "40294",
|
||
"key": "/library/metadata/40294",
|
||
"parentRatingKey": "40291",
|
||
"grandparentRatingKey": "40275",
|
||
"guid": "plex://episode/615580a9fa828e7f1a0caabd",
|
||
"parentGuid": "plex://season/615580a9fa828e7f1a0caab8",
|
||
"grandparentGuid": "plex://show/60e81fd8d8000e002d7d2976",
|
||
"type": "episode",
|
||
"title": "The World's Strongest Senior",
|
||
"titleSort": "World's Strongest Senior",
|
||
"grandparentKey": "/library/metadata/40275",
|
||
"parentKey": "/library/metadata/40291",
|
||
"librarySectionTitle": "动漫剧集",
|
||
"librarySectionID": 7,
|
||
"librarySectionKey": "/library/sections/7",
|
||
"grandparentTitle": "范马刃牙",
|
||
"parentTitle": "Combat Shadow Fighting Saga / Great Prison Battle Saga",
|
||
"originalTitle": "Baki Hanma",
|
||
"contentRating": "TV-MA",
|
||
"summary": "The world is shaken by news",
|
||
"index": 1,
|
||
"parentIndex": 1,
|
||
"audienceRating": 8.5,
|
||
"viewCount": 1,
|
||
"lastViewedAt": 1694320444,
|
||
"year": 2021,
|
||
"thumb": "/library/metadata/40294/thumb/1693544504",
|
||
"art": "/library/metadata/40275/art/1693952979",
|
||
"parentThumb": "/library/metadata/40291/thumb/1691115271",
|
||
"grandparentThumb": "/library/metadata/40275/thumb/1693952979",
|
||
"grandparentArt": "/library/metadata/40275/art/1693952979",
|
||
"duration": 1500000,
|
||
"originallyAvailableAt": "2021-09-30",
|
||
"addedAt": 1691115281,
|
||
"updatedAt": 1693544504,
|
||
"audienceRatingImage": "themoviedb://image.rating",
|
||
"Guid": [
|
||
{
|
||
"id": "imdb://tt14765720"
|
||
},
|
||
{
|
||
"id": "tmdb://3087250"
|
||
},
|
||
{
|
||
"id": "tvdb://8530933"
|
||
}
|
||
],
|
||
"Rating": [
|
||
{
|
||
"image": "themoviedb://image.rating",
|
||
"value": 8.5,
|
||
"type": "audience"
|
||
}
|
||
],
|
||
"Director": [
|
||
{
|
||
"id": 115144,
|
||
"filter": "director=115144",
|
||
"tag": "Keiya Saito",
|
||
"tagKey": "5f401c8d04a86500409ea6c1"
|
||
}
|
||
],
|
||
"Writer": [
|
||
{
|
||
"id": 115135,
|
||
"filter": "writer=115135",
|
||
"tag": "Tatsuhiko Urahata",
|
||
"tagKey": "5d7768e07a53e9001e6db1ce",
|
||
"thumb": "https://metadata-static.plex.tv/f/people/f6f90dc89fa87d459f85d40a09720c05.jpg"
|
||
}
|
||
]
|
||
}
|
||
}
|
||
"""
|
||
if not form:
|
||
return None
|
||
payload = form.get("payload")
|
||
if not payload:
|
||
return None
|
||
try:
|
||
message = json.loads(payload)
|
||
except Exception as e:
|
||
logger.debug(f"解析plex webhook出错：{str(e)}")
|
||
return None
|
||
eventType = message.get('event')
|
||
if not eventType:
|
||
return None
|
||
logger.debug(f"接收到plex webhook：{message}")
|
||
eventItem = schemas.WebhookEventInfo(event=eventType, channel="plex")
|
||
if message.get('Metadata'):
|
||
if message.get('Metadata', {}).get('type') == 'episode':
|
||
eventItem.item_type = "TV"
|
||
eventItem.item_name = "%s %s%s %s" % (
|
||
message.get('Metadata', {}).get('grandparentTitle'),
|
||
"S" + str(message.get('Metadata', {}).get('parentIndex')),
|
||
"E" + str(message.get('Metadata', {}).get('index')),
|
||
message.get('Metadata', {}).get('title'))
|
||
eventItem.item_id = message.get('Metadata', {}).get('key')
|
||
eventItem.season_id = message.get('Metadata', {}).get('parentIndex')
|
||
eventItem.episode_id = message.get('Metadata', {}).get('index')
|
||
|
||
if (message.get('Metadata', {}).get('summary')
|
||
and len(message.get('Metadata', {}).get('summary')) > 100):
|
||
eventItem.overview = str(message.get('Metadata', {}).get('summary'))[:100] + "..."
|
||
else:
|
||
eventItem.overview = message.get('Metadata', {}).get('summary')
|
||
else:
|
||
eventItem.item_type = "MOV" if message.get('Metadata',
|
||
{}).get('type') == 'movie' else "SHOW"
|
||
eventItem.item_name = "%s %s" % (
|
||
message.get('Metadata', {}).get('title'),
|
||
"(" + str(message.get('Metadata', {}).get('year')) + ")")
|
||
eventItem.item_id = message.get('Metadata', {}).get('key')
|
||
if len(message.get('Metadata', {}).get('summary')) > 100:
|
||
eventItem.overview = str(message.get('Metadata', {}).get('summary'))[:100] + "..."
|
||
else:
|
||
eventItem.overview = message.get('Metadata', {}).get('summary')
|
||
if message.get('Player'):
|
||
eventItem.ip = message.get('Player').get('publicAddress')
|
||
eventItem.client = message.get('Player').get('title')
|
||
# 这里给个空,防止拼消息的时候出现None
|
||
eventItem.device_name = ' '
|
||
if message.get('Account'):
|
||
eventItem.user_name = message.get("Account").get('title')
|
||
|
||
# 获取消息图片
|
||
if eventItem.item_id:
|
||
# 根据返回的item_id去调用媒体服务器获取
|
||
eventItem.image_url = self.get_remote_image_by_id(item_id=eventItem.item_id,
|
||
image_type="Backdrop")
|
||
|
||
eventItem.json_object = message
|
||
|
||
return eventItem
|
||
|
||
def get_plex(self):
|
||
"""
|
||
获取plex对象，以便直接操作
|
||
"""
|
||
return self._plex
|
||
|
||
def get_play_url(self, item_id: str) -> str:
|
||
"""
|
||
拼装媒体播放链接
|
||
:param item_id: 媒体的的ID
|
||
"""
|
||
return f'{self._playhost or self._host}web/index.html#!/server/{self._plex.machineIdentifier}/details?key={item_id}&X-Plex-Token={self._token}'
|
||
|
||
def get_resume(self, num: Optional[int] = 12) -> Optional[List[schemas.MediaServerPlayItem]]:
|
||
"""
|
||
获取继续观看的媒体
|
||
"""
|
||
if not self._plex:
|
||
return []
|
||
# 媒体库白名单
|
||
allow_library = ",".join(map(str, (lib.id for lib in self.get_librarys(hidden=True))))
|
||
params = {"contentDirectoryID": allow_library}
|
||
items = self._plex.fetchItems("/hubs/continueWatching/items",
|
||
container_start=0,
|
||
container_size=num,
|
||
maxresults=num,
|
||
params=params)
|
||
ret_resume = []
|
||
for item in items:
|
||
item_type = MediaType.MOVIE.value if item.TYPE == "movie" else MediaType.TV.value
|
||
if item_type == MediaType.MOVIE.value:
|
||
title = item.title
|
||
subtitle = str(item.year) if item.year else None
|
||
else:
|
||
title = item.grandparentTitle
|
||
subtitle = f"S{item.parentIndex}:E{item.index} - {item.title}"
|
||
link = self.get_play_url(item.key)
|
||
image = item.artUrl
|
||
ret_resume.append(schemas.MediaServerPlayItem(
|
||
id=item.key,
|
||
title=title,
|
||
subtitle=subtitle,
|
||
type=item_type,
|
||
image=image,
|
||
link=link,
|
||
percent=item.viewOffset / item.duration * 100 if item.viewOffset and item.duration else 0,
|
||
server_type='plex'
|
||
))
|
||
return ret_resume[:num]
|
||
|
||
def get_latest(self, num: Optional[int] = 20) -> Optional[List[schemas.MediaServerPlayItem]]:
|
||
"""
|
||
获取最近添加媒体
|
||
"""
|
||
if not self._plex:
|
||
return None
|
||
# 请求参数（除黑名单）
|
||
allow_library = ",".join(map(str, (lib.id for lib in self.get_librarys(hidden=True))))
|
||
params = {
|
||
"contentDirectoryID": allow_library,
|
||
"count": num,
|
||
"excludeContinueWatching": 1
|
||
}
|
||
ret_resume = []
|
||
sub_result = []
|
||
offset = 0
|
||
while True:
|
||
if len(ret_resume) >= num:
|
||
break
|
||
# 获取所有资料库
|
||
hubs = self._plex.fetchItems(
|
||
'/hubs/promoted',
|
||
container_start=offset,
|
||
container_size=num,
|
||
maxresults=num,
|
||
params=params
|
||
)
|
||
if len(hubs) == 0:
|
||
break
|
||
|
||
# 合并排序
|
||
for hub in hubs:
|
||
for item in hub.items():
|
||
sub_result.append(item)
|
||
sub_result.sort(key=lambda x: x.addedAt, reverse=True)
|
||
|
||
for item in sub_result:
|
||
if len(ret_resume) >= num:
|
||
break
|
||
item_type, title, image = "", "", ""
|
||
if item.TYPE == "movie":
|
||
item_type = MediaType.MOVIE.value
|
||
title = item.title
|
||
image = item.posterUrl
|
||
elif item.TYPE == "season":
|
||
item_type = MediaType.TV.value
|
||
title = "%s 第%s季" % (item.parentTitle, item.index)
|
||
image = item.posterUrl
|
||
elif item.TYPE == "episode":
|
||
item_type = MediaType.TV.value
|
||
title = "%s 第%s季 第%s集" % (item.grandparentTitle, item.parentIndex, item.index)
|
||
thumb = (item.parentThumb or item.grandparentThumb or '').lstrip('/')
|
||
image = (self._host + thumb + f"?X-Plex-Token={self._token}")
|
||
elif item.TYPE == "show":
|
||
item_type = MediaType.TV.value
|
||
title = "%s 共%s季" % (item.title, item.seasonCount)
|
||
image = item.posterUrl
|
||
link = self.get_play_url(item.key)
|
||
ret_resume.append(schemas.MediaServerPlayItem(
|
||
id=item.key,
|
||
title=title,
|
||
subtitle=str(item.year) if item.year else None,
|
||
type=item_type,
|
||
image=image,
|
||
link=link,
|
||
server_type='plex'
|
||
))
|
||
offset += num
|
||
return ret_resume[:num]
|
||
|
||
def get_data(self, endpoint: str, **kwargs) -> Optional[Response]:
|
||
"""
|
||
自定义从媒体服务器获取数据
|
||
:param endpoint: 端点
|
||
:param kwargs: 其他请求参数，如headers, cookies, proxies等
|
||
"""
|
||
return self.__request(method="get", endpoint=endpoint, **kwargs)
|
||
|
||
def post_data(self, endpoint: str, **kwargs) -> Optional[Response]:
|
||
"""
|
||
自定义从媒体服务器获取数据
|
||
:param endpoint: 端点
|
||
:param kwargs: 其他请求参数，如headers, cookies, proxies等
|
||
"""
|
||
return self.__request(method="post", endpoint=endpoint, **kwargs)
|
||
|
||
def put_data(self, endpoint: str, **kwargs) -> Optional[Response]:
|
||
"""
|
||
自定义从媒体服务器获取数据
|
||
:param endpoint: 端点
|
||
:param kwargs: 其他请求参数，如headers, cookies, proxies等
|
||
"""
|
||
return self.__request(method="put", endpoint=endpoint, **kwargs)
|
||
|
||
def __request(self, method: str, endpoint: str, **kwargs) -> Optional[Response]:
|
||
"""
|
||
自定义从媒体服务器获取数据
|
||
:param method: HTTP方法，如 get, post, put 等
|
||
:param endpoint: 端点
|
||
:param kwargs: 其他请求参数，如headers, cookies, proxies等
|
||
"""
|
||
if not self._session:
|
||
return None
|
||
try:
|
||
url = UrlUtils.adapt_request_url(host=self._host, endpoint=endpoint)
|
||
kwargs.setdefault("headers", self.__get_request_headers())
|
||
kwargs.setdefault("raise_exception", True)
|
||
request_method = getattr(RequestUtils(session=self._session), f"{method}_res", None)
|
||
if request_method:
|
||
return request_method(url=url, **kwargs)
|
||
else:
|
||
logger.error(f"方法 {method} 不存在")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"连接Plex出错：" + str(e))
|
||
return None
|
||
|
||
def __get_request_headers(self) -> dict:
|
||
"""获取请求头"""
|
||
return {
|
||
"X-Plex-Token": self._token,
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
def __adapt_plex_session(self) -> Session:
|
||
"""
|
||
创建并配置一个针对Plex服务的requests.Session实例
|
||
这个会话包括特定的头部信息，用于处理所有的Plex请求
|
||
"""
|
||
# 设置请求头部，通常包括验证令牌和接受/内容类型头部
|
||
headers = self.__get_request_headers()
|
||
session = Session()
|
||
session.headers = headers
|
||
return session
|
||
|
||
def close(self):
|
||
if self._session:
|
||
self._session.close()
|