mesonbuild/rewriter.py

# SPDX-License-Identifier: Apache-2.0
# Copyright 2016 The Meson development team

# This tool is used to manipulate an existing Meson build definition.
#
# - add a file to a target
# - remove files from a target
# - move targets
# - reindent?
from __future__ import annotations

from .ast import IntrospectionInterpreter, BUILD_TARGET_FUNCTIONS, AstConditionLevel, AstIDGenerator, AstIndentationGenerator, AstPrinter
from .ast.interpreter import IntrospectionBuildTarget, IntrospectionDependency, _symbol
from .interpreterbase import UnknownValue
from .interpreterbase.helpers import flatten
from mesonbuild.mesonlib import MesonException, pathname_sort_key, relpath, setup_vsenv
from . import mlog, environment
from functools import wraps
from .mparser import Token, ArrayNode, ArgumentNode, ArithmeticNode, AssignmentNode, BaseNode, StringNode, BooleanNode, DictNode, ElementaryNode, IdNode, FunctionNode, PlusAssignmentNode
from .mintro import IntrospectionEncoder
import json, os, re, sys, codecs
import typing as T
from pathlib import Path

if T.TYPE_CHECKING:
    import argparse
    from argparse import ArgumentParser, _FormatterClass

    from typing_extensions import TypeAlias

    from .mlog import AnsiDecorator

    RewriterKeysT: TypeAlias = dict[str, tuple[type, object | None, list[str] | None]]
    RewriterFuncT: TypeAlias = T.Callable[['Rewriter', dict[str, object]], None]

class RewriterException(MesonException):
    pass

# Note: when adding arguments, please also add them to the completion
# scripts in $MESONSRC/data/shell-completions/
def add_arguments(parser: ArgumentParser, formatter: _FormatterClass) -> None:
    parser.add_argument('-s', '--sourcedir', type=str, default='.', metavar='SRCDIR', help='Path to source directory.')
    parser.add_argument('-V', '--verbose', action='store_true', default=False, help='Enable verbose output')
    parser.add_argument('-S', '--skip-errors', dest='skip', action='store_true', default=False, help='Skip errors instead of aborting')
    subparsers = parser.add_subparsers(dest='type', title='Rewriter commands', description='Rewrite command to execute')

    # Target
    tgt_parser = subparsers.add_parser('target', aliases=['tgt'], help='Modify a target', formatter_class=formatter)
    tgt_parser.add_argument('-s', '--subdir', default='', dest='subdir', help='Subdirectory of the new target (only for the "add_target" action)')
    tgt_parser.add_argument('--type', dest='tgt_type', choices=rewriter_keys['target']['target_type'][2], default='executable',
                            help='Type of the target to add (only for the "add_target" action)')
    tgt_parser.add_argument('target', help='Name or ID of the target')
    tgt_parser.add_argument('operation', choices=['add', 'rm', 'add_target', 'rm_target', 'add_extra_files', 'rm_extra_files', 'info'],
                            help='Action to execute')
    tgt_parser.add_argument('sources', nargs='*', help='Sources to add/remove')

    # KWARGS
    kw_parser = subparsers.add_parser('kwargs', help='Modify keyword arguments', formatter_class=formatter)
    kw_parser.add_argument('operation', choices=rewriter_keys['kwargs']['operation'][2],
                           help='Action to execute')
    kw_parser.add_argument('function', choices=list(rewriter_func_kwargs.keys()),
                           help='Function type to modify')
    kw_parser.add_argument('id', help='ID of the function to modify (must be "/" for "project")')
    kw_parser.add_argument('kwargs', nargs='*', help='<keyword> <value> pairs, or list of <keyword> for "delete"')

    # Default options
    def_parser = subparsers.add_parser('default-options', aliases=['def'], help='Modify the project default options', formatter_class=formatter)
    def_parser.add_argument('operation', choices=rewriter_keys['default_options']['operation'][2],
                            help='Action to execute')
    def_parser.add_argument('options', nargs='*', help='<key> <value> pairs for "set"; list of <key> for "delete"')

    # JSON file/command
    cmd_parser = subparsers.add_parser('command', aliases=['cmd'], help='Execute a JSON array of commands', formatter_class=formatter)
    cmd_parser.add_argument('json', help='JSON string or file to execute')

class RequiredKeys:
    def __init__(self, keys: RewriterKeysT):
        self.keys = keys

    def __call__(self, f: RewriterFuncT) -> RewriterFuncT:
        @wraps(f)
        def wrapped(rewriter: 'Rewriter', cmd: dict[str, object]) -> None:
            for key, (typ, default, choices) in self.keys.items():
                if key not in cmd:
                    if default is not None:
                        cmd[key] = default
                    else:
                        raise RewriterException('Key "{}" is missing in object for {}'
                                                .format(key, f.__name__))
                if not isinstance(cmd[key], typ):
                    raise RewriterException('Invalid type of "{}". Required is {} but provided was {}'
                                            .format(key, typ.__name__, type(cmd[key]).__name__))
                if choices is not None:
                    if cmd[key] not in choices:
                        raise RewriterException('Invalid value of "{}": Possible values are {} but provided was "{}"'
                                                .format(key, choices, cmd[key]))
            return f(rewriter, cmd)
        return wrapped

class MTypeBase:
    node: BaseNode

    def __init__(self, node: T.Optional[BaseNode] = None):
        if node is None:
            self.node = self.new_node()
        else:
            self.node = node
        self.node_type = None
        for i in self.supported_nodes():
            if isinstance(self.node, i):
                self.node_type = i

    @classmethod
    def new_node(cls, value: T.Any = None) -> BaseNode:
        # Overwrite in derived class
        raise RewriterException('Internal error: new_node of MTypeBase was called')

    @classmethod
    def supported_nodes(cls) -> T.List[type]:
        # Overwrite in derived class
        return []

    def can_modify(self) -> bool:
        return self.node_type is not None

    def get_node(self) -> BaseNode:
        return self.node

    def add_value(self, value: T.Any) -> None:
        # Overwrite in derived class
        mlog.warning('Cannot add a value of type', mlog.bold(type(self).__name__), '--> skipping')

    def remove_value(self, value: T.Any) -> None:
        # Overwrite in derived class
        mlog.warning('Cannot remove a value of type', mlog.bold(type(self).__name__), '--> skipping')

    def remove_regex(self, value: T.Any) -> None:
        # Overwrite in derived class
        mlog.warning('Cannot remove a regex in type', mlog.bold(type(self).__name__), '--> skipping')

class MTypeStr(MTypeBase):
    def __init__(self, node: T.Optional[BaseNode] = None):
        super().__init__(node)

    @classmethod
    def new_node(cls, value: T.Optional[str] = None) -> BaseNode:
        if value is None:
            value = ''
        return StringNode(Token('string', '', 0, 0, 0, None, str(value)))

    @classmethod
    def supported_nodes(cls) -> T.List[type]:
        return [StringNode]

class MTypeBool(MTypeBase):
    def __init__(self, node: T.Optional[BaseNode] = None):
        super().__init__(node)

    @classmethod
    def new_node(cls, value: T.Optional[str] = None) -> BaseNode:
        return BooleanNode(Token('', '', 0, 0, 0, None, bool(value)))

    @classmethod
    def supported_nodes(cls) -> T.List[type]:
        return [BooleanNode]

class MTypeID(MTypeBase):
    def __init__(self, node: T.Optional[BaseNode] = None):
        super().__init__(node)

    @classmethod
    def new_node(cls, value: T.Optional[str] = None) -> BaseNode:
        if value is None:
            value = ''
        return IdNode(Token('', '', 0, 0, 0, None, str(value)))

    @classmethod
    def supported_nodes(cls) -> T.List[type]:
        return [IdNode]

class MTypeList(MTypeBase):
    node: ArrayNode

    def __init__(self, node: T.Optional[BaseNode] = None):
        super().__init__(node)

    @classmethod
    def new_node(cls, value: T.Optional[T.List[T.Any]] = None) -> ArrayNode:
        if value is None:
            value = []
        elif not isinstance(value, list):
            return cls._new_element_node(value)
        args = ArgumentNode(Token('', '', 0, 0, 0, None, ''))
        args.arguments = [cls._new_element_node(i) for i in value]
        return ArrayNode(_symbol('['), args, _symbol(']'))

    @classmethod
    def _new_element_node(cls, value: T.Any) -> BaseNode:
        # Overwrite in derived class
        raise RewriterException('Internal error: _new_element_node of MTypeList was called')

    def _ensure_array_node(self) -> None:
        if not isinstance(self.node, ArrayNode):
            tmp = self.node
            self.node = self.new_node()
            self.node.args.arguments = [tmp]

    @staticmethod
    def _check_is_equal(node: BaseNode, value: str) -> bool:
        # Overwrite in derived class
        return False

    @staticmethod
    def _check_regex_matches(node: BaseNode, regex: str) -> bool:
        # Overwrite in derived class
        return False

    def get_node(self) -> BaseNode:
        if isinstance(self.node, ArrayNode):
            if len(self.node.args.arguments) == 1:
                return self.node.args.arguments[0]
        return self.node

    @classmethod
    def supported_element_nodes(cls) -> T.List[T.Type]:
        # Overwrite in derived class
        return []

    @classmethod
    def supported_nodes(cls) -> T.List[T.Type]:
        return [ArrayNode] + cls.supported_element_nodes()

    def add_value(self, value: T.Any) -> None:
        if not isinstance(value, list):
            value = [value]
        self._ensure_array_node()
        for i in value:
            assert hasattr(self.node, 'args') # For mypy
            assert isinstance(self.node.args, ArgumentNode) # For mypy
            self.node.args.arguments += [self._new_element_node(i)]

    def _remove_helper(self, value: T.Any, equal_func: T.Callable[[T.Any, T.Any], bool]) -> None:
        def check_remove_node(node: BaseNode) -> bool:
            for j in value:
                if equal_func(i, j):
                    return True
            return False

        if not isinstance(value, list):
            value = [value]
        self._ensure_array_node()
        assert hasattr(self.node, 'args') # For mypy
        assert isinstance(self.node.args, ArgumentNode) # For mypy
        removed_list = []
        for i in self.node.args.arguments:
            if not check_remove_node(i):
                removed_list += [i]
        self.node.args.arguments = removed_list

    def remove_value(self, value: T.Any) -> None:
        self._remove_helper(value, self._check_is_equal)

    def remove_regex(self, regex: str) -> None:
        self._remove_helper(regex, self._check_regex_matches)

class MTypeStrList(MTypeList):
    def __init__(self, node: T.Optional[BaseNode] = None):
        super().__init__(node)

    @classmethod
    def _new_element_node(cls, value: str) -> StringNode:
        return StringNode(Token('string', '', 0, 0, 0, None, str(value)))

    @staticmethod
    def _check_is_equal(node: BaseNode, value: str) -> bool:
        if isinstance(node, StringNode):
            return bool(node.value == value)
        return False

    @staticmethod
    def _check_regex_matches(node: BaseNode, regex: str) -> bool:
        if isinstance(node, StringNode):
            return re.match(regex, node.value) is not None
        return False

    @classmethod
    def supported_element_nodes(cls) -> T.List[T.Type]:
        return [StringNode]

class MTypeIDList(MTypeList):
    def __init__(self, node: T.Optional[BaseNode] = None):
        super().__init__(node)

    @classmethod
    def _new_element_node(cls, value: str) -> IdNode:
        return IdNode(Token('', '', 0, 0, 0, None, str(value)))

    @staticmethod
    def _check_is_equal(node: BaseNode, value: str) -> bool:
        if isinstance(node, IdNode):
            return bool(node.value == value)
        return False

    @staticmethod
    def _check_regex_matches(node: BaseNode, regex: str) -> bool:
        if isinstance(node, StringNode):
            return re.match(regex, node.value) is not None
        return False

    @classmethod
    def supported_element_nodes(cls) -> T.List[T.Type]:
        return [IdNode]

rewriter_keys: T.Dict[str, RewriterKeysT] = {
    'default_options': {
        'operation': (str, None, ['set', 'delete']),
        'options': (dict, {}, None)
    },
    'kwargs': {
        'function': (str, None, None),
        'id': (str, None, None),
        'operation': (str, None, ['set', 'delete', 'add', 'remove', 'remove_regex', 'info']),
        'kwargs': (dict, {}, None)
    },
    'target': {
        'target': (str, None, None),
        'operation': (str, None, ['src_add', 'src_rm', 'target_rm', 'target_add', 'extra_files_add', 'extra_files_rm', 'info']),
        'sources': (list, [], None),
        'subdir': (str, '', None),
        'target_type': (str, 'executable', ['both_libraries', 'executable', 'jar', 'library', 'shared_library', 'shared_module', 'static_library']),
    }
}

rewriter_func_kwargs = {
    'dependency': {
        'language': MTypeStr,
        'method': MTypeStr,
        'native': MTypeBool,
        'not_found_message': MTypeStr,
        'required': MTypeBool,
        'static': MTypeBool,
        'version': MTypeStrList,
        'modules': MTypeStrList
    },
    'target': {
        'build_by_default': MTypeBool,
        'build_rpath': MTypeStr,
        'dependencies': MTypeIDList,
        'gui_app': MTypeBool,
        'link_with': MTypeIDList,
        'export_dynamic': MTypeBool,
        'implib': MTypeBool,
        'install': MTypeBool,
        'install_dir': MTypeStr,
        'install_rpath': MTypeStr,
        'pie': MTypeBool
    },
    'project': {
        'default_options': MTypeStrList,
        'meson_version': MTypeStr,
        'license': MTypeStrList,
        'license_files': MTypeStrList,
        'subproject_dir': MTypeStr,
        'version': MTypeStr
    }
}

class Rewriter:
    info_dump: T.Optional[T.Dict[str, T.Dict[str, T.Any]]]

    def __init__(self, sourcedir: str, generator: str = 'ninja', skip_errors: bool = False):
        self.sourcedir = sourcedir
        self.interpreter = IntrospectionInterpreter(sourcedir, '', generator, visitors = [AstIDGenerator(), AstIndentationGenerator(), AstConditionLevel()])
        self.skip_errors = skip_errors
        self.modified_nodes: T.List[BaseNode] = []
        self.to_remove_nodes: T.List[BaseNode] = []
        self.to_add_nodes: T.List[BaseNode] = []
        self.functions = {
            'default_options': self.process_default_options,
            'kwargs': self.process_kwargs,
            'target': self.process_target,
        }
        self.info_dump = None

    def analyze_meson(self) -> None:
        mlog.log('Analyzing meson file:', mlog.bold(os.path.join(self.sourcedir, environment.build_filename)))
        self.interpreter.analyze()
        mlog.log('  -- Project:', mlog.bold(self.interpreter.project_data['descriptive_name']))
        mlog.log('  -- Version:', mlog.cyan(self.interpreter.project_data['version']))

    def add_info(self, cmd_type: str, cmd_id: str, data: dict) -> None:
        if self.info_dump is None:
            self.info_dump = {}
        if cmd_type not in self.info_dump:
            self.info_dump[cmd_type] = {}
        self.info_dump[cmd_type][cmd_id] = data

    def print_info(self) -> None:
        if self.info_dump is None:
            return
        sys.stdout.write(json.dumps(self.info_dump, indent=2, cls=IntrospectionEncoder))

    def on_error(self) -> T.Tuple[AnsiDecorator, AnsiDecorator]:
        if self.skip_errors:
            return mlog.cyan('-->'), mlog.yellow('skipping')
        return mlog.cyan('-->'), mlog.red('aborting')

    def handle_error(self) -> None:
        if self.skip_errors:
            return None
        raise MesonException('Rewriting the meson.build failed')

    def all_assignments(self, varname: str) -> T.List[BaseNode]:
        assigned_values = []
        for ass in self.interpreter.all_assignment_nodes[varname]:
            if isinstance(ass, PlusAssignmentNode):
                continue
            assert isinstance(ass, AssignmentNode)
            assigned_values.append(ass.value)
        return assigned_values

    def find_target(self, target: str) -> T.Optional[IntrospectionBuildTarget]:
        for i in self.interpreter.targets:
            if target == i.id:
                return i

        potential_tgts = []
        for i in self.interpreter.targets:
            if target == i.name:
                potential_tgts.append(i)

        if not potential_tgts:
            potenial_tgts_1 = self.all_assignments(target)
            potenial_tgts_1 = [self.interpreter.node_to_runtime_value(el) for el in potenial_tgts_1]
            potential_tgts = [el for el in potenial_tgts_1 if isinstance(el, IntrospectionBuildTarget)]

        if not potential_tgts:
            return None
        elif len(potential_tgts) == 1:
            return potential_tgts[0]
        else:
            mlog.error('There are multiple targets matching', mlog.bold(target))
            for i in potential_tgts:
                mlog.error('  -- Target name', mlog.bold(i.name), 'with ID', mlog.bold(i.id))
            mlog.error('Please try again with the unique ID of the target', *self.on_error())
            self.handle_error()
            return None

    def find_dependency(self, dependency: str) -> T.Optional[IntrospectionDependency]:
        potential_deps = []
        for i in self.interpreter.dependencies:
            if i.name == dependency:
                potential_deps.append(i)

        checking_varnames = len(potential_deps) == 0

        if checking_varnames:
            potential_deps1 = self.all_assignments(dependency)
            potential_deps = [self.interpreter.node_to_runtime_value(el) for el in potential_deps1 if isinstance(el, FunctionNode) and el.func_name.value == 'dependency']

        if not potential_deps:
            return None
        elif len(potential_deps) == 1:
            return potential_deps[0]
        else:
            mlog.error('There are multiple dependencies matching', mlog.bold(dependency))
            for i in potential_deps:
                mlog.error('  -- Dependency name', i)
            if checking_varnames:
                mlog.error('Please try again with the name of the dependency', *self.on_error())
            self.handle_error()
            return None

    @RequiredKeys(rewriter_keys['default_options'])
    def process_default_options(self, cmd: T.Dict[str, T.Any]) -> None:
        # First, remove the old values
        kwargs_cmd: T.Dict[str, T.Any] = {
            'function': 'project',
            'id': "/",
            'operation': 'remove_regex',
            'kwargs': {
                'default_options': [f'{x}=.*' for x in cmd['options'].keys()]
            }
        }
        self.process_kwargs(kwargs_cmd)

        # Then add the new values
        if cmd['operation'] != 'set':
            return

        kwargs_cmd['operation'] = 'add'
        kwargs_cmd['kwargs']['default_options'] = []

        cdata = self.interpreter.coredata
        options = {
            **{str(k): v for k, v in cdata.optstore.items()},
        }

        for key, val in sorted(cmd['options'].items()):
            if key not in options:
                mlog.error('Unknown options', mlog.bold(key), *self.on_error())
                self.handle_error()
                continue

            try:
                val = options[key].validate_value(val)
            except MesonException as e:
                mlog.error('Unable to set', mlog.bold(key), mlog.red(str(e)), *self.on_error())
                self.handle_error()
                continue

            kwargs_cmd['kwargs']['default_options'] += [f'{key}={val}']

        self.process_kwargs(kwargs_cmd)

    @RequiredKeys(rewriter_keys['kwargs'])
    def process_kwargs(self, cmd: T.Dict[str, T.Any]) -> None:
        mlog.log('Processing function type', mlog.bold(cmd['function']), 'with id', mlog.cyan("'" + cmd['id'] + "'"))
        if cmd['function'] not in rewriter_func_kwargs:
            mlog.error('Unknown function type', cmd['function'], *self.on_error())
            return self.handle_error()
        kwargs_def = rewriter_func_kwargs[cmd['function']]

        # Find the function node to modify
        node = None
        arg_node = None
        if cmd['function'] == 'project':
            # msys bash may expand '/' to a path. It will mangle '//' to '/'
            # but in order to keep usage shell-agnostic, also allow `//` as
            # the function ID such that it will work in both msys bash and
            # other shells.
            if {'/', '//'}.isdisjoint({cmd['id']}):
                mlog.error('The ID for the function type project must be "/" or "//" not "' + cmd['id'] + '"', *self.on_error())
                return self.handle_error()
            node = self.interpreter.project_node
            arg_node = node.args
        elif cmd['function'] == 'target':
            tmp_tgt = self.find_target(cmd['id'])
            if not tmp_tgt:
                mlog.error('Unable to find the target', mlog.bold(cmd['id']), *self.on_error())
                return self.handle_error()
            node = tmp_tgt.node
            arg_node = node.args
        elif cmd['function'] == 'dependency':
            tmp_dep = self.find_dependency(cmd['id'])
            if not tmp_dep:
                mlog.error('Unable to find the dependency', mlog.bold(cmd['id']), *self.on_error())
                return self.handle_error()
            node = tmp_dep.node
            arg_node = node.args
        if not node:
            mlog.error('Unable to find the function node', *self.on_error())
            return self.handle_error()
        assert isinstance(node, FunctionNode)
        assert isinstance(arg_node, ArgumentNode)
        # Transform the key nodes to plain strings
        kwargs = {T.cast(IdNode, k).value: v for k, v in arg_node.kwargs.items()}

        # Print kwargs info
        if cmd['operation'] == 'info':
            info_data: T.Dict[str, T.Any] = {}
            for key, val in sorted(kwargs.items()):
                info_data[key] = None
                if isinstance(val, ElementaryNode):
                    info_data[key] = val.value
                elif isinstance(val, ArrayNode):
                    data_list = []
                    for i in val.args.arguments:
                        element = None
                        if isinstance(i, ElementaryNode):
                            element = i.value
                        data_list += [element]
                    info_data[key] = data_list
                elif isinstance(val, DictNode):
                    data_dict = {}
                    for k, v in val.args.kwargs.items():
                        if not isinstance(k, StringNode):
                            continue
                        value = None
                        if isinstance(v, ElementaryNode):
                            value = v.value
                        data_dict[k.value] = value
                    info_data[key] = data_dict

            self.add_info('kwargs', '{}#{}'.format(cmd['function'], cmd['id']), info_data)
            return # Nothing else to do

        # Modify the kwargs
        num_changed = 0
        for key, val in sorted(cmd['kwargs'].items()):
            if key not in kwargs_def:
                mlog.error('Cannot modify unknown kwarg', mlog.bold(key), *self.on_error())
                self.handle_error()
                continue

            if cmd['operation'] == 'delete':
                # Remove the key from the kwargs
                if key not in kwargs:
                    mlog.log('  -- Key', mlog.bold(key), 'is already deleted')
                    continue
                mlog.log('  -- Deleting', mlog.bold(key), 'from the kwargs')
                del kwargs[key]
            elif cmd['operation'] == 'set':
                # Replace the key from the kwargs
                mlog.log('  -- Setting', mlog.bold(key), 'to', mlog.yellow(str(val)))
                kwargs[key] = kwargs_def[key].new_node(val)
            else:
                # Modify the value from the kwargs

                if key not in kwargs:
                    kwargs[key] = None
                modifier = kwargs_def[key](kwargs[key])
                if not modifier.can_modify():
                    mlog.log('  -- Skipping', mlog.bold(key), 'because it is too complex to modify')
                    continue

                # Apply the operation
                val_str = str(val)
                if cmd['operation'] == 'add':
                    mlog.log('  -- Adding', mlog.yellow(val_str), 'to', mlog.bold(key))
                    modifier.add_value(val)
                elif cmd['operation'] == 'remove':
                    mlog.log('  -- Removing', mlog.yellow(val_str), 'from', mlog.bold(key))
                    modifier.remove_value(val)
                elif cmd['operation'] == 'remove_regex':
                    mlog.log('  -- Removing all values matching', mlog.yellow(val_str), 'from', mlog.bold(key))
                    modifier.remove_regex(val)

                # Write back the result
                kwargs[key] = modifier.get_node()

            num_changed += 1

        # Convert the keys back to IdNode's
        arg_node.kwargs = {IdNode(Token('', '', 0, 0, 0, None, k)): v for k, v in kwargs.items()}
        for k, v in arg_node.kwargs.items():
            k.level = v.level
        if num_changed > 0 and node not in self.modified_nodes:
            self.modified_nodes += [node]

    def find_assignment_node(self, node: BaseNode) -> T.Optional[AssignmentNode]:
        for k, v in self.interpreter.all_assignment_nodes.items():
            for ass in v:
                if ass.value == node:
                    return ass
        return None

    def affects_no_other_targets(self, candidate: BaseNode) -> bool:
        affected = self.interpreter.dataflow_dag.reachable({candidate}, False)
        affected_targets = [x for x in affected if isinstance(x, FunctionNode) and x.func_name.value in BUILD_TARGET_FUNCTIONS]
        return len(affected_targets) == 1

    def get_relto(self, target_node: BaseNode, node: BaseNode) -> Path:
        cwd = Path(os.getcwd())
        all_paths = self.interpreter.dataflow_dag.find_all_paths(node, target_node)
        # len(all_paths) == 0 would imply that data does not flow from node to
        # target_node. This would imply that adding sources to node would not
        # add the source to the target.
        assert all_paths
        if len(all_paths) > 1:
            return None
        return (cwd / next(x for x in all_paths[0] if isinstance(x, FunctionNode)).filename).parent

    def add_src_or_extra(self, op: str, target: IntrospectionBuildTarget, newfiles: T.List[str], to_sort_nodes: T.List[T.Union[FunctionNode, ArrayNode]]) -> None:
        assert op in {'src_add', 'extra_files_add'}

        if op == 'src_add':
            old: T.Set[T.Union[BaseNode, UnknownValue]] = set(target.source_nodes)
        elif op == 'extra_files_add':
            if target.extra_files is None:
                old = set()
            else:
                old = {target.extra_files}
            tgt_function: FunctionNode = target.node

        cwd = Path(os.getcwd())
        target_dir_abs = cwd / os.path.dirname(target.node.filename)
        source_root_abs = cwd / self.interpreter.source_root

        candidates1 = self.interpreter.dataflow_dag.reachable(old, True)
        # A node is a member of the set `candidates1` exactly if data from this node
        # flow into one of the `dest` nodes. We assume that this implies that if we
        # add `foo.c` to this node, then 'foo.c' will be added to one of these
        # nodes. This assumption is not always true:
        # ar = ['a.c', 'b.c']
        # srcs = ar[1]
        # executable('name', srcs)
        # Data flows from `ar` to `srcs`, but if we add 'foo.c':
        # ar = ['a.c', 'b.c', 'foo.c']
        # srcs = ar[1]
        # executable('name', srcs)
        # this does not add 'foo.c' to `srcs`. This is a known bug/limitation of
        # the meson rewriter that could be fixed by replacing `reachable` with a
        # more advanced analysis. But this is a lot of work and I think e.g.
        # `srcs = ar[1]` is rare in real-world projects, so I will just leave
        # this for now.

        candidates2 = {x for x in candidates1 if isinstance(x, (FunctionNode, ArrayNode))}

        # If we have this meson.build file:
        # shared = ['shared.c']
        # executable('foo', shared + ['foo.c'])
        # executable('bar', shared + ['bar.c'])
        # and we are tasked with adding 'new.c' to 'foo', we should do e.g this:
        # shared = ['shared.c']
        # executable('foo', shared + ['foo.c', 'new.c'])
        # executable('bar', shared + ['bar.c'])
        # but never this:
        # shared = ['shared.c', 'new.c']
        # executable('foo', shared + ['foo.c'])
        # executable('bar', shared + ['bar.c'])
        # We do this by removing the `['shared.c']`-node from `candidates2`.
        candidates2 = {x for x in candidates2 if self.affects_no_other_targets(x)}

        def path_contains_unknowns(candidate: BaseNode) -> bool:
            all_paths = self.interpreter.dataflow_dag.find_all_paths(candidate, target.node)
            for path in all_paths:
                for el in path:
                    if isinstance(el, UnknownValue):
                        return True
            return False

        candidates2 = {x for x in candidates2 if not path_contains_unknowns(x)}

        candidates2 = {x for x in candidates2 if self.get_relto(target.node, x) is not None}

        chosen: T.Union[FunctionNode, ArrayNode] = None
        new_kwarg_flag = False
        if len(candidates2) > 0:
            # So that files(['a', 'b']) gets modified to files(['a', 'b', 'c']) instead of files(['a', 'b'], 'c')
            if len({x for x in candidates2 if isinstance(x, ArrayNode)}) > 0:
                candidates2 = {x for x in candidates2 if isinstance(x, ArrayNode)}

            # We choose one more or less arbitrary candidate
            chosen = min(candidates2, key=lambda x: (x.lineno, x.colno))
        elif op == 'src_add':
            chosen = target.node
        elif op == 'extra_files_add':
            chosen = ArrayNode(_symbol('['), ArgumentNode(Token('', tgt_function.filename, 0, 0, 0, None, '[]')), _symbol(']'))

            # this is fundamentally error prone
            self.interpreter.dataflow_dag.add_edge(chosen, target.node)

            extra_files_idnode = IdNode(Token('string', tgt_function.filename, 0, 0, 0, None, 'extra_files'))
            if tgt_function not in self.modified_nodes:
                self.modified_nodes += [tgt_function]
            new_extra_files_node: BaseNode
            if target.node.args.get_kwarg_or_default('extra_files', None) is None:
                # Target has no extra_files kwarg, create one
                new_kwarg_flag = True
                new_extra_files_node = chosen
            else:
                new_kwarg_flag = True
                old_extra_files = target.node.args.get_kwarg_or_default('extra_files', None)
                target.node.args.kwargs = {k: v for k, v in target.node.args.kwargs.items() if not (isinstance(k, IdNode) and k.value == 'extra_files')}
                new_extra_files_node = ArithmeticNode('+', old_extra_files, _symbol('+'), chosen)

            tgt_function.args.kwargs[extra_files_idnode] = new_extra_files_node

        newfiles_relto = self.get_relto(target.node, chosen)
        old_src_list: T.List[T.Any] = flatten([self.interpreter.node_to_runtime_value(sn) for sn in old])

        if op == 'src_add':
            name = 'Source'
        elif op == 'extra_files_add':
            name = 'Extra file'
        # Generate the new String nodes
        to_append = []
        added = []

        old_src_list = [(target_dir_abs / x).resolve() if isinstance(x, str) else x.to_abs_path(source_root_abs) for x in old_src_list if not isinstance(x, UnknownValue)]
        for _newf in sorted(set(newfiles)):
            newf = Path(_newf)
            if os.path.isabs(newf):
                newf = Path(newf)
            else:
                newf = source_root_abs / newf
            if newf in old_src_list:
                mlog.log('  -- ', name, mlog.green(str(newf)), 'is already defined for the target --> skipping')
                continue

            mlog.log('  -- Adding ', name.lower(), mlog.green(str(newf)), 'at',
                     mlog.yellow(f'{chosen.filename}:{chosen.lineno}'))
            added.append(newf)
            mocktarget = self.interpreter.funcvals[target.node]
            assert isinstance(mocktarget, IntrospectionBuildTarget)
            # print("adding ", str(newf), 'to', mocktarget.name) todo: should we write something to stderr?

            path = relpath(newf, newfiles_relto)
            path = codecs.encode(path, 'unicode_escape').decode() # Because the StringNode constructor does the inverse
            token = Token('string', chosen.filename, 0, 0, 0, None, path)
            to_append += [StringNode(token)]

        assert isinstance(chosen, (FunctionNode, ArrayNode))
        arg_node = chosen.args
        # Append to the AST at the right place
        arg_node.arguments += to_append

        # Mark the node as modified
        if chosen not in to_sort_nodes:
            to_sort_nodes += [chosen]
        # If the extra_files array is newly created, i.e. if new_kwarg_flag is
        # True, don't mark it as its parent function node already is, otherwise
        # this would cause double modification.
        if chosen not in self.modified_nodes and not new_kwarg_flag:
            self.modified_nodes += [chosen]

    # Utility function to get a list of the sources from a node
    def arg_list_from_node(self, n: BaseNode) -> T.List[BaseNode]:
        args = []
        if isinstance(n, FunctionNode):
            args = list(n.args.arguments)
            if n.func_name.value in BUILD_TARGET_FUNCTIONS:
                args.pop(0)
        elif isinstance(n, ArrayNode):
            args = n.args.arguments
        elif isinstance(n, ArgumentNode):
            args = n.arguments
        return args

    def rm_src_or_extra(self, op: str, target: IntrospectionBuildTarget, to_be_removed: T.List[str], to_sort_nodes: T.List[T.Union[FunctionNode, ArrayNode]]) -> None:
        assert op in {'src_rm', 'extra_files_rm'}
        cwd = Path(os.getcwd())
        source_root_abs = cwd / self.interpreter.source_root

        # Helper to find the exact string node and its parent
        def find_node(src: str) -> T.Tuple[T.Optional[BaseNode], T.Optional[StringNode]]:
            if op == 'src_rm':
                nodes = self.interpreter.dataflow_dag.reachable(set(target.source_nodes), True).union({target.node})
            elif op == 'extra_files_rm':
                nodes = self.interpreter.dataflow_dag.reachable({target.extra_files}, True)
            for i in nodes:
                if isinstance(i, UnknownValue):
                    continue
                relto = self.get_relto(target.node, i)
                if relto is not None:
                    for j in self.arg_list_from_node(i):
                        if isinstance(j, StringNode):
                            if os.path.normpath(relto / j.value) == os.path.normpath(source_root_abs / src):
                                return i, j
            return None, None

        if op == 'src_rm':
            name = 'source'
        elif op == 'extra_files_rm':
            name = 'extra file'

        for i in to_be_removed:
            # Try to find the node with the source string
            root, string_node = find_node(i)
            if root is None:
                mlog.warning('  -- Unable to find', name, mlog.green(i), 'in the target')
                continue
            if not self.affects_no_other_targets(string_node):
                mlog.warning('  -- Removing the', name, mlog.green(i), 'is too compilicated')
                continue

            if not isinstance(root, (FunctionNode, ArrayNode)):
                raise NotImplementedError # I'm lazy

            # Remove the found string node from the argument list
            arg_node = root.args
            mlog.log('  -- Removing', name, mlog.green(i), 'from',
                     mlog.yellow(f'{string_node.filename}:{string_node.lineno}'))
            arg_node.arguments.remove(string_node)

            # Mark the node as modified
            if root not in to_sort_nodes:
                to_sort_nodes += [root]
            if root not in self.modified_nodes:
                self.modified_nodes += [root]

    @RequiredKeys(rewriter_keys['target'])
    def process_target(self, cmd: T.Dict[str, T.Any]) -> None:
        mlog.log('Processing target', mlog.bold(cmd['target']), 'operation', mlog.cyan(cmd['operation']))
        target = self.find_target(cmd['target'])
        if target is None and cmd['operation'] != 'target_add':
            mlog.error('Unknown target', mlog.bold(cmd['target']), *self.on_error())
            return self.handle_error()

        # Make source paths relative to the current subdir
        def rel_source(src: str) -> str:
            subdir = os.path.abspath(os.path.join(self.sourcedir, target.subdir))
            if os.path.isabs(src):
                return os.path.relpath(src, subdir)
            elif not os.path.exists(src):
                return src # Trust the user when the source doesn't exist
            # Make sure that the path is relative to the subdir
            return os.path.relpath(os.path.abspath(src), subdir)

        if target is not None:
            cmd['sources'] = [rel_source(x) for x in cmd['sources']]

        to_sort_nodes: T.List[T.Union[FunctionNode, ArrayNode]] = []

        if cmd['operation'] in {'src_add', 'extra_files_add'}:
            self.add_src_or_extra(cmd['operation'], target, cmd['sources'], to_sort_nodes)

        elif cmd['operation'] in {'src_rm', 'extra_files_rm'}:
            self.rm_src_or_extra(cmd['operation'], target, cmd['sources'], to_sort_nodes)

        elif cmd['operation'] == 'target_add':
            if target is not None:
                mlog.error('Can not add target', mlog.bold(cmd['target']), 'because it already exists', *self.on_error())
                return self.handle_error()

            id_base = re.sub(r'[- ]', '_', cmd['target'])
            target_id = id_base + '_exe' if cmd['target_type'] == 'executable' else '_lib'
            source_id = id_base + '_sources'
            filename = os.path.join(os.getcwd(), self.interpreter.source_root, cmd['subdir'], environment.build_filename)

            # Build src list
            src_arg_node = ArgumentNode(Token('string', filename, 0, 0, 0, None, ''))
            src_arr_node = ArrayNode(_symbol('['), src_arg_node, _symbol(']'))
            src_far_node = ArgumentNode(Token('string', filename, 0, 0, 0, None, ''))
            src_fun_node = FunctionNode(IdNode(Token('id', filename, 0, 0, 0, (0, 0), 'files')), _symbol('('), src_far_node, _symbol(')'))
            src_ass_node = AssignmentNode(IdNode(Token('id', filename, 0, 0, 0, (0, 0), source_id)), _symbol('='), src_fun_node)
            src_arg_node.arguments = [StringNode(Token('string', filename, 0, 0, 0, None, x)) for x in cmd['sources']]
            src_far_node.arguments = [src_arr_node]

            # Build target
            tgt_arg_node = ArgumentNode(Token('string', filename, 0, 0, 0, None, ''))
            tgt_fun_node = FunctionNode(IdNode(Token('id', filename, 0, 0, 0, (0, 0), cmd['target_type'])), _symbol('('), tgt_arg_node, _symbol(')'))
            tgt_ass_node = AssignmentNode(IdNode(Token('id', filename, 0, 0, 0, (0, 0), target_id)), _symbol('='), tgt_fun_node)
            tgt_arg_node.arguments = [
                StringNode(Token('string', filename, 0, 0, 0, None, cmd['target'])),
                IdNode(Token('string', filename, 0, 0, 0, None, source_id))
            ]

            src_ass_node.accept(AstIndentationGenerator())
            tgt_ass_node.accept(AstIndentationGenerator())
            self.to_add_nodes += [src_ass_node, tgt_ass_node]

        elif cmd['operation'] == 'target_rm':
            to_remove: BaseNode = self.find_assignment_node(target.node)
            if to_remove is None:
                to_remove = target.node
            self.to_remove_nodes += [to_remove]
            mlog.log('  -- Removing target', mlog.green(cmd['target']), 'at',
                     mlog.yellow(f'{to_remove.filename}:{to_remove.lineno}'))

        elif cmd['operation'] == 'info':
            # T.List all sources in the target

            cwd = Path(os.getcwd())
            source_root_abs = cwd / self.interpreter.source_root

            src_list = self.interpreter.nodes_to_pretty_filelist(source_root_abs, target.subdir, target.source_nodes)
            extra_files_list = self.interpreter.nodes_to_pretty_filelist(source_root_abs, target.subdir, [target.extra_files] if target.extra_files else [])

            src_list = ['unknown' if isinstance(x, UnknownValue) else relpath(x, source_root_abs) for x in src_list]
            extra_files_list = ['unknown' if isinstance(x, UnknownValue) else relpath(x, source_root_abs) for x in extra_files_list]

            test_data = {
                'name': target.name,
                'sources': src_list,
                'extra_files': extra_files_list
            }
            self.add_info('target', target.id, test_data)

        # Sort files
        for i in to_sort_nodes:
            if isinstance(i, FunctionNode) and i.func_name.value in BUILD_TARGET_FUNCTIONS:
                src_args = i.args.arguments[1:]
                target_name = [i.args.arguments[0]]
            else:
                src_args = i.args.arguments
                target_name = []
            unknown: T.List[BaseNode] = [x for x in src_args if not isinstance(x, StringNode)]
            sources: T.List[StringNode] = [x for x in src_args if isinstance(x, StringNode)]
            sources = sorted(sources, key=lambda x: pathname_sort_key(x.value))
            i.args.arguments = target_name + unknown + T.cast(T.List[BaseNode], sources)

    def process(self, cmd: T.Dict[str, T.Any]) -> None:
        if 'type' not in cmd:
            raise RewriterException('Command has no key "type"')
        if cmd['type'] not in self.functions:
            raise RewriterException('Unknown command "{}". Supported commands are: {}'
                                    .format(cmd['type'], list(self.functions.keys())))
        self.functions[cmd['type']](cmd)

    def apply_changes(self) -> None:
        assert all(hasattr(x, 'lineno') and hasattr(x, 'colno') and hasattr(x, 'filename') for x in self.modified_nodes)
        assert all(hasattr(x, 'lineno') and hasattr(x, 'colno') and hasattr(x, 'filename') for x in self.to_remove_nodes)
        assert all(isinstance(x, (ArrayNode, FunctionNode)) for x in self.modified_nodes)
        assert all(isinstance(x, (ArrayNode, AssignmentNode, FunctionNode)) for x in self.to_remove_nodes)
        # Sort based on line and column in reversed order
        work_nodes = [{'node': x, 'action': 'modify'} for x in self.modified_nodes]
        work_nodes += [{'node': x, 'action': 'rm'} for x in self.to_remove_nodes]
        work_nodes = sorted(work_nodes, key=lambda x: (T.cast(BaseNode, x['node']).lineno, T.cast(BaseNode, x['node']).colno), reverse=True)
        work_nodes += [{'node': x, 'action': 'add'} for x in self.to_add_nodes]

        # Generating the new replacement string
        str_list = []
        for i in work_nodes:
            new_data = ''
            if i['action'] == 'modify' or i['action'] == 'add':
                printer = AstPrinter()
                T.cast(BaseNode, i['node']).accept(printer)
                printer.post_process()
                new_data = printer.result.strip()
            data = {
                'file': T.cast(BaseNode, i['node']).filename,
                'str': new_data,
                'node': i['node'],
                'action': i['action']
            }
            str_list += [data]

        # Load build files
        files: T.Dict[str, T.Any] = {}
        for i in str_list:
            if i['file'] in files:
                continue
            fpath = os.path.realpath(T.cast(str, i['file']))
            fdata = ''
            # Create an empty file if it does not exist
            if not os.path.exists(fpath):
                with open(fpath, 'w', encoding='utf-8'):
                    pass
            with open(fpath, encoding='utf-8') as fp:
                fdata = fp.read()

            # Generate line offsets numbers
            m_lines = fdata.splitlines(True)
            offset = 0
            line_offsets = []
            for j in m_lines:
                line_offsets += [offset]
                offset += len(j)

            files[T.cast(str, i['file'])] = {
                'path': fpath,
                'raw': fdata,
                'offsets': line_offsets
            }

        # Replace in source code
        def remove_node(i: T.Dict[str, T.Any]) -> None:
            offsets = files[i['file']]['offsets']
            raw = files[i['file']]['raw']
            node = i['node']
            line = node.lineno - 1
            col = node.colno
            start = offsets[line] + col
            end = start
            if isinstance(node, (ArrayNode, FunctionNode)):
                end = offsets[node.end_lineno - 1] + node.end_colno

            # Only removal is supported for assignments
            elif isinstance(node, AssignmentNode) and i['action'] == 'rm':
                if isinstance(node.value, (ArrayNode, FunctionNode)):
                    remove_node({'file': i['file'], 'str': '', 'node': node.value, 'action': 'rm'})
                    raw = files[i['file']]['raw']
                while raw[end] != '=':
                    end += 1
                end += 1 # Handle the '='
                while raw[end] in {' ', '\n', '\t'}:
                    end += 1

            files[i['file']]['raw'] = raw[:start] + i['str'] + raw[end:]

        for i in str_list:
            if i['action'] in {'modify', 'rm'}:
                remove_node(i)
            elif i['action'] == 'add':
                files[T.cast(str, i['file'])]['raw'] += T.cast(str, i['str']) + '\n'

        # Write the files back
        for key, val in files.items():
            mlog.log('Rewriting', mlog.yellow(key))
            with open(val['path'], 'w', encoding='utf-8') as fp:
                fp.write(val['raw'])

target_operation_map = {
    'add': 'src_add',
    'rm': 'src_rm',
    'add_target': 'target_add',
    'rm_target': 'target_rm',
    'add_extra_files': 'extra_files_add',
    'rm_extra_files': 'extra_files_rm',
    'info': 'info',
}

def list_to_dict(in_list: T.List[str]) -> T.Dict[str, str]:
    result = {}
    it = iter(in_list)
    try:
        for i in it:
            # calling next(it) is not a mistake, we're taking the next element from
            # the iterator, avoiding the need to preprocess it into a sequence of
            # key value pairs.
            result[i] = next(it)
    except StopIteration:
        raise RewriterException('List of key/value pairs must have an even length.')
    return result

def list_to_dict_for_delete(args: T.List[str]) -> T.Dict[str, T.Optional[str]]:
    if len(args) % 2 == 0 and all(a == '' for a in args[1::2]):
        mlog.deprecation('Even-numbered arguments are all blank; '
                         'ignoring these for compatibility with Meson < 1.10')
        args = args[::2]
    return {a: None for a in args}

def generate_target(options: argparse.Namespace) -> T.List[T.Dict[str, T.Any]]:
    return [{
        'type': 'target',
        'target': options.target,
        'operation': target_operation_map[options.operation],
        'sources': options.sources,
        'subdir': options.subdir,
        'target_type': options.tgt_type,
    }]

def generate_kwargs(options: argparse.Namespace) -> T.List[T.Dict[str, T.Any]]:
    if options.operation == 'delete':
        kwargs = list_to_dict_for_delete(options.kwargs)
    else:
        kwargs = list_to_dict(options.kwargs)
    return [{
        'type': 'kwargs',
        'function': options.function,
        'id': options.id,
        'operation': options.operation,
        'kwargs': kwargs,
    }]

def generate_def_opts(options: argparse.Namespace) -> T.List[T.Dict[str, T.Any]]:
    if options.operation == 'delete':
        kwargs = list_to_dict_for_delete(options.options)
    else:
        kwargs = list_to_dict(options.options)
    return [{
        'type': 'default_options',
        'operation': options.operation,
        'options': kwargs,
    }]

def generate_cmd(options: argparse.Namespace) -> T.List[T.Dict[str, T.Any]]:
    if os.path.exists(options.json):
        with open(options.json, encoding='utf-8') as fp:
            return T.cast(T.List[T.Dict[str, T.Any]], json.load(fp))
    else:
        return T.cast(T.List[T.Dict[str, T.Any]], json.loads(options.json))

# Map options.type to the actual type name
cli_type_map = {
    'target': generate_target,
    'tgt': generate_target,
    'kwargs': generate_kwargs,
    'default-options': generate_def_opts,
    'def': generate_def_opts,
    'command': generate_cmd,
    'cmd': generate_cmd,
}

def run(options: argparse.Namespace) -> int:
    mlog.redirect(True)
    if not options.verbose:
        mlog.set_quiet()

    try:
        setup_vsenv()
        rewriter = Rewriter(options.sourcedir, skip_errors=options.skip)
        rewriter.analyze_meson()

        if options.type is None:
            mlog.error('No command specified')
            return 1

        commands = cli_type_map[options.type](options)

        if not isinstance(commands, list):
            raise TypeError('Command is not a list')

        for i, cmd in enumerate(commands):
            if not isinstance(cmd, object):
                raise TypeError('Command is not an object')
            rewriter.process(cmd)
            rewriter.apply_changes()

            if i == len(commands) - 1: # Improves the performance, is not necessary for correctness.
                break

            rewriter.modified_nodes = []
            rewriter.to_remove_nodes = []
            rewriter.to_add_nodes = []
            # The AST changed, so we need to update every information that was derived from the AST
            rewriter.interpreter = IntrospectionInterpreter(rewriter.sourcedir, '', rewriter.interpreter.backend, visitors = [AstIDGenerator(), AstIndentationGenerator(), AstConditionLevel()])
            rewriter.analyze_meson()

        rewriter.print_info()
        return 0
    except Exception as e:
        raise e
    finally:
        mlog.set_verbose()