YanLongChangAn
6cbacab98b
- 新增 asmr_bp 蓝图模块和 AsmrService 服务 - 新增 magnet_bp 蓝图模块和 MagnetService 服务 - 在应用初始化时注册新的蓝图模块 - 将 switch 语句替换为 if-elif 语句进行配置管理 - 在前端界面添加 ASMR 功能入口 - 增加 libtorrent 依赖包 - 修复字符串引号使用问题 - 增加文件重命名异常处理 - 扩展类型注解支持 Union 类型 - 增加请求超时时间到 30 秒 - 优化前端错误处理逻辑
47 lines
1.8 KiB
Python
47 lines
1.8 KiB
Python
import os
|
|
from typing import Optional, Callable, List, Union
|
|
|
|
|
|
class FolderService:
|
|
def __init__(
|
|
self,
|
|
paths: Union[List[str], str],
|
|
ignore_dirs: Optional[set] = None,
|
|
folder_callback: Optional[Callable] = None,
|
|
file_callback: Optional[Callable] = None,
|
|
empty_folder_callback: Optional[Callable] = None,
|
|
max_depth: Optional[int] = None
|
|
):
|
|
self.paths: List[str] = paths if type(paths) is not str else [paths]
|
|
self.ignore_dirs = ignore_dirs or set()
|
|
self.folder_callback = folder_callback
|
|
self.file_callback = file_callback
|
|
self.empty_folder_callback = empty_folder_callback
|
|
self.max_depth = max_depth
|
|
|
|
def _traverse_folder(self, path: Optional[str] = None, depth: Optional[int] = 0) -> None:
|
|
"""处理目录"""
|
|
for entry in os.scandir(path):
|
|
if entry.name in self.ignore_dirs:
|
|
continue
|
|
if entry.is_dir():
|
|
# 判断要加载的文件夹深度
|
|
if self.max_depth is not None and depth >= self.max_depth:
|
|
return
|
|
# 循环加载
|
|
self._traverse_folder(entry.path, depth + 1)
|
|
# 判断是空文件夹
|
|
if not os.listdir(entry.path) and self.empty_folder_callback:
|
|
self.empty_folder_callback(entry)
|
|
# 文件夹处理的回调
|
|
if self.folder_callback:
|
|
self.folder_callback(entry)
|
|
elif entry.is_file():
|
|
if self.file_callback:
|
|
self.file_callback(entry)
|
|
|
|
def process_folder(self) -> None:
|
|
for path in self.paths:
|
|
if os.path.exists(path):
|
|
self._traverse_folder(path)
|