YanLongChangAn
ade1b6c9b0
- 添加.gitignore和.flaskenv环境配置文件 - 创建Flask应用基础架构，包括models、services、utils模块 - 配置数据库模型User和Setting，集成SQLAlchemy和Alembic迁移 - 添加前端Vue项目结构，包含Element Plus组件库 - 配置前后端API路由和蓝prints模块 - 实现书籍下载服务BookService功能模块 - 添加代码规范配置.editorconfig、.oxfmtrc.json、.oxlintrc.json - 配置VSCode推荐插件和前端构建工具链 - 实现文件服务FileService和相关业务逻辑
95 lines
4.3 KiB
Python
95 lines
4.3 KiB
Python
import os
|
|
import re
|
|
import shutil
|
|
from collections import defaultdict
|
|
|
|
from app.constants import IGNORE_DIRS, TORRENT_ROOT_FOLDER
|
|
from app.services import FolderService
|
|
|
|
|
|
class VideoService:
|
|
def __init__(self, paths: list[str]):
|
|
self.paths: list[str] = paths
|
|
self.prefix_pattern = re.compile(r'(?i)[A-Z]+\.')
|
|
self.torrent_root_folder = "C:\\Users\\Localhost\\Desktop\\BT"
|
|
self.torrent_folder_depth = len(TORRENT_ROOT_FOLDER.split(os.sep))
|
|
self.new_paths: dict[str, dict[str, list[str]]] = defaultdict(dict)
|
|
self.new_paths_2: dict[str, list[str]] = defaultdict(dict)
|
|
self.current_folder = ""
|
|
self.result_list: list[str] = []
|
|
self.func_type = None
|
|
|
|
def process_torrent_file(self, entry: os.DirEntry):
|
|
name, ext = os.path.splitext(self.prefix_pattern.sub('.', entry.name))
|
|
if name not in self.new_paths[self.current_folder]:
|
|
self.new_paths[self.current_folder][name] = []
|
|
self.new_paths[self.current_folder][name].append(entry.path)
|
|
if name not in self.new_paths_2:
|
|
self.new_paths_2[name] = []
|
|
self.new_paths_2[name].append(entry.path)
|
|
# if self.func_type != "remove_deduplication":
|
|
# if ext != ".mp4":
|
|
# self.result_list.insert(0, f"重复文件：{name} {ext}")
|
|
# if len(self.new_paths[self.current_folder][name]) > 1:
|
|
# print(name, ext)
|
|
|
|
def get_init_folder(self, entry: os.DirEntry):
|
|
temp_path: list[str] = []
|
|
for path in self.paths:
|
|
if os.path.exists(path) and os.path.exists(os.path.join(path, entry.name)):
|
|
temp_path.append(os.path.join(path, entry.name))
|
|
self.current_folder = entry.name
|
|
FolderService(paths=temp_path, ignore_dirs=IGNORE_DIRS,
|
|
file_callback=self.process_torrent_file,
|
|
empty_folder_callback=self.remove_folder).process_folder()
|
|
if self.func_type != "remove_deduplication":
|
|
return
|
|
FolderService(paths=entry.path, ignore_dirs=IGNORE_DIRS,
|
|
file_callback=self.torrent_deduplication,
|
|
empty_folder_callback=self.remove_folder).process_folder()
|
|
|
|
def torrent_deduplication(self, entry: os.DirEntry):
|
|
without_modifier_file_name = self.prefix_pattern.sub('.', entry.name)
|
|
name, ext = os.path.splitext(without_modifier_file_name)
|
|
if name in self.new_paths[entry.path.split(os.sep)[self.torrent_folder_depth]]:
|
|
folder_arr = entry.path.split(os.sep)[self.torrent_folder_depth:-1]
|
|
folder_arr.insert(0, "done")
|
|
new_path = os.path.join(TORRENT_ROOT_FOLDER, *folder_arr)
|
|
print(os.path.join(new_path, entry.name))
|
|
os.makedirs(new_path, exist_ok=True)
|
|
try:
|
|
shutil.move(entry.path, os.path.join(new_path, entry.name))
|
|
self.result_list.insert(0, f"移动到文件夹：{os.path.join(new_path, entry.name)}")
|
|
except shutil.Error:
|
|
pass
|
|
|
|
def remove_folder(self, entry: os.DirEntry):
|
|
if entry.name == self.current_folder:
|
|
return
|
|
os.rmdir(entry.path)
|
|
self.result_list.insert(0, f"删除文件夹：{entry.path}")
|
|
|
|
def deduplication(self):
|
|
self.func_type = "remove_deduplication"
|
|
FolderService(paths=TORRENT_ROOT_FOLDER, ignore_dirs=IGNORE_DIRS | {"done"}, max_depth=1,
|
|
folder_callback=self.get_init_folder).process_folder()
|
|
return self.result_list
|
|
|
|
def find_deduplication(self):
|
|
self.func_type = "find_deduplication"
|
|
FolderService(paths=TORRENT_ROOT_FOLDER, ignore_dirs=IGNORE_DIRS | {"done"}, max_depth=1,
|
|
folder_callback=self.get_init_folder).process_folder()
|
|
for name in self.new_paths_2:
|
|
if len(self.new_paths_2[name]) > 1:
|
|
temp_name = []
|
|
for path_2 in self.new_paths_2[name]:
|
|
name_2 = os.path.splitext(os.path.basename(path_2))[0]
|
|
if name_2 not in temp_name:
|
|
temp_name.append(name_2)
|
|
else:
|
|
self.result_list.insert(0, name_2)
|
|
print(name_2)
|
|
if len(temp_name)>1 and name in temp_name:
|
|
self.result_list.insert(0, name)
|
|
return self.result_list
|