rule.py

from typing import List

from cdisc_rules_engine.enums.optional_condition_parameters import (
    OptionalConditionParameters,
)
from cdisc_rules_engine.enums.rule_types import RuleTypes
from cdisc_rules_engine.enums.sensitivity import Sensitivity


class Rule:
    """
    This class represents a rule DB record.
    A rule DB record represents a proprietary rule, but
    the structure will match the CDISC rules format
    """

    def __init__(self, record_params: dict):
        self.core_id: str = record_params["core_id"]
        self.reference: List[dict] = record_params["reference"]
        self.sensitivity: Sensitivity = record_params["sensitivity"]
        self.executability: str = record_params["executability"]
        self.category: str = record_params["category"]
        self.author: str = record_params["author"]
        self.description: str = record_params["description"]
        self.authority: dict = record_params["authority"]
        self.standards: dict = record_params["standards"]
        self.classes: dict = record_params.get("classes")
        self.domains: dict = record_params.get("domains")
        self.datasets: dict = record_params.get("datasets")
        self.entities: dict = record_params.get("entities")
        self.rule_type: RuleTypes = record_params["rule_type"]
        self.operations: List[dict] = record_params.get("operations")
        self.conditions: dict = record_params["conditions"]
        self.actions: dict = record_params["actions"]
        self.output_variables: dict = record_params.get("output variables")
        self.grouping_variables: List[str] = record_params.get("grouping_variables", [])

    @classmethod
    def _get_key(cls, obj: dict, space_key: str, default=None):
        """Get a value by key, supporting both space-format ('Rule Type') and
        underscore-format ('Rule_Type') for backward compatibility with the
        CDISC Library API which returns underscore keys."""
        if space_key in obj:
            return obj[space_key]
        return obj.get(space_key.replace(" ", "_"), default)

    @classmethod
    def from_cdisc_metadata(cls, rule_metadata: dict) -> dict:
        if cls.is_cdisc_rule_metadata(rule_metadata):
            authorities = rule_metadata.get("Authorities", [])
            scope = rule_metadata.get("Scope", {})
            outcome = rule_metadata.get("Outcome", {})
            executable_rule = {
                "core_id": rule_metadata.get("Core", {}).get("Id"),
                "author": "CDISC",
                "reference": cls.parse_references(authorities),
                "sensitivity": rule_metadata.get("Sensitivity"),
                "executability": (rule_metadata.get("Executability") or "").lower(),
                "description": rule_metadata.get("Description"),
                "authorities": authorities,
                "standards": cls.parse_standards(authorities),
                "classes": scope.get("Classes"),
                "domains": scope.get("Domains"),
                "entities": scope.get("Entities"),
                "rule_type": cls._get_key(rule_metadata, "Rule Type"),
                "conditions": cls.parse_conditions(rule_metadata.get("Check")),
                "actions": cls.parse_actions(outcome),
                "use_case": cls._get_key(scope, "Use Case"),
                "data_structures": cls._get_key(scope, "Data Structures"),
                "status": rule_metadata.get("Core", {}).get("Status", {}),
            }

            if "Operations" in rule_metadata:
                executable_rule["operations"] = rule_metadata.get("Operations")

            match_datasets = cls._get_key(rule_metadata, "Match Datasets")
            if match_datasets is not None:
                executable_rule["datasets"] = cls.parse_datasets(match_datasets)

            output_variables = cls._get_key(outcome, "Output Variables")
            if output_variables is not None:
                executable_rule["output_variables"] = output_variables

            grouping_variables = cls._get_key(rule_metadata, "Grouping Variables")
            if grouping_variables is not None:
                executable_rule["grouping_variables"] = grouping_variables

            return executable_rule
        else:
            return rule_metadata

    @classmethod
    def parse_standards(cls, authorities: List[dict]) -> List[dict]:
        standards = []
        for authority in authorities:
            for standard in authority.get("Standards", []):
                standards.append(
                    {
                        "Name": standard.get("Name"),
                        "Version": standard.get("Version"),
                        "Substandard": standard.get("Substandard"),
                    }
                )
        return standards

    @classmethod
    def parse_references(cls, authorities: List[dict]) -> list:
        references = []
        for authority in authorities:
            for standard in authority.get("Standards", []):
                references.append(standard.get("References"))
        return references

    @classmethod
    def is_cdisc_rule_metadata(cls, rule_metadata: dict) -> bool:
        return "Core" in rule_metadata

    @classmethod
    def parse_conditions(cls, conditions: dict | str) -> dict | str:
        if not conditions:
            raise ValueError("No check data provided")
        if isinstance(conditions, str):
            return conditions
        all_conditions = conditions.get("all")
        any_conditions = conditions.get("any")
        not_condition = conditions.get("not")
        conditions_json = {}
        if all_conditions:
            conditions_json["all"] = cls.build_conditions(all_conditions)
        if any_conditions:
            conditions_json["any"] = cls.build_conditions(any_conditions)
        if not_condition:
            conditions_json["not"] = cls.parse_conditions(not_condition)
        return conditions_json

    @classmethod
    def build_conditions(cls, conditions_data: List[dict]) -> dict:
        function = "get_dataset"
        conditions = []

        for condition in conditions_data:
            if "all" in condition:
                conditions.append({"all": cls.build_conditions(condition.get("all"))})
            elif "any" in condition:
                conditions.append({"any": cls.build_conditions(condition.get("any"))})
            elif "not" in condition:
                conditions.append({"not": cls.parse_conditions(condition.get("not"))})
            else:
                conditions.append(cls.build_condition(condition, function))

        return conditions

    @classmethod
    def build_condition(cls, condition: dict, variable_function: str) -> dict:
        data = {
            "name": variable_function,
            "operator": condition.get("operator"),
            "value": {
                "comparator": condition.get("value"),
            },
        }
        if condition.get("name"):
            data["value"]["target"] = condition.get("name")
        if condition.get("regex"):
            data["value"]["regex"] = condition.get("regex")
        if "variables" in condition:
            data["variables"] = condition["variables"]
        if "negative" in condition:
            data["value"]["negative"] = condition.get("negative")
        for optional_parameter in OptionalConditionParameters.values():
            if optional_parameter in condition:
                data["value"][optional_parameter] = condition.get(optional_parameter)
        return data

    @classmethod
    def parse_actions(cls, actions_data: dict) -> List[dict]:
        if not actions_data:
            raise ValueError("No actions data provided")
        action = "generate_dataset_error_objects"
        return [{"name": action, "params": {"message": actions_data.get("Message")}}]

    @classmethod
    def parse_datasets(cls, match_key_data: List[dict]) -> List[dict]:
        if not match_key_data:
            return None
        datasets = []
        for data in match_key_data:
            join_data = {
                "domain_name": data.get("Name"),
                "match_key": [
                    (
                        key
                        if isinstance(key, str)
                        else {k.lower(): v for k, v in key.items()}
                    )
                    for key in data.get("Keys", [])
                ],
                "wildcard": data.get("Wildcard", "**"),
            }
            if "Join_Type" in data:
                join_data["join_type"] = data.get("Join_Type")
            if "Child" in data:
                join_data["child"] = data.get("Child")
            datasets.append(join_data)
        return datasets