GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-07-08 00:01:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix storage api

Browse Source
This commit is contained in:
jxxghp
2024-06-30 19:41:32 +08:00
parent a7b906ada6
commit 02ad98c024
12 changed files with 319 additions and 553 deletions
Download Patch File Download Diff File Expand all files Collapse all files

953
app/modules/filemanager/__init__.py Normal file
View File

@@ -0,0 +1,953 @@
import re
from pathlib import Path
from threading import Lock
from typing import Optional, List, Tuple, Union, Dict
from jinja2 import Template
from app.core.config import settings
from app.core.context import MediaInfo
from app.core.meta import MetaBase
from app.core.metainfo import MetaInfo, MetaInfoPath
from app.helper.directory import DirectoryHelper
from app.helper.message import MessageHelper
from app.helper.module import ModuleHelper
from app.log import logger
from app.modules import _ModuleBase
from app.schemas import TransferInfo, ExistMediaInfo, TmdbEpisode, MediaDirectory, FileItem
from app.schemas.types import MediaType
from app.utils.system import SystemUtils
lock = Lock()
class FileTransferModule(_ModuleBase):
"""
文件整理模块
"""
_storage_schemas = []
def __init__(self):
super().__init__()
self.directoryhelper = DirectoryHelper()
self.messagehelper = MessageHelper()
def init_module(self) -> None:
# 加载模块
self._storage_schemas = ModuleHelper.load('app.modules.filetransfer.storage',
filter_func=lambda _, obj: hasattr(obj, 'schema'))
@staticmethod
def get_name() -> str:
return "文件整理"
def stop(self):
pass
def test(self) -> Tuple[bool, str]:
"""
测试模块连接性
"""
directoryhelper = DirectoryHelper()
# 检查下载目录
download_paths = directoryhelper.get_download_dirs()
if not download_paths:
return False, "下载目录未设置"
for d_path in download_paths:
path = d_path.path
if not path:
return False, f"下载目录 {d_path.name} 对应路径未设置"
download_path = Path(path)
if not download_path.exists():
return False, f"下载目录 {d_path.name} 对应路径 {path} 不存在"
# 检查媒体库目录
libaray_paths = directoryhelper.get_library_dirs()
if not libaray_paths:
return False, "媒体库目录未设置"
for l_path in libaray_paths:
path = l_path.path
if not path:
return False, f"媒体库目录 {l_path.name} 对应路径未设置"
library_path = Path(path)
if not library_path.exists():
return False, f"媒体库目录{l_path.name} 对应的路径 {path} 不存在"
# 检查硬链接条件
if settings.DOWNLOADER_MONITOR and settings.TRANSFER_TYPE == "link":
for d_path in download_paths:
link_ok = False
for l_path in libaray_paths:
if SystemUtils.is_same_disk(Path(d_path.path), Path(l_path.path)):
link_ok = True
break
if not link_ok:
return False, f"媒体库目录中未找到" \
f"与下载目录 {d_path.path} 在同一磁盘/存储空间/映射路径的目录，将无法硬链接"
return True, ""
def init_setting(self) -> Tuple[str, Union[str, bool]]:
pass
def recommend_name(self, meta: MetaBase, mediainfo: MediaInfo) -> Optional[str]:
"""
获取重命名后的名称
:param meta: 元数据
:param mediainfo: 媒体信息
:return: 重命名后的名称（含目录）
"""
# 重命名格式
rename_format = settings.TV_RENAME_FORMAT \
if mediainfo.type == MediaType.TV else settings.MOVIE_RENAME_FORMAT
# 获取重命名后的名称
path = self.get_rename_path(
template_string=rename_format,
rename_dict=self.__get_naming_dict(meta=meta,
mediainfo=mediainfo,
file_ext=Path(meta.title).suffix)
)
return str(path)
def list_files(self, fileitem: FileItem) -> Optional[List[FileItem]]:
"""
浏览文件
:param fileitem: 源文件
:return: 文件列表
"""
storage_oper = self.__get_storage_oper(fileitem.storage)
if not storage_oper:
logger.error(f"不支持 {fileitem.storage} 的文件浏览")
return None
return storage_oper.list(fileitem)
def create_folder(self, fileitem: FileItem, name: str) -> Optional[FileItem]:
"""
创建目录
:param fileitem: 源文件
:param name: 目录名
:return: 创建的目录
"""
storage_oper = self.__get_storage_oper(fileitem.storage)
if not storage_oper:
logger.error(f"不支持 {fileitem.storage} 的目录创建")
return None
return storage_oper.create_folder(fileitem, name)
def delete_file(self, fileitem: FileItem) -> bool:
"""
删除文件或目录
"""
storage_oper = self.__get_storage_oper(fileitem.storage)
if not storage_oper:
logger.error(f"不支持 {fileitem.storage} 的删除处理")
return False
return storage_oper.delete(fileitem)
def rename_file(self, fileitem: FileItem, name: str) -> bool:
"""
重命名文件或目录
"""
storage_oper = self.__get_storage_oper(fileitem.storage)
if not storage_oper:
logger.error(f"不支持 {fileitem.storage} 的重命名处理")
return False
return storage_oper.rename(fileitem, name)
def download_file(self, fileitem: FileItem, path: Path) -> bool:
"""
下载文件
"""
storage_oper = self.__get_storage_oper(fileitem.storage)
if not storage_oper:
logger.error(f"不支持 {fileitem.storage} 的下载处理")
return False
return storage_oper.download(fileitem, path)
def upload_file(self, fileitem: FileItem, path: Path) -> bool:
"""
上传文件
"""
storage_oper = self.__get_storage_oper(fileitem.storage)
if not storage_oper:
logger.error(f"不支持 {fileitem.storage} 的上传处理")
return False
return storage_oper.upload(fileitem, path)
def transfer(self, fileitem: FileItem, meta: MetaBase, mediainfo: MediaInfo,
transfer_type: str, target_storage: str = None, target_path: Path = None,
episodes_info: List[TmdbEpisode] = None,
scrape: bool = None) -> TransferInfo:
"""
文件整理
:param fileitem: 源文件
:param meta: 预识别的元数据，仅单文件整理时传递
:param mediainfo: 识别的媒体信息
:param transfer_type: 整理方式
:param target_storage: 目标存储
:param target_path: 目标路径
:param episodes_info: 当前季的全部集信息
:param scrape: 是否刮削元数据
:return: {path, target_path, message}
"""
# 目标路径不能是文件
if target_path and target_path.is_file():
logger.error(f"整理目标路径 {target_path} 是一个文件")
return TransferInfo(success=False,
path=fileitem.path,
message=f"{target_path} 不是有效目录")
# 获取目标路径
directoryhelper = DirectoryHelper()
if target_path:
dir_info = directoryhelper.get_library_dir(mediainfo, in_path=fileitem.path, to_path=target_path)
else:
dir_info = directoryhelper.get_library_dir(mediainfo, in_path=fileitem.path)
if dir_info:
# 是否需要刮削
if scrape is None:
need_scrape = dir_info.scrape
else:
need_scrape = scrape
# 拼装媒体库一、二级子目录
target_path = self.__get_dest_dir(mediainfo=mediainfo, target_dir=dir_info)
elif target_path:
# 自定义目标路径
need_scrape = scrape or False
else:
# 未找到有效的媒体库目录
logger.error(
f"{mediainfo.type.value} {mediainfo.title_year} 未找到有效的媒体库目录，无法整理文件，源路径：{fileitem.path}")
return TransferInfo(success=False,
path=fileitem.path,
message="未找到有效的媒体库目录")
logger.info(f"获取整理目标路径：{target_path}")
# 整理
return self.transfer_media(fileitem=fileitem,
in_meta=meta,
mediainfo=mediainfo,
transfer_type=transfer_type,
target_storage=target_storage,
target_path=target_path,
episodes_info=episodes_info,
need_scrape=need_scrape)
def __get_storage_oper(self, storage: str):
"""
获取存储操作对象
"""
for storage_schema in self._storage_schemas:
if storage_schema.schema == storage:
return storage_schema()
return None
def __list_files(self, fileitem: FileItem):
"""
浏览文件
"""
pass
def __transfer_command(self, fileitem: FileItem, target_storage: str,
target_file: Path, transfer_type: str) -> bool:
"""
使用系统命令处理单个文件
:param fileitem: 源文件
:param target_storage: 目标存储
:param target_file: 目标文件路径
:param transfer_type: 整理方式
"""
if fileitem.storage != "local" and target_storage != "local":
logger.error(f"不支持 {fileitem.storage} 到 {target_storage} 的文件整理")
return False
# 源操作对象
source_oper = self.__get_storage_oper(fileitem.storage)
# 目的操作对象
target_oper = self.__get_storage_oper(target_storage)
with lock:
if fileitem.storage == "local" and target_storage == "local":
# 本地到本地
if transfer_type == "copy":
return source_oper.copy(fileitem, target_file)
elif transfer_type == "move":
return source_oper.move(fileitem, target_file)
elif transfer_type == "link":
return source_oper.link(fileitem, target_file)
elif transfer_type == "softlink":
return source_oper.softlink(fileitem, target_file)
elif fileitem.storage == "local" and target_storage != "local":
# 本地到网盘
if transfer_type == "copy":
# 复制
filepath = Path(fileitem.path)
if not filepath.exists():
logger.error(f"文件 {filepath} 不存在")
return False
# TODO 根据目的路径创建文件夹
target_fileitem = target_oper.get_folder(target_file.parent)
if target_fileitem:
# 上传文件
return target_oper.upload(target_fileitem, filepath)
elif transfer_type == "move":
# 移动
filepath = Path(fileitem.path)
if not filepath.exists():
logger.error(f"文件 {filepath} 不存在")
return False
# TODO 根据目的路径获取文件夹
target_fileitem = target_oper.get_folder(target_file.parent)
if target_fileitem:
# 上传文件
result = target_oper.upload(target_fileitem, filepath)
if result:
# 删除源文件
return source_oper.delete(fileitem)
elif fileitem.storage != "local" and target_storage == "local":
# 网盘到本地
if transfer_type == "copy":
# 下载
return target_oper.download(fileitem, target_file)
elif transfer_type == "move":
# 下载
if target_oper.download(fileitem, target_file):
# 删除源文件
return source_oper.delete(fileitem)
return False
def __transfer_other_files(self, fileitem: FileItem, target_storage: str, target_file: Path,
transfer_type: str) -> int:
"""
根据文件名整理其他相关文件
:param fileitem: 源文件
:param target_storage: 目标存储
:param target_file: 目标路径
:param transfer_type: 整理方式
"""
retcode = self.__transfer_subtitles(fileitem=fileitem,
target_storage=target_storage,
target_file=target_file,
transfer_type=transfer_type)
if retcode != 0:
return retcode
retcode = self.__transfer_audio_track_files(fileitem=fileitem,
target_storage=target_storage,
target_file=target_file,
transfer_type=transfer_type)
if retcode != 0:
return retcode
return 0
def __transfer_subtitles(self, fileitem: FileItem, target_storage: str, target_file: Path,
transfer_type: str) -> int:
"""
根据文件名整理对应字幕文件
:param fileitem: 源文件
:param target_storage: 目标存储
:param target_file: 目标路径
:param transfer_type: 整理方式
"""
# 字幕正则式
_zhcn_sub_re = r"([.\[(](((zh[-_])?(cn|ch[si]|sg|sc))|zho?" \
r"|chinese|(cn|ch[si]|sg|zho?|eng)[-_&](cn|ch[si]|sg|zho?|eng)" \
r"|简[体中]?)[.\])])" \
r"|([\u4e00-\u9fa5]{0,3}[中双][\u4e00-\u9fa5]{0,2}[字文语][\u4e00-\u9fa5]{0,3})" \
r"|简体|简中|JPSC" \
r"|(?<![a-z0-9])gb(?![a-z0-9])"
_zhtw_sub_re = r"([.\[(](((zh[-_])?(hk|tw|cht|tc))" \
r"|繁[体中]?)[.\])])" \
r"|繁体中[文字]|中[文字]繁体|繁体|JPTC" \
r"|(?<![a-z0-9])big5(?![a-z0-9])"
_eng_sub_re = r"[.\[(]eng[.\])]"
# 比对文件名并整理字幕
org_path = Path(fileitem.path)
org_dir: Path = org_path.parent
# 列出所有字幕文件
storage_oper = self.__get_storage_oper(fileitem.storage)
if not storage_oper:
logger.error(f"不支持 {fileitem.storage} 的文件整理")
return 1
file_list: List[FileItem] = storage_oper.list(fileitem)
if len(file_list) == 0:
logger.debug(f"{org_dir} 目录下没有找到字幕文件...")
else:
logger.debug("字幕文件清单：" + str(file_list))
# 识别文件名
metainfo = MetaInfoPath(org_path)
for sub_item in file_list:
if f".{sub_item.extension.lower()}" not in settings.RMT_SUBEXT:
continue
# 识别字幕文件名
sub_file_name = re.sub(_zhtw_sub_re,
".",
re.sub(_zhcn_sub_re,
".",
sub_item.name,
flags=re.I),
flags=re.I)
sub_file_name = re.sub(_eng_sub_re, ".", sub_file_name, flags=re.I)
sub_metainfo = MetaInfoPath(Path(sub_item.path))
# 匹配字幕文件名
if (org_path.stem == Path(sub_file_name).stem) or \
(sub_metainfo.cn_name and sub_metainfo.cn_name == metainfo.cn_name) \
or (sub_metainfo.en_name and sub_metainfo.en_name == metainfo.en_name):
if metainfo.part and metainfo.part != sub_metainfo.part:
continue
if metainfo.season \
and metainfo.season != sub_metainfo.season:
continue
if metainfo.episode \
and metainfo.episode != sub_metainfo.episode:
continue
new_file_type = ""
# 兼容jellyfin字幕识别(多重识别), emby则会识别最后一个后缀
if re.search(_zhcn_sub_re, sub_item.name, re.I):
new_file_type = ".chi.zh-cn"
elif re.search(_zhtw_sub_re, sub_item.name,
re.I):
new_file_type = ".zh-tw"
elif re.search(_eng_sub_re, sub_item.name, re.I):
new_file_type = ".eng"
# 通过对比字幕文件大小 尽量整理所有存在的字幕
file_ext = f".{sub_item.extension}"
new_sub_tag_dict = {
".eng": ".英文",
".chi.zh-cn": ".简体中文",
".zh-tw": ".繁体中文"
}
new_sub_tag_list = [
new_file_type if t == 0 else "%s%s(%s)" % (new_file_type,
new_sub_tag_dict.get(
new_file_type, ""
),
t) for t in range(6)
]
for new_sub_tag in new_sub_tag_list:
new_file: Path = target_file.with_name(target_file.stem + new_sub_tag + file_ext)
# 如果字幕文件不存在, 直接整理字幕, 并跳出循环
try:
logger.debug(f"正在处理字幕：{sub_item.name}")
retcode = self.__transfer_command(fileitem=sub_item,
target_storage=target_storage,
target_file=new_file,
transfer_type=transfer_type)
if retcode == 0:
logger.info(f"字幕 {sub_item.name} {transfer_type}完成")
break
else:
logger.error(f"字幕 {sub_item.name} {transfer_type}失败，错误码 {retcode}")
return retcode
except Exception as error:
logger.info(f"字幕 {new_file} 出错了,原因: {str(error)}")
return 0
def __transfer_audio_track_files(self, fileitem: FileItem, target_storage: str, target_file: Path,
transfer_type: str) -> int:
"""
根据文件名整理对应音轨文件
:param fileitem: 源文件
:param target_storage: 目标存储
:param target_file: 目标路径
:param transfer_type: 整理方式
"""
org_path = Path(fileitem.path)
dir_name = org_path.parent
file_name = org_path.name
# 列出所有音轨文件
storage_oper = self.__get_storage_oper(fileitem.storage)
if not storage_oper:
logger.error(f"不支持 {fileitem.storage} 的文件整理")
return 1
file_list: List[FileItem] = storage_oper.list(fileitem)
# 匹配音轨文件
pending_file_list: List[FileItem] = [file for file in file_list if Path(file.name).stem == org_path.name
and f".{file.extension.lower()}" in settings.RMT_AUDIOEXT]
if len(pending_file_list) == 0:
logger.debug(f"{dir_name} 目录下没有找到匹配的音轨文件")
return 0
logger.debug("音轨文件清单：" + str(pending_file_list))
for track_file in pending_file_list:
track_ext = f".{track_file.extension}"
new_track_file = target_file.with_name(target_file.stem + track_ext)
try:
logger.info(f"正在整理音轨文件：{track_file} 到 {new_track_file}")
retcode = self.__transfer_command(fileitem=track_file,
target_storage=target_storage,
target_file=new_track_file,
transfer_type=transfer_type)
if retcode == 0:
logger.info(f"音轨文件 {file_name} {transfer_type}完成")
else:
logger.error(f"音轨文件 {file_name} {transfer_type}失败，错误码：{retcode}")
except Exception as error:
logger.error(f"音轨文件 {file_name} {transfer_type}失败：{str(error)}")
return 0
def __transfer_dir(self, fileitem: FileItem, transfer_type: str,
target_storage: str, target_path: Path) -> int:
"""
整理整个文件夹
:param fileitem: 源文件
:param transfer_type: 整理方式
:param target_storage: 目标存储
:param target_path: 目标路径
"""
logger.info(f"正在{transfer_type}目录：{fileitem.path} 到 {target_path}")
# 复制
retcode = self.__transfer_dir_files(fileitem=fileitem,
target_storage=target_storage,
target_path=target_path,
transfer_type=transfer_type)
if retcode == 0:
logger.info(f"文件 {fileitem.path} {transfer_type}完成")
else:
logger.error(f"文件{fileitem.path} {transfer_type}失败，错误码：{retcode}")
return retcode
def __transfer_dir_files(self, fileitem: FileItem, transfer_type: str,
target_storage: str, target_path: Path) -> int:
"""
按目录结构整理目录下所有文件
:param fileitem: 源文件
:param target_storage: 目标存储
:param target_path: 目标路径
:param transfer_type: 整理方式
"""
retcode = 0
# 列出所有文件
storage_oper = self.__get_storage_oper(fileitem.storage)
if not storage_oper:
logger.error(f"不支持 {fileitem.storage} 的文件整理")
return 1
file_list: List[FileItem] = storage_oper.list(fileitem)
# 整理文件
for item in file_list:
if item.type == "dir":
# 递归整理目录
new_path = target_path / item.name
retcode = self.__transfer_dir(fileitem=item,
transfer_type=transfer_type,
target_storage=target_storage,
target_path=new_path)
if retcode != 0:
return retcode
else:
# 整理文件
new_file = target_path / item.name
retcode = self.__transfer_command(fileitem=item,
target_storage=target_storage,
target_file=new_file,
transfer_type=transfer_type)
if retcode != 0:
return retcode
return retcode
def __transfer_file(self, fileitem: FileItem, target_storage: str, target_file: Path,
transfer_type: str, over_flag: bool = False) -> int:
"""
整理一个文件，同时处理其他相关文件
:param fileitem: 原文件
:param target_storage: 目标存储
:param target_file: 新文件
:param transfer_type: 整理方式
:param over_flag: 是否覆盖，为True时会先删除再整理
"""
if target_storage == "local" and (target_file.exists() or target_file.is_symlink()):
if not over_flag:
logger.warn(f"文件已存在：{target_file}")
return 0
else:
logger.info(f"正在删除已存在的文件：{target_file}")
target_file.unlink()
logger.info(f"正在整理文件：{fileitem.path} 到 {target_file}")
retcode = self.__transfer_command(fileitem=fileitem,
target_storage=target_storage,
target_file=target_file,
transfer_type=transfer_type)
if retcode == 0:
logger.info(f"文件 {fileitem.path} {transfer_type}完成")
else:
logger.error(f"文件 {fileitem.path} {transfer_type}失败，错误码：{retcode}")
return retcode
# 处理其他相关文件
return self.__transfer_other_files(fileitem=fileitem,
target_storage=target_storage,
target_file=target_file,
transfer_type=transfer_type)
@staticmethod
def __get_dest_dir(mediainfo: MediaInfo, target_dir: MediaDirectory) -> Path:
"""
根据设置并装媒体库目录
:param mediainfo: 媒体信息
:target_dir: 媒体库根目录
:typename_dir: 是否加上类型目录
"""
if not target_dir.media_type and target_dir.auto_category:
# 一级自动分类
download_dir = Path(target_dir.path) / mediainfo.type.value
else:
download_dir = Path(target_dir.path)
if not target_dir.category and target_dir.auto_category and mediainfo.category:
# 二级自动分类
download_dir = download_dir / mediainfo.category
return download_dir
def transfer_media(self,
fileitem: FileItem,
in_meta: MetaBase,
mediainfo: MediaInfo,
transfer_type: str,
target_storage: str,
target_path: Path,
episodes_info: List[TmdbEpisode] = None,
need_scrape: bool = False
) -> TransferInfo:
"""
识别并整理一个文件或者一个目录下的所有文件
:param fileitem: 整理的文件对象，可能是一个文件也可以是一个目录
:param in_meta：预识别元数据
:param mediainfo: 媒体信息
:param target_storage: 目标存储
:param target_path: 目标路径
:param transfer_type: 文件整理方式
:param episodes_info: 当前季的全部集信息
:param need_scrape: 是否需要刮削
:return: TransferInfo、错误信息
"""
# 检查目录路径
if fileitem.storage == "local" and not Path(fileitem.path).exists():
return TransferInfo(success=False,
path=fileitem.path,
message=f"{fileitem.path} 不存在")
if target_storage == "local":
# 检查目标路径
if not target_path.exists():
logger.info(f"目标路径不存在，正在创建：{target_path} ...")
target_path.mkdir(parents=True, exist_ok=True)
# 重命名格式
rename_format = settings.TV_RENAME_FORMAT \
if mediainfo.type == MediaType.TV else settings.MOVIE_RENAME_FORMAT
# 判断是否为文件夹
if fileitem.type == "dir":
# 整理整个目录，一般为蓝光原盘
new_path = self.get_rename_path(
path=target_path,
template_string=rename_format,
rename_dict=self.__get_naming_dict(meta=in_meta,
mediainfo=mediainfo)
).parent
# 整理目录
retcode = self.__transfer_dir(fileitem=fileitem,
target_storage=target_storage,
target_path=new_path,
transfer_type=transfer_type)
if retcode != 0:
logger.error(f"文件夹 {fileitem.path} 整理失败，错误码：{retcode}")
return TransferInfo(success=False,
message=f"错误码：{retcode}",
path=fileitem.path,
target_path=new_path)
logger.info(f"文件夹 {fileitem.path} 整理成功")
# 返回整理后的路径
return TransferInfo(success=True,
path=fileitem.path,
target_path=new_path,
total_size=fileitem.size,
need_scrape=need_scrape)
else:
# 整理单个文件
if mediainfo.type == MediaType.TV:
# 电视剧
if in_meta.begin_episode is None:
logger.warn(f"文件 {fileitem.path} 整理失败：未识别到文件集数")
return TransferInfo(success=False,
message=f"未识别到文件集数",
path=fileitem.path,
fail_list=[fileitem.path])
# 文件结束季为空
in_meta.end_season = None
# 文件总季数为1
if in_meta.total_season:
in_meta.total_season = 1
# 文件不可能超过2集
if in_meta.total_episode > 2:
in_meta.total_episode = 1
in_meta.end_episode = None
# 目的文件名
new_file = self.get_rename_path(
path=target_path,
template_string=rename_format,
rename_dict=self.__get_naming_dict(
meta=in_meta,
mediainfo=mediainfo,
episodes_info=episodes_info,
file_ext=f".{fileitem.extension}"
)
)
# 判断是否要覆盖
overflag = False
target_file = new_file
if target_storage == "local" \
and (new_file.exists() or new_file.is_symlink()):
if new_file.is_symlink():
target_file = new_file.readlink()
if not target_file.exists():
overflag = True
if not overflag:
# 目标文件已存在
logger.info(f"目标文件已存在，整理覆盖模式：{settings.OVERWRITE_MODE}")
match settings.OVERWRITE_MODE:
case 'always':
# 总是覆盖同名文件
overflag = True
case 'size':
# 存在时大覆盖小
if target_file.stat().st_size < fileitem.size:
logger.info(f"目标文件文件大小更小，将覆盖：{new_file}")
overflag = True
else:
return TransferInfo(success=False,
message=f"媒体库中已存在，且质量更好",
path=fileitem.path,
target_path=new_file,
fail_list=[fileitem.path])
case 'never':
# 存在不覆盖
return TransferInfo(success=False,
message=f"媒体库中已存在，当前设置为不覆盖",
path=fileitem.path,
target_path=new_file,
fail_list=[fileitem.path])
case 'latest':
# 仅保留最新版本
logger.info(f"仅保留最新版本，将覆盖：{new_file}")
overflag = True
else:
if settings.OVERWRITE_MODE == 'latest':
# 文件不存在，但仅保留最新版本
logger.info(f"整理覆盖模式：{settings.OVERWRITE_MODE}，仅保留最新版本")
self.delete_all_version_files(new_file)
# 整理文件
retcode = self.__transfer_file(fileitem=fileitem,
target_storage=target_storage,
target_file=new_file,
transfer_type=transfer_type,
over_flag=overflag)
if retcode != 0:
logger.error(f"文件 {fileitem.path} 整理失败，错误码：{retcode}")
return TransferInfo(success=False,
message=f"错误码：{retcode}",
path=fileitem.path,
target_path=new_file,
fail_list=[fileitem.path])
logger.info(f"文件 {fileitem.path} 整理成功")
return TransferInfo(success=True,
path=fileitem.path,
target_path=new_file,
file_count=1,
total_size=fileitem.size,
is_bluray=False,
file_list=[fileitem.path],
file_list_new=[str(new_file)],
need_scrape=need_scrape)
@staticmethod
def __get_naming_dict(meta: MetaBase, mediainfo: MediaInfo, file_ext: str = None,
episodes_info: List[TmdbEpisode] = None) -> dict:
"""
根据媒体信息，返回Format字典
:param meta: 文件元数据
:param mediainfo: 识别的媒体信息
:param file_ext: 文件扩展名
:param episodes_info: 当前季的全部集信息
"""
def __convert_invalid_characters(filename: str):
if not filename:
return filename
invalid_characters = r'\/:*?"<>|'
# 创建半角到全角字符的转换表
halfwidth_chars = "".join([chr(i) for i in range(33, 127)])
fullwidth_chars = "".join([chr(i + 0xFEE0) for i in range(33, 127)])
translation_table = str.maketrans(halfwidth_chars, fullwidth_chars)
# 将不支持的字符替换为对应的全角字符
for char in invalid_characters:
filename = filename.replace(char, char.translate(translation_table))
return filename
# 获取集标题
episode_title = None
if meta.begin_episode and episodes_info:
for episode in episodes_info:
if episode.episode_number == meta.begin_episode:
episode_title = episode.name
break
return {
# 标题
"title": __convert_invalid_characters(mediainfo.title),
# 英文标题
"en_title": __convert_invalid_characters(mediainfo.en_title),
# 原语种标题
"original_title": __convert_invalid_characters(mediainfo.original_title),
# 原文件名
"original_name": meta.title,
# 识别名称（优先使用中文）
"name": meta.name,
# 识别的英文名称（可能为空）
"en_name": meta.en_name,
# 年份
"year": mediainfo.year or meta.year,
# 资源类型
"resourceType": meta.resource_type,
# 特效
"effect": meta.resource_effect,
# 版本
"edition": meta.edition,
# 分辨率
"videoFormat": meta.resource_pix,
# 制作组/字幕组
"releaseGroup": meta.resource_team,
# 视频编码
"videoCodec": meta.video_encode,
# 音频编码
"audioCodec": meta.audio_encode,
# TMDBID
"tmdbid": mediainfo.tmdb_id,
# IMDBID
"imdbid": mediainfo.imdb_id,
# 豆瓣ID
"doubanid": mediainfo.douban_id,
# 季号
"season": meta.season_seq,
# 集号
"episode": meta.episode_seqs,
# 季集 SxxExx
"season_episode": "%s%s" % (meta.season, meta.episodes),
# 段/节
"part": meta.part,
# 剧集标题
"episode_title": __convert_invalid_characters(episode_title),
# 文件后缀
"fileExt": file_ext,
# 自定义占位符
"customization": meta.customization
}
@staticmethod
def get_rename_path(template_string: str, rename_dict: dict, path: Path = None) -> Path:
"""
生成重命名后的完整路径
"""
# 创建jinja2模板对象
template = Template(template_string)
# 渲染生成的字符串
render_str = template.render(rename_dict)
# 目的路径
if path:
return path / render_str
else:
return Path(render_str)
def media_exists(self, mediainfo: MediaInfo, **kwargs) -> Optional[ExistMediaInfo]:
"""
判断媒体文件是否存在于本地文件系统，只支持标准媒体库结构
:param mediainfo: 识别的媒体信息
:return: 如不存在返回None，存在时返回信息，包括每季已存在所有集{type: movie/tv, seasons: {season: [episodes]}}
"""
# 目的路径
dest_paths = DirectoryHelper().get_library_dirs()
# 检查每一个媒体库目录
for dest_path in dest_paths:
# 媒体分类路径
target_dir = self.__get_dest_dir(mediainfo=mediainfo, target_dir=dest_path)
if not target_dir.exists():
continue
# 重命名格式
rename_format = settings.TV_RENAME_FORMAT \
if mediainfo.type == MediaType.TV else settings.MOVIE_RENAME_FORMAT
# 获取相对路径（重命名路径）
meta = MetaInfo(mediainfo.title)
rel_path = self.get_rename_path(
template_string=rename_format,
rename_dict=self.__get_naming_dict(meta=meta,
mediainfo=mediainfo)
)
# 取相对路径的第1层目录
if rel_path.parts:
media_path = target_dir / rel_path.parts[0]
else:
continue
# 检查媒体文件夹是否存在
if not media_path.exists():
continue
# 检索媒体文件
media_files = SystemUtils.list_files(directory=media_path, extensions=settings.RMT_MEDIAEXT)
if not media_files:
continue
if mediainfo.type == MediaType.MOVIE:
# 电影存在任何文件为存在
logger.info(f"文件系统已存在：{mediainfo.title_year}")
return ExistMediaInfo(type=MediaType.MOVIE)
else:
# 电视剧检索集数
seasons: Dict[int, list] = {}
for media_file in media_files:
file_meta = MetaInfo(media_file.stem)
season_index = file_meta.begin_season or 1
episode_index = file_meta.begin_episode
if not episode_index:
continue
if season_index not in seasons:
seasons[season_index] = []
seasons[season_index].append(episode_index)
# 返回剧集情况
logger.info(f"{mediainfo.title_year} 文件系统已存在：{seasons}")
return ExistMediaInfo(type=MediaType.TV, seasons=seasons)
# 不存在
return None
@staticmethod
def delete_all_version_files(path: Path) -> bool:
"""
删除目录下的所有版本文件
:param path: 目录路径
"""
# 识别文件中的季集信息
meta = MetaInfoPath(path)
season = meta.season
episode = meta.episode
# 检索媒体文件
logger.warn(f"正在删除目标目录中其它版本的文件：{path.parent}")
media_files = SystemUtils.list_files(directory=path.parent, extensions=settings.RMT_MEDIAEXT)
if not media_files:
logger.info(f"目录中没有媒体文件：{path.parent}")
return False
# 删除文件
for media_file in media_files:
if str(media_file) == str(path):
continue
# 识别文件中的季集信息
filemeta = MetaInfoPath(media_file)
# 相同季集的文件才删除
if filemeta.season != season or filemeta.episode != episode:
continue
logger.info(f"正在删除文件：{media_file}")
media_file.unlink()
return True

117
app/modules/filemanager/storage/__init__.py Normal file
View File

@@ -0,0 +1,117 @@
from abc import ABCMeta, abstractmethod
from pathlib import Path
from typing import Optional, List, Any
from app import schemas
class StorageBase(metaclass=ABCMeta):
"""
存储基类
"""
transtype = {}
def support_transtype(self) -> dict:
"""
支持的整理方式
"""
return self.transtype
def is_support_transtype(self, transtype: str) -> bool:
"""
是否支持整理方式
"""
return transtype in self.transtype
@abstractmethod
def check(self) -> bool:
"""
检查存储是否可用
"""
pass
@abstractmethod
def list(self, fileitm: schemas.FileItem) -> Optional[List[schemas.FileItem]]:
"""
浏览文件
"""
pass
@abstractmethod
def create_folder(self, fileitm: schemas.FileItem, name: str) -> Optional[schemas.FileItem]:
"""
创建目录
"""
pass
@abstractmethod
def get_folder(self, path: Path) -> Optional[schemas.FileItem]:
"""
获取目录
"""
pass
@abstractmethod
def delete(self, fileitm: schemas.FileItem) -> bool:
"""
删除文件
"""
pass
@abstractmethod
def rename(self, fileitm: schemas.FileItem, name: str) -> bool:
"""
重命名文件
"""
pass
@abstractmethod
def download(self, fileitm: schemas.FileItem, path: Path):
"""
下载文件，保存到本地
"""
pass
@abstractmethod
def upload(self, fileitm: schemas.FileItem, path: Path) -> Optional[schemas.FileItem]:
"""
上传文件
"""
pass
@abstractmethod
def detail(self, fileitm: schemas.FileItem) -> Optional[schemas.FileItem]:
"""
获取文件详情
"""
pass
@abstractmethod
def copy(self, fileitm: schemas.FileItem, target_file: Path) -> bool:
"""
复制文件
"""
pass
@abstractmethod
def move(self, fileitm: schemas.FileItem, target_file: Path) -> bool:
"""
移动文件
"""
pass
@abstractmethod
def link(self, fileitm: schemas.FileItem, target_file: Path) -> bool:
"""
硬链接文件
"""
pass
@abstractmethod
def softlink(self, fileitm: schemas.FileItem, target_file: schemas.FileItem) -> bool:
"""
软链接文件
"""
pass

666
app/modules/filemanager/storage/alipan.py Normal file
View File

@@ -0,0 +1,666 @@
import base64
import json
import time
import uuid
from pathlib import Path
from typing import Optional, Tuple, List
from requests import Response
from app import schemas
from app.core.config import settings
from app.db.systemconfig_oper import SystemConfigOper
from app.log import logger
from app.modules.filemanager.storage import StorageBase
from app.schemas.types import SystemConfigKey, StorageSchema
from app.utils.http import RequestUtils
from app.utils.string import StringUtils
from app.utils.system import SystemUtils
class AliPan(StorageBase):
"""
阿里云相关操作
"""
# 存储类型
schema = StorageSchema.Alipan
# 支持的整理方式
transtype = {
"move": "移动"
}
_X_SIGNATURE = ('f4b7bed5d8524a04051bd2da876dd79afe922b8205226d65855d02b267422adb1'
'e0d8a816b021eaf5c36d101892180f79df655c5712b348c2a540ca136e6b22001')
_X_PUBLIC_KEY = ('04d9d2319e0480c840efeeb75751b86d0db0c5b9e72c6260a1d846958adceaf9d'
'ee789cab7472741d23aafc1a9c591f72e7ee77578656e6c8588098dea1488ac2a')
# 生成二维码
qrcode_url = ("https://passport.aliyundrive.com/newlogin/qrcode/generate.do?"
"appName=aliyun_drive&fromSite=52&appEntrance=web&isMobile=false"
"&lang=zh_CN&returnUrl=&bizParams=&_bx-v=2.0.31")
# 二维码登录确认
check_url = "https://passport.aliyundrive.com/newlogin/qrcode/query.do?appName=aliyun_drive&fromSite=52&_bx-v=2.0.31"
# 更新访问令牌
update_accessstoken_url = "https://auth.aliyundrive.com/v2/account/token"
# 创建会话
create_session_url = "https://api.aliyundrive.com/users/v1/users/device/create_session"
# 用户信息
user_info_url = "https://user.aliyundrive.com/v2/user/get"
# 浏览文件
list_file_url = "https://api.aliyundrive.com/adrive/v3/file/list"
# 创建目录或文件
create_folder_file_url = "https://api.aliyundrive.com/adrive/v2/file/createWithFolders"
# 文件详情
file_detail_url = "https://api.aliyundrive.com/v2/file/get"
# 删除文件
delete_file_url = " https://api.aliyundrive.com/v2/recyclebin/trash"
# 文件重命名
rename_file_url = "https://api.aliyundrive.com/v3/file/update"
# 获取下载链接
download_url = "https://api.aliyundrive.com/v2/file/get_download_url"
# 移动文件
move_file_url = "https://api.aliyundrive.com/v2/file/move"
# 上传文件完成
upload_file_complete_url = "https://api.aliyundrive.com/v2/file/complete"
def __init__(self):
self.systemconfig = SystemConfigOper()
def __handle_error(self, res: Response, apiname: str, action: bool = True):
"""
统一处理和打印错误信息
"""
if res is None:
logger.warn("无法连接到阿里云盘！")
return
try:
result = res.json()
except Exception as err:
logger.error(f"解析阿里云盘返回数据失败：{str(err)}")
return
code = result.get("code")
message = result.get("message")
display_message = result.get("display_message")
if code or message:
logger.warn(f"Aliyun {apiname}失败：{code} - {display_message or message}")
if action:
if code == "DeviceSessionSignatureInvalid":
logger.warn("设备已失效，正在重新建立会话...")
self.__create_session(self.__get_headers(self.__auth_params))
if code == "UserDeviceOffline":
logger.warn("设备已离线，尝试重新登录，如仍报错请检查阿里云盘绑定设备数量是否超限！")
self.__create_session(self.__get_headers(self.__auth_params))
if code == "AccessTokenInvalid":
logger.warn("访问令牌已失效，正在刷新令牌...")
self.__update_accesstoken(self.__auth_params, self.__auth_params.get("refreshToken"))
else:
logger.info(f"Aliyun {apiname}成功")
@property
def __auth_params(self):
"""
获取阿里云盘认证参数并初始化参数格式
"""
return self.systemconfig.get(SystemConfigKey.UserAliyunParams) or {}
def __update_params(self, params: dict):
"""
设置阿里云盘认证参数
"""
current_params = self.__auth_params
current_params.update(params)
self.systemconfig.set(SystemConfigKey.UserAliyunParams, current_params)
def __clear_params(self):
"""
清除阿里云盘认证参数
"""
self.systemconfig.delete(SystemConfigKey.UserAliyunParams)
def generate_qrcode(self) -> Optional[Tuple[dict, str]]:
"""
生成二维码
"""
res = RequestUtils(timeout=10).get_res(self.qrcode_url)
if res:
data = res.json().get("content", {}).get("data")
return {
"codeContent": data.get("codeContent"),
"ck": data.get("ck"),
"t": data.get("t")
}, ""
elif res is not None:
self.__handle_error(res, "生成二维码")
return {}, f"请求阿里云盘二维码失败：{res.status_code} - {res.reason}"
return {}, f"请求阿里云盘二维码失败：无法连接！"
def check_login(self, ck: str, t: str) -> Optional[Tuple[dict, str]]:
"""
二维码登录确认
"""
params = {
"t": t,
"ck": ck,
"appName": "aliyun_drive",
"appEntrance": "web",
"isMobile": "false",
"lang": "zh_CN",
"returnUrl": "",
"fromSite": "52",
"bizParams": "",
"navlanguage": "zh-CN",
"navPlatform": "MacIntel",
}
body = "&".join([f"{key}={value}" for key, value in params.items()])
status = {
"NEW": "请用阿里云盘 App 扫码",
"SCANED": "请在手机上确认",
"EXPIRED": "二维码已过期",
"CANCELED": "已取消",
"CONFIRMED": "已确认",
}
headers = {
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
}
res = RequestUtils(headers=headers, timeout=5).post_res(self.check_url, data=body)
if res:
data = res.json().get("content", {}).get("data") or {}
qrCodeStatus = data.get("qrCodeStatus")
data["tip"] = status.get(qrCodeStatus) or "未知"
if data.get("bizExt"):
try:
bizExt = json.loads(base64.b64decode(data["bizExt"]).decode('GBK'))
pds_login_result = bizExt.get("pds_login_result")
if pds_login_result:
data.pop('bizExt')
data.update({
'userId': pds_login_result.get('userId'),
'expiresIn': pds_login_result.get('expiresIn'),
'nickName': pds_login_result.get('nickName'),
'avatar': pds_login_result.get('avatar'),
'tokenType': pds_login_result.get('tokenType'),
"refreshToken": pds_login_result.get('refreshToken'),
"accessToken": pds_login_result.get('accessToken'),
"defaultDriveId": pds_login_result.get('defaultDriveId'),
"updateTime": time.time(),
})
self.__update_params(data)
self.user_info()
except Exception as e:
return {}, f"bizExt 解码失败：{str(e)}"
return data, ""
elif res is not None:
self.__handle_error(res, "登录确认")
return {}, f"阿里云盘登录确认失败：{res.status_code} - {res.reason}"
return {}, "阿里云盘登录确认失败：无法连接！"
def __update_accesstoken(self, params: dict, refresh_token: str) -> bool:
"""
更新阿里云盘访问令牌
"""
headers = self.__get_headers(params)
res = RequestUtils(headers=headers, timeout=10).post_res(
self.update_accessstoken_url, json={
"refresh_token": refresh_token,
"grant_type": "refresh_token"
})
if res:
data = res.json()
code = data.get("code")
if code in ["RefreshTokenExpired", "InvalidParameter.RefreshToken"]:
logger.warn("刷新令牌已过期，请重新登录！")
self.__clear_params()
return False
self.__update_params({
"accessToken": data.get('access_token'),
"expiresIn": data.get('expires_in'),
"updateTime": time.time()
})
logger.info(f"阿里云盘访问令牌已更新，accessToken={data.get('access_token')}")
return True
else:
self.__handle_error(res, "更新令牌", action=False)
return False
def __create_session(self, headers: dict):
"""
创建会话
"""
def __os_name():
"""
获取操作系统名称
"""
if SystemUtils.is_windows():
return 'Windows 操作系统'
elif SystemUtils.is_macos():
return 'MacOS 操作系统'
else:
return '类 Unix 操作系统'
res = RequestUtils(headers=headers, timeout=5).post_res(self.create_session_url, json={
'deviceName': f'MoviePilot {SystemUtils.platform}',
'modelName': __os_name(),
'pubKey': self._X_PUBLIC_KEY,
})
self.__handle_error(res, "创建会话", action=False)
@property
def __access_params(self) -> Optional[dict]:
"""
获取阿里云盘访问参数，如果超时则更新后返回
"""
params = self.__auth_params
if not params:
logger.warn("阿里云盘访问令牌不存在，请先扫码登录！")
return None
expires_in = params.get("expiresIn")
update_time = params.get("updateTime")
refresh_token = params.get("refreshToken")
if not expires_in or not update_time or not refresh_token:
logger.warn("阿里云盘访问令牌参数错误，请重新扫码登录！")
self.__clear_params()
return None
# 是否需要更新设备信息
update_device = False
# 判断访问令牌是否过期
if (time.time() - update_time) >= expires_in:
logger.info("阿里云盘访问令牌已过期，正在更新...")
if not self.__update_accesstoken(params, refresh_token):
# 更新失败
return None
update_device = True
# 生成设备ID
x_device_id = params.get("x_device_id")
if not x_device_id:
x_device_id = uuid.uuid4().hex
params['x_device_id'] = x_device_id
self.__update_params({"x_device_id": x_device_id})
update_device = True
# 更新设备信息重新创建会话
if update_device:
self.__create_session(self.__get_headers(params))
return params
def __get_headers(self, params: dict):
"""
获取请求头
"""
if not params:
return {}
return {
"Authorization": f"Bearer {params.get('accessToken')}",
"Content-Type": "application/json;charset=UTF-8",
"Accept": "application/json, text/plain, */*",
"Referer": "https://www.alipan.com/",
"User-Agent": settings.USER_AGENT,
"X-Canary": "client=web,app=adrive,version=v4.9.0",
"x-device-id": params.get('x_device_id'),
"x-signature": self._X_SIGNATURE
}
def check(self) -> bool:
"""
检查存储是否可用
"""
pass
def user_info(self) -> dict:
"""
获取用户信息（drive_id等）
"""
params = self.__access_params
if not params:
return {}
headers = self.__get_headers(params)
res = RequestUtils(headers=headers, timeout=10).post_res(self.user_info_url)
if res:
result = res.json()
self.__update_params({
"resourceDriveId": result.get("resource_drive_id"),
"backDriveId": result.get("backup_drive_id")
})
return result
else:
self.__handle_error(res, "获取用户信息")
return {}
def list(self, drive_id: str = None, parent_file_id: str = 'root', list_type: str = None,
limit: int = 100, order_by: str = 'updated_at', path: str = "/") -> List[schemas.FileItem]:
"""
浏览文件
limit 返回文件数量，默认 50，最大 100
order_by created_at/updated_at/name/size
parent_file_id 根目录为root
type all | file | folder
"""
params = self.__access_params
if not params:
return []
# 请求头
headers = self.__get_headers(params)
# 根目录处理
if not drive_id:
return [
schemas.FileItem(
fileid=parent_file_id,
drive_id=params.get("resourceDriveId"),
parent_fileid="root",
type="dir",
path="/资源库/",
name="资源库"
),
schemas.FileItem(
fileid=parent_file_id,
drive_id=params.get("backDriveId"),
parent_fileid="root",
type="dir",
path="/备份盘/",
name="备份盘"
)
]
# 返回数据
ret_items = []
# 分页获取
next_marker = None
while True:
if not parent_file_id or parent_file_id == "/":
parent_file_id = "root"
res = RequestUtils(headers=headers, timeout=10).post_res(self.list_file_url, json={
"drive_id": drive_id,
"type": list_type,
"limit": limit,
"order_by": order_by,
"parent_file_id": parent_file_id,
"marker": next_marker
}, params={
'jsonmask': ('next_marker,items(name,file_id,drive_id,type,size,created_at,updated_at,'
'category,file_extension,parent_file_id,mime_type,starred,thumbnail,url,'
'streams_info,content_hash,user_tags,user_meta,trashed,video_media_metadata,'
'video_preview_metadata,sync_meta,sync_device_flag,sync_flag,punish_flag')
})
if res:
result = res.json()
items = result.get("items")
if not items:
break
# 合并数据
ret_items.extend(items)
next_marker = result.get("next_marker")
if not next_marker:
# 没有下一页
break
else:
self.__handle_error(res, "浏览文件")
break
return [schemas.FileItem(
fileid=fileinfo.get("file_id"),
parent_fileid=fileinfo.get("parent_file_id"),
type="dir" if fileinfo.get("type") == "folder" else "file",
path=f"{path}{fileinfo.get('name')}" + ("/" if fileinfo.get("type") == "folder" else ""),
name=fileinfo.get("name"),
size=fileinfo.get("size"),
extension=fileinfo.get("file_extension"),
modify_time=StringUtils.str_to_timestamp(fileinfo.get("updated_at")),
thumbnail=fileinfo.get("thumbnail"),
drive_id=fileinfo.get("drive_id"),
) for fileinfo in ret_items]
def create_folder(self, fileitem: schemas.FileItem, name: str) -> Optional[schemas.FileItem]:
"""
创建目录
"""
params = self.__access_params
if not params:
return None
headers = self.__get_headers(params)
res = RequestUtils(headers=headers, timeout=10).post_res(self.create_folder_file_url, json={
"drive_id": fileitem.drive_id,
"parent_file_id": fileitem.parent_fileid,
"name": name,
"check_name_mode": "refuse",
"type": "folder"
})
if res:
"""
{
"parent_file_id": "root",
"type": "folder",
"file_id": "6673f2c8a88344741bd64ad192d7512b92087719",
"domain_id": "bj29",
"drive_id": "39146740",
"file_name": "test",
"encrypt_mode": "none"
}
"""
result = res.json()
return schemas.FileItem(
fileid=result.get("file_id"),
drive_id=result.get("drive_id"),
parent_fileid=result.get("parent_file_id"),
type=result.get("type"),
name=result.get("file_name"),
path=f"{fileitem.path}{result.get('file_name')}",
)
else:
self.__handle_error(res, "创建目录")
return None
def get_folder(self, path: Path) -> Optional[schemas.FileItem]:
"""
TODO 获取目录，不存在则创建
"""
pass
def delete(self, fileitem: schemas.FileItem) -> bool:
"""
删除文件
"""
params = self.__access_params
if not params:
return False
headers = self.__get_headers(params)
res = RequestUtils(headers=headers, timeout=10).post_res(self.delete_file_url, json={
"drive_id": fileitem.drive_id,
"file_id": fileitem.fileid
})
if res:
return True
else:
self.__handle_error(res, "删除文件")
return False
def detail(self, fileitem: schemas.FileItem) -> Optional[schemas.FileItem]:
"""
获取文件详情
"""
params = self.__access_params
if not params:
return None
headers = self.__get_headers(params)
res = RequestUtils(headers=headers, timeout=10).post_res(self.file_detail_url, json={
"drive_id": fileitem.drive_id,
"file_id": fileitem.fileid
})
if res:
result = res.json()
return schemas.FileItem(
fileid=result.get("file_id"),
drive_id=result.get("drive_id"),
parent_fileid=result.get("parent_file_id"),
type="file",
name=result.get("name"),
size=result.get("size"),
extension=result.get("file_extension"),
modify_time=StringUtils.str_to_timestamp(result.get("updated_at")),
thumbnail=result.get("thumbnail"),
path=f"{fileitem.path}{result.get('name')}"
)
else:
self.__handle_error(res, "获取文件详情")
return None
def rename(self, fileitem: schemas.FileItem, name: str) -> bool:
"""
重命名文件
"""
params = self.__access_params
if not params:
return False
headers = self.__get_headers(params)
res = RequestUtils(headers=headers, timeout=10).post_res(self.rename_file_url, json={
"drive_id": fileitem.drive_id,
"file_id": fileitem.fileid,
"name": name,
"check_name_mode": "refuse"
})
if res:
return True
else:
self.__handle_error(res, "重命名文件")
return False
def download(self, fileitem: schemas.FileItem, path: Path) -> bool:
"""
下载文件，保存到本地
"""
params = self.__access_params
if not params:
return False
headers = self.__get_headers(params)
res = RequestUtils(headers=headers, timeout=10).post_res(self.download_url, json={
"drive_id": fileitem.drive_id,
"file_id": fileitem.fileid
})
if res:
download_url = res.json().get("url")
if not download_url:
return False
res = RequestUtils().get_res(download_url)
if res:
with path.open("wb") as f:
f.write(res.content)
return True
else:
self.__handle_error(res, "获取下载链接")
return False
def upload(self, fileitem: schemas.FileItem, path: Path) -> Optional[schemas.FileItem]:
"""
上传文件，并标记完成
TODO 上传文件分片、秒传
"""
params = self.__access_params
if not params:
return None
headers = self.__get_headers(params)
res = RequestUtils(headers=headers, timeout=10).post_res(self.create_folder_file_url, json={
"drive_id": fileitem.drive_id,
"parent_file_id": fileitem.parent_fileid,
"name": path.name,
"check_name_mode": "refuse",
"create_scene": "file_upload",
"type": "file",
"part_info_list": [
{
"part_number": 1
}
],
"size": path.stat().st_size
})
if not res:
self.__handle_error(res, "创建文件")
return None
# 获取上传参数
result = res.json()
if result.get("exist"):
logger.info(f"文件{result.get('file_name')}已存在，无需上传")
return schemas.FileItem(
drive_id=result.get("drive_id"),
fileid=result.get("file_id"),
parent_fileid=result.get("parent_file_id"),
type="file",
name=result.get("file_name"),
path=f"{fileitem.path}{result.get('file_name')}"
)
file_id = result.get("file_id")
upload_id = result.get("upload_id")
part_info_list = result.get("part_info_list")
if part_info_list:
# 上传地址
upload_url = part_info_list[0].get("upload_url")
# 上传文件
res = RequestUtils(headers={
"Content-Type": "",
"User-Agent": settings.USER_AGENT,
"Referer": "https://www.alipan.com/",
"Accept": "*/*",
}).put_res(upload_url, data=path.read_bytes())
if not res:
self.__handle_error(res, "上传文件")
return None
# 标记文件上传完毕
res = RequestUtils(headers=headers, timeout=10).post_res(self.upload_file_complete_url, json={
"drive_id": fileitem.drive_id,
"file_id": file_id,
"upload_id": upload_id
})
if not res:
self.__handle_error(res, "标记上传状态")
return None
result = res.json()
return schemas.FileItem(
fileid=result.get("file_id"),
drive_id=result.get("drive_id"),
parent_fileid=result.get("parent_file_id"),
type="file",
name=result.get("name"),
path=f"{fileitem.path}{result.get('name')}",
)
else:
logger.warn("上传文件失败：无法获取上传地址！")
return None
def move(self, fileitem: schemas.FileItem, target_dir: schemas.FileItem) -> bool:
"""
移动文件
"""
params = self.__access_params
if not params:
return False
headers = self.__get_headers(params)
res = RequestUtils(headers=headers, timeout=10).post_res(self.move_file_url, json={
"drive_id": fileitem.drive_id,
"file_id": fileitem.fileid,
"to_parent_file_id": target_dir.fileid,
"check_name_mode": "refuse"
})
if res:
return True
else:
self.__handle_error(res, "移动文件")
return False
def copy(self, fileitm: schemas.FileItem, target_file: Path) -> bool:
"""
复制文件
"""
pass
def link(self, fileitm: schemas.FileItem, target_file: Path) -> bool:
"""
硬链接文件
"""
pass
def softlink(self, fileitm: schemas.FileItem, target_file: schemas.FileItem) -> bool:
"""
软链接文件
"""
pass

217
app/modules/filemanager/storage/local.py Normal file
View File

@@ -0,0 +1,217 @@
import shutil
from pathlib import Path
from typing import Optional, List
from app import schemas
from app.log import logger
from app.modules.filemanager.storage import StorageBase
from app.schemas.types import StorageSchema
from app.utils.system import SystemUtils
class LocalStorage(StorageBase):
"""
本地文件操作
"""
# 存储类型
schema = StorageSchema.Local
# 支持的整理方式
transtype = {
"copy": "复制",
"move": "移动",
"link": "硬链接",
"softlink": "软链接"
}
def check(self) -> bool:
"""
检查存储是否可用
"""
return True
def list(self, fileitem: schemas.FileItem) -> Optional[List[schemas.FileItem]]:
"""
浏览文件
"""
# 返回结果
ret_items = []
path = fileitem.path
if not fileitem.path or fileitem.path == "/":
if SystemUtils.is_windows():
partitions = SystemUtils.get_windows_drives() or ["C:/"]
for partition in partitions:
ret_items.append(schemas.FileItem(
type="dir",
path=partition + "/",
name=partition,
basename=partition
))
return ret_items
else:
path = "/"
else:
if SystemUtils.is_windows():
path = path.lstrip("/")
elif not path.startswith("/"):
path = "/" + path
# 遍历目录
path_obj = Path(path)
if not path_obj.exists():
logger.warn(f"目录不存在：{path}")
return []
# 如果是文件
if path_obj.is_file():
ret_items.append(schemas.FileItem(
type="file",
path=str(path_obj).replace("\\", "/"),
name=path_obj.name,
basename=path_obj.stem,
extension=path_obj.suffix[1:],
size=path_obj.stat().st_size,
modify_time=path_obj.stat().st_mtime,
))
return ret_items
# 扁历所有目录
for item in SystemUtils.list_sub_directory(path_obj):
ret_items.append(schemas.FileItem(
type="dir",
path=str(item).replace("\\", "/") + "/",
name=item.name,
basename=item.stem,
modify_time=item.stat().st_mtime,
))
# 遍历所有文件，不含子目录
for item in SystemUtils.list_sub_all(path_obj):
ret_items.append(schemas.FileItem(
type="file",
path=str(item).replace("\\", "/"),
name=item.name,
basename=item.stem,
extension=item.suffix[1:],
size=item.stat().st_size,
modify_time=item.stat().st_mtime,
))
return ret_items
def create_folder(self, fileitem: schemas.FileItem, name: str) -> Optional[schemas.FileItem]:
"""
创建目录
"""
if not fileitem.path:
return None
path_obj = Path(fileitem.path) / name
if not path_obj.exists():
path_obj.mkdir(parents=True, exist_ok=True)
return schemas.FileItem(
type="dir",
path=str(path_obj).replace("\\", "/") + "/",
name=name,
basename=name,
modify_time=path_obj.stat().st_mtime,
)
def get_folder(self, path: Path) -> Optional[schemas.FileItem]:
"""
获取目录
"""
pass
def detail(self, fileitm: schemas.FileItem) -> Optional[schemas.FileItem]:
"""
获取文件详情
"""
path_obj = Path(fileitm.path)
return schemas.FileItem(
type="file",
path=str(path_obj).replace("\\", "/"),
name=path_obj.name,
basename=path_obj.stem,
extension=path_obj.suffix[1:],
size=path_obj.stat().st_size,
modify_time=path_obj.stat().st_mtime,
)
def delete(self, fileitem: schemas.FileItem) -> bool:
"""
删除文件
"""
if not fileitem.path:
return False
path_obj = Path(fileitem.path)
if not path_obj.exists():
return False
if path_obj.is_file():
path_obj.unlink()
else:
shutil.rmtree(path_obj, ignore_errors=True)
return True
def rename(self, fileitem: schemas.FileItem, name: str) -> bool:
"""
重命名文件
"""
path_obj = Path(fileitem.path)
if not path_obj.exists():
return False
path_obj.rename(path_obj.parent / name)
def download(self, fileitem: schemas.FileItem, path: Path) -> bool:
"""
下载文件
"""
return False
def upload(self, fileitem: schemas.FileItem, path: Path) -> Optional[schemas.FileItem]:
"""
上传文件
"""
return None
def copy(self, fileitem: schemas.FileItem, target_file: Path) -> bool:
"""
复制文件
"""
file_path = Path(fileitem.path)
code, message = SystemUtils.copy(file_path, target_file)
if code != 0:
logger.error(f"复制文件失败：{message}")
return False
return True
def link(self, fileitem: schemas.FileItem, target_file: Path) -> bool:
"""
硬链接文件
"""
file_path = Path(fileitem.path)
code, message = SystemUtils.link(file_path, target_file)
if code != 0:
logger.error(f"硬链接文件失败：{message}")
return False
return True
def softlink(self, fileitem: schemas.FileItem, target_file: Path) -> bool:
"""
软链接文件
"""
file_path = Path(fileitem.path)
code, message = SystemUtils.copy(file_path, target_file)
if code != 0:
logger.error(f"软链接文件失败：{message}")
return False
return True
def move(self, fileitem: schemas.FileItem, target_file: Path) -> bool:
"""
移动文件
"""
file_path = Path(fileitem.path)
code, message = SystemUtils.move(file_path, target_file)
if code != 0:
logger.error(f"移动文件失败：{message}")
return False
return True

107
app/modules/filemanager/storage/rclone.py Normal file
View File

@@ -0,0 +1,107 @@
import subprocess
from pathlib import Path
from typing import Optional, List
from app import schemas
from app.log import logger
from app.modules.filemanager.storage import StorageBase
from app.schemas.types import StorageSchema
from app.utils.system import SystemUtils
class Rclone(StorageBase):
"""
rclone相关操作
"""
# 存储类型
schema = StorageSchema.Rclone
# 支持的整理方式
transtype = {
"move": "移动",
"copy": "复制"
}
@staticmethod
def __get_hidden_shell():
if SystemUtils.is_windows():
st = subprocess.STARTUPINFO()
st.dwFlags = subprocess.STARTF_USESHOWWINDOW
st.wShowWindow = subprocess.SW_HIDE
return st
else:
return None
def check(self) -> bool:
pass
def list(self, fileitm: schemas.FileItem) -> Optional[List[schemas.FileItem]]:
pass
def create_folder(self, fileitm: schemas.FileItem, name: str) -> Optional[schemas.FileItem]:
pass
def get_folder(self, path: Path) -> Optional[schemas.FileItem]:
"""
获取目录
"""
pass
def delete(self, fileitm: schemas.FileItem) -> bool:
pass
def rename(self, fileitm: schemas.FileItem, name: str) -> bool:
pass
def download(self, fileitm: schemas.FileItem, path: Path) -> bool:
pass
def upload(self, fileitm: schemas.FileItem, path: Path) -> Optional[schemas.FileItem]:
pass
def detail(self, fileitm: schemas.FileItem) -> Optional[schemas.FileItem]:
pass
def move(self, fileitm: schemas.FileItem, target_file: schemas.FileItem) -> bool:
"""
移动文件，target_file格式：rclone:path
"""
try:
retcode = subprocess.run(
[
'rclone', 'moveto',
fileitm.path,
f'{target_file}'
],
startupinfo=self.__get_hidden_shell()
).returncode
if retcode == 0:
return True
except Exception as err:
logger.error(f"移动文件失败：{err}")
return False
def copy(self, fileitm: schemas.FileItem, target_file: Path) -> bool:
"""
复制文件，target_file格式：rclone:path
"""
try:
retcode = subprocess.run(
[
'rclone', 'copyto',
fileitm.path,
f'{target_file}'
],
startupinfo=self.__get_hidden_shell()
).returncode
if retcode == 0:
return True
except Exception as err:
logger.error(f"复制文件失败：{err}")
return False
def link(self, fileitm: schemas.FileItem, target_file: Path) -> bool:
pass
def softlink(self, fileitm: schemas.FileItem, target_file: schemas.FileItem) -> bool:
pass

325
app/modules/filemanager/storage/u115.py Normal file
View File

@@ -0,0 +1,325 @@
import base64
from pathlib import Path
from typing import Optional, Tuple, List
import oss2
import py115
from py115 import Cloud
from py115.types import LoginTarget, QrcodeSession, QrcodeStatus, Credential
from app import schemas
from app.db.systemconfig_oper import SystemConfigOper
from app.log import logger
from app.modules.filemanager.storage import StorageBase
from app.schemas.types import SystemConfigKey, StorageSchema
from app.utils.http import RequestUtils
from app.utils.singleton import Singleton
class U115Pan(StorageBase, metaclass=Singleton):
"""
115相关操作
"""
# 存储类型
schema = StorageSchema.U115
# 支持的整理方式
transtype = {
"move": "移动"
}
cloud: Optional[Cloud] = None
_session: QrcodeSession = None
def __init__(self):
self.systemconfig = SystemConfigOper()
def __init_cloud(self) -> bool:
"""
初始化Cloud
"""
credential = self.__credential
if not credential:
logger.warn("115未登录，请先登录！")
return False
try:
if not self.cloud:
self.cloud = py115.connect(credential)
except Exception as err:
logger.error(f"115连接失败，请重新扫码登录：{str(err)}")
self.__clear_credential()
return False
return True
@property
def __credential(self) -> Optional[Credential]:
"""
获取已保存的115认证参数
"""
cookie_dict = self.systemconfig.get(SystemConfigKey.User115Params)
if not cookie_dict:
return None
return Credential.from_dict(cookie_dict)
def __save_credentail(self, credential: Credential):
"""
设置115认证参数
"""
self.systemconfig.set(SystemConfigKey.User115Params, credential.to_dict())
def __clear_credential(self):
"""
清除115认证参数
"""
self.systemconfig.delete(SystemConfigKey.User115Params)
def generate_qrcode(self) -> Optional[Tuple[dict, str]]:
"""
生成二维码
"""
try:
self.cloud = py115.connect()
self._session = self.cloud.qrcode_login(LoginTarget.Web)
image_bin = self._session.image_data
if not image_bin:
logger.warn("115生成二维码失败：未获取到二维码数据！")
return None
# 转换为base64图片格式
image_base64 = base64.b64encode(image_bin).decode()
return {
"codeContent": f"data:image/jpeg;base64,{image_base64}"
}, ""
except Exception as e:
logger.warn(f"115生成二维码失败：{str(e)}")
return {}, f"生成二维码失败：{str(e)}"
def check_login(self) -> Optional[Tuple[dict, str]]:
"""
二维码登录确认
"""
if not self._session:
return {}, "请先生成二维码！"
try:
if not self.cloud:
return {}, "请先生成二维码！"
status = self.cloud.qrcode_poll(self._session)
if status == QrcodeStatus.Done:
# 确认完成，保存认证信息
self.__save_credentail(self.cloud.export_credentail())
result = {
"status": 1,
"tip": "登录成功！"
}
elif status == QrcodeStatus.Waiting:
result = {
"status": 0,
"tip": "请使用微信或115客户端扫码"
}
elif status == QrcodeStatus.Expired:
result = {
"status": -1,
"tip": "二维码已过期，请重新刷新！"
}
self.cloud = None
elif status == QrcodeStatus.Failed:
result = {
"status": -2,
"tip": "登录失败，请重试！"
}
self.cloud = None
else:
result = {
"status": -3,
"tip": "未知错误，请重试！"
}
self.cloud = None
return result, ""
except Exception as e:
return {}, f"115登录确认失败：{str(e)}"
def storage(self) -> Optional[Tuple[int, int]]:
"""
获取存储空间
"""
if not self.__init_cloud():
return None
try:
return self.cloud.storage().space()
except Exception as e:
logger.error(f"获取115存储空间失败：{str(e)}")
return None
def check(self) -> bool:
"""
检查存储是否可用
"""
pass
def list(self, fileitem: schemas.FileItem) -> Optional[List[schemas.FileItem]]:
"""
浏览文件
"""
if not self.__init_cloud():
return None
try:
items = self.cloud.storage().list(dir_id=fileitem.parent_fileid)
return [schemas.FileItem(
fileid=item.file_id,
parent_fileid=item.parent_id,
type="dir" if item.is_dir else "file",
path=f"{fileitem.path}{item.name}" + ("/" if item.is_dir else ""),
name=item.name,
size=item.size,
extension=Path(item.name).suffix[1:],
modify_time=item.modified_time.timestamp() if item.modified_time else 0,
pickcode=item.pickcode
) for item in items]
except Exception as e:
logger.error(f"浏览115文件失败：{str(e)}")
return None
def create_folder(self, fileitem: schemas.FileItem, name: str) -> Optional[schemas.FileItem]:
"""
创建目录
"""
if not self.__init_cloud():
return None
try:
result = self.cloud.storage().make_dir(fileitem.parent_fileid, name)
return schemas.FileItem(
fileid=result.file_id,
parent_fileid=result.parent_id,
type="dir",
path=f"{fileitem.path}{name}/",
name=name,
modify_time=result.modified_time.timestamp() if result.modified_time else 0,
pickcode=result.pickcode
)
except Exception as e:
logger.error(f"创建115目录失败：{str(e)}")
return None
def get_folder(self, path: Path) -> Optional[schemas.FileItem]:
"""
TODO 获取目录，不存在则创建
"""
pass
def detail(self, fileitm: schemas.FileItem) -> Optional[schemas.FileItem]:
"""
获取文件详情
"""
pass
def delete(self, fileitem: schemas.FileItem) -> bool:
"""
删除文件
"""
if not self.__init_cloud():
return False
try:
self.cloud.storage().delete(fileitem.fileid)
return True
except Exception as e:
logger.error(f"删除115文件失败：{str(e)}")
return False
def rename(self, fileitem: schemas.FileItem, name: str) -> bool:
"""
重命名文件
"""
if not self.__init_cloud():
return False
try:
self.cloud.storage().rename(fileitem.fileid, name)
return True
except Exception as e:
logger.error(f"重命名115文件失败：{str(e)}")
return False
def download(self, fileitem: schemas.FileItem, path: Path) -> bool:
"""
获取下载链接
"""
if not self.__init_cloud():
return False
try:
ticket = self.cloud.storage().request_download(fileitem.pickcode)
if ticket:
res = RequestUtils(headers=ticket.headers).get_res(ticket.url)
if res:
with open(path, "wb") as f:
f.write(res.content)
return True
except Exception as e:
logger.error(f"115下载失败：{str(e)}")
return False
def upload(self, fileitem: schemas.FileItem, path: Path) -> Optional[schemas.FileItem]:
"""
上传文件
"""
if not self.__init_cloud():
return None
try:
ticket = self.cloud.storage().request_upload(dir_id=fileitem.fileid, file_path=str(path))
if ticket is None:
logger.warn(f"115请求上传出错")
return None
elif ticket.is_done:
logger.warn(f"115请求上传失败：文件已存在")
return None
else:
auth = oss2.Auth(**ticket.oss_token)
bucket = oss2.Bucket(
auth=auth,
endpoint=ticket.oss_endpoint,
bucket_name=ticket.bucket_name,
)
por = bucket.put_object_from_file(
key=ticket.object_key,
filename=str(path),
headers=ticket.headers,
)
result = por.resp.response.json()
if result:
fileitem = result.get('data')
logger.info(f"115上传文件成功：{fileitem}")
return schemas.FileItem(
fileid=fileitem.get('file_id'),
parent_fileid=fileitem.fileid,
type="file",
name=fileitem.get('file_name'),
path=f"{fileitem.path}{fileitem.get('file_name')}",
size=fileitem.get('file_size'),
extension=Path(fileitem.get('file_name')).suffix[1:],
pickcode=fileitem.get('pickcode')
)
else:
logger.warn(f"115上传文件失败：{por.resp.response.text}")
return None
except Exception as e:
logger.error(f"上传115文件失败：{str(e)}")
return None
def move(self, fileitem: schemas.FileItem, target_dir: schemas.FileItem) -> bool:
"""
移动文件
"""
if not self.__init_cloud():
return False
try:
self.cloud.storage().move(fileitem.fileid, target_dir.fileid)
return True
except Exception as e:
logger.error(f"移动115文件失败：{str(e)}")
return False
def copy(self, fileitm: schemas.FileItem, target_file: Path) -> bool:
pass
def link(self, fileitm: schemas.FileItem, target_file: Path) -> bool:
pass
def softlink(self, fileitm: schemas.FileItem, target_file: schemas.FileItem) -> bool:
pass