trestle.tasks.csv_to_oscal_cd

class CsvColumn:
    """CsvColumn."""

    _columns_required = [
        f'{COMPONENT_TITLE}',
        f'{COMPONENT_DESCRIPTION}',
        f'{COMPONENT_TYPE}',
        f'{RULE_ID}',
        f'{RULE_DESCRIPTION}',
        f'{PROFILE_SOURCE}',
        f'{PROFILE_DESCRIPTION}',
        f'{CONTROL_ID_LIST}',
        f'{NAMESPACE}',
    ]

    _columns_required_validation = [
        f'{COMPONENT_TITLE}',
        f'{COMPONENT_DESCRIPTION}',
        f'{COMPONENT_TYPE}',
        f'{RULE_ID}',
        f'{NAMESPACE}',
        f'{CHECK_ID}',
        f'{CHECK_DESCRIPTION}',
    ]

    _columns_optional = [
        f'{CHECK_ID}',
        f'{CHECK_DESCRIPTION}',
        f'{TARGET_COMPONENT}',
        f'{ORIGINAL_RISK_RATING}',
        f'{ADJUSTED_RISK_RATING}',
        f'{RISK_ADJUSTMENT}',
    ]

    _columns_parameter = [
        f'{PARAMETER_ID}',
        f'{PARAMETER_DESCRIPTION}',
        f'{PARAMETER_VALUE_ALTERNATIVES}',
        f'{PARAMETER_VALUE_DEFAULT}',
    ]

    _columns_ordered = _columns_required + _columns_parameter + _columns_optional

    @staticmethod
    def get_order(column_name: str) -> int:
        """Get order for column_name."""
        rval = sys.maxsize
        if column_name in CsvColumn._columns_ordered:
            rval = CsvColumn._columns_ordered.index(column_name)
        return rval

    @staticmethod
    def is_column_name_required(name: str) -> bool:
        """Is column name required."""
        return name in (CsvColumn._columns_required + CsvColumn._columns_required_validation)

    @staticmethod
    def is_column_name_optional(name: str) -> bool:
        """Is column name optional."""
        return name in (CsvColumn._columns_optional)

    @staticmethod
    def is_column_name_parameter(name: str) -> bool:
        """Is column name parameter."""
        return any(name.startswith(cname) for cname in CsvColumn._columns_parameter)

    @staticmethod
    def _get_column_names(column_list: List[str]) -> List[str]:
        """Generic method to get column names from a list.

        Args:
            column_list: The list of column names to return

        Returns:
            A copy of the column list
        """
        return list(column_list)

    @staticmethod
    def get_required_column_names() -> List[str]:
        """Get required column names."""
        return CsvColumn._get_column_names(CsvColumn._columns_required)

    @staticmethod
    def get_optional_column_names() -> List[str]:
        """Get optional column names."""
        return CsvColumn._get_column_names(CsvColumn._columns_optional)

    @staticmethod
    def get_parameter_column_names() -> List[str]:
        """Get parameter column names."""
        return CsvColumn._get_column_names(CsvColumn._columns_parameters)

    @staticmethod
    def get_required_column_names_validation() -> List[str]:
        """Get required column names validation."""
        return CsvColumn._get_column_names(CsvColumn._columns_required_validation)

    _rule_property_column_names = [
        f'{RULE_ID}',
        f'{RULE_DESCRIPTION}',
        f'{PARAMETER_ID}',
        f'{PARAMETER_DESCRIPTION}',
        f'{PARAMETER_VALUE_ALTERNATIVES}',
        f'{CHECK_ID}',
        f'{CHECK_DESCRIPTION}',
        f'{TARGET_COMPONENT}',
        f'{ORIGINAL_RISK_RATING}',
        f'{ADJUSTED_RISK_RATING}',
        f'{RISK_ADJUSTMENT}',
    ]

    @staticmethod
    def get_rule_property_column_names() -> List[str]:
        """Get rule property column names."""
        return CsvColumn._rule_property_column_names

    # columns required which do not become properties
    _columns_required_filtered = [
        f'{COMPONENT_TITLE}',
        f'{COMPONENT_DESCRIPTION}',
        f'{COMPONENT_TYPE}',
        f'{PROFILE_SOURCE}',
        f'{PROFILE_DESCRIPTION}',
        f'{CONTROL_ID_LIST}',
        f'{NAMESPACE}',
    ]

    # optional columns which do not become properties, initially
    _columns_optional_filtered = []

    _columns_filtered = _columns_required_filtered + _columns_optional_filtered

    @staticmethod
    def _get_filtered_column_names(column_names: List[str]) -> List[str]:
        """Get filtered column names excluding predefined filtered columns.

        Args:
            column_names: The list of column names to filter

        Returns:
            List of column names that are not in the filtered list
        """
        return [name for name in column_names if name not in CsvColumn._columns_filtered]

    @staticmethod
    def get_filtered_required_column_names() -> List[str]:
        """Get filtered required column names."""
        return CsvColumn._get_filtered_column_names(CsvColumn.get_required_column_names())

    @staticmethod
    def get_filtered_optional_column_names() -> List[str]:
        """Get filtered optional column names."""
        return CsvColumn._get_filtered_column_names(CsvColumn.get_optional_column_names())

    _check_property_column_names = [f'{RULE_ID}', f'{CHECK_ID}', f'{CHECK_DESCRIPTION}', f'{TARGET_COMPONENT}']

    @staticmethod
    def get_check_property_column_names() -> List[str]:
        """Get check property column names."""
        return CsvColumn._check_property_column_names

    # optional columns which do become properties, afterwards
    _columns_parameters = [f'{PARAMETER_ID}', f'{PARAMETER_DESCRIPTION}', f'{PARAMETER_VALUE_ALTERNATIVES}']

    # optional columns which require Param_Id be present in the row
    _columns_parameters_dependent = [
        f'{PARAMETER_DESCRIPTION}',
        f'{PARAMETER_VALUE_ALTERNATIVES}',
        f'{PARAMETER_VALUE_DEFAULT}',
    ]

class CsvToOscalComponentDefinition(TaskBase):
    """
    Task to create OSCAL ComponentDefinition json.

    Attributes:
        name: Name of the task.
    """

    name = 'csv-to-oscal-cd'

    def __init__(self, config_object: Optional[configparser.SectionProxy]) -> None:
        """
        Initialize trestle task csv-to-oscal-cd.

        Args:
            config_object: Config section associated with the task.
        """
        super().__init__(config_object)

    def print_info(self) -> None:
        """Print the help string."""
        name = self.name
        oscal_name = 'component_definition'
        # help note identifiers
        note01 = '1'
        note02 = '2'
        note03 = '3'
        note04 = '4'
        note05 = '5'
        # help generation
        logger.info(f'Help information for {name} task.')
        logger.info('')
        logger.info(f'Purpose: From csv produce OSCAL {oscal_name} file.')
        logger.info('')
        logger.info('')
        logger.info(f'Configuration flags sit under [task.{name}]:')
        text1 = '  title                = '
        text2 = '(required) the component definition title.'
        logger.info(text1 + text2)
        text1 = '  version              = '
        text2 = '(required) the component definition version.'
        logger.info(text1 + text2)
        text1 = '  csv-file             = '
        text2 = '(required) the path of the csv file.'
        text3 = ' [1st row are column headings; 2nd row are column descriptions; 3rd row and beyond is data]'
        logger.info(text1 + text2 + text3)
        text1 = '  required columns:      '
        for text2 in CsvColumn.get_required_column_names():
            if text2 in [f'{RULE_DESCRIPTION}', f'{PROFILE_SOURCE}', f'{PROFILE_DESCRIPTION}', f'{CONTROL_ID_LIST}']:
                text2 += f' (see note {note01})'
            logger.info(text1 + '$$' + text2)
            text1 = '                         '
        text1 = '  optional columns:      '
        for text2 in CsvColumn.get_optional_column_names():
            if text2 in [f'{ORIGINAL_RISK_RATING}', f'{ADJUSTED_RISK_RATING}', f'{RISK_ADJUSTMENT}']:
                text2 += f' (see note {note01})'
            elif text2 in [f'{TARGET_COMPONENT}']:
                text2 += f' (see note {note03})'
            else:
                text2 += f' (see note {note02})'
            logger.info(text1 + '$' + text2)
            text1 = '                         '
        for text2 in CsvColumn.get_parameter_column_names():
            text2 += f' (see notes {note01}, {note05})'
            logger.info(text1 + '$' + text2)
            text1 = '                         '
        text1 = '  comment columns:       '
        text2 = f'Informational (see note {note04})'
        logger.info(text1 + '#' + text2)
        text1 = '  output-dir           = '
        text2 = '(required) the path of the output directory for synthesized OSCAL .json files.'
        logger.info(text1 + text2)
        text1 = '  component-definition = '
        text2 = '(optional) the path of the existing component-definition OSCAL .json file.'
        logger.info(text1 + text2)
        text1 = '  class.column-name    = '
        text2 = f'(optional) the class to associate with the specified column name, e.g. class.{RULE_ID} = scc_class'
        logger.info(text1 + text2)
        text1 = '  output-overwrite     = '
        text2 = '(optional) true [default] or false; replace existing output when true.'
        logger.info(text1 + text2)
        text1 = '  validate-controls    = '
        text2 = '(optional) on, warn, or off [default]; validate controls exist in resolved profile.'
        logger.info(text1 + text2)
        # Notes
        text1 = ''
        text2 = ''
        logger.info(text1 + text2)
        text1 = 'Notes: '
        text2 = f'[{note01}] column is ignored for validation component type'
        logger.info(text1 + text2)
        text1 = '       '
        text2 = f'[{note02}] column is required for validation component type'
        logger.info(text1 + text2)
        text1 = '       '
        text2 = f'[{note03}] column is optional for validation component type'
        text3 = f', but may be needed to prevent {RULE_ID} collisions'
        logger.info(text1 + text2 + text3)
        text1 = '       '
        text2 = f'[{note04}] column name starting with # causes column to be ignored'
        logger.info(text1 + text2)
        text1 = '       '
        text2 = f'[{note05}] additional parameters are specified by adding a common suffix per set'
        text3 = f', for example: {PARAMETER_ID}_1, {PARAMETER_DESCRIPTION}_1, ...{PARAMETER_ID}_2...'
        logger.info(text1 + text2 + text3)

    def configure(self) -> bool:
        """Configure."""
        ok, msg = self._configure_csv_common()
        if not ok:
            logger.warning(msg)
            return False
        # config cd
        self._cd_path = None
        self._cd_file = self._config.get('component-definition')
        if self._cd_file is not None:
            self._cd_path = pathlib.Path(self._cd_file)
            if not self._cd_path.exists():
                logger.warning('"component-definition" not found')
                return False
        # validate_controls
        self._validate_controls = self._config.get('validate-controls', 'off')
        return True

    def get_class(self, name: str) -> str:
        """Get class value for specified name from config."""
        key = f'class.{name}'
        return self._config.get(key)

    def simulate(self) -> TaskOutcome:
        """Provide a simulated outcome."""
        return TaskOutcome('simulated-success')

    def execute(self) -> TaskOutcome:
        """Provide an executed outcome."""
        try:
            return self._execute()
        except Exception:
            logger.error(traceback.format_exc())
            return TaskOutcome('failure')

    def _execute(self) -> TaskOutcome:
        """Execute path core."""
        if not self.configure():
            return TaskOutcome('failure')
        # config output
        odir = self._config.get('output-dir')
        opth = pathlib.Path(odir)
        self._overwrite = self._config.getboolean('output-overwrite', True)
        # insure output dir exists
        opth.mkdir(exist_ok=True, parents=True)
        # calculate output file name & check writability
        oname = 'component-definition.json'
        ofile = opth / oname
        if not self._overwrite and pathlib.Path(ofile).exists():
            logger.warning(f'output: {ofile} already exists')
            return TaskOutcome('failure')
        # fetch existing component-definition, if any
        self._cd_mgr = _CdMgr(self._cd_path, self._title, self._timestamp, self._version)
        # fetch csv
        self._csv_mgr = _CsvMgr(self._csv_path)
        # create resolved profile -> catalog helper
        profile_list = self._csv_mgr.get_profile_list()
        self._resolved_profile_catalog_helper = _ResolvedProfileCatalogHelper(profile_list, self._workspace)
        self._unresolved_controls = []
        # calculate deletion, addition & modification rule lists
        rules = self._calculate_rules()
        # calculate deletion, addition & modification set-parameter lists
        set_params = self._calculate_set_params(rules[2])
        # calculate deletion, addition & modification control mapping lists
        control_mappings = self._calculate_control_mappings(rules[2])
        # rule set manager
        self._rule_set_id_mgr = _RuleSetIdMgr(self._cd_mgr.get_max_rule_set_number(), len(rules[1]))
        # rule additions, deletions & modifications (by row)
        self.rules_del(rules[0])
        self.rules_add(rules[1])
        self.rules_mod(rules[2])
        # set-parameters additions, deletions & modifications (by row)
        self.set_params_del(set_params[0])
        self.set_params_add(set_params[1])
        self.set_params_mod(set_params[2])
        # control mappings additions, deletions & modifications (by row)
        self.control_mappings_del(control_mappings[0])
        self.control_mappings_add(control_mappings[1])
        # note: control mappings mod is currently not possible
        # note: add/del user columns not currently supported
        if len(self._unresolved_controls) > 0:
            text = f'Unresolved controls: {self._unresolved_controls}'
            if self._validate_controls == 'warn':
                logger.warning(text)
            elif self._validate_controls == 'on':
                raise RuntimeError(text)
        # prepare new/revised component definition
        component_definition = self._cd_mgr.get_component_definition()
        # write OSCAL ComponentDefinition to file
        if self._verbose:
            logger.info(f'output: {ofile}')
        component_definition.oscal_write(pathlib.Path(ofile))
        return TaskOutcome('success')

    def _calculate_rules(self) -> tuple:
        """Calculate rules add, delete, modify."""
        cd_rules = self._cd_mgr.get_rule_keys()
        csv_rules = self._csv_mgr.get_rule_keys()
        del_rules = []
        add_rules = []
        mod_rules = []
        for key in cd_rules:
            if key in csv_rules:
                continue
            else:
                del_rules.append(key)
                logger.debug(f'rules del: {key}')
        for key in csv_rules:
            if key in cd_rules:
                mod_rules.append(key)
                logger.debug(f'rules mod: {key}')
            else:
                add_rules.append(key)
                logger.debug(f'rules add: {key}')
        return (del_rules, add_rules, mod_rules)

    def _calculate_set_params(self, mod_rules: List) -> tuple:
        """Calculate set parameters add, delete, modify."""
        cd_set_params = self._cd_mgr.get_set_params_keys()
        csv_set_params = self._csv_mgr.get_set_params_keys()
        del_set_params = []
        add_set_params = []
        mod_set_params = []
        for key in cd_set_params:
            rule_key = synthesize_rule_key(key[0], key[1], key[2], None, None)
            if rule_key not in mod_rules:
                continue
            if key in csv_set_params:
                continue
            else:
                del_set_params.append(key)
                logger.debug(f'params del: {key}')
        for key in csv_set_params:
            rule_key = synthesize_rule_key(key[0], key[1], key[2], None, None)
            if rule_key not in mod_rules:
                continue
            if key in cd_set_params:
                mod_set_params.append(key)
                logger.debug(f'params mod: {key}')
            else:
                add_set_params.append(key)
                logger.debug(f'params add: {key}')
        return (del_set_params, add_set_params, mod_set_params)

    def _calculate_control_mappings(self, mod_rules: List) -> tuple:
        """Calculate control mappings add, delete, modify."""
        cd_controls = self._cd_mgr.get_control_keys()
        csv_controls = self._csv_mgr.get_control_keys()
        del_control_mappings = []
        add_control_mappings = []
        mod_control_mappings = []
        for key in cd_controls:
            rule_key = synthesize_rule_key(key[0], key[1], key[2], None, None)
            if rule_key not in mod_rules:
                continue
            if key in csv_controls:
                continue
            else:
                del_control_mappings.append(key)
                logger.debug(f'ctl-maps del: {key}')
        for key in csv_controls:
            rule_key = synthesize_rule_key(key[0], key[1], key[2], None, None)
            if rule_key not in mod_rules:
                continue
            if key in cd_controls:
                mod_control_mappings.append(key)
                logger.debug(f'ctl-maps mod: {key}')
            else:
                add_control_mappings.append(key)
                logger.debug(f'ctl-maps add: {key}')
        return (del_control_mappings, add_control_mappings, mod_control_mappings)

    def _get_namespace(self, rule_key: tuple) -> str:
        """Get namespace."""
        return self._csv_mgr.get_value(rule_key, NAMESPACE).strip()

    def _get_prop_name(self, column_name: str) -> str:
        """Get property name."""
        return column_name.lstrip('$')

    def rules_del(self, del_rules: List[str]) -> None:
        """Delete rules."""
        for tokens in del_rules:
            component_title = tokens[0]
            component_type = tokens[1]
            rule_id = tokens[2]
            description = ''
            # component
            component = self._cd_mgr.get_component(component_title, component_type, description)
            # props
            deleted_props = self._delete_rule_props(component, rule_id)
            # In Pydantic v2, props with min_length=1 cannot be empty list, must be None
            component.props = deleted_props if deleted_props else None

    def _delete_rule_props(self, component: DefinedComponent, rule_id: str) -> List[Property]:
        """Delete rule props."""
        props = []
        # In Pydantic v2, props with min_length=1 becomes None when empty, not []
        if component.props is None:
            return props
        rule_set = _RuleSetHelper.get_rule_set(component.props, rule_id)
        for prop in component.props:
            if prop.remarks != rule_set:
                props.append(prop)
            elif prop.name in self._csv_mgr.get_parameter_id_column_names():
                # In Pydantic v2, prop.value is StringDatatype (RootModel)
                prop_value = prop.value.root if hasattr(prop.value, 'root') else prop.value
                self._delete_rule_set_parameter(component, prop_value)
            elif prop.name == RULE_ID:
                # In Pydantic v2, prop.value is StringDatatype (RootModel)
                prop_value = prop.value.root if hasattr(prop.value, 'root') else prop.value
                self._delete_rule_implemented_requirement(component, prop_value)
        return props

    def _control_implementation_generator(
        self, control_implementations: List[ControlImplementation]
    ) -> Iterator[ControlImplementation]:
        """Control implementation generator."""
        if control_implementations:
            for control_implementation in control_implementations:
                yield control_implementation

    def _set_parameter_generator(self, set_parameters: List[SetParameter]) -> Iterator[SetParameter]:
        """Set parameter generator."""
        if set_parameters:
            for set_parameter in set_parameters:
                yield set_parameter

    def _implemented_requirement_generator(
        self, implemented_requirements: List[ImplementedRequirement]
    ) -> Iterator[ImplementedRequirement]:
        """Implemented-requirement generator."""
        if implemented_requirements:
            for implemented_requirement in implemented_requirements:
                yield implemented_requirement

    def _delete_rule_set_parameter(self, component: DefinedComponent, parameter_id: str) -> None:
        """Delete rule set-parameter."""
        control_implementations = component.control_implementations
        for control_implementation in self._control_implementation_generator(control_implementations):
            if control_implementation.set_parameters:
                # Build new list with parameters that don't match parameter_id
                new_set_parameters = []
                for set_parameter in control_implementation.set_parameters:
                    if set_parameter.param_id != parameter_id:
                        _OscalHelper.add_set_parameter(new_set_parameters, set_parameter)
                # Only update if we have parameters, otherwise set to None
                if new_set_parameters:
                    control_implementation.set_parameters = new_set_parameters
                else:
                    control_implementation.set_parameters = None

    def _has_content(self, implemented_requirement: ImplementedRequirement) -> bool:
        """Check if implemented requirement has props or statements."""
        return len(as_list(implemented_requirement.props)) > 0 or len(as_list(implemented_requirement.statements)) > 0

    def _filter_implemented_requirements_by_rule(
        self, implemented_requirements: List[ImplementedRequirement], rule_id: str
    ) -> List[ImplementedRequirement]:
        """Filter implemented requirements by removing the specified rule."""
        new_implemented_requirements = []
        for implemented_requirement in implemented_requirements:
            self._delete_ir_props(implemented_requirement, rule_id)
            self._delete_ir_statements(implemented_requirement, rule_id)
            if self._has_content(implemented_requirement):
                new_implemented_requirements.append(implemented_requirement)
        return new_implemented_requirements

    def _delete_rule_implemented_requirement(self, component: DefinedComponent, rule_id: str) -> None:
        """Delete rule implemented_requirement."""
        control_implementations = component.control_implementations
        # Pydantic v2: Build new list to avoid empty list validation error
        new_control_implementations = []
        for control_implementation in self._control_implementation_generator(control_implementations):
            if control_implementation.implemented_requirements:
                new_implemented_requirements = self._filter_implemented_requirements_by_rule(
                    control_implementation.implemented_requirements, rule_id
                )
                # Pydantic v2: implemented_requirements is required with min_length=1
                # Only include control_implementation if it has implemented_requirements
                if new_implemented_requirements:
                    control_implementation.implemented_requirements = new_implemented_requirements
                    new_control_implementations.append(control_implementation)
        # Pydantic v2: Only assign if not empty to avoid validation error
        if new_control_implementations:
            component.control_implementations = new_control_implementations
        else:
            component.control_implementations = None

    def _delete_ir_statements(self, implemented_requirement: ImplementedRequirement, rule_id: str) -> None:
        """Delete implemented-requirement statements."""
        if implemented_requirement.statements:
            statements = implemented_requirement.statements
            new_statements = []
            for statement in statements:
                # Pydantic v2: Check if result is empty before assigning to avoid validation error
                deleted_props = self._delete_props(statement.props, rule_id)
                if deleted_props:
                    statement.props = deleted_props
                else:
                    statement.props = None
                if statement.props:
                    new_statements.append(statement)
            # Only set if not empty to avoid Pydantic v2 validation error
            if new_statements:
                implemented_requirement.statements = new_statements
            else:
                implemented_requirement.statements = None

    def _delete_ir_props(self, implemented_requirement: ImplementedRequirement, rule_id: str) -> None:
        """Delete implemented-requirement props."""
        if implemented_requirement.props:
            # Pydantic v2: Check if result is empty before assigning to avoid validation error
            deleted_props = self._delete_props(implemented_requirement.props, rule_id)
            if deleted_props:
                implemented_requirement.props = deleted_props
            else:
                implemented_requirement.props = None

    def _delete_props(self, props: List[Property], rule_id: str) -> List[property]:
        """Delete props."""
        rval = []
        if props:
            for prop in props:
                if prop.name == RULE_ID and prop.value == rule_id:
                    continue
                rval.append(prop)
        return rval

    def rules_add(self, add_rules: List[str]) -> None:
        """Add rules."""
        for rule_key in add_rules:
            component_title = self._csv_mgr.get_value(rule_key, COMPONENT_TITLE)
            component_type = self._csv_mgr.get_value(rule_key, COMPONENT_TYPE)
            component_description = self._csv_mgr.get_value(rule_key, COMPONENT_DESCRIPTION)
            # component
            component = self._cd_mgr.get_component(component_title, component_type, component_description)
            # props - build list before assignment to avoid empty list validation error
            new_props = self._create_rule_props(rule_key)
            props_list = as_list(component.props) + new_props
            component.props = props_list
            # additional props, when not validation component
            if not self._is_validation(rule_key):
                # control implementation
                source = self._csv_mgr.get_value(rule_key, PROFILE_SOURCE)
                description = self._csv_mgr.get_value(rule_key, PROFILE_DESCRIPTION)
                control_implementation = self._get_control_implementation(component, source, description)
                # set-parameters
                set_parameters = self._create_set_parameters(rule_key)
                if set_parameters:
                    set_params_list = as_list(control_implementation.set_parameters)
                    _OscalHelper.add_set_parameters(set_params_list, set_parameters)
                    control_implementation.set_parameters = set_params_list
                # control-mappings
                control_mappings = self._csv_mgr.get_value(rule_key, CONTROL_ID_LIST).split()
                self._add_rule_prop(control_implementation, control_mappings, rule_key)

    def _is_validation(self, rule_key: tuple) -> bool:
        """Check for validation component."""
        component_type = self._csv_mgr.get_value(rule_key, COMPONENT_TYPE)
        return is_validation(component_type)

    def _add_rule_prop(
        self, control_implementation: ControlImplementation, control_mappings: List[str], rule_key: tuple
    ) -> None:
        """Add rule prop."""
        namespace = self._get_namespace(rule_key)
        for control_mapping in control_mappings:
            control_id = derive_control_id(control_mapping)
            implemented_requirement = self._get_implemented_requirement(control_implementation, control_id)
            # create rule implementation (as property)
            name = RULE_ID
            prop = Property(
                name=name, value=self._csv_mgr.get_value(rule_key, name), ns=namespace, class_=self.get_class(name)
            )
            part_id = derive_part_id(control_mapping)
            if part_id is None:
                props = as_list(implemented_requirement.props) + [prop]
                implemented_requirement.props = props
            else:
                statement = self._get_statement(implemented_requirement, part_id)
                statement.props.append(prop)

    def _create_rule_props(self, rule_key: tuple) -> List[Property]:
        """Create rule props."""
        rule_set = self._rule_set_id_mgr.get_next_rule_set_id()
        row_number = self._csv_mgr.get_row_number(rule_key)
        rule_set_mgr = _RuleSetMgr(row_number, rule_set)
        namespace = self._get_namespace(rule_key)
        if self._is_validation(rule_key):
            column_names = CsvColumn.get_check_property_column_names()
        else:
            column_names = CsvColumn.get_rule_property_column_names()
        # req'd & optional props
        for column_name in column_names:
            prop_name = self._get_prop_name(column_name)
            prop_value = self._csv_mgr.get_value(rule_key, column_name).strip()
            rule_set_mgr.add_prop(prop_name, prop_value, namespace, self.get_class(prop_name))
        if not self._is_validation(rule_key):
            # parameter columns
            column_names = self._csv_mgr.get_parameter_column_names()
            for column_name in column_names:
                prop_name = self._get_prop_name(column_name)
                prop_value = self._csv_mgr.get_value(rule_key, column_name).strip()
                rule_set_mgr.add_prop(prop_name, prop_value, namespace, self.get_class(prop_name))
        # user props
        column_names = self._csv_mgr.get_user_column_names()
        for column_name in column_names:
            if column_name.startswith('#'):
                continue
            prop_name = self._get_prop_name(column_name)
            prop_value = self._csv_mgr.get_value(rule_key, column_name).strip()
            rule_set_mgr.add_prop(prop_name, prop_value, namespace, self.get_class(prop_name))
        return rule_set_mgr.get_props()

    def _get_control_implementation(
        self, component: DefinedComponent, source: str, description: str
    ) -> ControlImplementation:
        """Find or create control implementation."""
        control_implementations = as_list(component.control_implementations)
        for control_implementation in control_implementations:
            if control_implementation.source == source and control_implementation.description == description:
                return control_implementation
        # Use model_construct to bypass validation for empty implemented_requirements list
        control_implementation = ControlImplementation.model_construct(
            uuid=str(uuid.uuid4()), source=source, description=description, implemented_requirements=[]
        )
        control_implementations.append(control_implementation)
        component.control_implementations = control_implementations
        return control_implementation

    def _str_to_list(self, value: str) -> List[str]:
        """Transform string to list."""
        rval = []
        if ',' in value:
            values = value.split(',')
            # remove leading/trailing whitespace
            for v in values:
                rval.append(v.strip())
        else:
            rval.append(value)
        return rval

    def _create_set_parameters(self, rule_key: tuple) -> List[SetParameter]:
        """Create set parameters."""
        set_parameters = []
        for parameter_id_column_name in self._csv_mgr.get_parameter_id_column_names():
            suffix = parameter_id_column_name.replace(PARAMETER_ID, '')
            parameter_value_default_column_name = f'{PARAMETER_VALUE_DEFAULT}{suffix}'
            name = self._csv_mgr.get_value(rule_key, parameter_id_column_name)
            value = self._csv_mgr.get_value(rule_key, parameter_value_default_column_name)
            if name and value:
                try:
                    values = self._str_to_list(value)
                    set_parameter = SetParameter(param_id=name, values=values)
                    set_parameters.append(set_parameter)
                except Exception:
                    row_number = self._csv_mgr.get_row_number(rule_key)
                    text = (
                        f'row {row_number}: "{name}" is invalid for column {parameter_id_column_name} '
                        f'and/or "{",".join(values)}" is invalid for column {parameter_value_default_column_name}'
                    )
                    raise RuntimeError(text)
            elif name:
                row_number = self._csv_mgr.get_row_number(rule_key)
                text = f'row "{row_number}" missing value for "{parameter_value_default_column_name}"'
                logger.debug(text)
        return set_parameters

    def _get_implemented_requirement(
        self, control_implementation: ControlImplementation, control_id: str
    ) -> ImplementedRequirement:
        """Find or create implemented requirement."""
        if self._validate_controls != 'off':
            if not self._resolved_profile_catalog_helper.validate(control_id):
                if control_id not in self._unresolved_controls:
                    self._unresolved_controls.append(control_id)
        for implemented_requirement in control_implementation.implemented_requirements:
            if implemented_requirement.control_id == control_id:
                return implemented_requirement
        implemented_requirement = ImplementedRequirement(uuid=str(uuid.uuid4()), control_id=control_id, description='')
        control_implementation.implemented_requirements.append(implemented_requirement)
        return implemented_requirement

    def _get_statement(self, implemented_requirement: ImplementedRequirement, part_id: str) -> Statement:
        """Find or create statement."""
        statements = as_list(implemented_requirement.statements)
        for statement in statements:
            if statement.statement_id == part_id:
                return statement
        # Use model_construct to bypass validation for empty props list
        statement = Statement.model_construct(uuid=str(uuid.uuid4()), statement_id=part_id, description='', props=[])
        statements.append(statement)
        implemented_requirement.statements = statements
        return statement

    def rules_mod(self, mod_rules: List[str]) -> None:
        """Modify rules."""
        for rule_key in mod_rules:
            component_title = self._csv_mgr.get_value(rule_key, COMPONENT_TITLE)
            component_type = self._csv_mgr.get_value(rule_key, COMPONENT_TYPE)
            component_description = self._csv_mgr.get_value(rule_key, COMPONENT_DESCRIPTION)
            # component
            component = self._cd_mgr.get_component(component_title, component_type, component_description)
            # props
            component.props = self._modify_rule_props(component, rule_key)

    def _modify_rule_props(self, component: DefinedComponent, rule_key: tuple) -> List[Property]:
        """Modify rule props."""
        rule_id = self._csv_mgr.get_value(rule_key, RULE_ID)
        rule_set = _RuleSetHelper.get_rule_set(component.props, rule_id)
        rule_ns = self._csv_mgr.get_value(rule_key, NAMESPACE)
        column_names = CsvColumn.get_filtered_required_column_names() + CsvColumn.get_filtered_optional_column_names()
        # req'd & optional props
        for column_name in column_names:
            column_value = self._csv_mgr.get_value(rule_key, column_name).strip()
            class_ = self.get_class(column_name)
            self._cd_mgr.update_rule_definition(component, rule_set, column_name, column_value, rule_ns, class_)
        # parameter columns
        column_names = self._csv_mgr.get_parameter_column_names()
        for column_name in column_names:
            column_value = self._csv_mgr.get_value(rule_key, column_name).strip()
            class_ = self.get_class(column_name)
            self._cd_mgr.update_rule_definition(component, rule_set, column_name, column_value, rule_ns, class_)
        # user props
        column_names = self._csv_mgr.get_user_column_names()
        for column_name in column_names:
            column_value = self._csv_mgr.get_value(rule_key, column_name).strip()
            self._cd_mgr.update_rule_definition(component, rule_set, column_name, column_value, rule_ns, class_)
        return component.props

    def set_params_del(self, del_set_params: List[str]) -> None:
        """Set parameters delete."""
        for tokens in del_set_params:
            component_title = tokens[0]
            component_type = tokens[1]
            source = tokens[3]
            description = tokens[4]
            param_id = tokens[5]
            # ctl impl
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            if control_implementation:
                # Build new list with parameters that don't match param_id
                new_set_parameters = []
                for set_parameter in self._set_parameter_generator(control_implementation.set_parameters):
                    if set_parameter.param_id == param_id:
                        continue
                    _OscalHelper.add_set_parameter(new_set_parameters, set_parameter)
                # Only update if we have parameters, otherwise set to None
                if new_set_parameters:
                    control_implementation.set_parameters = new_set_parameters
                else:
                    control_implementation.set_parameters = None

    def set_params_add(self, add_set_params: List[str]) -> None:
        """Set parameters add."""
        for tokens in add_set_params:
            component_title = tokens[0]
            component_type = tokens[1]
            rule_id = tokens[2]
            source = tokens[3]
            description = tokens[4]
            param_id = tokens[5]
            # ctl impl
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            # add
            set_params_list = as_list(control_implementation.set_parameters)
            rule_key = synthesize_rule_key(component_title, component_type, rule_id, None, None)
            values = [self._csv_mgr.get_default_value_by_id(rule_key, param_id)]
            set_parameter = SetParameter(param_id=param_id, values=values)
            _OscalHelper.add_set_parameter(set_params_list, set_parameter)
            control_implementation.set_parameters = set_params_list

    def set_params_mod(self, mod_set_params: List[str]) -> None:
        """Set parameters modify."""
        for tokens in mod_set_params:
            component_title = tokens[0]
            component_type = tokens[1]
            rule_id = tokens[2]
            source = tokens[3]
            description = tokens[4]
            param_id = tokens[5]
            # ctl impl
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            if control_implementation:
                set_parameters = control_implementation.set_parameters
                for set_parameter in self._set_parameter_generator(set_parameters):
                    if set_parameter.param_id != param_id:
                        continue
                    rule_key = synthesize_rule_key(component_title, component_type, rule_id, None, None)
                    values = [self._csv_mgr.get_default_value_by_id(rule_key, param_id)]
                    replacement = SetParameter(param_id=param_id, values=values)
                    if set_parameter.values == replacement.values:
                        continue
                    logger.debug(f'params-mod: {rule_id} {param_id} {set_parameter.values} -> {replacement.values}')
                    set_parameter.values = replacement.values

    def _control_mappings_generator(self, control_mappings: List[str]) -> Iterator[List[str]]:
        """Control mappings generator."""
        for tokens in control_mappings:
            component_title = tokens[0]
            component_type = tokens[1]
            source = tokens[3]
            description = tokens[4]
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            if control_implementation:
                yield tokens

    def _process_target_requirement(
        self, implemented_requirement: ImplementedRequirement, rule_id: str, smt_id: str
    ) -> Optional[ImplementedRequirement]:
        """Process the target implemented requirement by removing the rule."""
        implemented_requirement.statements = _OscalHelper.remove_rule_statement(
            implemented_requirement.statements, rule_id, smt_id
        )
        # Get new props list after removing rule (handle None case)
        new_props = (
            _OscalHelper.remove_rule(implemented_requirement.props, rule_id) if implemented_requirement.props else []
        )
        # Only keep this requirement if it has props or statements
        if new_props or implemented_requirement.statements:
            # Only assign if new_props is not empty (Pydantic v2 requires min_length=1)
            if new_props:
                implemented_requirement.props = new_props
            return implemented_requirement
        return None

    def _filter_implemented_requirements(
        self, control_implementation: ControlImplementation, control_id: str, rule_id: str, smt_id: str
    ) -> List[ImplementedRequirement]:
        """Filter implemented requirements, removing the specified rule."""
        new_implemented_requirements = []
        for implemented_requirement in self._implemented_requirement_generator(
            control_implementation.implemented_requirements
        ):
            if implemented_requirement.control_id == control_id:
                processed = self._process_target_requirement(implemented_requirement, rule_id, smt_id)
                if processed:
                    new_implemented_requirements.append(processed)
            else:
                new_implemented_requirements.append(implemented_requirement)
        return new_implemented_requirements

    def _remove_empty_control_implementation(
        self, component_title: str, component_type: str, source: str, description: str
    ) -> None:
        """Remove control implementation from component if it's empty."""
        component = self._cd_mgr.get_component(component_title, component_type, None)
        if not component.control_implementations:
            return

        new_control_implementations = [
            ci
            for ci in component.control_implementations
            if not (ci.source == source and ci.description == description)
        ]
        # Only assign if list is not empty (Pydantic v2 requires min_length=1)
        if new_control_implementations:
            component.control_implementations = new_control_implementations
        else:
            component.control_implementations = None

    def control_mappings_del(self, del_control_mappings: List[str]) -> None:
        """Control mappings delete."""
        for tokens in self._control_mappings_generator(del_control_mappings):
            component_title = tokens[0]
            component_type = tokens[1]
            rule_id = tokens[2]
            source = tokens[3]
            description = tokens[4]
            smt_id = tokens[5]
            control_id = derive_control_id(smt_id)

            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )

            new_implemented_requirements = self._filter_implemented_requirements(
                control_implementation, control_id, rule_id, smt_id
            )

            # Update the list - it must have at least 1 item per OSCAL schema
            if new_implemented_requirements:
                control_implementation.implemented_requirements = new_implemented_requirements
            else:
                # If no implemented requirements left, remove the control_implementation from component
                self._remove_empty_control_implementation(component_title, component_type, source, description)

    def control_mappings_add(self, add_control_mappings: List[str]) -> None:
        """Control mappings add."""
        for tokens in self._control_mappings_generator(add_control_mappings):
            component_title = tokens[0]
            component_type = tokens[1]
            rule_id = tokens[2]
            source = tokens[3]
            description = tokens[4]
            smt_id = tokens[5]
            # ctl-id
            control_id = derive_control_id(smt_id)
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            implemented_requirement = self._get_implemented_requirement(control_implementation, control_id)
            # namespace
            rule_key = synthesize_rule_key(tokens[0], tokens[1], tokens[2], None, None)
            ns = self._get_namespace(rule_key)
            # create rule implementation (as property)
            name = RULE_ID
            prop = Property(name=name, value=rule_id, ns=ns, class_=self.get_class(name))
            if smt_id == control_id:
                # Initialize props if None, or use existing list
                if implemented_requirement.props is None:
                    implemented_requirement.props = [prop]
                else:
                    implemented_requirement.props.append(prop)
            else:
                statement = self._get_statement(implemented_requirement, smt_id)
                # Initialize props if None, or use existing list
                if statement.props is None:
                    statement.props = [prop]
                else:
                    statement.props.append(prop)