Handles input resolution for different source types. Normalizes all inputs to a local directory path for scanning.
| 82 | |
| 83 | |
| 84 | class InputHandler: |
| 85 | """ |
| 86 | Handles input resolution for different source types. |
| 87 | |
| 88 | Normalizes all inputs to a local directory path for scanning. |
| 89 | """ |
| 90 | |
| 91 | def __init__(self) -> None: |
| 92 | self._temp_dir: Path | None = None |
| 93 | |
| 94 | def resolve(self, input_path: str) -> tuple[Path, str]: |
| 95 | """ |
| 96 | Resolve input to a scannable directory. |
| 97 | |
| 98 | Args: |
| 99 | input_path: Path or URL to resolve |
| 100 | |
| 101 | Returns: |
| 102 | Tuple of (resolved_path, source_type) |
| 103 | source_type is one of: "git", "url", "zip", "file", "directory" |
| 104 | |
| 105 | Raises: |
| 106 | ValueError: If input type cannot be determined |
| 107 | FileNotFoundError: If local path doesn't exist |
| 108 | """ |
| 109 | input_path = input_path.strip() |
| 110 | |
| 111 | if self._is_git_url(input_path): |
| 112 | return self._clone_git(input_path), "git" |
| 113 | if self._is_file_url(input_path): |
| 114 | return self._download_file(input_path), "url" |
| 115 | if input_path.endswith(".zip"): |
| 116 | return self._extract_zip(Path(input_path)), "zip" |
| 117 | if input_path.endswith(".md"): |
| 118 | return self._wrap_single_file(Path(input_path)), "file" |
| 119 | if Path(input_path).is_dir(): |
| 120 | return Path(input_path).resolve(), "directory" |
| 121 | if Path(input_path).is_file(): |
| 122 | return self._wrap_single_file(Path(input_path)), "file" |
| 123 | raise ValueError( |
| 124 | f"Cannot determine input type for: {input_path}\n" |
| 125 | "Supported formats: Git URL, file URL, .zip file, .md file, or directory" |
| 126 | ) |
| 127 | |
| 128 | def cleanup(self) -> None: |
| 129 | """Clean up temporary files created during resolution.""" |
| 130 | if self._temp_dir and self._temp_dir.exists(): |
| 131 | shutil.rmtree(self._temp_dir, ignore_errors=True) |
| 132 | self._temp_dir = None |
| 133 | |
| 134 | def temp_dir_for_cleanup(self) -> Path | None: |
| 135 | """Return the temp directory path if one was created (for caller to clean up after graph).""" |
| 136 | return self._temp_dir |
| 137 | |
| 138 | def _get_temp_dir(self) -> Path: |
| 139 | """Get or create a temporary directory for this session.""" |
| 140 | if not self._temp_dir: |
| 141 | self._temp_dir = Path(tempfile.mkdtemp(prefix="skillspector_")) |
no outgoing calls