busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: Implement recursive storage listing and directory browsing for backups, and add a migration option to fix cross-filters.

Browse Source
This commit is contained in:
busya
2026-02-25 20:01:33 +03:00
parent 1d8eadf796
commit 82d2cb9fe3
12 changed files with 533 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
backend/src/plugins/storage/plugin.py
View File

@@ -212,13 +212,21 @@ class StoragePlugin(PluginBase):
# @PURPOSE: Lists all files and directories in a specific category and subpath.
# @PARAM: category (Optional[FileCategory]) - The category to list.
# @PARAM: subpath (Optional[str]) - Nested path within the category.
# @PARAM: recursive (bool) - Whether to scan nested subdirectories recursively.
# @PRE: Storage root must exist.
# @POST: Returns a list of StoredFile objects.
# @RETURN: List[StoredFile] - List of file and directory metadata objects.
def list_files(self, category: Optional[FileCategory] = None, subpath: Optional[str] = None) -> List[StoredFile]:
def list_files(
self,
category: Optional[FileCategory] = None,
subpath: Optional[str] = None,
recursive: bool = False,
) -> List[StoredFile]:
with belief_scope("StoragePlugin:list_files"):
root = self.get_storage_root()
logger.info(f"[StoragePlugin][Action] Listing files in root: {root}, category: {category}, subpath: {subpath}")
logger.info(
f"[StoragePlugin][Action] Listing files in root: {root}, category: {category}, subpath: {subpath}, recursive: {recursive}"
)
files = []
categories = [category] if category else list(FileCategory)
@@ -235,17 +243,37 @@ class StoragePlugin(PluginBase):
continue
logger.debug(f"[StoragePlugin][Action] Scanning directory: {target_dir}")
if recursive:
for current_root, dirs, filenames in os.walk(target_dir):
dirs[:] = [d for d in dirs if "Logs" not in d]
for filename in filenames:
file_path = Path(current_root) / filename
if "Logs" in str(file_path):
continue
stat = file_path.stat()
files.append(
StoredFile(
name=filename,
path=str(file_path.relative_to(root)),
size=stat.st_size,
created_at=datetime.fromtimestamp(stat.st_ctime),
category=cat,
mime_type=None,
)
)
continue
# Use os.scandir for better performance and to distinguish files vs dirs
with os.scandir(target_dir) as it:
for entry in it:
# Skip logs
if "Logs" in entry.path:
continue
stat = entry.stat()
is_dir = entry.is_dir()
files.append(StoredFile(
name=entry.name,
path=str(Path(entry.path).relative_to(root)),
@@ -341,4 +369,4 @@ class StoragePlugin(PluginBase):
# [/DEF:get_file_path:Function]
# [/DEF:StoragePlugin:Class]
# [/DEF:StoragePlugin:Module]
# [/DEF:StoragePlugin:Module]