"""Utility functions for tkfilter.""" from __future__ import annotations import inspect import os import re from collections.abc import Sequence from types import ModuleType from typing import Any, Callable def split_file_path(path: str) -> tuple[str, str, str, str]: """Return dirname, basename, filebody, ext. This local helper avoids depending on tklib.tkutils from the tkfilter core. """ dirname = os.path.dirname(path) basename = os.path.basename(path) filebody, ext = os.path.splitext(basename) return dirname, basename, filebody, ext def module_name_from_path(path: str) -> str: """Return importable module name from a Python file path.""" _, _, filebody, _ = split_file_path(path) return filebody def is_error_result(value: Any) -> bool: """Return True when a filter result represents an error. Older filter code often returns a string containing "Error" instead of raising an exception. Dict/list/None values are treated as non-error here; None simply means "not matched" in check_file_type(). """ if value is None: return False if isinstance(value, str): return "error" in value.lower() return False def is_sequence_like(value: Any) -> bool: """Return True for list/tuple/ndarray-like sequences, excluding strings.""" if isinstance(value, (str, bytes)): return False return isinstance(value, Sequence) or hasattr(value, "__len__") def call_compatible(func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: """Call func while dropping unsupported keyword arguments. Existing filters are not completely uniform: some functions accept app and cparams, some do not, some accept is_print, etc. This helper keeps the caller simple while preserving the old filter API. """ try: sig = inspect.signature(func) except (TypeError, ValueError): return func(*args, **kwargs) params = sig.parameters accepts_var_kw = any(p.kind == p.VAR_KEYWORD for p in params.values()) if accepts_var_kw: return func(*args, **kwargs) allowed = { name for name, p in params.items() if p.kind in (p.POSITIONAL_OR_KEYWORD, p.KEYWORD_ONLY) } filtered_kwargs = {k: v for k, v in kwargs.items() if k in allowed} return func(*args, **filtered_kwargs) def get_module_attr(module: ModuleType, name: str, default: Any = None) -> Any: """Safe getattr wrapper for plugin modules.""" return getattr(module, name, default) def should_skip_data_file(path: str, skip_temporary: bool = True, skip_output: bool = True) -> bool: """Return True for common temporary/output files that should be skipped.""" _, basename, filebody, _ = split_file_path(path) if skip_temporary and (not filebody or filebody.startswith("~")): return True if skip_output and re.search(r"-out\.", basename, flags=re.IGNORECASE): return True return False