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

fix typing

Browse Source
This commit is contained in:
jxxghp 2023-06-14 08:26:35 +08:00
parent 1f832633ea
commit 9d2d3f58a2
14 changed files with 129 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
app/chain/download.py
View File

@ -29,8 +29,8 @@ class DownloadChain(ChainBase):
msg_text = ""
if torrent.site_name:
msg_text = f"站点：{torrent.site_name}"
if meta.resource:
msg_text = f"{msg_text}\n质量：{meta.resource}"
if meta.resource_term:
msg_text = f"{msg_text}\n质量：{meta.resource_term}"
if torrent.size:
if str(torrent.size).isdigit():
size = StringUtils.str_filesize(torrent.size)

4
app/chain/transfer.py
View File

@ -133,8 +133,8 @@ class TransferChain(ChainBase):
msg_str = f"类型：{mediainfo.type.value}"
if mediainfo.category:
msg_str = f"{msg_str}，类别：{mediainfo.category}"
if meta.resource:
msg_str = f"{msg_str}，质量：{meta.resource}"
if meta.resource_term:
msg_str = f"{msg_str}，质量：{meta.resource_term}"
msg_str = f"{msg_str}， 大小：{file_size}"
# 发送
self.post_message(title=msg_title, text=msg_str, image=mediainfo.get_message_image())

161
app/core/meta/metabase.py
View File

@ -182,45 +182,6 @@ class MetaBase(object):
self.type = MediaType.TV
self._subtitle_flag = True
def is_in_season(self, season: Union[list, int, str]) -> bool:
"""
是否包含季
"""
if isinstance(season, list):
if self.end_season is not None:
meta_season = list(range(self.begin_season, self.end_season + 1))
else:
if self.begin_season is not None:
meta_season = [self.begin_season]
else:
meta_season = [1]
return set(meta_season).issuperset(set(season))
else:
if self.end_season is not None:
return self.begin_season <= int(season) <= self.end_season
else:
if self.begin_season is not None:
return int(season) == self.begin_season
else:
return int(season) == 1
def is_in_episode(self, episode: Union[list, int, str]) -> bool:
"""
是否包含集
"""
if isinstance(episode, list):
if self.end_episode is not None:
meta_episode = list(range(self.begin_episode, self.end_episode + 1))
else:
meta_episode = [self.begin_episode]
return set(meta_episode).issuperset(set(episode))
else:
if self.end_episode is not None:
return self.begin_episode <= int(episode) <= self.end_episode
else:
return int(episode) == self.begin_episode
@property
def season(self) -> str:
"""
@ -248,19 +209,6 @@ class MetaBase(object):
else:
return ""
@property
def season_name(self) -> str:
"""
返回begin_season 的Sxx
"""
if self.begin_season is not None:
return "S%s" % str(self.begin_season).rjust(2, "0")
else:
if self.type == MediaType.TV:
return "S01"
else:
return ""
@property
def season_seq(self) -> str:
"""
@ -289,40 +237,6 @@ class MetaBase(object):
else:
return [self.begin_season]
def set_season(self, sea: Union[list, int, str]):
"""
更新季
"""
if not sea:
return
if isinstance(sea, list):
if len(sea) == 1 and str(sea[0]).isdigit():
self.begin_season = int(sea[0])
self.end_season = None
elif len(sea) > 1 and str(sea[0]).isdigit() and str(sea[-1]).isdigit():
self.begin_season = int(sea[0])
self.end_season = int(sea[-1])
elif str(sea).isdigit():
self.begin_season = int(sea)
self.end_season = None
def set_episode(self, ep: Union[list, int, str]):
"""
更新集
"""
if not ep:
return
if isinstance(ep, list):
if len(ep) == 1 and str(ep[0]).isdigit():
self.begin_episode = int(ep[0])
self.end_episode = None
elif len(ep) > 1 and str(ep[0]).isdigit() and str(ep[-1]).isdigit():
self.begin_episode = int(ep[0])
self.end_episode = int(ep[-1])
elif str(ep).isdigit():
self.begin_episode = int(ep)
self.end_episode = None
@ property
def episode(self) -> str:
"""
@ -402,7 +316,7 @@ class MetaBase(object):
return ""
@property
def resource(self) -> str:
def resource_term(self) -> str:
"""
返回资源类型字符串，含分辨率
"""
@ -450,3 +364,76 @@ class MetaBase(object):
返回音频编码
"""
return self.audio_encode or ""
def is_in_season(self, season: Union[list, int, str]) -> bool:
"""
是否包含季
"""
if isinstance(season, list):
if self.end_season is not None:
meta_season = list(range(self.begin_season, self.end_season + 1))
else:
if self.begin_season is not None:
meta_season = [self.begin_season]
else:
meta_season = [1]
return set(meta_season).issuperset(set(season))
else:
if self.end_season is not None:
return self.begin_season <= int(season) <= self.end_season
else:
if self.begin_season is not None:
return int(season) == self.begin_season
else:
return int(season) == 1
def is_in_episode(self, episode: Union[list, int, str]) -> bool:
"""
是否包含集
"""
if isinstance(episode, list):
if self.end_episode is not None:
meta_episode = list(range(self.begin_episode, self.end_episode + 1))
else:
meta_episode = [self.begin_episode]
return set(meta_episode).issuperset(set(episode))
else:
if self.end_episode is not None:
return self.begin_episode <= int(episode) <= self.end_episode
else:
return int(episode) == self.begin_episode
def set_season(self, sea: Union[list, int, str]):
"""
更新季
"""
if not sea:
return
if isinstance(sea, list):
if len(sea) == 1 and str(sea[0]).isdigit():
self.begin_season = int(sea[0])
self.end_season = None
elif len(sea) > 1 and str(sea[0]).isdigit() and str(sea[-1]).isdigit():
self.begin_season = int(sea[0])
self.end_season = int(sea[-1])
elif str(sea).isdigit():
self.begin_season = int(sea)
self.end_season = None
def set_episode(self, ep: Union[list, int, str]):
"""
更新集
"""
if not ep:
return
if isinstance(ep, list):
if len(ep) == 1 and str(ep[0]).isdigit():
self.begin_episode = int(ep[0])
self.end_episode = None
elif len(ep) > 1 and str(ep[0]).isdigit() and str(ep[-1]).isdigit():
self.begin_episode = int(ep[0])
self.end_episode = int(ep[-1])
elif str(ep).isdigit():
self.begin_episode = int(ep)
self.end_episode = None

16
app/modules/emby/__init__.py
View File

@ -5,7 +5,7 @@ from app.core.context import MediaInfo
from app.log import logger
from app.modules import _ModuleBase
from app.modules.emby.emby import Emby
from app.schemas.context import ExistMediaInfo
from app.schemas.context import ExistMediaInfo, RefreshMediaItem
from app.utils.types import MediaType
@ -65,12 +65,12 @@ class EmbyModule(_ModuleBase):
:return: 成功或失败
"""
items = [
{
"title": mediainfo.title,
"year": mediainfo.year,
"type": mediainfo.type,
"category": mediainfo.category,
"target_path": file_path
}
RefreshMediaItem(
title=mediainfo.title,
year=mediainfo.year,
type=mediainfo.type,
category=mediainfo.category,
target_path=file_path
)
]
return self.emby.refresh_library_by_items(items)

19
app/modules/emby/emby.py
View File

@ -5,6 +5,7 @@ from typing import List, Optional, Union, Dict
from app.core.config import settings
from app.log import logger
from app.schemas.context import RefreshMediaItem
from app.utils.http import RequestUtils
from app.utils.singleton import Singleton
from app.utils.string import StringUtils
@ -346,7 +347,7 @@ class Emby(metaclass=Singleton):
return False
return False
def refresh_library_by_items(self, items: List[dict]) -> bool:
def refresh_library_by_items(self, items: List[RefreshMediaItem]) -> bool:
"""
按类型、名称、年份来刷新媒体库
:param items: 已识别的需要刷新媒体库的媒体信息列表
@ -357,8 +358,6 @@ class Emby(metaclass=Singleton):
logger.info(f"开始刷新Emby媒体库...")
library_ids = []
for item in items:
if not item:
continue
library_id = self.__get_emby_library_id_by_item(item)
if library_id and library_id not in library_ids:
library_ids.append(library_id)
@ -370,24 +369,24 @@ class Emby(metaclass=Singleton):
return self.__refresh_emby_library_by_id(library_id)
logger.info(f"Emby媒体库刷新完成")
def __get_emby_library_id_by_item(self, item: dict) -> Optional[str]:
def __get_emby_library_id_by_item(self, item: RefreshMediaItem) -> Optional[str]:
"""
根据媒体信息查询在哪个媒体库，返回要刷新的位置的ID
:param item: {title, year, type, category, target_path}
"""
if not item.get("title") or not item.get("year") or not item.get("type"):
if not item.title or not item.year or not item.type:
return None
if item.get("type") != MediaType.MOVIE.value:
item_id = self.__get_emby_series_id_by_name(item.get("title"), item.get("year"))
if item.type != MediaType.MOVIE.value:
item_id = self.__get_emby_series_id_by_name(item.title, item.year)
if item_id:
# 存在电视剧，则直接刷新这个电视剧就行
return item_id
else:
if self.get_movies(item.get("title"), item.get("year")):
if self.get_movies(item.title, item.year):
# 已存在，不用刷新
return None
# 查找需要刷新的媒体库ID
item_path = Path(item.get("target_path"))
item_path = Path(item.target_path)
for folder in self._folders:
# 找同级路径最多的媒体库（要求容器内映射路径与实际一致）
max_comm_path = ""
@ -416,7 +415,7 @@ class Emby(metaclass=Singleton):
return match_id if match_num == 1 else folder.get("Id")
# 如果找不到，只要路径中有分类目录名就命中
for subfolder in folder.get("SubFolders"):
if subfolder.get("Path") and re.search(r"[/\\]%s" % item.get("category"),
if subfolder.get("Path") and re.search(r"[/\\]%s" % item.category,
subfolder.get("Path")):
return folder.get("Id")
# 刷新根目录

2
app/modules/filetransfer/__init__.py
View File

@ -468,7 +468,7 @@ class FileTransferModule(_ModuleBase):
# 集号
"episode": meta.episode_seqs,
# 季集 SxxExx
"season_episode": "%s%s" % (meta.season_name, meta.episode),
"season_episode": "%s%s" % (meta.season, meta.episode),
# 段/节
"part": meta.part,
# 文件后缀

11
app/modules/jellyfin/__init__.py
View File

@ -64,13 +64,4 @@ class JellyfinModule(_ModuleBase):
:param file_path: 文件路径
:return: 成功或失败
"""
items = [
{
"title": mediainfo.title,
"year": mediainfo.year,
"type": mediainfo.type,
"category": mediainfo.category,
"target_path": file_path
}
]
return self.jellyfin.refresh_library_by_items(items)
return self.jellyfin.refresh_root_library()

12
app/modules/jellyfin/jellyfin.py
View File

@ -300,18 +300,6 @@ class Jellyfin(metaclass=Singleton):
logger.error(f"连接Library/Refresh出错：" + str(e))
return False
def refresh_library_by_items(self, items: List[dict]) -> bool:
"""
按类型、名称、年份来刷新媒体库，Jellyfin没有刷单个项目的API，这里直接刷新整个库
:param items: 已识别的需要刷新媒体库的媒体信息列表
"""
# 没找到单项目刷新的对应的API，先按全库刷新
if not items:
return False
if not self._host or not self._apikey:
return False
return self.refresh_root_library()
def get_iteminfo(self, itemid: str) -> dict:
"""
获取单个项目详情

16
app/modules/plex/__init__.py
View File

@ -5,7 +5,7 @@ from app.core.context import MediaInfo
from app.log import logger
from app.modules import _ModuleBase
from app.modules.plex.plex import Plex
from app.schemas.context import ExistMediaInfo
from app.schemas.context import ExistMediaInfo, RefreshMediaItem
from app.utils.types import MediaType
@ -64,12 +64,12 @@ class PlexModule(_ModuleBase):
:return: 成功或失败
"""
items = [
{
"title": mediainfo.title,
"year": mediainfo.year,
"type": mediainfo.type,
"category": mediainfo.category,
"target_path": file_path
}
RefreshMediaItem(
title=mediainfo.title,
year=mediainfo.year,
type=mediainfo.type,
category=mediainfo.category,
target_path=file_path
)
]
return self.plex.refresh_library_by_items(items)

15
app/modules/plex/plex.py
View File

@ -8,6 +8,7 @@ from plexapi.server import PlexServer
from app.core.config import settings
from app.log import logger
from app.schemas.context import RefreshMediaItem
from app.utils.singleton import Singleton
@ -164,7 +165,7 @@ class Plex(metaclass=Singleton):
return False
return self._plex.library.update()
def refresh_library_by_items(self, items: List[dict]) -> bool:
def refresh_library_by_items(self, items: List[RefreshMediaItem]) -> bool:
"""
按路径刷新媒体库 item: target_path
"""
@ -172,7 +173,7 @@ class Plex(metaclass=Singleton):
return False
result_dict = {}
for item in items:
file_path = item.get("target_path")
file_path = item.target_path
lib_key, path = self.__find_librarie(file_path, self._libraries)
# 如果存在同一剧集的多集,key(path)相同会合并
result_dict[path] = lib_key
@ -186,18 +187,18 @@ class Plex(metaclass=Singleton):
self._plex.query(f'/library/sections/{lib_key}/refresh?path={quote_plus(path)}')
@staticmethod
def __find_librarie(path: str, libraries: List[dict]) -> Tuple[str, str]:
def __find_librarie(path: Path, libraries: List[dict]) -> Tuple[str, str]:
"""
判断这个path属于哪个媒体库
多个媒体库配置的目录不应有重复和嵌套,
"""
def is_subpath(_path: str, _parent: str) -> bool:
def is_subpath(_path: Path, _parent: Path) -> bool:
"""
判断_path是否是_parent的子目录下
"""
_path = Path(_path).resolve()
_parent = Path(_parent).resolve()
_path = _path.resolve()
_parent = _parent.resolve()
return _path.parts[:len(_parent.parts)] == _parent.parts
if path is None:
@ -207,7 +208,7 @@ class Plex(metaclass=Singleton):
for lib in libraries:
if hasattr(lib, "locations") and lib.locations:
for location in lib.locations:
if is_subpath(path, location):
if is_subpath(path, Path(location)):
return lib.key, location
except Exception as err:
logger.error(f"查找媒体库出错：{err}")

2
app/modules/slack/slack.py
View File

@ -262,7 +262,7 @@ class Slack:
meta = MetaInfo(torrent.title, torrent.description)
link = torrent.page_url
title = f"{meta.season_episode} " \
f"{meta.resource} " \
f"{meta.resource_term} " \
f"{meta.release_team}"
title = re.sub(r"\s+", " ", title).strip()
free = torrent.get_volume_factor_string()

2
app/modules/telegram/telegram.py
View File

@ -149,7 +149,7 @@ class Telegram(metaclass=Singleton):
meta = MetaInfo(torrent.title, torrent.description)
link = torrent.page_url
title = f"{meta.season_episode} " \
f"{meta.resource} " \
f"{meta.resource_term} " \
f"{meta.release_team}"
title = re.sub(r"\s+", " ", title).strip()
free = torrent.get_volume_factor_string()

2
app/modules/wechat/wechat.py
View File

@ -215,7 +215,7 @@ class WeChat(metaclass=Singleton):
meta = MetaInfo(title=torrent.title, subtitle=torrent.description)
torrent_title = f"{index}.【{torrent.site_name}】" \
f"{meta.season_episode} " \
f"{meta.resource} " \
f"{meta.resource_term} " \
f"{meta.release_team} " \
f"{StringUtils.str_filesize(torrent.size)} " \
f"{torrent.get_volume_factor_string()} " \

13
app/schemas/context.py
View File

@ -121,3 +121,16 @@ class NotExistMediaInfo(BaseModel):
total_episodes: int = 0
# 开始集
start_episode: int = 0
class RefreshMediaItem(BaseModel):
# 标题
title: str
# 年份
year: Optional[str] = None
# 类型
type: Optional[MediaType] = None
# 类别
category: Optional[str] = None
# 目录
target_path: Optional[Path] = None