YanLongChangAn
208209e05f
- 修改文件重命名逻辑，移除空格和连字符处理 - 更新UI端口依赖从 sass-embedded 到 sass - 调整请求基础URL配置为局域网地址 - 重构后端运行配置以支持外部访问 - 添加种子列表路由和提取名称功能 - 移除按名称查询单个种子的路由 - 实现标准化名称提取算法 - 添加批量获取所有种子的功能
157 lines
6.9 KiB
Python
157 lines
6.9 KiB
Python
import os
|
|
import re
|
|
from collections import defaultdict
|
|
from typing import List, Optional
|
|
|
|
from app.constants import IGNORE_DIRS, SEPARATORS, SPECIAL_PREFIXES, SPECIAL_DIRS
|
|
from app.services.folder_service import FolderService
|
|
from app.utils import to_json_serializable
|
|
|
|
|
|
class TorrentService:
|
|
def __init__(self, paths: List[str]):
|
|
self.paths: List[str] = paths
|
|
self.torrent_root_folder = ["C:\\Users\\Localhost\\Desktop\\BT"]
|
|
self.prefix_pattern = re.compile(r'(?i)[A-Z]+\.')
|
|
self.torrent_list = defaultdict(set)
|
|
self.list_of_duplicate_torrent: List[str] = []
|
|
self.result_list = []
|
|
self.torrent_name = defaultdict(lambda: defaultdict(set))
|
|
self.search_name = ""
|
|
|
|
def process_torrent_file(self, entry: os.DirEntry[str]) -> None:
|
|
temp_name = entry.name.replace("-4K", "")
|
|
without_modifier_file_name = self.prefix_pattern.sub('.', temp_name)
|
|
name, ext = os.path.splitext(without_modifier_file_name)
|
|
if name not in self.torrent_list:
|
|
self.torrent_list[name].add(os.path.basename(os.path.dirname(entry.path)))
|
|
else:
|
|
for torrent_path in self.torrent_list[name]:
|
|
if not os.path.exists(os.path.join(os.path.dirname(entry.path), f"{name}A{ext}")):
|
|
self.list_of_duplicate_torrent.insert(0, name)
|
|
print(f"data: {entry.name} 已有种子: {torrent_path}")
|
|
|
|
def generate_torrent_list(self):
|
|
folder_obj = FolderService(paths=self.paths, ignore_dirs=IGNORE_DIRS, file_callback=self.process_torrent_file)
|
|
folder_obj.process_folder()
|
|
return self.torrent_list
|
|
|
|
def deduplication(self):
|
|
self.generate_torrent_list()
|
|
return self.list_of_duplicate_torrent
|
|
|
|
def has_torrent_callback(self, entry: os.DirEntry):
|
|
temp_name = entry.name
|
|
temp_name = temp_name.replace("-4K", "")
|
|
without_modifier_file_name = self.prefix_pattern.sub('.', temp_name)
|
|
name, ext = os.path.splitext(without_modifier_file_name)
|
|
if name not in self.torrent_list:
|
|
self.result_list.insert(0, f"{os.path.basename(os.path.dirname(entry.path))} {entry.name} 该文件无种子")
|
|
|
|
def has_torrent(self):
|
|
FolderService(paths=self.torrent_root_folder, ignore_dirs=IGNORE_DIRS,
|
|
file_callback=self.process_torrent_file).process_folder()
|
|
FolderService(paths=self.paths, ignore_dirs=IGNORE_DIRS,
|
|
file_callback=self.has_torrent_callback).process_folder()
|
|
return self.result_list
|
|
|
|
def statistics_file_callback(self, entry: os.DirEntry):
|
|
path_arr = entry.path.split(os.sep)
|
|
torrent_sub_type = {"0_多人", "FWAY", "0_女同", "_temp"}
|
|
torrent_type = {"4k2", "4K原版", "456k", "1024", "2048", "done", "FC2", "hhd800", "other", "高清中文字幕",
|
|
"三级写真", "无码流出", "亚洲有码原创"}
|
|
if path_arr[-2] not in torrent_type and path_arr[-2] not in torrent_sub_type:
|
|
if path_arr[-3] in torrent_sub_type:
|
|
# print(path_arr[-3])
|
|
if path_arr[-4] in torrent_type:
|
|
# print(path_arr[-4])
|
|
pass
|
|
else:
|
|
print(f"有奇怪的东西混入{path_arr}")
|
|
pass
|
|
elif path_arr[-3] in torrent_type:
|
|
self.torrent_name[path_arr[-2]][path_arr[-3]].add(entry.name)
|
|
pass
|
|
else:
|
|
print(f"有奇怪的东西混入2{path_arr}")
|
|
|
|
def statistics(self):
|
|
FolderService(paths=self.paths, ignore_dirs=IGNORE_DIRS,
|
|
file_callback=self.statistics_file_callback).process_folder()
|
|
return to_json_serializable(self.torrent_name)
|
|
|
|
def process_file_name(self, filename: str) -> Optional[str]:
|
|
name, ext = os.path.splitext(filename)
|
|
ext = ext.lower()
|
|
name = name.upper()
|
|
name.maketrans({"-": "", " ":""})
|
|
# 处理特殊前缀
|
|
for prefix in SPECIAL_PREFIXES:
|
|
if name.startswith(prefix):
|
|
# name = name.replace(prefix, "")
|
|
return f"{prefix}-{name[len(prefix):]}{ext}"
|
|
return name
|
|
|
|
def get_by_name_file_callback(self, entry: os.DirEntry):
|
|
path_arr = entry.path.split(os.sep)
|
|
torrent_sub_type = {"0_多人", "FWAY", "0_女同", "_temp"}
|
|
torrent_type = {"4k2", "4K原版", "456k", "1024", "2048", "done", "FC2", "hhd800", "other", "高清中文字幕",
|
|
"三级写真", "无码流出", "亚洲有码原创"}
|
|
if path_arr[-2] not in torrent_type and path_arr[-2] not in torrent_sub_type:
|
|
if path_arr[-3] in torrent_sub_type:
|
|
# print(path_arr[-3])
|
|
if path_arr[-4] in torrent_type:
|
|
# print(path_arr[-4])
|
|
pass
|
|
else:
|
|
print(f"有奇怪的东西混入{path_arr}")
|
|
pass
|
|
elif path_arr[-3] in torrent_type:
|
|
if self.search_name in path_arr[-2]:
|
|
print(entry.name)
|
|
# self.result_list.insert(0, entry.path)
|
|
self.torrent_name[path_arr[-2]][path_arr[-3]].add(entry.name)
|
|
pass
|
|
else:
|
|
print(f"有奇怪的东西混入2{path_arr}")
|
|
|
|
def get_by_name(self, name):
|
|
self.search_name = name
|
|
FolderService(paths=self.paths, ignore_dirs=IGNORE_DIRS,
|
|
file_callback=self.get_by_name_file_callback).process_folder()
|
|
return to_json_serializable(self.torrent_name)
|
|
|
|
def extract_name_from_filename(self, filename: str) -> str:
|
|
"""根据torrent文件名提取标准化名称"""
|
|
temp_name = filename.replace("-4K", "")
|
|
without_modifier_file_name = self.prefix_pattern.sub('.', temp_name)
|
|
name, ext = os.path.splitext(without_modifier_file_name)
|
|
name = name.upper()
|
|
name = name.replace("-", "")
|
|
# print(name.translate(str.maketrans({"-": "", " ":""})))
|
|
|
|
for prefix in SPECIAL_PREFIXES:
|
|
if name.startswith(prefix):
|
|
return f"{prefix}-{name[len(prefix):]}"
|
|
|
|
# print(name)
|
|
new_name = []
|
|
for i, char in enumerate(name):
|
|
# print(i, char, not name[i - 1].isnumeric())
|
|
if i > 0 and not name[i - 1].isnumeric() and char.isnumeric():
|
|
new_name.append(f"-{char}")
|
|
else:
|
|
new_name.append(char)
|
|
|
|
return f"{''.join(new_name)}"
|
|
|
|
def get_all_file_callback(self, entry: os.DirEntry):
|
|
# path_arr = entry.path.split(os.sep)
|
|
new_name = self.extract_name_from_filename(entry.name)
|
|
self.result_list.append(new_name)
|
|
|
|
|
|
def get_all(self):
|
|
FolderService(paths=self.paths, ignore_dirs=IGNORE_DIRS,
|
|
file_callback=self.get_all_file_callback).process_folder()
|
|
return self.result_list |