(self, input: dict, request: Request)
| 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]) |
nothing calls this directly
no test coverage detected