code: def upload_working_dir_if_needed(
      runtime_env: Dict[str, Any],
      scratch_dir: Optional[str] = os.getcwd(),
      logger: Optional[logging.Logger] = default_logger,
      upload_fn=None,
  ) -> Dict[str, Any]:
      """Uploads the working_dir and replaces it with a URI.
  
      If the working_dir is already a URI, this is a no-op.
      """
      working_dir = runtime_env.get("working_dir")
      if working_dir is None:
          return runtime_env
  
      if not isinstance(working_dir, str) and not isinstance(
          working_dir, Path
      ):
          raise TypeError(
              "working_dir must be a string or Path (either a local path "
              f"or remote URI), got {type(working_dir)}."
          )
  
      if isinstance(working_dir, Path):
          working_dir = str(working_dir)
  
      # working_dir is already a URI -- just pass it through.
      try:
          protocol, path = parse_uri(working_dir)
      except ValueError:
          protocol, path = None, None
  
      if protocol is not None:
          if (
              protocol in Protocol.remote_protocols()
              and not path.endswith(".zip")
          ):
              raise ValueError(
                  "Only .zip files supported for remote URIs."
              )
          return runtime_env
  
      excludes = runtime_env.get("excludes", None)
      try:
          working_dir_uri = get_uri_for_directory(
              working_dir, excludes=excludes
          )
      except ValueError:  # working_dir is not a directory
          package_path = Path(working_dir)
          if not package_path.exists() or package_path.suffix != ".zip":
              raise ValueError(
                  f"directory {package_path} must be an existing "
                  "directory or a zip package"
              )
  
          pkg_uri = get_uri_for_package(package_path)
          try:
              upload_package_to_gcs(pkg_uri, package_path.read_bytes())
          except Exception as e:
              raise RuntimeEnvSetupError(
                  f"Failed to upload package {package_path} to the Ray cluster: {e}"
              ) from e
          runtime_env["working_dir"] = pkg_uri
          return runtime_env
      if upload_fn is None:
          try:
              upload_package_if_needed(
                  working_dir_uri,
                  scratch_dir,
                  working_dir,
                  include_parent_dir=False,
                  excludes=excludes,
                  logger=logger,
              )
          except Exception as e:
              raise RuntimeEnvSetupError(
                  f"Failed to upload working_dir {working_dir} to the Ray cluster: {e}"
              ) from e
      else:
          upload_fn(working_dir, excludes=excludes)
  
      runtime_env["working_dir"] = working_dir_uri
      return runtime_env
  
file_path: ray/python/ray/_private/runtime_env/working_dir.py
imports: import logging
  import os
  from pathlib import Path
  from typing import Any, Dict, List, Optional
  from ray._private.runtime_env.context import RuntimeEnvContext
  from ray._private.runtime_env.packaging import Protocol, delete_package, download_and_unpack_package, get_local_dir_from_uri, get_uri_for_directory, get_uri_for_package, parse_uri, upload_package_if_needed, upload_package_to_gcs
  from ray._private.runtime_env.plugin import RuntimeEnvPlugin
  from ray._private.utils import get_directory_size_bytes, try_to_create_directory
  from ray.exceptions import RuntimeEnvSetupError
label_desc: Use image node_id 1 for calling a global function with example usage: upload_working_dir_if_needed(runtime_env, scratch_dir, logger, upload_fn) and returns: runtime_env, runtime_env, runtime_env, runtime_env