feat: generation of python files for custom search command

docs/generated_files.md

@@ -24,5 +24,6 @@ The following table describes the files generated by UCC framework.
 | inputs.xml | output/<YOUR_ADDON_NAME>/default/data/ui/views | Generates inputs.xml based on inputs configuration present in globalConfig, in `default/data/ui/views/inputs.xml` folder |
 | _redirect.xml | output/<YOUR_ADDON_NAME>/default/data/ui/views | Generates ta_name_redirect.xml file, if oauth is mentioned in globalConfig, in `default/data/ui/views/` folder. |
 | _.html | output/<YOUR_ADDON_NAME>/default/data/ui/alerts | Generates `alert_name.html` file based on alerts configuration present in globalConfig, in `default/data/ui/alerts` folder. |
+| _.py | output/<YOUR_ADDON_NAME>/bin | Generates Python files for custom search commands provided in the globalConfig. |
 | globalConfig.json | <source_dir> | Generates globalConfig.json file in the source code if globalConfig is not present in source directory at build time. |
 

splunk_add_on_ucc_framework/generators/file_const.py

@@ -24,6 +24,7 @@
     RedirectXml,
 )
 from splunk_add_on_ucc_framework.generators.html_files import AlertActionsHtml
+from splunk_add_on_ucc_framework.generators.python_files import CustomCommandPy
 from splunk_add_on_ucc_framework.generators.conf_files import (
     AlertActionsConf,
     AppConf,
@@ -95,4 +96,9 @@ class FileClass(NamedTuple):
         AlertActionsHtml,
         ["default", "data", "ui", "alerts"],
     ),
+    FileClass(
+        "_.py",
+        CustomCommandPy,
+        ["bin"],
+    ),
 ]

splunk_add_on_ucc_framework/generators/python_files/__init__.py

@@ -0,0 +1,19 @@
+#
+# Copyright 2025 Splunk Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from ..file_generator import FileGenerator
+from .create_custom_command_python import CustomCommandPy
+
+__all__ = ["FileGenerator", "CustomCommandPy"]

splunk_add_on_ucc_framework/generators/python_files/create_custom_command_python.py

@@ -0,0 +1,115 @@
+#
+# Copyright 2025 Splunk Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from typing import Any, Dict, List
+
+from splunk_add_on_ucc_framework.generators.file_generator import FileGenerator
+
+
+class CustomCommandPy(FileGenerator):
+    __description__ = "Generates Python files for custom search commands provided in the globalConfig."
+
+    def argument_generator(
+        self, argument_list: List[str], arg: Dict[str, Any]
+    ) -> List[str]:
+        validate_str = ""
+        validate = arg.get("validate", {})
+        if validate:
+            validate_type = validate["type"]
+            if validate_type in ("Integer", "Float"):
+                min_val = validate.get("minimum")
+                max_val = validate.get("maximum")
+                args = []
+                if min_val is not None:
+                    args.append(f"minimum={min_val}")
+                if max_val is not None:
+                    args.append(f"maximum={max_val}")
+                validate_args = ", ".join(args)
+                validate_str = (
+                    f", validate=validators.{validate_type}({validate_args})"
+                    if args
+                    else f", validate=validators.{validate_type}()"
+                )
+            elif validate_type:
+                validate_str = f", validate=validators.{validate_type}()"
+
+        if arg["default"] is None:
+            arg_str = (
+                f"{arg['name']} = Option(name='{arg['name']}', "
+                f"require={arg.get('require')}"
+                f"{validate_str})"
+            )
+        else:
+            arg_str = (
+                f"{arg['name']} = Option(name='{arg['name']}', "
+                f"require={arg.get('require')}"
+                f"{validate_str}, "
+                f"default='{arg.get('default', '')}')"
+            )
+        argument_list.append(arg_str)
+        return argument_list
+
+    def _set_attributes(self, **kwargs: Any) -> None:
+        self.commands_info = []
+        for command in self._global_config.custom_search_commands:
+            argument_list: List[str] = []
+            imported_file_name = command["fileName"].replace(".py", "")
+            template = command["commandType"].replace(" ", "_") + ".template"
+            for argument in command["arguments"]:
+                argument_dict = {
+                    "name": argument["name"],
+                    "require": argument.get("required", False),
+                    "validate": argument.get("validate"),
+                    "default": argument.get("defaultValue"),
+                }
+                self.argument_generator(argument_list, argument_dict)
+            self.commands_info.append(
+                {
+                    "imported_file_name": imported_file_name,
+                    "file_name": command["commandName"],
+                    "class_name": command["commandName"].title(),
+                    "description": command.get("description"),
+                    "syntax": command.get("syntax"),
+                    "template": template,
+                    "list_arg": argument_list,
+                }
+            )
+
+    def generate(self) -> Dict[str, str]:
+        if not self.commands_info:
+            return {}
+
+        generated_files = {}
+        for command_info in self.commands_info:
+            file_name = command_info["file_name"] + ".py"
+            file_path = self.get_file_output_path(["bin", file_name])
+            self.set_template_and_render(
+                template_file_path=["custom_command"],
+                file_name=command_info["template"],
+            )
+            rendered_content = self._template.render(
+                imported_file_name=command_info["imported_file_name"],
+                class_name=command_info["class_name"],
+                description=command_info["description"],
+                syntax=command_info["syntax"],
+                list_arg=command_info["list_arg"],
+            )
+            self.writer(
+                file_name=file_name,
+                file_path=file_path,
+                content=rendered_content,
+            )
+            generated_files.update({file_name: file_path})
+        return generated_files

splunk_add_on_ucc_framework/templates/custom_command/dataset_processing.template

@@ -0,0 +1,33 @@
+import sys
+import import_declare_test
+
+from splunklib.searchcommands import \
+    dispatch, EventingCommand, Configuration, Option, validators
+from {{imported_file_name}} import transform
+
+@Configuration()
+class {{class_name}}Command(EventingCommand):
+    {% if syntax or description%}
+    """
+
+    {% if syntax %}
+    ##Syntax
+    {{syntax}}
+    {% endif %}
+
+    {% if description %}
+    ##Description
+    {{description}}
+    {% endif %}
+
+    """
+    {% endif %}
+
+    {% for arg in list_arg %}
+    {{arg}}
+    {% endfor %}
+
+    def transform(self, events):
+       return transform(self, events)
+
+dispatch({{class_name}}Command, sys.argv, sys.stdin, sys.stdout, __name__)

splunk_add_on_ucc_framework/templates/custom_command/generating.template

@@ -0,0 +1,32 @@
+import sys
+import import_declare_test
+
+from splunklib.searchcommands import \
+    dispatch, GeneratingCommand, Configuration, Option, validators
+from {{imported_file_name}} import generate
+
+@Configuration()
+class {{class_name}}Command(GeneratingCommand):
+    {% if syntax or description%}
+    """
+
+    {% if syntax %}
+    ##Syntax
+    {{syntax}}
+    {% endif %}
+
+    {% if description %}
+    ##Description
+    {{description}}
+    {% endif %}
+
+    """
+    {% endif %}
+    {% for arg in list_arg %}
+    {{arg}}
+    {% endfor %}
+
+    def generate(self):
+       return generate(self)
+
+dispatch({{class_name}}Command, sys.argv, sys.stdin, sys.stdout, __name__)

splunk_add_on_ucc_framework/templates/custom_command/streaming.template

@@ -0,0 +1,33 @@
+import sys
+import import_declare_test
+
+from splunklib.searchcommands import \
+    dispatch, StreamingCommand, Configuration, Option, validators
+from {{imported_file_name}} import stream
+
+@Configuration()
+class {{class_name}}Command(StreamingCommand):
+    {% if syntax or description%}
+    """
+
+    {% if syntax %}
+    ##Syntax
+    {{syntax}}
+    {% endif %}
+
+    {% if description %}
+    ##Description
+    {{description}}
+    {% endif %}
+
+    """
+    {% endif %}
+
+    {% for arg in list_arg %}
+    {{arg}}
+    {% endfor %}
+
+    def stream(self, events):
+        return stream(self, events)
+
+dispatch({{class_name}}Command, sys.argv, sys.stdin, sys.stdout, __name__)

tests/smoke/test_ucc_build.py

@@ -168,6 +168,12 @@ def test_ucc_generate_with_everything(caplog):
             ("bin", "example_input_three.py"),
             ("bin", "example_input_four.py"),
             ("bin", "import_declare_test.py"),
+            ("bin", "countmatchescommand.py"),
+            ("bin", "countmatches.py"),
+            ("bin", "filter.py"),
+            ("bin", "filtercommand.py"),
+            ("bin", "generatetext.py"),
+            ("bin", "generatetextcommand.py"),
             ("bin", "splunk_ta_uccexample_rh_account.py"),
             ("bin", "splunk_ta_uccexample_rh_example_input_one.py"),
             ("bin", "splunk_ta_uccexample_rh_example_input_two.py"),

tests/testdata/expected_addons/expected_output_global_config_everything/Splunk_TA_UCCExample/bin/countmatches.py

@@ -0,0 +1,4 @@
+def stream(self, records):
+        for record in records:
+           # write custom logic for the search command
+            yield record

tests/testdata/expected_addons/expected_output_global_config_everything/Splunk_TA_UCCExample/bin/countmatchescommand.py

@@ -0,0 +1,17 @@
+import sys
+import import_declare_test
+
+from splunklib.searchcommands import \
+    dispatch, StreamingCommand, Configuration, Option, validators
+from countmatches import stream
+
+@Configuration()
+class CountmatchescommandCommand(StreamingCommand):
+
+    fieldname = Option(name='fieldname', require=True, validate=validators.Fieldname())
+    pattern = Option(name='pattern', require=True, validate=validators.RegularExpression())
+
+    def stream(self, events):
+        return stream(self, events)
+
+dispatch(CountmatchescommandCommand, sys.argv, sys.stdin, sys.stdout, __name__)

tests/testdata/expected_addons/expected_output_global_config_everything/Splunk_TA_UCCExample/bin/filter.py

@@ -0,0 +1,36 @@
+def transform(self, records):
+        contains = self.contains
+        replace_array = self.replace_array
+
+        if contains and replace_array:
+            arr = replace_array.split(",")
+            if len(arr) != 2:
+                raise ValueError("Please provide only two arguments, separated by comma for 'replace'")
+
+            for record in records:
+                _raw = record.get("_raw")
+                if contains in _raw:
+                    record["_raw"] = _raw.replace(arr[0], arr[1])
+                    yield record
+            return
+
+        if contains:
+            for record in records:
+                _raw = record.get("_raw")
+                if contains in _raw:
+                    yield record
+            return
+
+        if replace_array:
+            arr = replace_array.split(",")
+            if len(arr) != 2:
+                raise ValueError("Please provide only two arguments, separated by comma for 'replace'")
+
+            for record in records:
+                _raw = record.get("_raw")
+                record["_raw"] = _raw.replace(arr[0], arr[1])
+                yield record
+            return
+
+        for record in records:
+            yield record

tests/testdata/expected_addons/expected_output_global_config_everything/Splunk_TA_UCCExample/bin/filtercommand.py

@@ -0,0 +1,26 @@
+import sys
+import import_declare_test
+
+from splunklib.searchcommands import \
+    dispatch, EventingCommand, Configuration, Option, validators
+from filter import transform
+
+@Configuration()
+class FiltercommandCommand(EventingCommand):
+    """
+
+    ##Syntax
+    | filtercommand contains='value1' replace='value to be replaced,value to replace with'
+
+    ##Description
+    It filters records from the events stream returning only those which has :code:`contains` in them and replaces :code:`replace_array[0]` with :code:`replace_array[1]`.
+
+    """
+
+    contains = Option(name='contains', require=False)
+    replace_array = Option(name='replace_array', require=False)
+
+    def transform(self, events):
+       return transform(self, events)
+
+dispatch(FiltercommandCommand, sys.argv, sys.stdin, sys.stdout, __name__)

tests/testdata/expected_addons/expected_output_global_config_everything/Splunk_TA_UCCExample/bin/generatetext.py

@@ -0,0 +1,8 @@
+import time
+import logging
+
+
+def generate(self):
+    logging.debug("Generating %d events with text %s" % (self.count, self.text))
+    for i in range(1, self.count + 1):
+        yield {'_serial': i, '_time': time.time(), '_raw': str(i) + '. ' + self.text}