# -*- coding: utf-8 -*-

import copy
import shutil
import socket
import stat
from collections.abc import Sequence
from ucagent.util.log import info, warning
import os
from typing import List, Tuple, Union
import json
import importlib
import re
import time
import inspect
import fnmatch
import ast
from pathlib import Path
import yaml
from collections import OrderedDict
import traceback


def fmt_time_deta(sec: Union[int, float, str, None], abbr: bool = False) -> str:
    """
    Format time duration in seconds to a human-readable string.

    Args:
        sec: Time duration in seconds.
        abbr: Whether to use abbreviated format.

    Returns:
        Formatted string representing the time duration.
    """
    if sec is None:
        return "N/A"
    if isinstance(sec, str):
        if sec.isdigit():
            sec = int(sec)
        else:
            return sec
    sec = int(sec)
    s = sec % 60
    m = (sec // 60) % 60
    h = (sec // 3600) % 24
    deta_time = f"{h:02d}:{m:02d}:{s:02d}"
    if abbr:
        if h > 0:
            deta_time = f"{h}h {m:02d}m {s:02d}s"
        elif m > 0:
            deta_time = f"{m}m {s:02d}s"
        else:
            deta_time = f"{s}s"
    return deta_time


def fmt_time_stamp(sec: Union[int, float], fmt: str = "%Y-%m-%d %H:%M:%S") -> str:
    """
    Format a time duration in seconds to a string.

    Args:
        sec: Time duration in seconds.
        fmt: Format string (default is "%Y-%m-%d %H:%M:%S").

    Returns:
        Formatted time string.
    """
    if sec is None:
        return "N/A"
    if isinstance(sec, str):
        return sec
    if isinstance(sec, (int, float)):
        return time.strftime(fmt, time.localtime(sec))
    raise ValueError(f"Unsupported type for sec: {type(sec)}. Expected int or float.")


def is_text_file(file_path: str) -> bool:
    """
    Check if a file is a text file by attempting to read it.

    Args:
        file_path: Path to the file.

    Returns:
        True if the file is a text file, False otherwise.
    """
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            f.read(1000)  # Read a small portion of the file
            return True
    except UnicodeDecodeError:
        return False
    except Exception:
        return False


def get_file_size(file_path):
    """
    Get the size of a file in bytes.
    :param file_path: Path to the file.
    :return: Size of the file in bytes.
    """
    try:
        return os.path.getsize(file_path)
    except OSError:
        return 0  # Return 0 if the file does not exist or is inaccessible


def bytes_to_human_readable(size):
    """
    Convert bytes to a human-readable format.
    :param size: Size in bytes.
    :return: Human-readable string representation of the size.
    """
    if size < 1024:
        return f"{size} B"
    elif size < 1024 ** 2:
        return f"{size / 1024:.2f} KB"
    elif size < 1024 ** 3:
        return f"{size / (1024 ** 2):.2f} MB"
    else:
        return f"{size / (1024 ** 3):.2f} GB"


def get_sub_str(text, start_str, end_str):
    """
    Extract a substring from text between two delimiters.
    :param text: The input text.
    :param start_str: The starting delimiter.
    :param end_str: The ending delimiter.
    :return: The extracted substring or None if not found.
    """
    start_index = text.find(start_str)
    if start_index == -1:
        return None
    start_index += len(start_str)
    
    end_index = text.find(end_str, start_index)
    if end_index == -1:
        return None
    
    return start_str + text[start_index:end_index].strip() + end_str


def str_has_blank(text: str) -> bool:
    """
    Check if a string contains any whitespace characters.
    :param text: The input string.
    :return: True if the string contains whitespace, False otherwise.
    """
    return any(char.isspace() for char in text)


def str_remove_blank(text: str) -> str:
    """
    Remove all whitespace characters from a string.
    :param text: The input string.
    :return: The string with all whitespace characters removed.
    """
    return "".join(text.split())


def str_replace_to(text: str, old: list, new: str) -> str:
    """
    Replace all occurrences of any string in a list with a new string.
    :param text: The input string.
    :param old: List of strings to be replaced.
    :param new: The string to replace with.
    :return: The modified string.
    """
    for o in old:
        text = text.replace(o, new)
    return text


def nested_keys_as_list(
    ndata: dict, leaf: str, keynames: List[str], ex_ignore_names=["line", "pline", "nline"]
) -> Tuple[List[str], List[str]]:
    """Convert nested dictionary keys to a list of paths up to a specified leaf node."""
    broken_leaf = []
    def _nest_dict_leafs(
        data,
        ret_list,
        ret_map,
        prefix="",
        stop_key="",
        leaf_key="",
        ignore_keys=[],
        parent_key="",
    ):
        child_count = [len(data[k]) for k in data.keys() if not k in ex_ignore_names]
        is_processed = False
        for key, value in data.items():
            if isinstance(value, dict):
                new_prefix = f"{prefix}/{key}" if prefix else key
                if key in ignore_keys:
                    new_prefix = prefix
                    parent_key = key
                if key != stop_key:
                    _nest_dict_leafs(
                        value,
                        ret_list,
                        ret_map,
                        new_prefix,
                        stop_key,
                        leaf_key,
                        ignore_keys,
                        parent_key,
                    )
            elif not is_processed:
                new_prefix = prefix
                if key not in ignore_keys:
                    new_prefix = f"{prefix}/{key}" if prefix else key
                if parent_key == leaf_key:
                    ret_list.append(f"{new_prefix}")
                    ret_map[new_prefix] = data
                    is_processed = True
                else:
                    if child_count and child_count[0] < 1:
                        broken_leaf.append((parent_key, new_prefix, value))
                        is_processed = True

    ret_data = []
    ret_lblock = OrderedDict()
    stop_keys = keynames + [""]
    stop_key_map = {k: stop_keys[i + 1] for i, k in enumerate(keynames)}
    _nest_dict_leafs(
        ndata,
        ret_data,
        ret_lblock,
        stop_key=stop_key_map[leaf],
        leaf_key=leaf,
        ignore_keys=keynames + ex_ignore_names,
        parent_key=keynames[0],
    )
    return ret_data, broken_leaf, ret_lblock


def parse_nested_keys(
    target_file: str,
    keyname_list: List[str],
    prefix_list: List[str],
    subfix_list: List[str],
    ignore_chars: List[str] = ["<", ">"],
) -> dict:
    """Parse the function points and checkpoints from a file."""
    assert os.path.exists(target_file), f"File {target_file} does not exist. You need to provide a valid file path."
    assert len(keyname_list) > 0, "Prefix must be provided."
    assert "line" not in keyname_list, "'line' is a reserved key name."
    assert len(prefix_list) == len(subfix_list), "Prefix and subfix lists must have the same length."
    assert len(prefix_list) == len(keyname_list), "Prefix and keyname lists must have the same length."
    pre_values = [None] * len(prefix_list)
    key_dict = OrderedDict()
    def get_pod_next_key(i: int):
        nkey = keyname_list[i + 1] if i < len(keyname_list) - 1 else None
        if i == 0:
            return key_dict, nkey
        # Check if parent level exists
        if pre_values[i - 1] is None:
            return None, nkey
        return pre_values[i - 1][keyname_list[i]], nkey
    with open(target_file, 'r') as f:
        index = 1
        lines = f.readlines()
        pre_pod = {}
        for line in lines:
            line = str_remove_blank(line.strip())
            for i, key in enumerate(keyname_list):
                prefix = prefix_list[i]
                subfix = subfix_list[i]
                pre_key = keyname_list[i - 1] if i > 0 else None
                pre_prf = prefix_list[i - 1] if i > 0 else None
                if not prefix in line:
                    continue
                # find prefix+*+subfix in line
                assert line.count(prefix) == 1, f"At line ({index}): '{line}' should contain exactly one {key} '{prefix}'"
                current_key = rm_blank_in_str(str_replace_to(get_sub_str(line, prefix, subfix), ignore_chars, ""))
                pod, next_key = get_pod_next_key(i)
                # Enhanced error message with context
                if pod is None:
                    raise ValueError(
                        f"At line ({index}): Found {key} tag '{prefix}' but its parent {pre_key} tag '{pre_prf}' "
                        f"was not found in previous lines. Please ensure proper nesting: each '{prefix}' must be "
                        f"preceded by a '{pre_prf}' tag.\nCurrent line content: {line}"
                    )
                assert current_key not in pod, f"At line ({index}): '{current_key}' is defined multiple times."
                pline = pre_pod.get("line", index - 5) # default 5 lines before if no previous pod
                pod[current_key] = {"line": index, "pline": pline, "nline": index + 5} # default 5 more lines for a node
                if pre_pod:
                    pre_pod["nline"] = index
                pre_pod = pod[current_key]
                if next_key is not None:
                    pod[current_key][next_key] = {}
                pre_values[i] = pod[current_key]
            index += 1
    return key_dict


def load_json_file(path: str):
    """
    Load a JSON file from the specified path.
    :param path: Path to the JSON file.
    :return: Parsed JSON data.
    """
    assert os.path.exists(path), f"JSON file {path} does not exist."
    json_file = os.path.join(path)
    with open(json_file, "r", encoding="utf-8") as f:
        try:
            data = json.load(f)
            return data
        except json.JSONDecodeError as e:
            raise ValueError(f"Error decoding JSON from file {json_file}: {e}")
        except Exception as e:
            raise RuntimeError(f"Unexpected error while loading JSON file {json_file}: {e}")


def save_json_file(path: str, data):
    """
    Save data to a JSON file at the specified path.
    :param path: Path to the JSON file.
    :param data: Data to be saved (should be JSON serializable).
    """
    dir_name = os.path.dirname(path)
    if dir_name and not os.path.exists(dir_name):
        os.makedirs(dir_name)
    with open(path, 'w', encoding='utf-8') as f:
        try:
            json.dump(data, f, indent=4, ensure_ascii=False)
        except TypeError as e:
            raise ValueError(f"Data provided is not JSON serializable: {e}")
        except Exception as e:
            raise RuntimeError(f"Unexpected error while saving JSON file {path}: {e}")

def get_abs_path_cwd_ucagent(workspace, path):
    """
    Get the absolute path of a file or directory in the workspace.
    :param workspace: The workspace directory.
    :param path: The relative or absolute path to be resolved.
    :return: The absolute path.
    """
    ucagent_path = os.path.abspath(workspace + os.sep + ".ucagent")
    if not os.path.exists(ucagent_path):
        os.makedirs(ucagent_path)
    return os.path.abspath(ucagent_path + os.sep + path)


def save_ucagent_info(workspace, info: dict):
    """
    Save UCAgent information to a JSON file in the workspace.
    :param workspace: The workspace directory where the file will be saved.
    :param info: The UCAgent information to be saved.
    """
    assert os.path.exists(workspace), f"Workspace {workspace} does not exist."
    info_path = get_abs_path_cwd_ucagent(workspace, "ucagent_info.json")
    save_json_file(info_path, info)


def load_ucagent_info(workspace) -> dict:
    """
    Load UCAgent information from a JSON file in the workspace.
    :param workspace: The workspace directory where the file is located.
    :return: The loaded UCAgent information.
    """
    if not os.path.exists(workspace):
        return {}
    info_path = get_abs_path_cwd_ucagent(workspace, "ucagent_info.json")
    if not os.path.exists(info_path):
        return {}
    return load_json_file(info_path)


def load_toffee_report(
    result_json_path: str,
    workspace: str,
    run_test_success: bool,
    return_all_checks: bool,
) -> dict:
    """
    Load a Toffee JSON report from the specified path.
    :param path: Path to the Toffee JSON report file.
    :return: Parsed Toffee report data.
    """
    assert os.path.exists(result_json_path), f"Toffee report file {result_json_path} does not exist."
    ret_data = {
        "run_test_success": run_test_success,
    }
    try:
        data = load_json_file(result_json_path)
    except Exception as e:
        raise RuntimeError(f"Failed to load JSON file {result_json_path}: {e}")
    # Extract relevant information from the JSON data
    # tests
    test_abstract_info = data.get("test_abstract_info", {})
    if not isinstance(test_abstract_info, dict):
        raise ValueError(f"Expected test_abstract_info to be a dict, got {type(test_abstract_info)}")
    try:
        tests = get_toffee_json_test_case(workspace, test_abstract_info)
    except Exception as e:
        raise RuntimeError(f"Failed to parse test case information: {e}")
    if not isinstance(tests, list):
        raise ValueError(f"Expected tests to be a list, got {type(tests)}")
    if not tests:
        # Handle empty test cases
        tests_map = {}
        fails = []
    else:
        try:
            # Check if all items in tests are proper tuples with at least 2 elements
            for i, test_item in enumerate(tests):
                if not isinstance(test_item, (list, tuple)) or len(test_item) < 2:
                    raise ValueError(f"Test item {i} is not a proper tuple/list with at least 2 elements: {test_item}")
            
            tests_map = {k[0]: k[1] for k in tests}
            fails = [k[0] for k in tests if k[1] == "FAILED"]
        except Exception as e:
            raise RuntimeError(f"Failed to process test results: {e}. Tests data: {tests}")
    ret_data["tests"] = {
        "total": len(tests),
        "fails": len(fails),
    }
    ret_data["tests"]["test_cases"] = tests_map
    # coverages
    # functional coverage
    fc_data = data.get("coverages", {}).get("functional", {})
    ret_data["total_funct_point"] = fc_data.get("point_num_total", 0)
    ret_data["total_check_point"] = fc_data.get("bin_num_total",   0)
    ret_data["failed_funct_point"] = ret_data["total_funct_point"] - fc_data.get("point_num_hints", 0)
    ret_data["failed_check_point"] = ret_data["total_check_point"] - fc_data.get("bin_num_hints",   0)
    # failed bins:
    # groups->points->bins
    bins_fail = []
    bins_unmarked = []
    bins_funcs = {}
    failed_funcs_bins = {}
    bins_funcs_reverse = {}
    bins_all = []
    for g in fc_data.get("groups", []):
        for p in g.get("points", []):
            cv_funcs = p.get("functions", {})
            for b in p.get("bins", []):
                bin_full_name = rm_blank_in_str("%s/%s/%s" % (g["name"], p["name"], b["name"]))
                bin_is_fail = b["hints"] == 0
                if bin_is_fail:
                    bins_fail.append(bin_full_name)
                test_funcs = cv_funcs.get(b["name"], [])
                if len(test_funcs) < 1:
                    bins_unmarked.append(bin_full_name)
                else:
                    for tf in test_funcs:
                        func_key = rm_workspace_prefix(workspace, tf)
                        if func_key not in bins_funcs:
                            bins_funcs[func_key] = []
                        if func_key in fails:
                            if func_key not in failed_funcs_bins:
                                failed_funcs_bins[func_key] = []
                            failed_funcs_bins[func_key].append(bin_full_name)
                        bins_funcs[func_key].append(bin_full_name)
                        if bin_full_name not in bins_funcs_reverse:
                            bins_funcs_reverse[bin_full_name] = []
                        bins_funcs_reverse[bin_full_name].append(
                            [func_key, tests_map.get(func_key, "Unknown")]
                        )
                # all bins
                bins_all.append(bin_full_name)
    ret_data["failed_test_case_with_check_point_list"] = failed_funcs_bins
    if return_all_checks:
        ret_data["all_check_point_list"] = bins_all
    if len(bins_fail) > 0:
        ret_data["failed_check_point_list"] = bins_fail
    ret_data["unmarked_check_points"] = len(bins_unmarked)
    if len(bins_unmarked) > 0:
        ret_data["unmarked_check_point_list"] = bins_unmarked
    # functions with no check points
    test_fc_no_check_points = []
    for f, _ in tests:
        if f not in bins_funcs:
            test_fc_no_check_points.append(f)
    ret_data["test_function_with_no_check_point_mark"] = len(test_fc_no_check_points)
    if len(test_fc_no_check_points) > 0:
        ret_data["test_function_with_no_check_point_mark_list"] = test_fc_no_check_points
    return ret_data


def del_report_keys(report: dict, keys: List[str]) -> dict:
    """
    Delete specified keys from a report dictionary.
    :param report: The report dictionary.
    :param keys: List of keys to be deleted.
    :return: The modified report dictionary.
    """
    if not keys:
        return report
    for key in keys:
        if "." in key:
            sub_report = report
            parts = key.split(".")
            for p in parts[:-1]:
                if p in sub_report and isinstance(sub_report[p], dict):
                    sub_report = sub_report[p]
                else:
                    sub_report = None
                    break
            if sub_report is not None and parts[-1] in sub_report:
                del sub_report[parts[-1]]
        else:
            if key in report:
                del report[key]
    return report


def get_toffee_json_test_case(workspace: str, item: dict) -> str:
    """
    Get the test case file and word from a toffee JSON item.
    :param workspace: The workspace directory where the test case files are located.
    :param item: A dictionary representing a test case item from the toffee JSON report.
    :return: A tuple containing the relative path to the test case file and the status word.
    """
    ret = []
    for k, v in item.items():
        key = k.replace(os.path.abspath(workspace), "")
        if key.startswith(os.sep):
            key = key[1:]
        ret.append((key, v))
    return ret


def get_unity_chip_doc_marks(path: str, leaf_node: str, mini_leaf_count: int = 0, error_char_list=["*", "?"], return_line_block: bool = False):
    """
    Get the Unity chip documentation marks from a file.
    :param path: Path to the file containing Unity chip documentation.
    :param leaf_node: The leaf node type to consider in the documentation hierarchy.
    :param mini_leaf_count: The minimum number of leaf nodes required.
    :return: key_name_list
    :return: key_name_list, key_line_blocks
    """
    keynames = ["FG", "FC", "CK", "BG", "TC"]
    assert leaf_node in keynames, f"Invalid leaf_node '{leaf_node}'. Must be one of {keynames}."
    prefix   = ["<FG-", "<FC-", "<CK-", "<BG-", "<TC-"]
    subfix   = [">"]* len(prefix)
    data = parse_nested_keys(path, keynames, prefix, subfix)
    tindex = keynames.index(leaf_node)
    klist, blist, klines = nested_keys_as_list(data, leaf_node, keynames)
    assert len(klist) >= mini_leaf_count, f"Need {mini_leaf_count} {leaf_node} at least, but find {len(klist)}"
    fmsg = ", ".join([f"{b[1]} at line {b[2]} need sub node '<{leaf_node}-*>'" for b in blist])
    assert len(blist) == 0, f"Incomplete label '<{leaf_node}-*>' detected: `{fmsg}`, delete the incomplete labels or fix it according to the format requirements: " + \
                            f"{' '.join([x+'*>' for x in prefix[:tindex+1]])}"
    invalid_char_keys = []
    finded_keys = set()
    for k in klist:
        for ec in error_char_list:
            if ec in k:
                invalid_char_keys.append(k)
                finded_keys.add(k)
    if len(invalid_char_keys) > 0:
        invalid_char_keys = ", ".join(invalid_char_keys)
        finded_keys = ", ".join(finded_keys)
        raise ValueError(f"Invalid characters {finded_keys} found in keys: {invalid_char_keys}")
    if not return_line_block:
        return klist
    return klist, get_file_blocks(path, klines)


def get_file_blocks(file_path: str, line_info) -> dict:
    blocks = OrderedDict()
    with open(file_path, "r") as f:
        lines = f.read().splitlines()
        for k, v in line_info.items():
            blocks[k] = []
            line = v.get("line", -1)
            if line < 0:
                continue
            pline = v.get("pline", line)
            nline = v.get("nline", line)
            nsize = max(1, nline - pline + 1)
            lnfmt = f"%0" + str(max(len(str(nline)), len(str(pline)))) + "d: %s"
            for i, l in enumerate(lines[pline - 1: pline - 1 + nsize]):
                if i == 0 and pline != line:
                    l = "..."
                if i == (nsize - 1) and nline != line:
                    l = "..."
                blocks[k].append(lnfmt % (i + pline, l))
    return blocks


def merge_file_blocks(blocks: list) -> list:
    if len(blocks) < 2:
        return blocks
    line_content = {}
    for block in blocks:
        for lines in block.values():
            for line in lines:
                if ": " in line:
                    n, c = line.split(": ", 1)
                    if c != "...":
                        line_content[int(n)] = line
    recorded_lines = set()
    result = []
    for block in blocks:
        for key, lines in block.items():
            new_lines = []
            no_list = []
            for line in lines:
                if ": " in line:
                    n, _ = line.split(": ", 1)
                    n = int(n)
                    no_list.append(n)
                    if n in recorded_lines:
                        continue
                    recorded_lines.add(n)
                    new_lines.append(line_content.get(n, line))
                else:
                    new_lines.append(line)
            if new_lines:
                mean_value = -1
                if no_list:
                    mean_value = no_list[len(no_list) // 2]
                result.append(({key: new_lines}, mean_value))
    # sort by mean line number
    result.sort(key=lambda x: x[1])
    return [result[0] for result in result]


def rm_workspace_prefix(workspace: str, path: str) -> dict:
    """
    Remove the workspace prefix from the keys in a dictionary.
    :param workspace: The workspace directory to be removed from the keys.
    :param path: The path to the file or directory.
    :return: A path with the workspace prefix removed.
    """
    workspace = os.path.abspath(workspace)
    if path.startswith(os.sep):
        path = path[1:]
    abs_path = os.path.abspath(os.path.join(workspace, path))
    assert abs_path.startswith(workspace), f"Path {abs_path} is not under workspace {workspace}."
    path = abs_path.replace(workspace, "")
    if path.startswith(os.sep):
        path = path[1:]
    return path if path else "."


def import_class_from_str(class_path: str, modue: None = None):
    """
    Import a class from a string like 'module.submodule.ClassName'
    """
    if "." not in class_path:
        assert modue is not None, "Module must be provided if class_path does not contain a dot."
        return getattr(modue, class_path)
    module_path, class_name = class_path.rsplit(".", 1)
    module = importlib.import_module(module_path)
    return getattr(module, class_name)


def append_python_path(py_path: list):
    """
    Append paths to sys.path for Python module imports.
    :param py_path: List of paths to be added to sys.path.
    """
    import sys
    if isinstance(py_path, str):
        py_path = [py_path]
    for p in py_path:
        if not os.path.exists(p):
            raise FileNotFoundError(f"Path {p} does not exist.")
        if os.path.isfile(p):
            p = os.path.dirname(p)
        p = os.path.abspath(p)
        if p not in sys.path:
            sys.path.append(p)


def import_python_file(file_path: str, py_path: list = []):
    """
    Import a Python file as a module.
    :param file_path: Path to the Python file to be imported.
    :return: The imported module.
    """
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"File {file_path} does not exist.")
    module_name = os.path.splitext(os.path.basename(file_path))[0]
    if py_path:
        append_python_path(py_path)
    spec = importlib.util.spec_from_file_location(module_name, file_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def render_template(template: str, kwargs) -> str:
    """
    Render a template string with the provided keyword arguments.
    :param template: The template string to be rendered.
    :param kwargs: Keyword arguments to be used in the template.
    :return: The rendered string.
    """
    if not isinstance(template, str):
        return template
    tvalue = template.strip()
    if (tvalue.count("{") == tvalue.count("}") == 1) and \
       (tvalue.startswith("{") and tvalue.endswith("}")):
        key = tvalue.replace("}", "").replace("{", "").strip()
        if isinstance(kwargs, dict):
            target = kwargs.get(key)
        else:
            target = getattr(kwargs, key, None)
        if target is not None:
            return target
        return template
    else:
        for k in re.findall(r"\{[^{}]*\}", template):
            key = str(k).replace("}", "").replace("{", "").strip()
            if isinstance(kwargs, dict):
                target = kwargs.get(key)
            else:
                target = getattr(kwargs, key, None)
            if target is not None:
                template = template.replace(k, str(target))
        return template


def fill_template(data, template_data):
    if template_data is None:
        return data
    if isinstance(data, str):
        return render_template(data, template_data)
    elif isinstance(data, list):
        return [fill_template(d, template_data) for d in data]
    elif isinstance(data, (dict, OrderedDict)):
        ret = OrderedDict()
        for k, v in data.items():
            k = render_template(k, template_data)
            v = render_template(v, template_data)
            ret[k] = v
        return ret
    return data


def find_files_by_regex(workspace, pattern):
    """
    Find files in a workspace that match a given regex pattern.
    """
    matched_files = []
    assert os.path.exists(workspace), f"Workspace {workspace} does not exist."
    abs_workspace = os.path.abspath(workspace)
    def __find(p):
        regex = re.compile(p)
        for root, dirs, files in os.walk(abs_workspace):
            for filename in files:
                if regex.search(filename):
                    f = os.path.abspath(os.path.join(root, filename))
                    matched_files.append(
                        f.removeprefix(abs_workspace + os.sep)
                    )
    if isinstance(pattern, str):
        pattern = [pattern]
    for p in pattern:
        __find(p)
    return list(set(matched_files))


def find_files_by_glob(workspace, pattern):
    """Find files in a workspace that match a given glob pattern."""
    import glob

    assert os.path.exists(workspace), f"Workspace {workspace} does not exist."
    if isinstance(pattern, str):
        pattern = [pattern]
    abs_workspace = os.path.abspath(workspace)
    ret = set()

    def __find(p):
        for f in glob.glob(os.path.join(abs_workspace, p), recursive=True):
            ret.add(f.removeprefix(abs_workspace + os.sep))

    for p in pattern:
        __find(p)
    return list(ret)


def find_files_by_pattern(workspace, pattern, ignore_warn=False):
    """Find files in a workspace that match a given pattern, which can be either a glob or regex."""

    def is_regex_pattern(s: str) -> bool:
        try:
            re.compile(s)
            return True
        except re.error:
            return False
    if isinstance(pattern, str):
        pattern = [pattern]
    ret = []
    for p in pattern:
        if os.path.isfile(workspace + os.path.sep + p):
            ret.append(p)
            continue
        # first try glob
        new_p = find_files_by_glob(workspace, p)
        # if no files found, try regex
        if not new_p and is_regex_pattern(p):
            new_p += find_files_by_regex(workspace, p)
        if len(new_p) < 1 and not ignore_warn:
            warning(f"No files found in workspace {workspace} matching pattern: {p}")
            continue
        ret += new_p
    return list(set(ret))


def dump_as_json(data):
    """
    Convert a dictionary to a JSON string with pretty formatting.
    """
    if isinstance(data, str):
        return data
    return json.dumps(data, indent=4, ensure_ascii=False) #.replace("\\n", "\n").replace("\\", "")

def copytree_incremental(src_dir, dst_dir, skip_existing=True, enable_skill_list=[], disable_skill_list=[]):
    """
    Incremental copying of directories with skill-based filtering.
    
    :param src_dir: source directory
    :param dst_dir: destination directory
    :param skip_existing: whether to skip existing files (True=skip, False=overwrite)
    :param enable_skill_list: list of skills to include (only copy these)
    :param disable_skill_list: list of skills to exclude (copy all except these)
    :return: list of copied skills
    """
    if not os.path.exists(src_dir):
        raise ValueError(f"Source directory {src_dir} does not exist")
    if not os.path.exists(dst_dir):
        os.makedirs(dst_dir)

    # if both are not empty, enable_skill_list is first
    if enable_skill_list and disable_skill_list:
        disable_skill_list = []

    copied_skills = []
    
    # recursively discover skills by SKILL.md
    skill_dir_map = {}
    for root, _, files in os.walk(src_dir):
        if "SKILL.md" not in files:
            continue
        skill_name = os.path.basename(root)
        skill_dir_map[skill_name] = root
    all_skills = list(skill_dir_map.keys())
    
    # determine which skills to copy based on enable_skill_list and disable_skill_list
    skills_to_copy = []
    if enable_skill_list:
        skills_to_copy = [skill for skill in enable_skill_list if skill in skill_dir_map]
    elif disable_skill_list:
        skills_to_copy = [skill for skill in all_skills if skill not in disable_skill_list]
    else:
        skills_to_copy = all_skills
    
    # copy skill
    for skill in skills_to_copy:
        src_skill_dir = skill_dir_map.get(skill)
        if not src_skill_dir or not os.path.exists(src_skill_dir):
            continue
        skill_rel_dir = os.path.relpath(src_skill_dir, src_dir)
        dst_skill_dir = os.path.join(dst_dir, skill_rel_dir)
        if not os.path.exists(dst_skill_dir):
            os.makedirs(dst_skill_dir)
        for root, dirs, files in os.walk(src_skill_dir):
            rel_path = os.path.relpath(root, src_skill_dir)
            dst_path = os.path.join(dst_skill_dir, rel_path) if rel_path != '.' else dst_skill_dir
            for dirname in dirs:
                dst_subdir = os.path.join(dst_path, dirname)
                if not os.path.exists(dst_subdir):
                    os.makedirs(dst_subdir)
            file_copied = False
            for filename in files:
                src_file = os.path.join(root, filename)
                dst_file = os.path.join(dst_path, filename)
                if os.path.exists(dst_file) and skip_existing:
                    continue
                else:
                    shutil.copy2(src_file, dst_file)
                    file_copied = True
            if file_copied and skill not in copied_skills:
                copied_skills.append(skill)
    
    return copied_skills


def find_skill_dir_by_name(root_dir, target_dir_name):
    if not root_dir or not target_dir_name:
        return None
    if not os.path.isdir(root_dir):
        return None

    for root, dirs, _ in os.walk(root_dir):
        if target_dir_name in dirs:
            return os.path.join(root, target_dir_name)
    return None

def render_template_dir(workspace, template_dir, kwargs):
    """
    Render all template files in a directory with the provided keyword arguments.
    :param workspace: The workspace directory where the templates are located.
    :param template_dir: The directory containing the template files.
    :param kwargs: Keyword arguments to be used in the templates.
    :return: A dictionary mapping file names to rendered content.
    """
    assert os.path.exists(workspace), f"Workspace {workspace} does not exist."
    assert os.path.exists(template_dir), f"Template directory {template_dir} does not exist."
    import jinja2
    import shutil
    dst_dir = os.path.join(workspace, os.path.basename(template_dir))
    if os.path.exists(dst_dir):
        shutil.rmtree(dst_dir)
    shutil.copytree(template_dir, dst_dir)
    rendered_files = []
    env = jinja2.Environment(loader=jinja2.FileSystemLoader(dst_dir), keep_trailing_newline=True)
    for root, _, files in os.walk(dst_dir):
        for fname in files:
            abs_path = os.path.join(root, fname)
            new_fname = jinja2.Template(fname).render(**kwargs)
            new_abs_path = os.path.join(root, new_fname)
            if new_fname != fname:
                os.rename(abs_path, new_abs_path)
                abs_path = new_abs_path
            if "/__pycache__/" in abs_path or not is_text_file(abs_path):
                continue
            info(f"Rendering template file: {abs_path}")
            with open(abs_path, "r", encoding="utf-8") as f:
                content = f.read()
            template = env.from_string(content)
            rendered_content = template.render(**kwargs)
            with open(abs_path, "w", encoding="utf-8") as f:
                f.write(rendered_content)
            rendered_files.append(os.path.relpath(abs_path, workspace))
    return rendered_files


def get_template_path(template_name: str, lang:str=None, template_path:str=None) -> str:
    """
    Get the absolute path to a template file.
    :param template_name: The name of the template file.
    :return: The absolute path to the template file.
    """
    if not template_name:
        return None
    if not template_path:
        assert lang is not None, "Language must be specified if template_path is not provided."
        current_dir = os.path.dirname(os.path.abspath(__file__))
        template_path = os.path.abspath(os.path.join(current_dir, "../lang", lang, "template"))
    else:
        assert os.path.exists(template_path), f"Template path {template_path} does not exist."
    tmp = os.path.join(template_path, template_name)
    assert os.path.exists(tmp), f"Template {template_name} does not exist at {template_path}."
    return tmp


def append_time_str(data: str):
    time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    return data + "\nNow time: " + time_str


def fill_dlist_none(data, value, keys=None, json_keys=[]):
    def _conver_json(v):
        assert isinstance(v, str)
        if not v:
            return value
        v = fix_json_string(v)
        try:
            json.loads(v)
            return v
        except json.JSONDecodeError as e:
            from .log import warning
            v = f"Find Invalid JSON string: {repr(v)} - {e}, set as empty JSON object."
            warning(v)
            return json.dumps({"error": v})
    _keys = keys
    if keys is not None:
        if isinstance(keys, str):
            _keys = [keys]
    if data is None:
        return value
    if not isinstance(data, (dict, list)):
        return data
    if isinstance(data, dict):
        for k, v in data.items():
            if v is None:
                if _keys is not None and k not in _keys:
                    continue
            if k in json_keys and isinstance(v, str):
                data[k] = _conver_json(v)
            else:
                data[k] = fill_dlist_none(v, value, _keys, json_keys)
    elif isinstance(data, list):
        for i, v in enumerate(data):
            data[i] = fill_dlist_none(v, value, _keys, json_keys)
    return data


def get_ai_message_tool_call(msg):
    lines = []
    def _format_tool_args(tc) -> list[str]:
        lines = [
            f"  {tc.get('name', 'Tool')} ({tc.get('id')})",
            f" Call ID: {tc.get('id')}",
        ]
        if tc.get("error"):
            lines.append(f"  Error: {tc.get('error')}")
        lines.append("  Args:")
        args = tc.get("args")
        if isinstance(args, str):
            lines.append(f"    {args}")
        elif isinstance(args, dict):
            for arg, value in args.items():
                lines.append(f"    {arg}: {value}")
        return lines
    if msg.tool_calls:
        lines.append("Tool Calls:")
        for tc in msg.tool_calls:
            lines.extend(_format_tool_args(tc))
    if msg.invalid_tool_calls:
        lines.append("Invalid Tool Calls:")
        for itc in msg.invalid_tool_calls:
            lines.extend(_format_tool_args(itc))
    return "\n".join(lines) if lines else None


def get_func_arg_list(func):
    """
    Get the argument names of a function.
    :param func: The function to inspect.
    :return: A list of argument names.
    """
    if not callable(func):
        raise ValueError("Provided object is not callable.")
    sig = inspect.signature(func)
    return [
        param.name
        for param in sig.parameters.values()
        if param.kind
        in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
    ]


def get_target_from_file(target_file, func_pattern, ex_python_path = [], dtype="FUNC"):
    """
    Import target file and get objects (functions, classes, or all) that match the given pattern.
    :param target_file: Path to the Python file to import.
    :param func_pattern: Pattern to match object names. Can be:
                        - Exact string: "func_A1" or "ClassA"
                        - Glob pattern: "func_A*" or "Class*"
                        - Regex pattern: r"func_[A-Z]\\d+" or r"Class[A-Z]+"
    :param ex_python_path: Additional Python paths to add to sys.path for import.
    :param dtype: Type of objects to retrieve. Options:
                - "FUNC": Only functions
                - "CLASS": Only classes
                - "ALL": All objects (functions, classes, variables, etc.)
    :return: List of objects that match the pattern and type criteria.
    """
    import sys
    import importlib.util
    import fnmatch
    import re
    import types
    # Validate input parameters
    valid_dtypes = ["FUNC", "CLASS", "ALL"]
    if dtype not in valid_dtypes:
        raise ValueError(f"Invalid dtype '{dtype}'. Must be one of {valid_dtypes}.")
    # Validate target file exists
    if not os.path.exists(target_file):
        raise FileNotFoundError(f"Target file {target_file} does not exist.")
    # Add extra Python paths if provided
    if isinstance(ex_python_path, str):
        ex_python_path = [ex_python_path]
    elif not isinstance(ex_python_path, list):
        ex_python_path = list(ex_python_path)
    ex_python_path.append(os.path.dirname(target_file))  # Ensure the target file's directory is included
    ex_python_path = list(set(ex_python_path))  # Remove duplicates
    for path in ex_python_path:
        info(f"Adding '{path}' to sys.path for import.")
        if os.path.exists(path) and path not in sys.path:
            sys.path.insert(0, path)
    try:
        # Import the target file as a module
        module_name = os.path.splitext(os.path.basename(target_file))[0]
        spec = importlib.util.spec_from_file_location(module_name, target_file)
        if spec is None:
            raise ImportError(f"Could not create module spec for {target_file}")
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        # Helper function to check object type
        def is_target_type(obj, target_dtype):
            if target_dtype == "FUNC":
                return callable(obj)
            elif target_dtype == "CLASS":
                return isinstance(obj, type) and not isinstance(obj, types.ModuleType)
            elif target_dtype == "ALL":
                return True
            return False
        # Get all objects from the module based on type
        all_objects = []
        for name in dir(module):
            obj = getattr(module, name)
            # Skip private/protected members and built-ins
            if name.startswith("_"):
                continue
            # Check if object is defined in this module (not imported)
            if hasattr(obj, "__module__") and obj.__module__ != module_name:
                continue
            # For classes, also check if they're defined in this file
            if isinstance(obj, type):
                if not hasattr(obj, "__module__") or obj.__module__ != module_name:
                    continue
            # Check if object matches the target dtype
            if is_target_type(obj, dtype):
                all_objects.append((name, obj))
        # Filter objects based on pattern
        matched_objects = []

        # Determine if pattern is regex or glob
        def is_regex_pattern(pattern):
            """Check if pattern contains regex special characters"""
            regex_chars = set("[]()+?^${}\\|.")
            return any(char in pattern for char in regex_chars)

        if is_regex_pattern(func_pattern):
            # Treat as regex pattern
            try:
                regex = re.compile(func_pattern)
                for name, obj in all_objects:
                    if regex.match(name):
                        matched_objects.append(obj)
            except re.error as e:
                raise ValueError(f"Invalid regex pattern '{func_pattern}': {e}")
        else:
            # Treat as glob pattern or exact string
            for name, obj in all_objects:
                if fnmatch.fnmatch(name, func_pattern):
                    matched_objects.append(obj)
        return matched_objects
    except Exception as e:
        raise ImportError(f"Failed to import and process {target_file}: {e}")


def list_files_by_mtime(
    directory: str | os.PathLike[str],
    max_files: int = 100,
    subdir: str | Sequence[str] | None = None,
    ignore_patterns: str = "*.pyc,*.log,*.tmp,*.fst,*.dat,*.vcd,*.bin,*.ini,.*",
) -> list[tuple[float, float, str]]:
    """列出目录中的文件并按修改时间倒序排列"""
    root = Path(directory).resolve()
    if not root.is_dir():
        return []

    now = time.time()
    patterns = [pattern.strip() for pattern in ignore_patterns.split(",") if pattern.strip()]
    files: list[tuple[float, float, str]] = []

    def collect(current: Path) -> None:
        try:
            for entry in current.iterdir():
                try:
                    stat_result = entry.stat()
                except OSError as error:
                    warning(f"Error processing path {entry}: {error}")
                    continue

                if stat.S_ISDIR(stat_result.st_mode):
                    collect(entry)
                    continue

                if not stat.S_ISREG(stat_result.st_mode):
                    continue

                try:
                    relative = entry.relative_to(root)
                except ValueError:
                    relative = Path(os.path.relpath(entry, root))

                relative_str = str(relative)
                if any(fnmatch.fnmatch(relative_str, pattern) for pattern in patterns):
                    continue

                files.append((now - stat_result.st_mtime, stat_result.st_mtime, relative_str))
        except OSError as error:
            warning(f"Error listing directory {current}: {error}")

    if subdir is None:
        targets = [root]
    elif isinstance(subdir, str):
        targets = [root / subdir]
    else:
        targets = [root / item for item in subdir]

    for target in targets:
        if not target.is_dir():
            continue
        collect(target)

    files.sort(key=lambda x: x[0])
    return files[:max_files]


def fix_json_string(json_str):
    try:
        json.loads(json_str)
        return json_str
    except json.JSONDecodeError:
        pass
    try:
        py_obj = ast.literal_eval(json_str)
        return json.dumps(py_obj)
    except (SyntaxError, ValueError):
        pass
    fixed = json_str
    in_string = False
    quote_char = None
    i = 0
    result = []
    while i < len(fixed):
        char = fixed[i]
        if char in ["'", '"']:
            if not in_string:
                in_string = True
                quote_char = char
                result.append('"')
            elif char == quote_char and (i == 0 or fixed[i - 1] != "\\"):
                in_string = False
                result.append('"')
            else:
                result.append(char)
        else:
            result.append(char)
        i += 1
    fixed = "".join(result)
    fixed = re.sub(r"([{,])\s*([a-zA-Z0-9_]+)\s*:", r'\1"\2":', fixed)
    fixed = re.sub(r",\s*([}\]])", r"\1", fixed)
    try:
        json.loads(fixed)
        return fixed
    except json.JSONDecodeError:
        return json_str


def import_and_instance_tools(class_list: List[str], module=None):
    """
    Import a list of classes from their string representations.
    :param class_list: List of class strings in the format 'module.ClassName'.
    :param module: Optional module to import from if class_list does not contain a dot.
    :return: A list of imported classes.
    """
    if not class_list:
        return []

    def _attach_call_count(instance):
        if hasattr(instance, "call_count"):
            return instance
        warning(f"Attaching call_count to tool instance of type {type(instance)}")
        instance.__dict__["call_count"] = 0

        def get_new_invoke(old_inv):
            def new_invoke(self, input, config=None, **kwargs):
                self.call_count += 1
                return old_inv(input, config, **kwargs)
            return new_invoke
        def get_new_ainvoke(old_ainv):
            def new_ainvoke(self, input, config=None, **kwargs):
                self.call_count += 1
                return old_ainv(input, config, **kwargs)
            return new_ainvoke
        object.__setattr__(instance, 'invoke', get_new_invoke(object.__getattribute__(instance, "invoke")))
        object.__setattr__(instance, 'ainvoke', get_new_ainvoke(object.__getattribute__(instance, "ainvoke")))
        return instance
    tools = []
    for cls in class_list:
        if "." not in cls:
            assert module is not None, "Module must be provided if class does not contain a dot."
            tools.append(_attach_call_count(getattr(module, cls)()))
        else:
            module_path, class_name = cls.rsplit(".", 1)
            mod = importlib.import_module(module_path)
            tools.append(_attach_call_count(getattr(mod, class_name)()))
    return tools


def convert_tools(tools):
    from langgraph.prebuilt.tool_node import ToolNode
    llm_builtin_tools: list[dict] = []
    if isinstance(tools, ToolNode):
        tool_classes = list(tools.tools_by_name.values())
        tool_node = tools
    else:
        llm_builtin_tools = [t for t in tools if isinstance(t, dict)]
        tool_node = ToolNode([t for t in tools if not isinstance(t, dict)])
        tool_classes = list(tool_node.tools_by_name.values())
    return llm_builtin_tools + tool_classes


def copy_indent_from(src: list, dst: list):
    """
    Copy the indentation from the source string to the destination string.
    :param src: The source string from which to copy the indentation.
    :param dst: The destination string to which the indentation will be applied.
    :return: The destination string with the copied indentation.
    """
    if not src or not dst:
        return dst
    ret = []
    indent = 0
    for s, d in zip(src, dst):
        if not s or not d:
            ret.append(d)
            continue
        indent = len(s) - len(s.lstrip())
        ret.append(" " * indent + d.lstrip())
    if len(src) < len(dst):
        for d in dst[len(src) :]:
            ret.append(" " * indent + d)
    return ret


def _install_uvicorn_log_bridge(logger):
    """Attach a handler to the real uvicorn loggers that delegates to *logger*.

    uvicorn's module-level ``logger = logging.getLogger("uvicorn.error")`` is
    evaluated at import time, before the monkey-patch in create_verify_mcps
    runs.  This means runtime log calls bypass the patched getLogger and go to
    the standard logger.  By adding a handler *after* construction we ensure
    runtime messages (startup, connection, shutdown) are forwarded to the TUI.
    """
    import logging

    class _BridgeHandler(logging.Handler):
        def emit(self, record):
            try:
                logger.log(record.levelno, record.getMessage())
            except Exception:
                pass

    handler = _BridgeHandler()
    handler.setLevel(logging.DEBUG)
    for name in ("uvicorn.error", "uvicorn.access"):
        uv_logger = logging.getLogger(name)
        uv_logger.addHandler(handler)


def create_verify_mcps(mcp_tools: list, host: str, port: int, logger=None):
    import logging
    __old_getLogger = logging.getLogger
    def __getLogger(name):
        return logger
    if logger:
        logging.getLogger = __getLogger
    try:
        from mcp.server.fastmcp import FastMCP
        from ucagent.tools.uctool import to_fastmcp
        from ucagent.util.log import info
        fastmcp_tools = []
        for tool in mcp_tools:
            fastmcp_tools.append(to_fastmcp(tool))
        # Start the FastMCP server
        info(f"create FastMCP server with tools: {[tool.name for tool in fastmcp_tools]}")
        mcp = FastMCP("UnityTest", tools=fastmcp_tools, host=host, port=port)
        s = mcp.settings
        info(f"FastMCP server started at {s.host}:{s.port}")
        starlette_app = mcp.streamable_http_app()
        import uvicorn
        config = uvicorn.Config(
            starlette_app,
            host=mcp.settings.host,
            port=mcp.settings.port,
            log_level=mcp.settings.log_level.lower(),
            timeout_keep_alive=300,
            timeout_graceful_shutdown=60,
        )
        server = uvicorn.Server(config)
    finally:
        logging.getLogger = __old_getLogger

    if logger:
        _install_uvicorn_log_bridge(logger)

    return server, __old_getLogger


def start_verify_mcps(server, old_getLogger):
    import logging
    from ucagent.util.log import info
    import anyio
    async def _run():
        await server.serve()
    try:
        anyio.run(_run)
    except Exception as e:
        info(f"FastMCP server exit with: {e}")
    info("FastMCP server stopped.")
    # logging.getLogger was already restored in create_verify_mcps; this is kept
    # for safety in case old_getLogger is still the real function (no-op then).
    if old_getLogger is not None:
        logging.getLogger = old_getLogger


def stop_verify_mcps(server):
    from ucagent.util.log import info
    if server is not None:
        info("Stopping FastMCP server...")
        server.should_exit = True
    else:
        info("FastMCP server is not running.")


def get_diff(old_lines, new_lines, file_name):
    import difflib
    diff = difflib.unified_diff(
        old_lines,
        new_lines,
        fromfile=file_name + "(old)",
        tofile=file_name + "(new)",
    )
    if not diff:
        return "\n[DIFF]\nNo changes detected."
    return "\n[DIFF]\n" + "".join(diff)


def max_str(str_data, max_size=10):
    if len(str_data) <= max_size:
        return str_data
    return str_data[:max_size] + "..."


def ordered_dict_representer(dumper, data):
    return dumper.represent_dict(data.items())
yaml.add_representer(OrderedDict, ordered_dict_representer)


def yam_str(data: dict) -> str:
    """
    Convert a dictionary to a YAML-formatted string.
    """
    class LiteralStr(str):
        """Custom string class for literal scalar representation"""
        pass
    def represent_literal_str(dumper, data):
        """Custom representer for literal strings"""
        if "\n" in data:
            # Use literal style (|) for multi-line strings
            return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="|")
        else:
            # Use default style for single-line strings
            return dumper.represent_scalar("tag:yaml.org,2002:str", data)

    def process_strings(obj):
        if isinstance(obj, dict):
            ret = OrderedDict()
            for k, v in obj.items():
                ret[k] = process_strings(v)
            return ret
        elif isinstance(obj, list):
            return [process_strings(item) for item in obj]
        elif isinstance(obj, str) and "\n" in obj:
            return LiteralStr(obj)
        else:
            return obj

    processed_data = process_strings(data)
    yaml.add_representer(LiteralStr, represent_literal_str)
    try:
        return yaml.dump(
            processed_data,
            allow_unicode=True,
            default_flow_style=False,
            width=float("inf"),  # Prevent line wrapping
            indent=2,
        )
    finally:
        if LiteralStr in yaml.representer.Representer.yaml_representers:
            del yaml.representer.Representer.yaml_representers[LiteralStr]


def rm_blank_in_str(input_str: str) -> str:
    """Remove blank lines from a string."""
    assert isinstance(input_str, str), "Input must be a string."
    return "".join([c.strip() for c in input_str.split()])


def parse_marks_from_file(file_path: str, tag: str) -> dict:
    """Parse marks from a file based on a given tag.

    Args:
        file_path (str): The path to the file to parse.
        tag (str): The tag to filter marks. eg ABC means <ABC>value</ABC>

    Returns:
        dict: marks that match the given tag.
    """
    ret = {
        "detail": [],
    }
    tag = tag.strip()
    assert os.path.exists(file_path), f"File {file_path} does not exist."
    with open(file_path, "r", encoding="utf-8") as f:
        lines = f.readlines()
        for i, line in enumerate(lines):
            line = rm_blank_in_str(line).strip()
            if f"<{tag}>" not in line:
                continue
            assert f"</{tag}>" in line, f"Line {i+1}: Missing closing tag </{tag}>. Note: tags cannot span multiple lines."
            assert line.index(f"<{tag}>") < line.index(f"</{tag}>"), f"Line {i+1}: Malformed tags. Ensure <{tag}> appears before </{tag}>."
            assert line.count(f"<{tag}>") == 1 and line.count(f"</{tag}>") == 1, f"Line {i+1}: Multiple <{tag}> or </{tag}> tags found. Only one pair is allowed per line."
            value = line.split(f"<{tag}>", 1)[1].split(f"</{tag}>", 1)[0].strip()
            ret["detail"].append({
                "line": i + 1,
                "value": value,
            })
    ret["count"] = len(ret["detail"])
    ret["marks"] = [d["value"] for d in ret["detail"]]
    return ret


def parse_line_ignore_file(file_path: str) -> dict:
    """Parse ignore lines from a file.

    Args:
        file_path (str): The path to the file to parse.
    Returns:
        dict: A dictionary with the ignore lines and their count.
    """
    ret = {
        "detail": [],
    }
    assert os.path.exists(file_path), f"File {file_path} does not exist."
    with open(file_path, "r", encoding="utf-8") as f:
        lines = f.readlines()
        for i, line in enumerate(lines):
            line = rm_blank_in_str(line).strip()
            if not line or line.startswith("#"):
                continue
            value = line.split("#", 1)[0].strip()
            ret["detail"].append({
                "line": i + 1,
                "value": value,
            })
    ret["count"] = len(ret["detail"])
    ret["marks"] = [d["value"] for d in ret["detail"]]
    return ret


def parse_un_coverage_json(file_path: str, workspace: str) -> dict:
    """Parse Unity test coverage data from a file.

    Args:
        file_path (str): The path to the file to parse (related to workspace).

    Returns:
        dict: A dictionary with the coverage data and statistics.
    """
    ret = OrderedDict({
        "lines_total": 0,
        "lines_covered": 0,
        "lines_uncovered": 0,
        "coverage_rate": 0.0,
        "uncoverage_detail": [],
    })
    if file_path.startswith(os.sep):
        file_path = file_path[1:]
    file_path = os.path.abspath(os.path.join(workspace, file_path))
    assert os.path.exists(file_path), f"File {file_path} does not exist."
    data = json.load(open(file_path, "r", encoding="utf-8"))
    ret["lines_total"] = data["overview"]["total"]["line"]
    ret["lines_uncovered"] = data["overview"]["miss"]["line"]
    ret["lines_covered"] = ret["lines_total"] - ret["lines_uncovered"]
    # Parse uncovered lines details
    un_covered = data.get("uncovered", {}).get("data", {})
    if ret["lines_total"] > 0:
        ret["coverage_rate"] = float(ret["lines_covered"]) / float(ret["lines_total"])
    if ret["lines_uncovered"] > 0 and un_covered:
        for cpath, data in un_covered.items():
            if data["total"]["line"] == 0:
                continue
            for module_name, cover_lines in data["modules"].items():
                if cover_lines["miss"]["line"] == 0:
                    continue
                cpath = rm_workspace_prefix(workspace, cpath)
                lines = cover_lines["line"]
                ret["uncoverage_detail"].append(OrderedDict({
                        "module_name": module_name,
                        "lines_uncovered": cpath  + ":" + ','.join(lines),
                    }))
    return ret


def is_str_array_eq(str_list1, str_list2):
    a = sorted([s.strip() for s in str_list1 if s and s.strip()])
    b = sorted([s.strip() for s in str_list2 if s and s.strip()])
    return a == b


def get_str_array_diff(str_list1, str_list2):
    a = sorted([s.strip() for s in str_list1 if s and s.strip()])
    b = sorted([s.strip() for s in str_list2 if s and s.strip()])
    only_in_1 = [s for s in a if s not in b]
    only_in_2 = [s for s in b if s not in a]
    return only_in_1, only_in_2


def clean_report_with_keys(
    report: dict, keys: list = None, default_keys=["all_check_point_list"]
) -> dict:
    data = copy.deepcopy(report)
    target_keys = []
    if keys is not None:
        target_keys = keys
    return del_report_keys(data, list(set(target_keys + default_keys)))


def description_bug_doc():
    return [
        "[Bug Analysis Document Format] (see Guide_Doc/dut_bug_analysis.md for details)",
        "  Tag hierarchy: <FG-GROUP> / <FC-FUNCTION> / <CK-CHECKPOINT> / <BG-BUGNAME-XX> / <TC-FAILEDTESTCASE>",
        "  - Confidence(XX): integer 0~100, indicating confidence level (0=known ignore/placeholder, 100=confirmed bug)",
        "  - <TC-*> format: <TC-test_xxx.py::[ClassName::]test_func_name>, ClassName is optional",
        "  - Each <BG-*> must have at least one FAILED <TC-*> test case",
        "  - Failed checkpoints should also be recorded as bugs, using 'assert False' as placeholder if needed",
        "  Format example:",
        "    <FG-LOGIC>",
        "            <FC-ADD>",
        "                <CK-BASIC>",
        "                    <BG-ADD_OVERFLOW-80> Addition overflow handling error, 80% confidence",
        "                       <TC-test_add.py::test_add_overflow> Overflow boundary test",
        "                       <TC-test_add.py::test_add_max_value> Max value test",
        "                   Bug root cause analysis:",
        "                   ```verilog",
        "                     // Adder.v line 10, bit-width error",
        "                     10: output [WIDTH-2:0] sum,  // BUG: should be [WIDTH-1:0]",
        "                   ```",
        "                   Fix suggestion:",
        "                   ```verilog",
        "                     10: output [WIDTH-1:0] sum,  // FIX: restore correct bit-width",
        "                   ```",
    ]


def description_func_doc():
    return [
        "[Functions and Checkpoints Document Format] (see Guide_Doc/dut_functions_and_checks.md for details)",
        "  Tag hierarchy: <FG-GROUP> / <FC-FUNCTION> / <CK-CHECKPOINT>, tags must be on separate lines",
        "  Format example:",
        "    <FG-LOGIC>",
        "          Group description...",
        "            <FC-ADD>",
        "               Function description: Performs addition of two numbers.",
        "                <CK-BASIC>",
        "                  Checkpoint description: Verifies basic addition functionality.",
        "                <CK-OVERFLOW>",
        "                  Checkpoint description: Verifies addition overflow handling.",
        "             <FC-MUL>",
        "                ...",
        "    <FG-MEMORY>",
        "          ...",
    ]


def check_file_block(file_blocks, workspace, checker=None):
    """
    Check if the file blocks exist in the workspace.

    Args:
        file_blocks (dict): The file blocks to check. eg: {'file1.py': {"k1": [line_from, line_to], 'k2': [line_from, line_to]}, ...}
        workspace (str): The workspace directory.
        checker (callable, optional): A function to further check each code block. It should accept the file block string as input.
    """
    assert isinstance(file_blocks, dict), "file_blocks must be a dictionary."
    ret_map = {}
    for f, blocks in file_blocks.items():
        fpath = os.path.abspath(os.path.join(workspace, f))
        assert os.path.exists(fpath), f"File {f} does not exist in workspace {workspace}."
        if not blocks:
            continue
        assert isinstance(blocks, dict), f"Blocks for file {f} must be a dictionary."
        with open(fpath, "r", encoding="utf-8") as fr:
            lines = fr.readlines()
            line_count = len(lines)
        for k, v in blocks.items():
            assert isinstance(v, list) and len(v) == 2, f"Block {k} in file {f} must be a list of two integers [line_from, line_to]."
            line_from, line_to = v
            assert isinstance(line_from, int) and isinstance(line_to, int), f"Block {k} in file {f} must contain integers."
            assert 1 <= line_from <= line_count, f"Block {k} in file {f}: line_from {line_from} is out of range (1-{line_count})."
            assert 1 <= line_to <= line_count, f"Block {k} in file {f}: line_to {line_to} is out of range (1-{line_count})."
            assert line_from <= line_to, f"Block {k} in file {f}: line_from {line_from} must be less than or equal to line_to {line_to}."
        def _get_code_block_key(line_index):
            for k, v in blocks.items():
                line_from, line_to = v
                if line_from <= line_index <= line_to:
                    return k
            return None

        record_map = {k: "" for k in blocks.keys()}
        for index, line in enumerate(lines, start=1):
            block_key = _get_code_block_key(index)
            if block_key is None:
                continue
            # Remove comments and check if line is empty
            line = line.split("#", 1)[0]
            if not line.strip():
                continue
            if not line.endswith("\n"):
                line += "\n"
            record_map[block_key] += line
        if callable(checker):
            for k in blocks.keys():
                record_map[k] = checker(record_map[k])
        ret_map[f] = record_map
    return ret_map


def tc_list_as_loc_blocks(func_list, target_tc_prefix="", ignore_tc_prefix=""):
    func_file_blocks = {}
    for tc in func_list:
        tc_fname, (file_path, line_from, line_to) = parse_test_case_name(tc)
        tc_name = tc_fname.split("::", 1)[-1]
        if target_tc_prefix and not tc_name.startswith(target_tc_prefix):
            continue
        if ignore_tc_prefix and tc_name.startswith(ignore_tc_prefix):
            continue
        if file_path not in func_file_blocks:
            func_file_blocks[file_path] = {}
        func_file_blocks[file_path][tc] = [line_from, line_to]
    return func_file_blocks


def parse_test_case_name(tc):
    # file.py:xx-yy::[ClassName::]test_func
    tc_file, tc_name = tc.split("::", 1)
    tc_rfile, line_range = tc_file.split(":")
    tc_file = tc_rfile.split("/tests/", 1)[-1]
    a, b = line_range.split("-", 1)
    assert a.isdigit() and b.isdigit(), f"Invalid line range in test case '{tc}'."
    return f"{tc_file}::{tc_name}", (tc_rfile, int(a), int(b))


def description_mark_function_doc(
    func_list=[], workspace=None, func_RunTestCases=None, timeout_RunTestCases=0
):
    """
    Description for marking functions in test cases.
    """
    simple_msg = (
        "You need to use `mark_function` at the beginning of test functions to associate them with checkpoints. "
        "Example: env.dut.fc_cover['FG-GROUP'].mark_function('FC-FUNCTION', "
        "test_function_name, ['CK-CHECK1', 'CK-CHECK2']). "
        "If a test case covers checkpoints of multiple functions, call mark_function multiple times. "
        "If the test case is redundant, delete it. (See Guide_Doc/dut_test_case.md)"
    )

    if len(func_list) > 0:
        assert workspace is not None, "workspace must be provided if func_list is empty."
        func_file_blocks = {}
        func_test_cases = {}
        for tc in func_list:
            tc_name, (file_path, line_from, line_to) = parse_test_case_name(tc)
            func_test_cases[tc] = tc_name
            if file_path not in func_file_blocks:
                func_file_blocks[file_path] = {}
            func_file_blocks[file_path][tc] = [line_from, line_to]
        blocks = {}
        for _, v in check_file_block(func_file_blocks, workspace, lambda x: ".mark_function" in x).items():
            blocks.update(v)
        no_mark_tc_list = []
        er_mark_tc_list = []
        nf_mark_tc_list = []
        for tc in func_list:
            if tc not in blocks:
                nf_mark_tc_list.append(tc)
            elif blocks[tc] == True:
                er_mark_tc_list.append(tc)
            else:
                no_mark_tc_list.append(tc)
        if len(nf_mark_tc_list) > 0:
            warning(f"Test cases not found in workspace {workspace}: {nf_mark_tc_list}")
        emsg = ""
        if len(er_mark_tc_list) > 0:
            tc_to_run = " ".join([func_test_cases[tc] for tc in er_mark_tc_list])
            tc_msg = f"Test cases ({', '.join(er_mark_tc_list)}) already called 'mark_function' but encountered errors. " + \
                    f"Please call RunTestCases('{tc_to_run}') to see detailed errors and verify that function point, test case, and checkpoint names match the documentation."
            if func_RunTestCases is not None:
                warning(f"Running test RunTestCases('{tc_to_run}') to get detailed error messages...")
                _, run_msg = func_RunTestCases(pytest_args=tc_to_run, timeout=timeout_RunTestCases, return_line_coverage=False, raw_return=True, detail=True)
                tc_msg = f"Test cases ({', '.join(er_mark_tc_list)}) already called 'mark_function' but encountered errors:\n STDOUT:\n{run_msg['STDOUT']}\nSTDERR:\n{run_msg['STDERR']}\n" + \
                         f"Note: If you cannot find the root cause, call RunTestCases('{tc_to_run}') to get more detailed information."
            emsg += tc_msg
        if len(no_mark_tc_list) > 0:
            emsg += f"Test cases not marked with 'mark_function': {', '.join(no_mark_tc_list)}. {simple_msg}"
        return emsg
    return simple_msg


def check_source_code_in_tc(
    workspace, report, checker, target_tc_prefix="", ignore_tc_preifx=""
):
    """Check source code in test cases"""
    test_cases = report.get("tests", {}).get("test_cases", {})
    if target_tc_prefix:
        test_cases = {k : v for k, v in test_cases.items() if str(k.split("::")[-1]).startswith(target_tc_prefix)}
    if ignore_tc_preifx:
        test_cases = {k : v for k, v in test_cases.items() if not str(k.split("::")[-1]).startswith(ignore_tc_preifx)}
    if not test_cases:
        warning(f"no test cases find in test report")
        warning("target_tc_prefix: " + target_tc_prefix)
        warning("ignore_tc_preifx: " + ignore_tc_preifx)
        warning("raw test cases: " + ", ".join(report.get("tests", {}).get("test_cases", {}).keys()))
        return False, {"error": "[No Test Cases] No matching test cases found in the test report. Please check that test files exist and are named with 'test_' prefix, and that the filter prefix settings are correct."}
    # file.py:line1-line2::[class::]test_case_name
    # block fmt: {'file1.py': {"k1": [line_from, line_to], 'k2': [line_from, line_to]}, ...}
    file_blocks = {}
    for k in test_cases.keys():
        p = k.split("::")[0]
        try:
            path, lins = p.split(":")
            if path not in file_blocks:
                file_blocks[path] = {}
            line_s, line_t = lins.split("-")
            file_blocks[path][k] = [int(line_s), int(line_t)]
        except Exception as e:
            raise ValueError(f"Invalid test case format '{k}'. Expected format: 'file.py:line1-line2::[class::]test_case_name'. Error: {e}")
    ret = {}
    for _, b in check_file_block(file_blocks, workspace, checker).items():
        ret.update(b)
    return True, ret


def check_has_assert_in_tc(workspace, report, target_tc_prefix="", ignore_tc_preifx=""):
    """Check tc has assert or not"""

    def has_assert(text_str):
        for key in ["assert", "pytest.raises"]:
            if len([l for l in text_str.splitlines() if key in l.strip()]) > 0:
                return True
        return False

    try:
        failed_tc = []
        ret, msg = check_source_code_in_tc(
            workspace, report, has_assert, target_tc_prefix, ignore_tc_preifx
        )
        if not ret:
            return ret, msg
        for k, v in msg.items():
            if not v:
                failed_tc.append(k)
        if not failed_tc:
            return True, "All test cases have assert statements."
        failed_str = list_str_abbr(failed_tc)
        return False, {
            "error": f"[Missing Assertions] The following {len(failed_tc)} test cases do not contain assert statements: {failed_str}. "
            + "[Problem] Every test case MUST contain at least one assert statement to verify DUT behavior, otherwise it cannot be determined whether the test truly passes. "
            + "[Solution] Add assertions in test functions, format: assert output == expected_output, 'error description'. "
            + "You can also use 'with pytest.raises(ExpectedException): ...' to verify exceptions. "
            + "Note: Do not use 'self.assertEqual' or other unittest methods; this framework only supports assert and pytest.raises.",
        }
    except Exception as e:
        warning(f"check_has_assert_in_tc error: {e}")
        warning(traceback.format_exc())
        return False, {"error": str(e)}


def replace_bash_var(in_str, data: dict):
    """
    Replace bash-like variables in the input string with values from the data dictionary.

    Args:
        in_str (str): template str, eg: "Hello, $(name: Bob)!"
        data (dict): data eg: {'name': 'Alice'}

    Returns:
        str: replaced str eg: "Hello, Alice!"
    """
    pattern = r"\$\(\s*(?P<key>\w+)\s*:\s*(?P<default>.*?)\s*\)"

    def replace_match(match):
        key = match.group("key").strip()
        default = match.group("default").strip()
        return str(data.get(key, default)) if default else str(data.get(key))

    return re.sub(pattern, replace_match, in_str)


def tips_of_get_coverage_data_path(dut_name: str):
    return f"""
If 'get_coverage_data_path' not find in the template, you should define it like this:
def get_coverage_data_path(request, new_path:bool):
    return get_file_in_tmp_dir(request, current_path_file("data/"), "{dut_name}.dat",  new_path=new_path)
"""


def make_llm_tool_ret(ret, check_pass=True):
    """Convert the return value to a LLM tool return format."""
    if isinstance(ret, str):
        return ret
    ret_str =  yam_str(ret)
    if isinstance(ret, dict) and check_pass:
        for k in ["check_pass", "complete"]:
            if not ret.get(k, True):
                ret_str += "\n[Important]\nYou MUST check the error messages (or suggestions) above, and fix it."
                break
    return ret_str


def list_str_abbr(data: list, max_items=50):
    """Convert a list to a string representation with a maximum number of items."""
    if not isinstance(data, list):
        return str(data)
    subfix = ", ..."
    if len(data) <= max_items:
        subfix = ""
    return ", ".join([str(d) for d in data[:max_items]]) + subfix


def get_fixture_scope(dut_func_or_dut_code):
    """Get the scope of a pytest fixture function.
    Args:
        dut_func_or_dut_code: The fixture function or its source code as a string.
    Returns:
        The scope of the fixture ('function', 'class', 'module', 'session') or None if not found.
    """
    if isinstance(dut_func_or_dut_code, str):
        source_code = dut_func_or_dut_code
        dut_func = None
    else:
        dut_func = dut_func_or_dut_code
        source_code = inspect.getsource(dut_func)
    if hasattr(dut_func, "_pytestfixturefunction"):
        fixture_def = dut_func._pytestfixturefunction
        scope = getattr(fixture_def, "scope", None)
        if scope is None:
            # Try to get scope from the fixture definition
            if hasattr(fixture_def, "_scope"):
                scope = fixture_def._scope
        return scope
    # check fixture scope in source code
    if "@pytest.fixture" in source_code:
        # Extract the fixture decorator line
        fixture_pattern = r"@pytest\.fixture\([^)]*\)"
        matches = re.findall(fixture_pattern, source_code)
        if matches:
            for match in matches:
                # Check if scope is specified
                if "scope" in match:
                    # Extract scope value
                    scope_pattern = r'scope\s*=\s*["\'](\w+)["\']'
                    scope_match = re.search(scope_pattern, match)
                    if scope_match:
                        return scope_match.group(1)
    return None


def markdown_headers(workspace, markdown_file, levels=(1, 2, 3, 4, 5, 6)):
    """Extract headers from a markdown file.
    Args:
        markdown_file: The path to the markdown file.
    Returns:
        A list of headers found in the markdown file.
    """
    if isinstance(levels, int):
        levels = (levels,)
    file_path = os.path.abspath(workspace + os.sep + markdown_file)
    if not os.path.isfile(file_path):
        raise Exception(f"File not found: {file_path}")
    pattern = re.compile(r"^(#{1,6})\s+(.*)", re.MULTILINE)
    headers = []
    with open(file_path, "r", encoding="utf-8") as f:
        content = f.read()
        matches = pattern.findall(content)
        for match in matches:
            level = len(match[0])
            if level in levels:
                title = match[1].strip()
                headers.append((level, title))
    return headers


def markdown_get_miss_headers(workspace, markdown_file, ref_markdown_file, levels=2):
    """Get missing headers from a markdown file.
    Args:
        markdown_file: The path to the markdown file.
        ref_markdown_file: The path to the reference markdown file.
        levels: The header levels to check (default is 2).
    Returns:
        A list of missing headers and diff messages.
    """

    def has_head(s_list, lev, t_head):
        for k, v in s_list:
            if k == lev and t_head in v:
                return True
        return False

    missed_msg = "Target headers:\n"
    missed_headers = []
    source_headers = markdown_headers(workspace, markdown_file, levels)
    for lev, head in markdown_headers(workspace, ref_markdown_file, levels):
        if not has_head(source_headers, lev, head):
            missed_headers.append((lev, head))
            missed_msg += f"Level {lev}, {head}: Missed\n"
        else:
            missed_msg += f"Level {lev}, {head}: Present\n"
    if missed_headers:
        missed_msg += "Source headers:\n"
        for lev, head in source_headers:
            missed_msg += f"Level {lev}, {head}\n"
    return missed_headers, missed_msg


def range_list_merge(range1: list, range2: list) -> list:
    """Merge two lists of ranges.

    Args:
        range1 (list): The first list of ranges.
        range2 (list): The second list of ranges.

    Returns:
        list: The merged list of ranges.
    """
    all_ranges = range1 + range2
    if not all_ranges:
        return []
    # Sort ranges by start line
    all_ranges.sort(key=lambda x: x[0])
    merged_ranges = []
    current_start, current_end = all_ranges[0]
    for start, end in all_ranges[1:]:
        if start <= current_end + 1:
            current_end = max(current_end, end)
        else:
            merged_ranges.append((current_start, current_end))
            current_start, current_end = start, end
    merged_ranges.append((current_start, current_end))
    return merged_ranges


def parse_line_CK_map_file(workspace, file_path: str) -> dict:
    """Parse mapped lines of CK from a file.

    Args:
        file_path (str): The path to the file to parse.
    Returns:
        dict: A dictionary with the CK and its mapped lines.
    """
    # mapped lines format:
    # FGROUP/FC-FUNCTION/CK-CHECK: line_start1-line_end1,line_start2-line_end2,...
    # eg:
    #  FGROUP1/FC-FUNCTION1/CK-CHECK1: 10-20,30-40,45-45
    ret = {}
    real_file_path = os.path.abspath(workspace + os.sep + file_path)
    assert os.path.exists(real_file_path), f"File {real_file_path} does not exist."
    with open(real_file_path, "r", encoding="utf-8") as f:
        lines = f.readlines()
        for i, line in enumerate(lines):
            line = rm_blank_in_str(line).strip()
            if not line or line.startswith("#"):
                continue
            value = line.split("#", 1)[0].strip()
            assert ":" in value, f"{file_path} at line {i+1}: Missing ':' separator. Expected format 'FG-GROUP/FC-FUNCTION/CK-CHECK: line_start1-line_end1,...'"
            key, line_ranges_str = value.split(":", 1)
            key = key.strip()
            assert "/" in key, f"{file_path} at line {i+1}: Invalid key format. Expected 'FG-GROUP/FC-FUNCTION/CK-CHECK'. key: {key}"
            line_ranges = line_ranges_str.split(",")
            line_list = []
            for lr in line_ranges:
                lr = lr.strip()
                assert "-" in lr, f"{file_path} at line {i+1}: Invalid line range format '{lr}'. Expected 'line_start-line_end', eg: 10-20, 14-14"
                start_str, end_str = lr.split("-", 1)
                assert start_str.isdigit() and end_str.isdigit(), f"{file_path} at line {i+1}: Line range '{lr}' must contain integers"
                start_line = int(start_str)
                end_line = int(end_str)
                assert start_line <= end_line, f"{file_path} at line {i+1}: Line range '{lr}' start line must be less than or equal to end line"
                line_list.append((start_line, end_line))
            # Merge line ranges
            pre_list = []
            if key in ret:
                pre_list = ret[key]
            ret[key] = range_list_merge(pre_list, line_list)
    return ret


def get_un_mapped_lines(
    workspace, source_file: str, ck_line_map: dict, max_example_lines: int = 20
) -> list:
    """Get unmapped lines from a source file based on CK line mapping.

    Args:
        source_file (str): The path to the source file.
        ck_line_map (dict): The CK line mapping.

    Returns:
        list: A list of unmapped line numbers.
        example_str: eg: "1000: line_content\n1005: line_content\n..."
    """
    real_file_path = os.path.abspath(workspace + os.sep + source_file)
    assert os.path.exists(real_file_path), f"File {real_file_path} does not exist."
    with open(real_file_path, "r", encoding="utf-8") as f:
        lines = f.readlines()
        total_lines = len(lines)
    range_list = []
    for _, v in ck_line_map.items():
        range_list.extend(v)
    range_list = range_list_merge([], range_list)
    mapped_lines = set()
    for start_line, end_line in range_list:
        for line_num in range(start_line, end_line + 1):
            mapped_lines.add(line_num)
    unmapped_lines = [line_num for line_num in range(1, total_lines + 1) if line_num not in mapped_lines]
    unmapped_lines = [line_num for line_num in unmapped_lines if lines[line_num -1].strip()]
    tline = "line"
    line_size = max([len(f"{line_num}") for line_num in unmapped_lines[:max_example_lines]] + [len(tline)])
    if len(unmapped_lines) > 0:
        example_str = f"{tline}: line_content\n"
        example_str += "\n".join(f"{line_num:>{line_size}}: {lines[line_num-1].rstrip()}" for line_num in unmapped_lines[:max_example_lines])
        if len(unmapped_lines) > max_example_lines:
            example_str += f"\n... (and {len(unmapped_lines) - max_example_lines} more lines)"
    else:
        example_str = "All lines are mapped."
    return unmapped_lines, example_str


def is_ucagent_complete(workspace=".", need_agent_exit=False):
    """Check UCAgent is complete from file"""
    status_data = load_ucagent_info(workspace)
    if not status_data.get("all_completed", False):
        return False
    if need_agent_exit:
        return status_data.get("is_agent_exit", False)
    return True


def get_ucagent_hook_msg(
    msg_continue,
    msg_cmp,
    msg_exit,
    msg_init,
    msg_wait_hm="",
    workspace=".",
    need_agent_exit=False,
):
    """Get UCAgent hook message from file"""
    status_data = load_ucagent_info(workspace)
    if not status_data:
        return msg_init
    if status_data.get("is_agent_exit", False):
        return msg_exit
    if not need_agent_exit:
        if status_data.get("all_completed", False):
            return msg_cmp
    if status_data.get("is_wait_human_check", False):
        return msg_wait_hm
    return msg_continue


def get_interaction_messages(key, config_file=None):
    """Get interaction prompts from default cfg"""
    # [config_file.yaml::]continue_prompt_keys[|stop_prompt_keys]
    from ucagent.util.config import get_config
    import os

    if "::" in key:
        config_file, key = key.split("::", 1)
    if config_file:
        if not os.path.isfile(config_file):
            print(f"Config file '{config_file}' not found.")
            return False, None, None
    continue_key = key
    if "|" in key:
        continue_key, stop_key = key.split("|", 1)
    else:
        stop_key = None
    cfg = get_config(config_file)
    continue_value = os.environ.get(continue_key, None)
    if continue_value is None:
        continue_value = cfg.get_value("hooks." + continue_key, None)
    stop_value = os.environ.get(stop_key, None) if stop_key else None
    if stop_value is None and stop_key:
        stop_value = cfg.get_value("hooks." + stop_key, None)
    return True, continue_value, stop_value


def is_run_report_pass(report, stdout, stderr):
    run_pass = report.get("run_test_success", False)
    if run_pass:
        return True, ""
    return False, {
        "error": "[Run Failed] Running test cases / generating report failed! Check STDOUT and STDERR output to identify the cause (common issues: import errors, syntax errors, undefined fixtures, DUT compilation failures, etc.).",
        "STDOUT": stdout,
        "STDERR": stderr,
    }


def get_tools_from_cfg(tool_list, cfg: dict):
    """Get tools from configuration"""
    ignore_tools = cfg.get("ignore_tools", [])
    selected_tools = cfg.get("selected_tools", [])
    tools = []
    for t in tool_list:
        ignored = False
        for ig_t in ignore_tools:
            if "*" in ig_t:
                if fnmatch.fnmatch(t.name, ig_t):
                    warning(f"Tool {t.name} is ignored by configuration.")
                    ignored = True
                    break
            else:
                if t.name == ig_t:
                    warning(f"Tool {t.name} is ignored by configuration.")
                    ignored = True
                    break
        if ignored:
            continue
        if selected_tools:
            selected = False
            for sg_t in selected_tools:
                if "*" in sg_t:
                    if fnmatch.fnmatch(t.name, sg_t):
                        tools.append(t)
                        selected = True
                        break
                else:
                    if t.name == sg_t:
                        tools.append(t)
                        selected = True
                        break
            if not selected:
                warning(f"Tool {t.name} is not selected by configuration.")
        else:
            tools.append(t)
    return tools


def find_available_port(start_port=5000, end_port=65000):
    """Find an available port in the given range."""
    for port in range(start_port, end_port + 1):
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            try:
                s.bind(("", port))
                return port
            except OSError:
                continue
    raise RuntimeError(f"No available port found in range {start_port}-{end_port}.")


def is_port_free(host: str, port: int) -> bool:
    """Return True if *host:port* is available to bind."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        try:
            s.bind((host, port))
            return True
        except OSError:
            return False


def chmode_ro_by_pattern(
    workspace, pattern_list: str, ignore_list: list = ["__pycache__"]
) -> list:
    """Change file mode to read-only."""
    file_list = []
    path_list = []
    for file_path_pattern in pattern_list:
        for p in find_files_by_pattern(workspace, file_path_pattern):
            file_path = os.path.abspath(os.path.join(workspace, p))
            if os.path.exists(file_path):
                if os.path.isdir(file_path):
                    for dirpath, dirnames, filenames in os.walk(file_path):
                        for filename in filenames:
                            file_full_path = os.path.join(dirpath, filename)
                            file_list.append(file_full_path)
                        for dirname in dirnames:
                            dir_full_path = os.path.join(dirpath, dirname)
                            path_list.append(dir_full_path)
                    path_list.append(file_path)
                else:
                    file_list.append(file_path)
            else:
                warning(f"File not found for: {file_path}")
    all_list = file_list + path_list
    return chmode_ro(all_list, ignore_list)


def chmode_ro(path_list: list, ignore_list: list = ["__pycache__"]):
    mfile = stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH
    mpath = mfile | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
    for file_path in path_list:
        ignored = False
        for ig in ignore_list:
            if ig in file_path:
                ignored = True
                break
            if "*" in ig:
                if fnmatch.fnmatch(os.path.basename(file_path), ig):
                    ignored = True
                    break
        if ignored:
            info(f"File ignored for setting mode to read-only: {file_path}")
            continue
        if os.path.isdir(file_path):
            os.chmod(file_path, mpath)
        else:
            os.chmod(file_path, mfile)
    info(f"Set file mode to read-only completed ({len(path_list)} files).")
    return path_list


def chmode_rw(path_list: list, ignore_list: list = ["__pycache__"]):
    """Set file mode to read-write."""
    mfile = (
        stat.S_IREAD
        | stat.S_IWRITE
        | stat.S_IRGRP
        | stat.S_IWGRP
        | stat.S_IROTH
        | stat.S_IWOTH
    )
    mpath = mfile | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
    for file_path in path_list:
        ignored = False
        for ig in ignore_list:
            if ig in file_path:
                ignored = True
                break
            if "*" in ig:
                if fnmatch.fnmatch(os.path.basename(file_path), ig):
                    ignored = True
                    break
        if ignored:
            info(f"File ignored for setting mode to read-write: {file_path}")
            continue
        if os.path.exists(file_path):
            if os.path.isdir(file_path):
                os.chmod(file_path, mpath)
            else:
                os.chmod(file_path, mfile)
        else:
            warning(f"File not found for setting mode to read-write: {file_path}")
    info(f"Set file mode to read-write completed ({len(path_list)} files).")
    return path_list


def get_xml_tag_list(workspace, xml_file, tag_name: str) -> list:
    """Get list of tag values from an XML file.

    Args:
        xml_file (str): The path to the XML file.
        tag_name (str): The name of the tag to extract.

    Returns:
        list: A list of tag values.
    """
    import xml.etree.ElementTree as ET

    target = os.path.abspath(workspace + os.path.sep + xml_file)
    if os.path.exists(target) == False:
        raise Exception(f"XML file not found in workspace: {xml_file}")
    content = "<root>" + open(target, "r", encoding="utf-8").read() + "</root>"
    ret = []
    tree = ET.ElementTree(ET.fromstring(content))
    root = tree.getroot()
    for elem in root.iter(tag_name):
        ret.append(elem.text.strip())
    return ret


def match_pattern_list(name: str, pattern_list: list) -> bool:
    """Check if the name matches any pattern in the pattern list."""
    for pattern in pattern_list:
        if "*" in pattern:
            if fnmatch.fnmatch(name, pattern):
                return True
        else:
            if pattern in name:
                return True
    return False


def sync_dir_to(source_dir, target_dir, ignore_pattern_list=[]):
    """Sync source directory to target directory with incremental updates and deletion support.

    Args:
        source_dir: Source directory path
        target_dir: Target directory path
        ignore_pattern_list: List of patterns to ignore during sync

    Returns:
        target_dir: The target directory path

    Features:
        - Only copies files if they don't exist in target or have newer modification time
        - Removes files/directories in target that don't exist in source
        - Recursively syncs subdirectories
    """
    if not os.path.exists(source_dir):
        raise Exception(f"Source directory '{source_dir}' does not exist.")
    if not os.path.isdir(source_dir):
        raise Exception(f"Source path '{source_dir}' is not a directory.")
    if not os.path.exists(target_dir):
        os.makedirs(target_dir)
    # Track items in source (excluding ignored ones)
    source_items = set()
    # Sync items from source to target
    for item in os.listdir(source_dir):
        if match_pattern_list(item, ignore_pattern_list):
            continue
        source_items.add(item)
        s = os.path.join(source_dir, item)
        d = os.path.join(target_dir, item)
        if os.path.isdir(s):
            # Recursively sync subdirectories
            sync_dir_to(s, d, ignore_pattern_list)
        else:
            # Check if file needs to be copied
            need_copy = False
            if not os.path.exists(d):
                need_copy = True
                info(f"New file to copy: {item}")
            else:
                # Compare modification times
                source_mtime = os.path.getmtime(s)
                target_mtime = os.path.getmtime(d)
                if source_mtime > target_mtime:
                    need_copy = True
                    info(f"Updated file to copy: {item}")
            if need_copy:
                shutil.copy2(s, d)
    # Remove items in target that don't exist in source
    for item in os.listdir(target_dir):
        if match_pattern_list(item, ignore_pattern_list):
            continue
        if item not in source_items:
            d = os.path.join(target_dir, item)
            if os.path.isdir(d):
                shutil.rmtree(d)
                info(f"Removed directory from target: {item}")
            else:
                os.remove(d)
                info(f"Removed file from target: {item}")
    return target_dir

def copy_skill_files(cfg, workspace, root_dir):
    """Copy skill files to workspace,include default skills and additional skills.
    Args:
        cfg: Configuration object
        workspace: Workspace directory path
        root_dir: Root directory path
    """
    dst_path = get_workspace_skill_root(workspace)
    src_paths=[]
    # default skills path
    default_skill_path = os.path.join(root_dir,"lang",cfg.lang,"skills")
    src_paths.append(default_skill_path)
    # additional skills path
    if cfg.skill.extra_skill_path:
        extra_skills_path = os.path.abspath(cfg.skill.extra_skill_path)
        src_paths.append(extra_skills_path)
    # Copy skills to workspace
    for src_path in src_paths:
        if os.path.exists(src_path):
            try:
                copied_skills = copytree_incremental(src_path, dst_path, skip_existing=True,enable_skill_list=cfg.skill.enable_skill_list,disable_skill_list=cfg.skill.disable_skill_list)
                if copied_skills:
                    info(f"Copy {len(copied_skills)} new skill file(s)")
            except Exception as e:
                warning(f"Failed to copy skills: {e}")
        else:
            info(f"Skills not found at {src_path}, skipping")


def find_most_similar_strings(a: Union[str, List[str]], b: List[str]) -> Union[int, List[Tuple[str, int]]]:
    """
    Find the index of the most similar string in array b.
    If input a is a single string, returns the index directly.
    If input a is a list of strings, returns list of (string, index) tuples.

    Args:
        a: Single string or list of strings to be matched
        b: List of candidate strings to compare against

    Returns:
        The most similar string if a is single string,
        or list of tuples (string_from_a, index_of_most_similar_in_b) if a is list
    """
    import difflib
    # Handle single string input case
    if isinstance(a, str):
        max_similarity = -1
        best_index = 0
        for idx, b_item in enumerate(b):
            similarity = difflib.SequenceMatcher(None, a, b_item).ratio()
            if similarity > max_similarity:
                max_similarity = similarity
                best_index = idx
        return b[best_index]
    # Handle list input case
    result = []
    for a_item in a:
        max_similarity = -1
        best_index = 0
        for idx, b_item in enumerate(b):
            similarity = difflib.SequenceMatcher(None, a_item, b_item).ratio()
            if similarity > max_similarity:
                max_similarity = similarity
                best_index = idx
        result.append((a_item, best_index))
    return result

def get_workspace_skill_root(workspace):
    return get_abs_path_cwd_ucagent(workspace, "skills")


def get_func_params_regex(source_code: str) -> list[str]:
    """
    Extract function parameter names from the source code of a Python function using regular expressions.
    """
    pattern = r'def\s+\w+\s*\(([^)]*)\)'
    match = re.search(pattern, source_code)
    if not match:
        return []
    params_str = match.group(1).strip()
    if not params_str:
        return []
    params = []
    current = ""
    bracket_count = 0
    for char in params_str:
        if char == ',' and bracket_count == 0:
            params.append(current.strip())
            current = ""
        else:
            if char in '([{':
                bracket_count += 1
            elif char in ')]}':
                bracket_count -= 1
            current += char
    if current:
        params.append(current.strip())
    return params