Added replace_objects and replace_container_contents options

Fixes #2 and #15

Signed-off-by: MaxGelbakhiani <[email protected]>

action.yml

@@ -35,6 +35,14 @@ inputs:
     description: Number of epochs for object to stay valid
     required: false
     default: 0
+  REPLACE_OBJECTS:
+    description: Replace existing objects with the same attributes in the container
+    required: false
+    default: 'True'
+  REPLACE_CONTAINER_CONTENTS:
+    description: Remove all the old existing objects in the container after the new objects are uploaded
+    required: false
+    default: 'False'
 
 outputs:
   OUTPUT_CONTAINER_URL:
@@ -93,11 +101,14 @@ runs:
         NEOFS_ATTRIBUTES: ${{ inputs.NEOFS_ATTRIBUTES }}
         URL_PREFIX: ${{ inputs.URL_PREFIX }}
         LIFETIME: ${{ inputs.LIFETIME }}
+        REPLACE_OBJECTS: ${{ inputs.REPLACE_OBJECTS }}
+        REPLACE_CONTAINER_CONTENTS: ${{ inputs.REPLACE_CONTAINER_CONTENTS }}
         GITHUB_ACTION_PATH: ${{ github.action_path }}
       run: |
         source "$GITHUB_ACTION_PATH/venv/bin/activate" && NEOFS_CLI_PASSWORD=$NEOFS_WALLET_PASSWORD python "$GITHUB_ACTION_PATH/push-to-neofs.py" \
         --lifetime "$LIFETIME" --neofs_domain "$NEOFS_NETWORK_DOMAIN" --attributes "$NEOFS_ATTRIBUTES" \
         --cid "$STORE_OBJECTS_CID" --files-dir "$PATH_TO_FILES_DIR" --url_path_prefix "$URL_PREFIX" \
+        --replace-objects "$REPLACE_OBJECTS" --replace-container-contents "$REPLACE_CONTAINER_CONTENTS" \
         --wallet "$GITHUB_ACTION_PATH/wallet.json"
         BASE_URL="https://$NEOFS_HTTP_GATE/$STORE_OBJECTS_CID"
         if [ -z "$URL_PREFIX" ]; then

push-to-neofs.py

@@ -1,14 +1,25 @@
 import os
+import re
 import subprocess
 import argparse
 import magic
+import json
+import distutils.util
 
 FILE_PATH = "FilePath"  # the key for the attribute, is the path for the static page and allure report zip files
 CONTENT_TYPE = "ContentType"
 NEOFS_WALLET_PASSWORD_ENV_NAME = "NEOFS_WALLET_PASSWORD"
 PORT_8080 = 8080
 
 
+def str_to_bool(value):
+    """Convert a string representation of a boolean value to a boolean."""
+    try:
+        return bool(distutils.util.strtobool(value))
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"Invalid boolean value: {value}")
+
+
 def parse_args():
     parser = argparse.ArgumentParser(description="Process allure reports")
     parser.add_argument(
@@ -69,6 +80,20 @@ def parse_args():
         help="Timeout for the put each file to neofs, in seconds. Default is 600 seconds",
         default=600,
     )
+    parser.add_argument(
+        "--replace-objects",
+        required=False,
+        type=str_to_bool,
+        help="Replace existing objects with the same attributes in the container",
+        default=True,
+    )
+    parser.add_argument(
+        "--replace-container-contents",
+        required=False,
+        type=str_to_bool,
+        help="Remove all the old existing objects in the container after the new objects are uploaded",
+        default=False,
+    )
 
     return parser.parse_args()
 
@@ -88,45 +113,34 @@ def get_rpc_endpoint(neofs_domain: str) -> str:
     return f"{neofs_domain}:{PORT_8080}"
 
 
-def push_file(
-    directory: str,
-    subdir: str,
-    url_path_prefix: str,
-    filename: str,
-    attributes: str,
-    base_cmd: str,
-    put_timeout: int,
-) -> None:
-    filepath = os.path.join(subdir, filename)
-    mime_type = magic.from_file(filepath, mime=True)
-    relative_path = os.path.relpath(filepath, os.path.dirname(directory))
-
-    if url_path_prefix is not None and url_path_prefix != "":
-        neofs_path_attr = os.path.join(url_path_prefix, relative_path)
-    else:
-        neofs_path_attr = relative_path
-
-    base_cmd_with_file = f"{base_cmd} --file {filepath} --attributes {FILE_PATH}={neofs_path_attr},{CONTENT_TYPE}={mime_type}"
-
-    if attributes is not None and attributes != "":
-        base_cmd_with_file += f",{attributes}"
+def neofs_cli_execute(cmd: str, json_output: bool = False, timeout: int = None):
+    """
+    Executes a given command and returns its output.
 
-    print(f"Neofs cli cmd is: {base_cmd_with_file}")
+    :param cmd: Command to execute.
+    :param json_output: Specifies if the command output is JSON.
+    :param timeout: Optional timeout for command execution.
+    :return: Command output as a string or a JSON object.
+    """
 
     try:
         compl_proc = subprocess.run(
-            base_cmd_with_file,
+            cmd,
             check=True,
             text=True,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
-            timeout=put_timeout,
+            timeout=timeout,
             shell=True,
         )
 
         print(f"RC: {compl_proc.returncode}")
         print(f"Output: {compl_proc.stdout}")
-        print(f"Error: {compl_proc.stderr}")
+
+        if json_output:
+            return json.loads(compl_proc.stdout)
+        else:
+            return compl_proc.stdout.splitlines()
 
     except subprocess.CalledProcessError as e:
         raise Exception(
@@ -138,6 +152,117 @@ def push_file(
         )
 
 
+def search_objects_in_container(endpoint: str,
+                                wallet: str,
+                                password: str,
+                                cid: str,
+                                filters: str) -> list[str]:
+    cmd = (
+        f"NEOFS_CLI_PASSWORD={password} neofs-cli --rpc-endpoint {endpoint} "
+        f"--wallet {wallet} object search --cid {cid} --filters '{filters}'"
+    )
+    output_filter_re = re.compile(r"^Found \d+ objects\.$")
+    stdout_list = neofs_cli_execute(cmd)
+    filtered_lines = [line for line in stdout_list if not output_filter_re.search(line)]
+    return filtered_lines
+
+
+def list_objects_in_container(endpoint: str,
+                              wallet: str,
+                              password: str,
+                              cid: str) -> list[str]:
+    cmd = (
+        f"NEOFS_CLI_PASSWORD={password} neofs-cli --rpc-endpoint {endpoint} "
+        f"--wallet {wallet} container list-objects --cid {cid}"
+    )
+    return neofs_cli_execute(cmd)
+
+
+def delete_objects(
+    endpoint: str,
+    wallet: str,
+    password: str,
+    cid: str,
+    oids: list[str],
+) -> None:
+    for oid in oids:
+        cmd = (
+            f"NEOFS_CLI_PASSWORD={password} neofs-cli --rpc-endpoint {endpoint} "
+            f"--wallet {wallet} object delete --cid {cid} --oid '{oid}'"
+        )
+        neofs_cli_execute(cmd)
+
+
+def compile_attributes(file_path: str, content_type: str = None,
+                       attributes: str = None, output_format: str = "str") -> str:
+    attrs = {
+        FILE_PATH: file_path,
+    }
+    if content_type:
+        attrs[CONTENT_TYPE] = content_type
+    if attributes:
+        attrs.update(dict([attr.split('=') for attr in attributes.split(',')]))
+    if output_format == "str":
+        return ','.join([f"{k}={v}" for k, v in attrs.items()])
+    elif output_format == "filter_str":
+        return ','.join([f"{k} EQ {v}" for k, v in attrs.items()])
+    elif output_format == "dict":
+        return attrs
+
+
+def get_file_info(directory: str, url_path_prefix: str):
+    base_path = os.path.abspath(directory)
+    file_infos = []
+
+    for subdir, dirs, files in os.walk(base_path):
+        for filename in files:
+            filepath = os.path.join(subdir, filename)
+            mime_type = magic.from_file(filepath, mime=True)
+            relative_path = os.path.relpath(filepath, os.path.dirname(directory))
+
+            if url_path_prefix is not None and url_path_prefix != "":
+                neofs_path_attr = os.path.join(url_path_prefix, relative_path)
+            else:
+                neofs_path_attr = relative_path
+
+            file_infos.append({
+                'filepath': filepath,
+                'mime_type': mime_type,
+                'neofs_path_attr': neofs_path_attr,
+            })
+
+    return file_infos
+
+
+def push_file(
+    endpoint: str,
+    wallet: str,
+    password: str,
+    cid: str,
+    file_info: dict,
+    attributes: str,
+    put_timeout: int,
+    expiration_epoch: int = None,
+) -> None:
+    filepath = file_info['filepath']
+    mime_type = file_info['mime_type']
+    neofs_path_attr = file_info['neofs_path_attr']
+
+    attrs = compile_attributes(neofs_path_attr, mime_type, attributes)
+
+    base_cmd = (
+        f"NEOFS_CLI_PASSWORD={password} neofs-cli --rpc-endpoint {endpoint} "
+        f"--wallet {wallet}  object put --cid {cid} --timeout {put_timeout}s"
+    )
+    if expiration_epoch:
+        base_cmd += f" --expire-at {expiration_epoch}"
+
+    cmd = f"{base_cmd} --file {filepath} --attributes {attrs}"
+    print(f"Neofs cli cmd is: {cmd}")
+
+    neofs_cli_execute(cmd, timeout=put_timeout)
+
+
 def push_files_to_neofs(
     directory: str,
     endpoint: str,
@@ -148,33 +273,49 @@ def push_files_to_neofs(
     lifetime: int,
     put_timeout: int,
     password: str,
+    replace_objects: bool,
+    replace_container_contents: bool
 ) -> None:
     if not os.path.exists(directory):
         raise Exception(f"Directory '{directory}' does not exist.")
     if not os.listdir(directory):
         raise Exception(f"Directory '{directory}' is empty.")
 
-    base_cmd = (
-        f"NEOFS_CLI_PASSWORD={password} neofs-cli --rpc-endpoint {endpoint} "
-        f"--wallet {wallet}  object put --cid {cid} --timeout {put_timeout}s"
-    )
     if lifetime is not None and lifetime > 0:
         current_epoch = get_current_epoch(endpoint)
         expiration_epoch = current_epoch + lifetime
-        base_cmd += f" --expire-at {expiration_epoch}"
 
-    base_path = os.path.abspath(directory)
-    for subdir, dirs, files in os.walk(base_path):
-        for filename in files:
-            push_file(
-                base_path,
-                subdir,
-                url_path_prefix,
-                filename,
-                attributes,
-                base_cmd,
-                put_timeout,
+    files = get_file_info(directory, url_path_prefix)
+    flat_existing_objects = []
+    if replace_container_contents:
+        flat_existing_objects = list_objects_in_container(endpoint, wallet, password, cid)
+    elif replace_objects:
+        existing_objects = []
+        for file in files:
+            search_attrs = compile_attributes(
+                file['neofs_path_attr'], output_format="filter_str"
             )
+            obj_to_delete = search_objects_in_container(
+                endpoint, wallet, password, cid, search_attrs
+            )
+            existing_objects.append(obj_to_delete)
+        flat_existing_objects = [obj for sublist in existing_objects for obj in
+                                 (sublist if isinstance(sublist, list) else [sublist])]
+
+    for file in files:
+        push_file(
+            endpoint,
+            wallet,
+            password,
+            cid,
+            file,
+            attributes,
+            put_timeout,
+            expiration_epoch,
+        )
+
+    if flat_existing_objects:
+        delete_objects(endpoint, wallet, password, cid, flat_existing_objects)
 
 
 if __name__ == "__main__":
@@ -192,4 +333,6 @@ def push_files_to_neofs(
         args.lifetime,
         args.put_timeout,
         neofs_password,
+        args.replace_objects,
+        args.replace_container_contents,
     )