feat(cli): add extraction command

keep/cli/cli.py

@@ -853,7 +853,6 @@ def delete_mapping(info: Info, mapping_id: int):
     )
     # Check the response
     if response.ok:
-        response = response.json()
         click.echo(
             click.style(f"Mapping rule {mapping_id} deleted successfully", bold=True)
         )
@@ -865,6 +864,191 @@ def delete_mapping(info: Info, mapping_id: int):
         )
 
 
+@cli.group()
+@pass_info
+def extraction(info: Info):
+    """Manage extraction."""
+    pass
+
+
+@extraction.command(name="list")
+@pass_info
+def list_extraction(info: Info):
+    """List extractions."""
+    resp = make_keep_request(
+        "GET",
+        info.keep_api_url + "/extraction",
+        headers={"x-api-key": info.api_key, "accept": "application/json"},
+    )
+    if not resp.ok:
+        raise Exception(f"Error getting extractions: {resp.text}")
+
+    extractions = resp.json()
+    if len(extractions) == 0:
+        click.echo(click.style("No extractions found.", bold=True))
+        return
+
+    # Create a new table
+    table = PrettyTable()
+    # Add column headers
+    table.field_names = [
+        "ID",
+        "Name",
+        "Description",
+        "Priority",
+        "Attribute",
+        "Condition",
+        "Disabled",
+        "Regex",
+        "Pre",
+        "Created By",
+        "Creation Time",
+        "Updated By",
+        "Update Time",
+    ]
+
+    # Add rows for each extraction
+    for e in extractions:
+        table.add_row(
+            [
+                e["id"],
+                e["name"],
+                e["description"],
+                e["priority"],
+                e["attribute"],
+                e["condition"],
+                e["disabled"],
+                e["regex"],
+                e["pre"],
+                e["created_by"],
+                e["created_at"],
+                e["updated_by"],
+                e["updated_at"],
+            ]
+        )
+    print(table)
+
+
+@extraction.command(name="create")
+@click.option(
+    "--name",
+    "-n",
+    type=str,
+    help="The name of the extraction.",
+    required=True,
+)
+@click.option(
+    "--description",
+    "-d",
+    type=str,
+    help="The description of the extraction.",
+    required=False,
+    default="",
+)
+@click.option(
+    "--priority",
+    "-p",
+    type=click.IntRange(0, 100),
+    help="The priority of the extraction, higher priority means this rule will execute first.",
+    required=False,
+    default=0,
+)
+@click.option(
+    "--pre",
+    type=bool,
+    help="Whether this rule should be applied before or after the alert is standardized",
+    required=False,
+    default=False,
+)
+@click.option(
+    "--attribute",
+    "-a",
+    type=str,
+    help="Event attribute name to extract from.",
+    required=True,
+    default="",
+)
+@click.option(
+    "--regex",
+    "-r",
+    type=str,
+    help="The regex rule to extract by. Regex format should be like python regex pattern for group matching.",
+    required=True,
+    default="",
+)
+@click.option(
+    "--condition",
+    "-c",
+    type=str,
+    help="CEL based condition",
+    required=True,
+    default="",
+)
+@pass_info
+def create(
+        info: Info, name: str, description: str, priority: int, pre: bool, attribute: str, regex: str, condition: str
+):
+    """Create a extraction rule."""
+    response = make_keep_request(
+        "POST",
+        info.keep_api_url + "/extraction",
+        headers={"x-api-key": info.api_key, "accept": "application/json"},
+        json={
+            "name": name,
+            "description": description,
+            "priority": priority,
+            "pre": pre,
+            "attribute": attribute,
+            "regex": regex,
+            "condition": condition
+        },
+    )
+
+    # Check the response
+    if response.ok:
+        click.echo(
+            click.style(f"Extraction rule {name} created successfully", bold=True)
+        )
+    else:
+        click.echo(
+            click.style(
+                f"Error creating extraction rule {name}: {response.text}",
+                bold=True,
+            )
+        )
+
+
+@extraction.command(name="delete")
+@click.option(
+    "--extraction-id",
+    type=int,
+    help="The ID of the extraction to delete.",
+    required=True,
+)
+@pass_info
+def delete_extraction(info: Info, extraction_id: int):
+    """Delete a extraction with a specified ID."""
+
+    # Delete the extraction with the specified ID
+    response = make_keep_request(
+        "DELETE",
+        info.keep_api_url + f"/extraction/{extraction_id}",
+        headers={"x-api-key": info.api_key, "accept": "application/json"},
+    )
+
+    # Check the response
+    if response.ok:
+        click.echo(
+            click.style(f"Extraction rule {extraction_id} deleted successfully", bold=True)
+        )
+    else:
+        click.echo(
+            click.style(
+                f"Error deleting extraction rule {extraction_id}: {response.text}", bold=True
+            )
+        )
+
+
 @cli.group()
 @pass_info
 def provider(info: Info):