Fix local Windows path computation errors

- simplify URI analysis
- fix URI usage in the runner

Author: popescu-v

khiops/core/internals/filesystems.py

@@ -8,7 +8,6 @@
 
 import json
 import os
-import platform
 import shutil
 import warnings
 from abc import ABC, abstractmethod
@@ -59,8 +58,8 @@ def is_local_resource(uri_or_path):
     `bool`
         `True` if a URI refers to a local path
     """
-    uri_info = urlparse(uri_or_path, allow_fragments=False)
-    return len(uri_info.scheme) <= 1 or uri_info.scheme == "file"
+    # Check if Path is local Windows path
+    return isinstance(create_resource(uri_or_path), LocalFilesystemResource)
 
 
 def create_resource(uri_or_path):
@@ -80,15 +79,30 @@ def create_resource(uri_or_path):
     `FilesystemResource`
         The URI resource object, its class depends on the URI.
     """
-    uri_info = urlparse(uri_or_path, allow_fragments=False)
-    if uri_info.scheme == "s3":
-        return AmazonS3Resource(uri_or_path)
-    elif uri_info.scheme == "gs":
-        return GoogleCloudStorageResource(uri_or_path)
-    elif is_local_resource(uri_or_path):
-        return LocalFilesystemResource(uri_or_path)
+    if (index := uri_or_path.find("://")) > 0:
+        scheme = uri_or_path[:index]
+        if len(scheme) > 1:  # any 1-char scheme is considered a windows path
+            uri_info = urlparse(uri_or_path, allow_fragments=False)
+            if uri_info.scheme == "s3":
+                return AmazonS3Resource(uri_or_path)
+            elif uri_info.scheme == "gs":
+                return GoogleCloudStorageResource(uri_or_path)
+            else:
+                raise ValueError(f"Unsupported URI scheme {uri_info.scheme}")
+        elif scheme == "file":
+            uri_info = urlparse(uri_or_path, allow_fragments=False)
+
+            # Reject URI if authority is not empty
+            if uri_info.netloc:
+                raise ValueError(
+                    f"Non-empty 'autority' in local-path URI '{uri_or_path}': "
+                    f"'{uri_info.netloc}'"
+                )
+            return LocalFilesystemResource(uri_or_path)
+        else:
+            return LocalFilesystemResource(uri_or_path)
     else:
-        raise ValueError(f"Unsupported URI scheme {uri_info.scheme}")
+        return LocalFilesystemResource(uri_or_path)
 
 
 def parent_path(path):
@@ -411,12 +425,8 @@ def __init__(self, uri):
         # Obtain the local from the URI
         # Case where the scheme is in fact a windows drive
         #   => Build the proper path with drive
-        if (
-            len(self.uri_info.scheme) == 1
-            and self.uri_info.scheme.isalpha()
-            and platform.system() == "Windows"
-        ):
-            self.path = os.path.join(f"{self.uri_info.scheme}:\\", self.uri_info.path)
+        if len(self.uri_info.scheme) == 1 and self.uri_info.scheme.isalpha():
+            self.path = f"{self.uri_info.scheme}:{self.uri_info.path}"
         # Case of the "file" scheme
         elif self.uri_info.scheme == "file":
             # If invalid second colon in path (eg. "/C:/Users"):

khiops/core/internals/runner.py

@@ -59,7 +59,9 @@ def _get_dir_status(a_dir):
         # Remove initial slash on windows systems
         # urllib's url2pathname does not work properly
         a_dir_res = fs.create_resource(os.path.normpath(a_dir))
-        a_dir_path = a_dir_res.uri_info.path
+
+        # a_dir_res is a LocalFilesystemResource already
+        a_dir_path = a_dir_res.path
         if platform.system() == "Windows":
             if a_dir_path.startswith("/"):
                 a_dir_path = a_dir_path[1:]
@@ -98,31 +100,6 @@ def _check_samples_dir(samples_dir):
         )
 
 
-def _extract_path_from_uri(uri):
-    res = fs.create_resource(uri)
-    if platform.system() == "Windows":
-        # Case of file:///<LETTER>:/<REST_OF_PATH>:
-        #   Eliminate first slash ("/") from path if the first component
-        if (
-            res.uri_info.scheme == ""
-            and res.uri_info.path[0] == "/"
-            and res.uri_info.path[1].isalpha()
-            and res.uri_info.path[2] == ":"
-        ):
-            path = res.uri_info.path[1:]
-        # Case of C:/<REST_OF_PATH>:
-        #   Just use the original path
-        elif len(res.uri_info.scheme) == 1:
-            path = uri
-        # Otherwise return URI path as-is
-        else:
-            path = res.uri_info.path
-
-    else:
-        path = res.uri_info.path
-    return path
-
-
 def _khiops_env_file_exists(env_dir):
     """Check ``khiops_env`` exists relative to the specified environment dir"""
     khiops_env_path = os.path.join(env_dir, "khiops_env")
@@ -399,7 +376,7 @@ def root_temp_dir(self):
     def root_temp_dir(self, dir_path):
         # Check existence, directory status and permissions for local paths
         if fs.is_local_resource(dir_path):
-            real_dir_path = _extract_path_from_uri(dir_path)
+            real_dir_path = fs.create_resource(dir_path).path
             if os.path.exists(real_dir_path):
                 if os.path.isfile(real_dir_path):
                     raise KhiopsEnvironmentError(
@@ -439,7 +416,7 @@ def create_temp_file(self, prefix, suffix):
         # Local resource: Effectively create the file with the python file API
         if fs.is_local_resource(self.root_temp_dir):
             # Extract the path from the potential URI
-            root_temp_dir_path = _extract_path_from_uri(self.root_temp_dir)
+            root_temp_dir_path = fs.create_resource(self.root_temp_dir).path
 
             # Create the temporary file
             tmp_file_fd, tmp_file_path = tempfile.mkstemp(
@@ -470,7 +447,7 @@ def create_temp_dir(self, prefix):
         """
         # Local resource: Effectively create the directory with the python file API
         if fs.is_local_resource(self.root_temp_dir):
-            root_temp_dir_path = _extract_path_from_uri(self.root_temp_dir)
+            root_temp_dir_path = fs.create_resource(self.root_temp_dir).path
             temp_dir = tempfile.mkdtemp(prefix=prefix, dir=root_temp_dir_path)
         # Remote resource: Just return a highly probable unique path
         else: