import os import re from collections import defaultdict from typing import List from app.constants import IGNORE_DIRS from app.services 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)) 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)