fix: prevent S3 path conflicts using tempfile (#569)

## Problem

S3 downloads were sometimes failing with `NotADirectoryError` and
`FileExistsError` when S3 buckets contained objects with conflicting
naming patterns that cannot be represented in traditional filesystem
hierarchies.

**Example conflict:**
- S3 object: `foo` (file)
- S3 object: `foo/documents` (file requiring foo to be a directory)

This created a race condition where download order determined
success/failure

## Solution

Used tempfile to create unique download paths for each S3 object:

**Before:**
```
S3: "foo" → Local: /downloads/foo
S3: "foo/documents" → Local: /downloads/foo/documents
Conflict: foo cannot be both file and directory
```

**After:**
```
S3: "foo" → Local: /downloads/a1b2c3d4e5f6/foo
S3: "foo/documents" → Local: /downloads/9g8h7i6j5k4l/documents
No conflicts: Each file gets unique directory
```

## Future Work

This PR targets only the s3 downloads. I think it would make sense to
use tempfiles for all downloads (as in [PR
#571](#571)),
but that requires more extensive changes to implement cleanly. This fix
provides immediate relief from the path conflict issues while we work on
the more comprehensive tempfile solution.

Author: CyMule

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 1.2.2
+
+* **Fix**: prevent S3 path conflicts using tempfile for directory isolation
+
 ## 1.2.1
 
 * **Fix**: Embeddings are properly assigned when embedding in batches

test/integration/connectors/expected_results/s3-minio/directory_structure.json renamed to test/integration/connectors/expected_results/s3-minio/expected_s3_keys.json

@@ -1,5 +1,5 @@
 {
-  "directory_structure": [
+  "s3_keys": [
     "wiki_movie_plots_small.csv"
   ]
 }

test/integration/connectors/expected_results/s3-specialchar/directory_structure.json renamed to test/integration/connectors/expected_results/s3-specialchar/expected_s3_keys.json

@@ -1,5 +1,5 @@
 {
-  "directory_structure": [
+  "s3_keys": [
     "Why_is_the_sky_blue?.txt",
     "[test]?*.txt"
   ]

test/integration/connectors/expected_results/s3/directory_structure.json renamed to test/integration/connectors/expected_results/s3/expected_s3_keys.json

@@ -1,5 +1,5 @@
 {
-  "directory_structure": [
+  "s3_keys": [
     "2023-Jan-economic-outlook.pdf",
     "Silent-Giant-(1).pdf",
     "page-with-formula.pdf",

test/integration/connectors/utils/validation/source.py

@@ -167,11 +167,24 @@ def run_expected_download_files_validation(
 
 
 def run_directory_structure_validation(expected_output_dir: Path, download_files: list[str]):
-    directory_record = expected_output_dir / "directory_structure.json"
-    with directory_record.open("r") as directory_file:
-        directory_file_contents = json.load(directory_file)
-    directory_structure = directory_file_contents["directory_structure"]
-    assert directory_structure == download_files
+    s3_keys_file = expected_output_dir / "expected_s3_keys.json"
+    
+    if s3_keys_file.exists():
+        with s3_keys_file.open("r") as f:
+            s3_keys = json.load(f)["s3_keys"]
+        
+        expected_filenames = {Path(s3_key).name for s3_key in s3_keys}
+        actual_filenames = {Path(download_file).name for download_file in download_files}
+        
+        assert expected_filenames == actual_filenames, (
+            f"Expected filenames: {sorted(expected_filenames)}, "
+            f"Got filenames: {sorted(actual_filenames)}"
+        )
+    else:
+        directory_record = expected_output_dir / "directory_structure.json"
+        with directory_record.open("r") as f:
+            directory_structure = json.load(f)["directory_structure"]
+        assert directory_structure == download_files
 
 
 def update_fixtures(

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "1.2.1"  # pragma: no cover
+__version__ = "1.2.2"  # pragma: no cover

unstructured_ingest/interfaces/downloader.py

@@ -36,9 +36,11 @@ class Downloader(BaseProcess, BaseConnector, ABC):
     def get_download_path(self, file_data: FileData) -> Optional[Path]:
         if not file_data.source_identifiers:
             return None
+        
         rel_path = file_data.source_identifiers.relative_path
         if not rel_path:
             return None
+        
         rel_path = rel_path[1:] if rel_path.startswith("/") else rel_path
         return self.download_dir / Path(rel_path)
 

unstructured_ingest/processes/connectors/fsspec/fsspec.py

@@ -264,12 +264,31 @@ class FsspecDownloaderConfig(DownloaderConfig):
 
 @dataclass
 class FsspecDownloader(Downloader):
+    TEMP_DIR_PREFIX = "unstructured_"
+    
     protocol: str
     connection_config: FsspecConnectionConfigT
     connector_type: str = CONNECTOR_TYPE
     download_config: Optional[FsspecDownloaderConfigT] = field(
         default_factory=lambda: FsspecDownloaderConfig()
     )
+    
+    def get_download_path(self, file_data: FileData) -> Optional[Path]:
+        has_source_identifiers = file_data.source_identifiers is not None
+        has_filename = has_source_identifiers and file_data.source_identifiers.filename
+        
+        if not (has_source_identifiers and has_filename):
+            return None
+        
+        filename = file_data.source_identifiers.filename
+        
+        mkdir_concurrent_safe(self.download_dir)
+        
+        temp_dir = tempfile.mkdtemp(
+            prefix=self.TEMP_DIR_PREFIX, 
+            dir=self.download_dir
+        )
+        return Path(temp_dir) / filename
 
     def is_async(self) -> bool:
         with self.connection_config.get_client(protocol=self.protocol) as client: