"""
Label Ruleset Execution Script
Applies validated rulesets to processed data to generate classification labels.
Supports train/val/test splits and provides comprehensive execution statistics.
Key Features:
- Field availability validation at execution time
- Priority-based rule evaluation (first match wins)
- Comprehensive statistics tracking
- Fail-safe error handling
- Support for multiple job types (training, validation, testing, calibration)
Usage:
python label_ruleset_execution.py --job-type training
"""
import argparse
import json
import logging
from datetime import datetime
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
import pandas as pd
# Configure logging
logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)
[docs]
class RulesetFieldValidator:
"""Validates field availability in actual data at execution time."""
[docs]
def validate_fields(self, ruleset: dict, data_df: pd.DataFrame) -> Dict[str, Any]:
"""
Validates all field references exist in actual data.
This is an EXECUTION-TIME validator that checks:
- All required fields exist in DataFrame
- All fields used in rules exist in DataFrame
- Field null percentages (data quality check)
Args:
ruleset: Validated ruleset configuration
data_df: Actual DataFrame to check
Returns:
Dictionary with validation results:
- valid: bool
- missing_fields: List[str]
- warnings: List[str]
"""
result = {"valid": True, "missing_fields": [], "warnings": []}
field_config = ruleset.get("field_config", {})
required_fields = set(field_config.get("required_fields", []))
rules = ruleset.get("ruleset", [])
# Extract all field references from rules
used_fields = set()
for rule in rules:
if not rule.get("enabled", True):
continue
fields = self._extract_fields_from_conditions(rule.get("conditions", {}))
used_fields.update(fields)
# Check field availability in data
available_fields = set(data_df.columns)
# Check required fields exist
missing_required = required_fields - available_fields
if missing_required:
result["valid"] = False
result["missing_fields"].extend(list(missing_required))
logger.error(f"Required fields missing in data: {missing_required}")
# Check used fields exist
missing_used = used_fields - available_fields
if missing_used:
result["valid"] = False
result["missing_fields"].extend(list(missing_used))
logger.error(f"Fields used in rules but not in data: {missing_used}")
# Check for high null percentages
for field in used_fields & available_fields:
null_pct = data_df[field].isnull().sum() / len(data_df)
if null_pct > 0.5:
result["warnings"].append(
f"Field '{field}' has {null_pct:.1%} null values"
)
logger.warning(f"Field '{field}' has {null_pct:.1%} null values")
return result
def _extract_fields_from_conditions(self, condition: dict) -> List[str]:
"""Recursively extract all field names from a condition."""
fields = []
if "all_of" in condition:
for subcond in condition["all_of"]:
fields.extend(self._extract_fields_from_conditions(subcond))
elif "any_of" in condition:
for subcond in condition["any_of"]:
fields.extend(self._extract_fields_from_conditions(subcond))
elif "none_of" in condition:
for subcond in condition["none_of"]:
fields.extend(self._extract_fields_from_conditions(subcond))
elif "field" in condition:
fields.append(condition["field"])
return fields
[docs]
class RuleEngine:
"""
Evaluates validated rules against data rows to produce labels.
Extended for multilabel support.
Optimized for:
- Batch processing (vectorized where possible)
- Priority-based evaluation (first match wins)
- Efficient condition checking
- Minimal memory footprint
- Multilabel sparse representation
"""
def __init__(self, validated_ruleset: dict):
"""
Initialize rule engine with multilabel support.
Args:
validated_ruleset: Pre-validated ruleset from RulesetGenerator
"""
# Extract configuration
self.label_config = validated_ruleset["label_config"]
self.field_config = validated_ruleset["field_config"]
self.ruleset = validated_ruleset["ruleset"]
self.metadata = validated_ruleset.get("metadata", {})
# Filter to enabled rules only (already sorted by priority)
self.active_rules = [r for r in self.ruleset if r.get("enabled", True)]
# Determine label type
self.label_type = self.label_config.get("output_label_type", "binary")
# Get output column names (normalize to list)
output_label_name = self.label_config["output_label_name"]
if isinstance(output_label_name, str):
# Single-label: string → list of one
self.output_columns = [output_label_name]
else:
# Multilabel: already a list
self.output_columns = output_label_name
# Multilabel-specific configuration
self.sparse_representation = self.label_config.get(
"sparse_representation", True
)
# Common configuration
self.default_label = self.label_config["default_label"]
self.evaluation_mode = self.label_config.get("evaluation_mode", "priority")
# Statistics tracking (per column)
self.rule_match_counts = {
col: {r["rule_id"]: 0 for r in self.active_rules}
for col in self.output_columns
}
self.default_label_counts = {col: 0 for col in self.output_columns}
self.total_evaluated = 0
[docs]
def evaluate_row(self, row: pd.Series):
"""
Evaluate rules against a single row.
Returns:
- Single-label mode: int or str (label value)
- Multilabel mode: Dict[str, Any] (column → value mapping)
"""
self.total_evaluated += 1
if self.label_type in ["binary", "multiclass"]:
return self._evaluate_row_single_label(row)
else:
return self._evaluate_row_multilabel(row)
def _evaluate_row_single_label(self, row: pd.Series):
"""Evaluate rules for single-label mode (backward compatible)."""
# Single column name (output_columns is list of one)
output_col = self.output_columns[0]
for rule in self.active_rules:
try:
if self._evaluate_conditions(rule["conditions"], row):
rule_id = rule["rule_id"]
output_label = rule["output_label"]
# output_label should be int/str for single-label
self.rule_match_counts[output_col][rule_id] += 1
return output_label
except (KeyError, TypeError, ValueError, AttributeError) as e:
logger.warning(
f"Error evaluating rule {rule.get('rule_id', '<unknown>')} "
f"on row {getattr(row, 'name', '<unknown>')}: "
f"{type(e).__name__}: {e}"
)
continue
self.default_label_counts[output_col] += 1
return self.default_label
def _evaluate_row_multilabel(self, row: pd.Series) -> Dict[str, Any]:
"""Evaluate rules for multilabel mode with sparse representation."""
import numpy as np
# Initialize all columns with NaN (sparse) or default (dense)
if self.sparse_representation:
result = {col: np.nan for col in self.output_columns}
else:
# Handle per-column default_label
if isinstance(self.default_label, dict):
result = {col: self.default_label[col] for col in self.output_columns}
else:
result = {col: self.default_label for col in self.output_columns}
# Evaluate rules in priority order
for rule in self.active_rules:
try:
if not self._evaluate_conditions(rule["conditions"], row):
continue
# Rule matched - get output
output_label = rule.get("output_label")
if isinstance(output_label, dict):
# Multilabel: dict mapping column → value
for col, value in output_label.items():
if col not in result:
continue
# Get per-column default for comparison
if isinstance(self.default_label, dict):
col_default = self.default_label.get(col)
else:
col_default = self.default_label
# Only set if not already set (priority order)
if pd.isna(result[col]) or result[col] == col_default:
result[col] = value
self.rule_match_counts[col][rule["rule_id"]] += 1
except (KeyError, TypeError, ValueError, AttributeError) as e:
logger.warning(
f"Error evaluating rule {rule.get('rule_id', '<unknown>')} "
f"on row {getattr(row, 'name', '<unknown>')}: "
f"{type(e).__name__}: {e}"
)
continue
# Fill remaining NaN columns with default if dense mode
for col in result:
if pd.isna(result[col]):
self.default_label_counts[col] += 1
if not self.sparse_representation:
# Handle per-column default_label
if isinstance(self.default_label, dict):
result[col] = self.default_label[col]
else:
result[col] = self.default_label
return result
[docs]
def evaluate_batch(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Evaluate rules for entire DataFrame.
Returns:
DataFrame with label column(s) added
"""
if self.label_type in ["binary", "multiclass"]:
# Single column result (backward compatible)
output_col = self.output_columns[0]
df[output_col] = df.apply(self.evaluate_row, axis=1)
return df
else:
# Multi-column result (multilabel)
results = df.apply(self.evaluate_row, axis=1, result_type="expand")
# Add all label columns to original dataframe
for col in self.output_columns:
df[col] = results[col]
return df
[docs]
def get_statistics(self) -> Dict[str, Any]:
"""Get execution statistics with multilabel support."""
if self.label_type in ["binary", "multiclass"]:
# Single-label statistics (backward compatible)
output_col = self.output_columns[0]
return {
"total_evaluated": self.total_evaluated,
"rule_match_counts": self.rule_match_counts[output_col],
"default_label_count": self.default_label_counts[output_col],
"rule_match_percentages": {
rule_id: (count / self.total_evaluated * 100)
if self.total_evaluated > 0
else 0
for rule_id, count in self.rule_match_counts[output_col].items()
},
"default_label_percentage": (
self.default_label_counts[output_col] / self.total_evaluated * 100
if self.total_evaluated > 0
else 0
),
}
else:
# Multilabel statistics (per column)
stats = {
"label_type": "multilabel",
"total_evaluated": self.total_evaluated,
"per_column_statistics": {},
}
for col in self.output_columns:
col_stats = {
"rule_match_counts": self.rule_match_counts[col],
"default_label_count": self.default_label_counts[col],
"rule_match_percentages": {
rule_id: (count / self.total_evaluated * 100)
if self.total_evaluated > 0
else 0
for rule_id, count in self.rule_match_counts[col].items()
},
"default_label_percentage": (
self.default_label_counts[col] / self.total_evaluated * 100
if self.total_evaluated > 0
else 0
),
}
stats["per_column_statistics"][col] = col_stats
return stats
def _evaluate_conditions(self, conditions: dict, row: pd.Series) -> bool:
"""
Recursively evaluate nested conditions.
Args:
conditions: Condition dictionary with logical operators
row: DataFrame row as Series
Returns:
Boolean indicating whether conditions are satisfied
"""
# Handle logical operators
if "all_of" in conditions:
return all(
self._evaluate_conditions(cond, row) for cond in conditions["all_of"]
)
elif "any_of" in conditions:
return any(
self._evaluate_conditions(cond, row) for cond in conditions["any_of"]
)
elif "none_of" in conditions:
return not any(
self._evaluate_conditions(cond, row) for cond in conditions["none_of"]
)
# Handle leaf condition (field comparison)
else:
return self._evaluate_leaf_condition(conditions, row)
def _evaluate_leaf_condition(self, condition: dict, row: pd.Series) -> bool:
"""
Evaluate a single leaf condition (field comparison).
Args:
condition: Single condition with field, operator, value
row: DataFrame row as Series
Returns:
Boolean indicating whether condition is satisfied
"""
field = condition["field"]
operator = condition["operator"]
expected_value = condition["value"]
# Get actual value from row
if field not in row.index:
return False
actual_value = row[field]
# Handle null values
if pd.isna(actual_value):
if operator == "is_null":
return True
elif operator == "is_not_null":
return False
else:
return False # Null doesn't match comparisons
# Apply operator
return self._apply_operator(operator, actual_value, expected_value)
def _apply_operator(self, operator: str, actual: Any, expected: Any) -> bool:
"""Apply comparison operator."""
# Comparison operators
if operator == "equals":
return actual == expected
elif operator == "not_equals":
return actual != expected
elif operator == ">":
return float(actual) > float(expected)
elif operator == ">=":
return float(actual) >= float(expected)
elif operator == "<":
return float(actual) < float(expected)
elif operator == "<=":
return float(actual) <= float(expected)
# Collection operators
elif operator == "in":
return actual in expected
elif operator == "not_in":
return actual not in expected
# String operators
elif operator == "contains":
return str(expected) in str(actual)
elif operator == "not_contains":
return str(expected) not in str(actual)
elif operator == "starts_with":
return str(actual).startswith(str(expected))
elif operator == "ends_with":
return str(actual).endswith(str(expected))
elif operator == "regex_match":
import re
return bool(re.search(expected, str(actual)))
# Null operators
elif operator == "is_null":
return False # Already handled null case
elif operator == "is_not_null":
return True # Already handled null case
else:
raise ValueError(f"Unsupported operator: {operator}")
def _detect_file_format(file_path: Path) -> str:
"""
Detect file format based on extension.
Args:
file_path: Path to file
Returns:
File format: 'csv', 'tsv', or 'parquet'
"""
suffix = file_path.suffix.lower()
if suffix in [".csv", ".csv.gz"]:
return "csv"
elif suffix in [".tsv", ".tsv.gz"]:
return "tsv"
elif suffix in [".parquet", ".pq"]:
return "parquet"
else:
# Default to CSV
return "csv"
def _read_dataframe(file_path: Path) -> pd.DataFrame:
"""
Read DataFrame from file, automatically detecting format.
Args:
file_path: Path to data file
Returns:
DataFrame
"""
file_format = _detect_file_format(file_path)
if file_format == "csv":
return pd.read_csv(file_path)
elif file_format == "tsv":
return pd.read_csv(file_path, sep="\t")
elif file_format == "parquet":
return pd.read_parquet(file_path)
else:
raise ValueError(f"Unsupported file format: {file_format}")
def _write_dataframe(df: pd.DataFrame, file_path: Path, file_format: str):
"""
Write DataFrame to file in specified format.
Args:
df: DataFrame to write
file_path: Output file path
file_format: Format to write ('csv', 'tsv', 'parquet')
"""
file_path.parent.mkdir(parents=True, exist_ok=True)
if file_format == "csv":
df.to_csv(file_path, index=False)
elif file_format == "tsv":
df.to_csv(file_path, sep="\t", index=False)
elif file_format == "parquet":
df.to_parquet(file_path, index=False)
else:
raise ValueError(f"Unsupported file format: {file_format}")
[docs]
def apply_field_types(df: pd.DataFrame, field_config: dict) -> pd.DataFrame:
"""
Apply field type conversions based on ruleset field_config.
This ensures data types match rule expectations, avoiding type mismatch issues
(e.g., string '1' vs float 1.0) that can cause rule evaluation failures.
Args:
df: Input DataFrame
field_config: Field configuration from validated ruleset containing field_types
Returns:
DataFrame with corrected types
"""
field_types = field_config.get("field_types", {})
for field, expected_type in field_types.items():
if field not in df.columns:
continue
try:
if expected_type == "float":
# Convert to numeric, coercing errors to NaN
df[field] = pd.to_numeric(df[field], errors="coerce")
logger.info(f"Converted field '{field}' to float")
elif expected_type == "int":
# Convert to numeric Int64 (nullable integer type)
df[field] = pd.to_numeric(df[field], errors="coerce").astype("Int64")
logger.info(f"Converted field '{field}' to int")
elif expected_type == "bool":
def parse_bool(val):
if pd.isna(val):
return val
str_val = str(val).lower().strip()
if str_val in ("true", "1", "yes", "t"):
return True
elif str_val in ("false", "0", "no", "f", ""):
return False
return pd.NA # Invalid boolean value
df[field] = df[field].map(parse_bool)
logger.info(f"Converted field '{field}' to bool")
elif expected_type == "string":
df[field] = df[field].astype(str)
logger.info(f"Converted field '{field}' to string")
except Exception as e:
logger.warning(f"Could not convert field '{field}' to {expected_type}: {e}")
return df
[docs]
def main(
input_paths: Dict[str, str],
output_paths: Dict[str, str],
environ_vars: Dict[str, str],
job_args: argparse.Namespace,
logger: Optional[Callable[[str], None]] = None,
) -> Dict[str, pd.DataFrame]:
"""
Main logic for ruleset execution.
Supports multiple file formats: CSV, TSV, Parquet (auto-detected).
Args:
input_paths: Dictionary with keys:
- "validated_ruleset": Path to validated ruleset JSON
- "input_data": Directory with train/val/test splits
output_paths: Dictionary with keys:
- "processed_data": Directory for output with labels
- "execution_report": Path for execution statistics
- "rule_match_statistics": Optional path for detailed statistics
environ_vars: Environment variables
job_args: Command line arguments (job_type)
logger: Optional logger function
Returns:
Dictionary of processed DataFrames by split name
"""
log = logger or print
# 1. Load validated ruleset from directory
ruleset_dir = Path(input_paths["validated_ruleset"])
ruleset_path = ruleset_dir / "validated_ruleset.json"
with open(ruleset_path, "r") as f:
validated_ruleset = json.load(f)
log(f"[INFO] Loaded validated ruleset v{validated_ruleset.get('version')}")
log(f"[INFO] Rules: {validated_ruleset['metadata']['enabled_rules']} enabled")
# 2. Initialize field validator
field_validator = RulesetFieldValidator()
log(f"[INFO] Initialized field validator")
# 3. Initialize rule engine
rule_engine = RuleEngine(validated_ruleset)
log(f"[INFO] Initialized rule engine")
# 4. Determine splits to process
input_dir = Path(input_paths["input_data"])
output_dir = Path(output_paths["processed_data"])
output_dir.mkdir(parents=True, exist_ok=True)
if job_args.job_type == "training":
splits = ["train", "val", "test"]
else:
splits = [job_args.job_type]
# 5. Process each split
processed_splits = {}
split_statistics = {}
# Get preferred format from environment (optional)
preferred_format = environ_vars.get("PREFERRED_INPUT_FORMAT", "").lower()
if preferred_format and preferred_format not in ["csv", "tsv", "parquet"]:
log(f"[WARNING] Invalid PREFERRED_INPUT_FORMAT '{preferred_format}', ignoring")
preferred_format = ""
for split_name in splits:
log(f"[INFO] Processing {split_name} split...")
# Load data - try multiple formats
split_dir = input_dir / split_name
if not split_dir.exists():
log(f"[WARNING] Split directory not found: {split_dir}")
continue
# Find any data file with supported extensions (no filename assumptions)
data_file = None
input_format = None
# Define supported extensions with their priority order
supported_extensions = [".csv", ".parquet", ".tsv", ".pq", ".csv.gz", ".tsv.gz"]
# If preferred format specified, reorder to check it first
if preferred_format:
format_extensions = {
"csv": [".csv", ".csv.gz"],
"tsv": [".tsv", ".tsv.gz"],
"parquet": [".parquet", ".pq"],
}
preferred_exts = format_extensions.get(preferred_format, [])
# Put preferred extensions first, then others
other_exts = [
ext for ext in supported_extensions if ext not in preferred_exts
]
supported_extensions = preferred_exts + other_exts
log(
f"[INFO] Looking for '{preferred_format}' format first for {split_name}"
)
# Search for files with supported extensions
found_files = []
for ext in supported_extensions:
matching_files = list(split_dir.glob(f"*{ext}"))
if matching_files:
found_files.extend(matching_files)
# Use first match from this extension
data_file = matching_files[0]
input_format = _detect_file_format(data_file)
if len(matching_files) > 1:
log(
f"[WARNING] Multiple {ext} files found in {split_dir}, using: {data_file.name}"
)
else:
log(
f"[INFO] Found data file: {data_file.name} (format: {input_format})"
)
break
if data_file is None:
log(
f"[WARNING] No data file with supported extensions found in {split_dir}"
)
log(f"[WARNING] Supported extensions: {', '.join(supported_extensions)}")
continue
df = _read_dataframe(data_file)
log(f"[INFO] Loaded {split_name}: {df.shape} (format: {input_format})")
# Apply field type conversions from field_config
df = apply_field_types(df, validated_ruleset["field_config"])
log(f"[INFO] Applied field type conversions for {split_name}")
# Validate field availability in data
validation_result = field_validator.validate_fields(validated_ruleset, df)
if not validation_result["valid"]:
error_msg = f"Field validation failed for {split_name}: {validation_result['missing_fields']}"
log(f"[ERROR] {error_msg}")
if environ_vars.get("FAIL_ON_MISSING_FIELDS", "true").lower() == "true":
raise ValueError(error_msg)
else:
log(f"[WARNING] Skipping {split_name} due to validation failure")
continue
# Log warnings if any
for warning in validation_result.get("warnings", []):
log(f"[WARNING] {warning}")
log(f"[INFO] Field validation passed for {split_name}")
# Apply rules to generate labels
df = rule_engine.evaluate_batch(df)
# Compute label distribution (handles both single-label and multilabel)
label_dist = {}
for col in rule_engine.output_columns:
if col in df.columns:
label_dist[col] = df[col].value_counts().to_dict()
if rule_engine.label_type in ["binary", "multiclass"]:
# Single-label: flatten dict
label_dist = label_dist.get(rule_engine.output_columns[0], {})
log(f"[INFO] {split_name} label distribution: {label_dist}")
# Save statistics
split_statistics[split_name] = {
"total_rows": len(df),
"label_distribution": label_dist,
"execution_stats": rule_engine.get_statistics(),
}
# Reset engine statistics for next split
rule_engine.rule_match_counts = {
col: {r["rule_id"]: 0 for r in rule_engine.active_rules}
for col in rule_engine.output_columns
}
rule_engine.default_label_counts = {
col: 0 for col in rule_engine.output_columns
}
rule_engine.total_evaluated = 0
# Save labeled data in same format as input
output_split_dir = output_dir / split_name
output_split_dir.mkdir(exist_ok=True)
# Determine output extension based on input format
if input_format == "csv":
output_file = output_split_dir / f"{split_name}_processed_data.csv"
elif input_format == "tsv":
output_file = output_split_dir / f"{split_name}_processed_data.tsv"
elif input_format == "parquet":
output_file = output_split_dir / f"{split_name}_processed_data.parquet"
else:
output_file = output_split_dir / f"{split_name}_processed_data.csv"
_write_dataframe(df, output_file, input_format)
log(f"[INFO] Saved {output_file} (format: {input_format})")
processed_splits[split_name] = df
# 6. Save execution report
execution_report = {
"ruleset_version": validated_ruleset.get("version"),
"ruleset_timestamp": validated_ruleset.get("generated_timestamp"),
"execution_timestamp": datetime.now().isoformat(),
"label_config": validated_ruleset["label_config"],
"split_statistics": split_statistics,
"total_rules_evaluated": validated_ruleset["metadata"]["enabled_rules"],
}
report_dir = Path(output_paths["execution_report"])
report_dir.mkdir(parents=True, exist_ok=True)
report_path = report_dir / "execution_report.json"
with open(report_path, "w") as f:
json.dump(execution_report, f, indent=2)
log(f"[INFO] Saved execution report: {report_path}")
# 7. Save detailed rule match statistics in execution_report folder
stats_path = report_dir / "rule_match_statistics.json"
with open(stats_path, "w") as f:
json.dump(split_statistics, f, indent=2)
log(f"[INFO] Saved rule match statistics: {stats_path}")
log("[INFO] Ruleset execution complete")
return processed_splits
if __name__ == "__main__":
import sys
import traceback
import os
try:
# Parse command line arguments
parser = argparse.ArgumentParser(
description="Execute validated rulesets on processed data to generate labels"
)
parser.add_argument(
"--job-type",
type=str,
required=True,
choices=["training", "validation", "testing", "calibration"],
help="Job type: training (all splits), validation, testing, or calibration",
)
args = parser.parse_args()
# Set up paths using container paths
input_paths = {
"validated_ruleset": "/opt/ml/processing/input/validated_ruleset",
"input_data": "/opt/ml/processing/input/data",
}
output_paths = {
"processed_data": "/opt/ml/processing/output/processed_data",
"execution_report": "/opt/ml/processing/output/execution_report",
}
# Get configuration from environment variables
environ_vars = {
"FAIL_ON_MISSING_FIELDS": os.environ.get("FAIL_ON_MISSING_FIELDS", "true"),
"ENABLE_RULE_MATCH_TRACKING": os.environ.get(
"ENABLE_RULE_MATCH_TRACKING", "true"
),
"ENABLE_PROGRESS_LOGGING": os.environ.get(
"ENABLE_PROGRESS_LOGGING", "true"
),
"PREFERRED_INPUT_FORMAT": os.environ.get("PREFERRED_INPUT_FORMAT", ""),
}
# Configure detailed logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
# Log key parameters
logger.info("Starting ruleset execution with parameters:")
logger.info(f" Job Type: {args.job_type}")
logger.info(
f" Fail on Missing Fields: {environ_vars['FAIL_ON_MISSING_FIELDS']}"
)
logger.info(
f" Rule Match Tracking: {environ_vars['ENABLE_RULE_MATCH_TRACKING']}"
)
logger.info(f" Progress Logging: {environ_vars['ENABLE_PROGRESS_LOGGING']}")
# Execute the main processing logic
result = main(
input_paths=input_paths,
output_paths=output_paths,
environ_vars=environ_vars,
job_args=args,
logger=logger.info,
)
# Log completion summary
total_splits = len(result)
logger.info(
f"Ruleset execution completed successfully. Processed {total_splits} split(s)"
)
sys.exit(0)
except Exception as e:
logger.error(f"Error in ruleset execution script: {str(e)}")
logger.error(traceback.format_exc())
sys.exit(1)