flaskProject/app/services/torrent_service.py

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 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 = name.replace("-", "")
        # 处理特殊前缀
        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)