| 1 | from helpers.api import ApiHandler, Request, Response |
| 2 | from helpers.backup import BackupService |
| 3 | from typing import Dict, Any |
| 4 | |
| 5 | |
| 6 | class BackupPreviewGrouped(ApiHandler): |
| 7 | @classmethod |
| 8 | def requires_auth(cls) -> bool: |
| 9 | return True |
| 10 | |
| 11 | @classmethod |
| 12 | def requires_loopback(cls) -> bool: |
| 13 | return False |
| 14 | |
| 15 | async def process(self, input: dict, request: Request) -> dict | Response: |
| 16 | try: |
| 17 | # Get input parameters |
| 18 | include_patterns = input.get("include_patterns", []) |
| 19 | exclude_patterns = input.get("exclude_patterns", []) |
| 20 | include_hidden = input.get("include_hidden", True) |
| 21 | max_depth = input.get("max_depth", 3) |
| 22 | search_filter = input.get("search_filter", "") |
| 23 | |
| 24 | # Support legacy string patterns format for backward compatibility |
| 25 | patterns_string = input.get("patterns", "") |
| 26 | if patterns_string and not include_patterns: |
| 27 | lines = [line.strip() for line in patterns_string.split('\n') |
| 28 | if line.strip() and not line.strip().startswith('#')] |
| 29 | for line in lines: |
| 30 | if line.startswith('!'): |
| 31 | exclude_patterns.append(line[1:]) |
| 32 | else: |
| 33 | include_patterns.append(line) |
| 34 | |
| 35 | if not include_patterns: |
| 36 | return { |
| 37 | "success": True, |
| 38 | "groups": [], |
| 39 | "stats": {"total_groups": 0, "total_files": 0, "total_size": 0}, |
| 40 | "total_files": 0, |
| 41 | "total_size": 0 |
| 42 | } |
| 43 | |
| 44 | # Create metadata object for testing |
| 45 | metadata = { |
| 46 | "include_patterns": include_patterns, |
| 47 | "exclude_patterns": exclude_patterns, |
| 48 | "include_hidden": include_hidden |
| 49 | } |
| 50 | |
| 51 | backup_service = BackupService() |
| 52 | all_files = await backup_service.test_patterns(metadata, max_files=10000) |
| 53 | |
| 54 | # Apply search filter if provided |
| 55 | if search_filter.strip(): |
| 56 | search_lower = search_filter.lower() |
| 57 | all_files = [f for f in all_files if search_lower in f["path"].lower()] |
| 58 | |
| 59 | # Group files by directory structure |
| 60 | groups: Dict[str, Dict[str, Any]] = {} |
| 61 | total_size = 0 |
| 62 | |
| 63 | for file_info in all_files: |
| 64 | path = file_info["path"] |
| 65 | total_size += file_info["size"] |
| 66 | |
| 67 | # Split path and limit depth |
| 68 | path_parts = path.strip('/').split('/') |
| 69 | |
| 70 | # Limit to max_depth for grouping |
| 71 | if len(path_parts) > max_depth: |
| 72 | group_path = '/' + '/'.join(path_parts[:max_depth]) |
| 73 | is_truncated = True |
| 74 | else: |
| 75 | group_path = '/' + '/'.join(path_parts[:-1]) if len(path_parts) > 1 else '/' |
| 76 | is_truncated = False |
| 77 | |
| 78 | if group_path not in groups: |
| 79 | groups[group_path] = { |
| 80 | "path": group_path, |
| 81 | "files": [], |
| 82 | "file_count": 0, |
| 83 | "total_size": 0, |
| 84 | "is_truncated": False, |
| 85 | "subdirectories": set() |
| 86 | } |
| 87 | |
| 88 | groups[group_path]["files"].append(file_info) |
| 89 | groups[group_path]["file_count"] += 1 |
| 90 | groups[group_path]["total_size"] += file_info["size"] |
| 91 | groups[group_path]["is_truncated"] = groups[group_path]["is_truncated"] or is_truncated |
| 92 | |
| 93 | # Track subdirectories for truncated groups |
| 94 | if is_truncated and len(path_parts) > max_depth: |
| 95 | next_dir = path_parts[max_depth] |
| 96 | groups[group_path]["subdirectories"].add(next_dir) |
| 97 | |
| 98 | # Convert groups to sorted list and add display info |
| 99 | sorted_groups = [] |
| 100 | for group_path, group_info in sorted(groups.items()): |
| 101 | group_info["subdirectories"] = sorted(list(group_info["subdirectories"])) |
| 102 | |
| 103 | # Limit displayed files for UI performance |
| 104 | if len(group_info["files"]) > 50: |
| 105 | group_info["displayed_files"] = group_info["files"][:50] |
| 106 | group_info["additional_files"] = len(group_info["files"]) - 50 |
| 107 | else: |
| 108 | group_info["displayed_files"] = group_info["files"] |
| 109 | group_info["additional_files"] = 0 |
| 110 | |
| 111 | sorted_groups.append(group_info) |
| 112 | |
| 113 | return { |
| 114 | "success": True, |
| 115 | "groups": sorted_groups, |
| 116 | "stats": { |
| 117 | "total_groups": len(sorted_groups), |
| 118 | "total_files": len(all_files), |
| 119 | "total_size": total_size, |
| 120 | "search_applied": bool(search_filter.strip()), |
| 121 | "max_depth": max_depth |
| 122 | }, |
| 123 | "total_files": len(all_files), |
| 124 | "total_size": total_size |
| 125 | } |
| 126 | |
| 127 | except Exception as e: |
| 128 | return { |
| 129 | "success": False, |
| 130 | "error": str(e) |
| 131 | } |