Added replace_existing option

Fixes #2

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

action.yml

@@ -34,7 +34,12 @@ inputs:
   LIFETIME:
     description: Number of epochs for object to stay valid
     required: false
-    default: 0
+    default: '0'
+  REPLACE_EXISTING:
+    description: Replace existing objects with the same attributes in the container
+    required: false
+    default: 'False'
+
 
 outputs:
   OUTPUT_CONTAINER_URL:
@@ -93,11 +98,13 @@ runs:
         NEOFS_ATTRIBUTES: ${{ inputs.NEOFS_ATTRIBUTES }}
         URL_PREFIX: ${{ inputs.URL_PREFIX }}
         LIFETIME: ${{ inputs.LIFETIME }}
+        REPLACE_EXISTING: ${{ inputs.REPLACE_EXISTING }}
         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-existing "$REPLACE_EXISTING" \
         --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,7 +1,9 @@
 import os
+import re
 import subprocess
 import argparse
 import magic
+import json
 
 FILE_PATH = "FilePath"  # the key for the attribute, is the path for the static page and allure report zip files
 CONTENT_TYPE = "ContentType"
@@ -69,6 +71,13 @@ def parse_args():
         help="Timeout for the put each file to neofs, in seconds. Default is 600 seconds",
         default=600,
     )
+    parser.add_argument(
+        "--replace-object",
+        required=False,
+        type=bool,
+        help="Replace object with the same name in the container",
+        default=None,
+    )
 
     return parser.parse_args()
 
@@ -88,45 +97,32 @@ 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}"
+def neofs_cli_execute(cmd: str, json_output: bool = False, timeout: int = None):
+    """
+    Executes a given command and returns its output.
 
-    if attributes is not None and attributes != "":
-        base_cmd_with_file += f",{attributes}"
-
-    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,
             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 +134,104 @@ 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 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, attributes: str = None, output_format: str = "str") -> str:
+    attrs = {
+        FILE_PATH: file_path,
+        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 +242,45 @@ def push_files_to_neofs(
     lifetime: int,
     put_timeout: int,
     password: str,
+    replace_object: 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)
+
+    if replace_object:
+        existing_objects = []
+        for file in files:
+            search_attrs = compile_attributes(
+                file['neofs_path_attr'], file['mime_type'], attributes, 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])]
+        if flat_existing_objects:
+            delete_objects(endpoint, wallet, password, cid, flat_existing_objects)
+
+    for file in files:
+        push_file(
+            endpoint,
+            wallet,
+            password,
+            cid,
+            file,
+            attributes,
+            put_timeout,
+            expiration_epoch,
+        )
 
 
 if __name__ == "__main__":
@@ -192,4 +298,5 @@ def push_files_to_neofs(
         args.lifetime,
         args.put_timeout,
         neofs_password,
+        args.replace_object,
     )