Copy local file with cancellation support.
(
url: str, temp_path: Path, cancel_event: threading.Event
)
| 211 | |
| 212 | |
| 213 | def _copy_file_with_progress( |
| 214 | url: str, temp_path: Path, cancel_event: threading.Event |
| 215 | ) -> DownloadResult: |
| 216 | """Copy local file with cancellation support.""" |
| 217 | try: |
| 218 | parsed_url = urllib.parse.urlparse(url) |
| 219 | |
| 220 | # File URL - copy local file |
| 221 | logger.debug(f"Parsing file URL: {url}") |
| 222 | logger.debug(f"Parsed URL path: {parsed_url.path}") |
| 223 | |
| 224 | # Handle both Unix and Windows file URLs |
| 225 | if os.name == "nt": # Windows |
| 226 | # On Windows, file:///C:/path becomes /C:/path, so remove leading slash |
| 227 | if ( |
| 228 | parsed_url.path.startswith("/") |
| 229 | and len(parsed_url.path) > 3 |
| 230 | and parsed_url.path[2] == ":" |
| 231 | ): |
| 232 | source_path = Path(parsed_url.path[1:]) |
| 233 | else: |
| 234 | source_path = Path(parsed_url.path) |
| 235 | else: # Unix-like |
| 236 | source_path = Path(parsed_url.path) |
| 237 | |
| 238 | logger.debug(f"Resolved file path: {source_path}") |
| 239 | |
| 240 | if not source_path.exists(): |
| 241 | raise FileNotFoundError(f"Source file not found: {source_path}") |
| 242 | |
| 243 | # Check for cancellation before copy |
| 244 | if cancel_event.is_set(): |
| 245 | cancelled_error = RuntimeError(f"Copy cancelled for {url}") |
| 246 | logger.warning(f"Copy cancelled for {url}", exc_info=cancelled_error) |
| 247 | return DownloadResult(url, temp_path, cancelled_error) |
| 248 | |
| 249 | shutil.copy2(source_path, temp_path) |
| 250 | return DownloadResult(url, temp_path) # Success |
| 251 | |
| 252 | except KeyboardInterrupt as e: |
| 253 | # Set cancel event and interrupt main thread |
| 254 | cancel_event.set() |
| 255 | _thread.interrupt_main() |
| 256 | logger.warning(f"Copy interrupted by KeyboardInterrupt for {url}", exc_info=e) |
| 257 | return DownloadResult(url, temp_path, e) |
| 258 | except Exception as e: |
| 259 | if not cancel_event.is_set(): |
| 260 | logger.error(f"Copy failed for {url}: {e}", exc_info=e) |
| 261 | else: |
| 262 | logger.warning(f"Copy failed for {url} (cancelled): {e}", exc_info=e) |
| 263 | return DownloadResult(url, temp_path, e) |
| 264 | |
| 265 | |
| 266 | def clear_session_cache() -> None: |
no test coverage detected