xts_tools/ci/utils.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Copyright (c) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

from __future__ import annotations
import os
import re
import json
import sys
import logging
from enum import Enum
from pathlib import Path

CODEBASE = os.path.dirname(os.path.dirname(os.path.dirname(
    os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))


class InterfaceRepo(str, Enum):
    SDK_JS = "sdk-js"
    SDK_C = "sdk_c"
    DRIVERS = "driver_interface"


class ChangeFileEntity:
    def __init__(self, name, path):
        self.name = name
        self.path = path
        self.add = []
        self.modified = []
        self.delete = []
        self._already_match_utils = False

    def addAddPaths(self, add_list):
        self.add += list(map(lambda x: os.path.join(self.path, x), add_list))
        self.add.sort()

    def addModifiedPaths(self, modified_list):
        self.modified += list(map(lambda x: os.path.join(self.path, x), modified_list))
        self.modified.sort()

    def addRenamePathsto(self, rename_list):
        for list in rename_list:
            self.add += [os.path.join(self.path, list[1])]
            self.delete += [os.path.join(self.path, list[0])]
            self.add.sort()
            self.delete.sort()

    def addDeletePaths(self, delete_list):
        self.delete += list(map(lambda x: os.path.join(self.path, x), delete_list))
        self.delete.sort()

    def isEmpty(self):
        if self.add:
            return False
        if self.modified:
            return False
        if self.delete:
            return False
        return True

    def get_already_match_utils(self):
        return self._already_match_utils

    def set_already_match_utils(self, already_match_utils):
        self._already_match_utils = already_match_utils

    def __str__(self):
        add_str = '\n  '.join(self.add) if self.add else 'None'
        modified_str = '\n  '.join(self.modified) if self.modified else 'None'
        delete_str = '\n  '.join(self.delete) if self.delete else 'None'

        return (f"ChangeFileEntity(\n"
                f"  name: {self.name},\n"
                f"  path: {self.path},\n"
                f"  add: [\n    {add_str}\n  ],\n"
                f"  modified: [\n    {modified_str}\n  ],\n"
                f"  delete: [\n    {delete_str}\n  ]\n"
                f")")


class MatchConfig:
    config_path = os.path.join(CODEBASE, "test/xts/tools/config")
    MACTH_CONFIG_PATH = os.path.join(config_path, "ci_match_config.json")
    exception_path = {}
    all_com_path = {}
    skip_judge_build_path = {}
    temple_list = []
    xts_path_list = []
    interface_path_dict = {}

    INTERFACE_BUNDLE_NAME_PATH = os.path.join(config_path, "ci_api_part_name.json")
    INTERFACE_WHITELIST_PATH = os.path.join(config_path, "ci_api_ignore.json")
    interface_bundle_conf = {}
    interface_whitelist_conf = {}

    WHITE_LIST_PATH = os.path.join(config_path, "ci_target_white_list.json")
    white_list_repo = {}

    uncompile_suites = {}
    verify_suites = {}

    @classmethod
    def initialization(cls):
        if not cls.exception_path:
            print("MatchConfig initialization begin...")
            if not os.path.exists(cls.MACTH_CONFIG_PATH):
                print("warning: Reading the configuration file is abnormal because {} not exist".format(
                    cls.MACTH_CONFIG_PATH))
            with open(cls.MACTH_CONFIG_PATH, 'r') as file:
                rules_data = json.load(file)
                cls.exception_path = rules_data['exception_path']
                cls.all_com_path = rules_data['all_com_path']
                cls.skip_judge_build_path = rules_data['skip_judge_build_path']
                cls.temple_list = rules_data['temple_list']
                cls.xts_path_list = rules_data['xts_path_list']
                cls.interface_path_dict = rules_data['interface_path_dict']
        print("MatchConfig initialization end.")
    
    @classmethod
    def interface_initialization(cls):
        if not cls.interface_bundle_conf:
            print("INTERFACE_BUNDLE_NAME initialization begin...")
            if not os.path.exists(cls.INTERFACE_BUNDLE_NAME_PATH):
                print("warning: Reading the configuration file is abnormal because {} not exist".format(
                    cls.INTERFACE_BUNDLE_NAME_PATH))
                return
            with open(cls.INTERFACE_BUNDLE_NAME_PATH, 'r') as file:
                cls.interface_bundle_conf = json.load(file)
            print("INTERFACE_BUNDLE_NAME initialization end.")

        if not cls.interface_whitelist_conf:
            print("INTERFACE_WHITELIST initialization begin...")
            if not os.path.exists(cls.INTERFACE_WHITELIST_PATH):
                print("warning: Reading the configuration file is abnormal because {} not exist".format(
                    cls.INTERFACE_WHITELIST_PATH))
                return
            with open(cls.INTERFACE_WHITELIST_PATH, 'r') as file:
                cls.interface_whitelist_conf = json.load(file)
            print("INTERFACE_WHITELIST initialization end.")
    
    @classmethod
    def get_interface_bundle_config(cls, repo_name: InterfaceRepo):
        if not cls.interface_bundle_conf:
            cls.interface_initialization()
        return cls.interface_bundle_conf.get(repo_name, {})

    @classmethod
    def get_interface_whitelist_config(cls):
        if not cls.interface_whitelist_conf:
            cls.interface_initialization()
        return cls.interface_whitelist_conf

    @classmethod
    def get_interface_path_mapping(cls):
        if not cls.interface_path_dict:
            cls.initialization()
        return cls.interface_path_dict

    @classmethod
    def get_interface_path(cls, repo_name: InterfaceRepo):
        return MatchConfig.get_interface_path_mapping().get(repo_name)

    @classmethod
    def get_exception_path(cls):
        if not cls.exception_path:
            cls.initialization()
        return cls.exception_path

    @classmethod
    def get_all_com_path(cls):
        if not cls.all_com_path:
            cls.initialization()
        return cls.all_com_path

    @classmethod
    def get_skip_judge_build_path(cls):
        if not cls.skip_judge_build_path:
            cls.initialization()
        return cls.skip_judge_build_path

    @classmethod
    def get_temple_list(cls):
        if not cls.temple_list:
            cls.initialization()
        return cls.temple_list

    @classmethod
    def get_verify_suites(cls, xts_root_dir):
        root_tgt = PathUtils.get_root_target(xts_root_dir)

        if root_tgt not in cls.verify_suites:
            verify_path = os.path.join(xts_root_dir, 'ci_verify_suites.json')

            try:
                with open(verify_path, 'r') as file:
                    cls.verify_suites[root_tgt] = json.load(file)

                if not isinstance(cls.verify_suites[root_tgt], list):
                    print(f'Config file: {verify_path} malformed, ' \
                          f'list expected, got: {type(cls.verify_suites[root_tgt]).__name__}.')
                    cls.verify_suites[root_tgt] = [root_tgt]

            except Exception as err:
                print(f'Error reading config file: {verify_path}, reason: {type(err).__name__} - {err}.')
                cls.verify_suites[root_tgt] = [root_tgt]

        return cls.verify_suites[root_tgt]

    @classmethod
    def get_xts_path_list(cls):
        if not cls.xts_path_list:
            cls.initialization()
        return cls.xts_path_list

    @classmethod
    def initialization_white_list(cls):
        if not cls.white_list_repo:
            print("WhiteList initialization begin...")
            if not os.path.exists(cls.WHITE_LIST_PATH):
                print("warning: Reading the configuration file is abnormal because {} not exist".format(
                    cls.WHITE_LIST_PATH))
            with open(cls.WHITE_LIST_PATH, 'r') as file:
                white_file = json.load(file)
                white_repos = white_file["repo_list"]
                for white_repo in white_repos:
                    cls.white_list_repo[white_repo["path"]] = white_repo
        print("WhiteList initialization end.")

    @classmethod
    def get_white_list_repo(cls):
        if not cls.white_list_repo:
            cls.initialization_white_list()
        return cls.white_list_repo

    @classmethod
    def get_uncompile_suite_list(cls, xts_root_dir, device_type):
        root_tgt = PathUtils.get_root_target(xts_root_dir)
        if root_tgt not in cls.uncompile_suites:
            uncompile_path = os.path.join(xts_root_dir, 'ci_uncompile_suite.json')
            if not os.path.exists(uncompile_path):
                print('Get uncompile testsuite failed because {} not exist'.format(uncompile_path))
                return []
            with open(uncompile_path, 'r') as file:
                cls.uncompile_suites[root_tgt] = json.load(file)
        if device_type in cls.uncompile_suites[root_tgt]:
            return cls.uncompile_suites[root_tgt][device_type]
        elif isinstance(cls.uncompile_suites[root_tgt], dict):
            return []
        else:
            return cls.uncompile_suites[root_tgt]


class XTSTargetUtils:
    @staticmethod
    def get_suite_name(xts_root_dir):
        valid_suites = {'acts', 'dcts', 'hats', 'hits'}
        suite_name = os.environ.get('XTS_SUITENAME') or \
                     os.environ.get('xts_suitename') or \
                     os.path.normpath(xts_root_dir).split(os.sep)[-1]
        return suite_name if suite_name in valid_suites else 'acts'

    @staticmethod
    def filter_suite_targets(suite_name, targets: list):
        results = []
        if not targets:
            return results
        for tgt in set(targets):
            # canonical
            if ':' in tgt:
                if f'test/xts/{suite_name}' in tgt:
                    results.append(tgt)
                else:
                    print(f'[ERROR] {tgt} not belong to xts {suite_name} suite')
                    sys.exit(1)
            else:
                # shorthand
                results.append(tgt)
        return results

    @staticmethod
    def get_current_Build(xts_root_dir, current_dir):
        while PathUtils.is_parent_path(xts_root_dir, current_dir):
            # 当前目录是否包含需跳过的keywords
            if PathUtils.isMatchRules(current_dir, MatchConfig.get_skip_judge_build_path()):
                current_dir = os.path.dirname(current_dir)
                continue
            # 检查当前目录下是否存在BUILD.gn文件
            build_gn_path = os.path.join(current_dir, 'BUILD.gn')
            if os.path.exists(build_gn_path):
                return build_gn_path
                # 如果没有找到，向上一层目录移动
            current_dir = os.path.dirname(current_dir)
        # xts仓最外层均有BUILD.gn文件
        return current_dir

    # 路径获取target
    @staticmethod
    def getTargetfromPath(xts_root_dir, path) -> list:
        if path == xts_root_dir:
            root_target = PathUtils.get_all_build_target(xts_root_dir)
            return root_target
        build_file = XTSTargetUtils.get_current_Build(xts_root_dir, path)
        targets = XTSTargetUtils.getTargetFromBuild(build_file)
        if targets == None:
            return XTSTargetUtils.getTargetfromPath(xts_root_dir, os.path.dirname(os.path.dirname(build_file)))
        return targets

    @staticmethod
    def getTargetFromBuild(build_File) -> list:
        pattern = re.compile(r'(\b(?:' + '|'.join(
            re.escape(word) for word in MatchConfig.get_temple_list()) + r')\b)\s*\(\s*"([^"]*)"\)')
        with open(build_File, 'r', encoding='utf-8') as file:
            content = file.read()
        matches = pattern.findall(content)
        targets = [match[1] for match in matches]
        relative_path = os.path.relpath(os.path.dirname(build_File), CODEBASE)
        if len(targets) > 1:
            deps = XTSTargetUtils.getDepsinBuild(content)
            # 编译本gn中未被依赖的目标
            targets = [item for item in targets if item not in deps]
        return [f"{relative_path}:{item}" for item in targets]

    @staticmethod
    def getDepsinBuild(build):
        # 定义正则表达式模式来匹配deps数组
        pattern = re.compile(r'deps\s*=\s*\[\s*(?P<deps>.*?)\s*\]', re.DOTALL)
        # pattern = r'\s*deps\s*=\s*<deps>'
        # 搜索文本中的匹配项
        matches = pattern.findall(build)
        all_deps = []

        for match in matches:
            # 分割字符串并去除双引号和空格
            deps_list = [dep.strip('\n').strip().strip('"').lstrip(':') for dep in match.split(',')]
            all_deps.extend(deps_list)

        return all_deps

    '''
    {
        "部件A": ["用例A1", "用例A2", ... "用例Am"],
        "部件B": ["用例B1", "用例B2", ... "用例Bn"],
    }
    '''
    @staticmethod
    def getPathsByBundle(bundle, test_home, filter=None) -> list:
        matching_files = []
        # 遍历根目录及其子目录
        for root, dirs, files in os.walk(test_home):
            if PathUtils.isMatchRules(root, MatchConfig.get_exception_path()):
                continue
            for file in files:
                if file != 'BUILD.gn':
                    continue
                file_path = os.path.join(root, file)
                # 读取文件内容
                content = ""
                with open(file_path, 'r', encoding='utf-8') as f:
                    content = f.read()
                # 检查是否包含bundle
                for bundle_ in bundle:
                    part_name = f'part_name = "{bundle_}"'
                    if part_name not in content:
                        continue
                    if not filter:
                        matching_files.append(root)
                        break
                    testsuite_list = filter.get(bundle_)
                    if not testsuite_list:
                        continue
                    isHapNameMatch = False
                    for testsuite in testsuite_list:
                        hap_name = f'hap_name = "{testsuite}"'
                        if hap_name in content:
                            isHapNameMatch = True
                            break
                    if isHapNameMatch:
                        matching_files.append(root)
                        break
        return matching_files

    @staticmethod
    def del_uncompile_target(xts_root_dir, device_type, targets) -> list:
        ci_target = set()
        uncompile_suite_list = MatchConfig.get_uncompile_suite_list(xts_root_dir, device_type)
        print("Config uncompile testsuite: {}".format(uncompile_suite_list))
        for path_target in targets:
            if path_target not in uncompile_suite_list:
                ci_target.add(path_target)
        print("Accurte compile target: {}".format(ci_target))
        return list(ci_target)


class PathUtils:

    # 路径列表简化
    @staticmethod
    def removeSubandDumpPath(path_list: list) -> list:
        # 排序，确保父目录在子目录之前,减少运算
        path_list.sort()
        # 存储最小集
        minimal_paths_set = set()
        # 记录已存在的父目录的全部未添加编译的子目录
        parent_dirs = {}

        for path in path_list:
            # 检查当前路径或其父路径是否已经在最小集中
            isinclude = False
            for m_path in minimal_paths_set:
                if PathUtils.is_parent_path(m_path, path):
                    isinclude = True
                    break
            # 添加逻辑
            if not isinclude:
                PathUtils.addPathClean(path, minimal_paths_set, parent_dirs)

        return list(minimal_paths_set)

    @staticmethod
    def addPathClean(path, minimal_paths_set, parent_dirs):
        # 检查当前路径的首层父目录是否在最小集中
        parent_path = os.path.dirname(path)
        if parent_path in parent_dirs:
            # 在-原list修改
            subdirs = parent_dirs[parent_path]
        else:
            # 不在-记录父目录及本目录
            subdirs = [os.path.join(parent_path, d) for d in os.listdir(parent_path) if
                       os.path.isdir(os.path.join(parent_path, d))]
            parent_dirs[parent_path] = subdirs
        subdirs.remove(path)
        minimal_paths_set.add(path)
        # 检查是否替换为添加其直接父目录
        if len(subdirs) == 0:
            del parent_dirs[parent_path]
            # minimal_paths_sets删除parent_path子目录
            for d in os.listdir(parent_path):
                p = os.path.join(parent_path, d)
                if os.path.isdir(p) and p in minimal_paths_set:
                    minimal_paths_set.remove(os.path.join(parent_path, d))
            PathUtils.addPathClean(parent_path, minimal_paths_set, parent_dirs)

    @staticmethod
    def get_current_exist(root_path, path) -> str:
        current_dir = path
        while PathUtils.is_parent_path(root_path, current_dir):
            if os.path.exists(current_dir):
                return current_dir
            current_dir = os.path.dirname(current_dir)
        # 根目录必然存在
        return root_path

    @staticmethod
    def is_parent_path(parent_path, child_path):
        # 获取公共路径
        common_path = os.path.commonpath([parent_path, child_path])
        return common_path == parent_path

    @staticmethod
    def get_all_build_target(xts_root_dir, full_flag = 0):
        return MatchConfig.get_verify_suites(xts_root_dir) \
            if full_flag == 0 \
            else [PathUtils.get_root_target(xts_root_dir)]

    @staticmethod
    def get_root_target(xts_root_dir):
        suite_name = os.path.basename(xts_root_dir)
        target = f'xts_{suite_name}'
        return target

    @staticmethod
    def isMatchRules(file, rules):
        string_rules = rules["string_rules"]
        re_rules = rules["re_rules"]
        for rule in string_rules:
            if rule in file:
                return True
        for rule in re_rules:
            if re.compile(rule).search(file):
                return True
        return False

    @staticmethod
    def isTargetContains(targetFiles, file) -> bool:
        for f in targetFiles:
            if PathUtils.is_parent_path(f, file):
                return True
        return False


class Singleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]


class XTSLogger(metaclass = Singleton):
    """
    Wrapper class of logging.Logger.

    By default, the logger writes into stdout, not to a file.
    
    Examples:
        logger = XTSLogger()
        logger.logging_phase = "PHASE STRING"
        logger.info("hello world.")
    """
    def __init__(self, name = "xts_logger", level = logging.INFO,
                 format = "[XTS %(levelname)s] %(message)s"):
        if hasattr(self, "_logger"):
            return
        
        self._logger = logging.getLogger(name)
        self._logger.propagate = False
        self._logging_phase = None
        self._logger.setLevel(level)

        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setFormatter(logging.Formatter(format))
        console_handler.setLevel(level)
        self._logger.addHandler(console_handler)

    @property
    def logger(self):
        return self._logger
    
    @property
    def logging_phase(self):
        return self._logging_phase
    
    @logging_phase.setter
    def logging_phase(self, phase: str | None):
        self._logging_phase = phase


    def add_file_handler(self, fpath: str, level = logging.INFO,
                         format = "[XTS %(levelname)s] %(message)s"):
        """
        Add FileHandler to the internal logger.

        Args:
            fpath: Log file path, default access mode is 'w'.
            level: Log level, default to INFO.
            format: Format string for Formatter.

        Returns:
            The new FileHandler object.
        """
        abs_fpath = os.path.abspath(fpath)
        handlers = {
            os.path.abspath(h.baseFilename): h
                for h in self._logger.handlers
                    if isinstance(h, logging.FileHandler)
        }

        file_handler = handlers.get(abs_fpath)
        if file_handler:
            return file_handler
        
        file_handler = logging.FileHandler(abs_fpath, 'w')
        file_handler.setFormatter(logging.Formatter(format))
        file_handler.setLevel(level)
        self._logger.addHandler(file_handler)
        return file_handler

    def remove_file_handler(self, fpath: str):
        abs_fpath = os.path.abspath(fpath)
        for handler in list(self._logger.handlers):
            if not isinstance(handler, logging.FileHandler):
                continue
            
            handler_fpath = os.path.abspath(handler.baseFilename)

            if (abs_fpath == handler_fpath):
                handler.close()
                self._logger.removeHandler(handler)
                return

    def _process_msg(self, msg):
        return msg if not self._logging_phase \
                else f"[{self.logging_phase}] {msg}"
    
    def debug(self, msg, *args, **kwargs):
        self._logger.debug(self._process_msg(msg), *args, **kwargs)
    
    def info(self, msg, *args, **kwargs):
        self._logger.info(self._process_msg(msg), *args, **kwargs)

    def warning(self, msg, *args, **kwargs):
        self._logger.warning(self._process_msg(msg), *args, **kwargs)

    def error(self, msg, *args, **kwargs):
        self._logger.error(self._process_msg(msg), *args, **kwargs)

    def critical(self, msg, *args, **kwargs):
        self._logger.critical(self._process_msg(msg), *args, **kwargs)


class WhitelistProcessor:
    """
    Simple yet handy whitelist conf processor.
    """
    class MatchPattern:
        def __init__(self, original: str, regex_str, reserve_flag: bool):
            self.original = original
            self.regex = re.compile(regex_str)
            self.dir_flag = original.endswith('/')
            self.reserve_flag = reserve_flag

    MPat_t = MatchPattern

    class PathType(Enum):
        FILE = 0
        DIR = 1

    PType_t = PathType

    class PathSegment:
        def __init__(self, path: str, ptype: WhitelistProcessor.PType_t):
            self.path = path
            self.type = ptype

    PSeg_t = PathSegment

    def __init__(self, config: dict, repo_map: dict, codebase: str):
        """
        Initialize the processor with a whitelist configuration and repo map.

        - config: dict where keys are repo names and values pattern lists.
        - repo_map: dict where keys are repo names and values are their
                    base directories (relative to codebase).
        - codebase: string abs path to the project root.
        """
        self.config = config
        self.repo_map = repo_map
        self.codebase = codebase
        self.parsed_patterns = {}
        self.invalid_patterns = []
        self._parse_config(repo_map)

    def _parse_config(self, repo_map: dict):
        for repo_name, patterns in self.config.items():
            base_dir = repo_map.get(repo_name)
            if not base_dir:
                logger = XTSLogger()
                logger.warning(f"Repo name '{repo_name}' not found in repo map, skipping its whitelist patterns.")
                continue

            self.parsed_patterns[base_dir] = []
            for pat in patterns:
                # negation pattern
                reserve_flag = False
                clean_pat = pat
                while clean_pat.startswith('!'):
                    reserve_flag = not reserve_flag
                    clean_pat = clean_pat[1:]

                regex_str = self._pattern_to_regex(clean_pat)
                if regex_str:
                    self.parsed_patterns[base_dir].append(
                        self.MatchPattern(clean_pat, regex_str, reserve_flag)
                    )
                else:
                    self.invalid_patterns.append((base_dir, pat))

        if self.invalid_patterns:
            logger = XTSLogger()
            for base, pat in self.invalid_patterns:
                logger.warning(f"Invalid whitelist pattern ignored: '{pat}' in directory '{base}'")

    def _pattern_to_regex(self, pattern: str):
        # disallow patterns with leading slash
        if pattern.startswith('/'):
            return None

        # treat any *** or more as globstar
        p = re.sub(r'\*{3,}', '**', pattern)

        dir_flag = p.endswith('/')
        p = p.rstrip('/')
        if not p:
            return "^$" if dir_flag else None

        segments = p.split('/')
        res_parts = []

        for i, seg in enumerate(segments):
            if seg == '**':
                # this completely kills the game
                if len(segments) == 1:
                    res_parts.append(r'.*')
                elif i == 0:
                    # leading globstar (**/)
                    res_parts.append(r'(?:.*/)?')
                elif i == len(segments) - 1:
                    # trailing globstar (/**)
                    res_parts.append(r'.*')
                else:
                    # globstar in middle (/**/)
                    res_parts.append(r'(?:.*/)?')
            else:
                seg_res = re.escape(seg)
                # downgrade ** to * within the segment
                seg_res = seg_res.replace(r'\*\*', r'[^/]*')
                # handle * within the segment
                seg_res = seg_res.replace(r'\*', r'[^/]*')
                res_parts.append(seg_res)

        regex_str = "^"
        for i, part in enumerate(res_parts):
            if i > 0:
                # add slash if not handled already
                if res_parts[i - 1] != r'(?:.*/)?':
                    regex_str += '/'
            regex_str += part
        regex_str += "$"

        return regex_str

    def reserve_file(self, fpath):
        """
        Check if a file path should be reserved (kept in the results).
        """
        path = Path(fpath)

        for base_dir, patterns in self.parsed_patterns.items():
            # type hint
            patterns: list[WhitelistProcessor.MPat_t] = patterns
            base_path = Path(base_dir)
            try:
                rel_path = path.relative_to(base_path).as_posix()
                if rel_path == '.':
                    rel_path = ""
            except ValueError:
                # not under current base_dir
                continue

            logger = XTSLogger()
            parts = rel_path.split('/') if rel_path else []
            path_segments = [
                self.PSeg_t('/'.join(parts[:i]), self.PType_t.DIR)
                for i in range(1, len(parts))
            ]
            path_segments.append(self.PSeg_t(rel_path, self.PType_t.FILE))

            for pat in patterns:
                for seg in path_segments:
                    if self._check_pattern_match(seg, pat):
                        if pat.reserve_flag:
                            logger.info(f"Reserve file: {fpath} due to pattern (!{base_path}/{pat.original}) match.")
                            return True
                        logger.info(f"Ignore file: {fpath} due to pattern ({base_path}/{pat.original}) match.")
                        return False
        return True

    def _check_pattern_match(self, segment: PathSegment, pattern: MatchPattern):
        """
        1. If pattern ends with /, it ONLY matches a directory.
        2. If pattern does NOT end with /, it can match both a file and a directory.
        3. For symlinks: if pattern ends with /, check if it points to a directory.
           If it does not end with /, it matches as a normal file.
        Returns:
            - True (pattern structurally matches segment path)
            - False (pattern mismatch)
        """
        seg_path = segment.path
        seg_type = segment.type
        dir_flag = pattern.dir_flag

        if not pattern.regex.match(seg_path):
            return False

        if seg_type == self.PType_t.FILE:
            return not dir_flag
        return True

    def filter_files(self, files):
        """
        Filter a list of files, removing ignored ones.
        """
        return [f for f in files if self.reserve_file(f)]