target s3

Author: CyMule

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

@@ -169,18 +169,25 @@ def run_expected_download_files_validation(
 
 def run_directory_structure_validation(expected_output_dir: Path, download_files: list[str]):
     s3_keys_file = expected_output_dir / "expected_s3_keys.json"
-    with s3_keys_file.open("r") as f:
-        s3_keys = json.load(f)["s3_keys"]
 
-    expected_paths = []
-    for s3_key in s3_keys:
-        hash_based_path = generate_hash_based_path(s3_key)
-        if hash_based_path:
-            expected_paths.append(str(hash_based_path))
-    
-    expected_paths.sort()
-    download_files.sort()
-    assert expected_paths == download_files
+    if s3_keys_file.exists():
+        with s3_keys_file.open("r") as f:
+            s3_keys = json.load(f)["s3_keys"]
+        
+        expected_paths = []
+        for s3_key in s3_keys:
+            hash_based_path = generate_hash_based_path(s3_key)
+            if hash_based_path:
+                expected_paths.append(str(hash_based_path))
+        
+        expected_paths.sort()
+        download_files.sort()
+        assert expected_paths == download_files
+    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/interfaces/downloader.py

@@ -8,7 +8,6 @@
 from unstructured_ingest.data_types.file_data import FileData
 from unstructured_ingest.interfaces.connector import BaseConnector
 from unstructured_ingest.interfaces.process import BaseProcess
-from unstructured_ingest.utils.filesystem import generate_hash_based_path
 
 
 class DownloaderConfig(BaseModel):
@@ -42,7 +41,8 @@ def get_download_path(self, file_data: FileData) -> Optional[Path]:
         if not rel_path:
             return None
 
-        return generate_hash_based_path(rel_path, self.download_dir)
+        rel_path = rel_path[1:] if rel_path.startswith("/") else rel_path
+        return self.download_dir / Path(rel_path)
 
     @staticmethod
     def is_float(value: str):

unstructured_ingest/processes/connectors/fsspec/fsspec.py

@@ -29,7 +29,7 @@
     UploaderConfig,
 )
 from unstructured_ingest.processes.connectors.fsspec.utils import sterilize_dict
-from unstructured_ingest.utils.filesystem import mkdir_concurrent_safe
+from unstructured_ingest.utils.filesystem import generate_hash_based_path, mkdir_concurrent_safe
 
 if TYPE_CHECKING:
     from fsspec import AbstractFileSystem
@@ -270,6 +270,16 @@ class FsspecDownloader(Downloader):
     download_config: Optional[FsspecDownloaderConfigT] = field(
         default_factory=lambda: FsspecDownloaderConfig()
     )
+    
+    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
+        
+        return generate_hash_based_path(rel_path, self.download_dir)
 
     def is_async(self) -> bool:
         with self.connection_config.get_client(protocol=self.protocol) as client: