Scrapling/scrapling/engines/toolbelt/custom.py

"""
Functions related to custom types or type checking
"""
import inspect
import logging
from dataclasses import dataclass, field

from scrapling.parser import Adaptor, SQLiteStorageSystem
from scrapling.core._types import Any, List, Type, Union, Optional, Dict


@dataclass(frozen=True)
class Response:
    url: str
    text: str
    content: bytes
    status: int
    reason: str
    encoding: str = 'utf-8'  # default encoding
    cookies: Dict = field(default_factory=dict)
    headers: Dict = field(default_factory=dict)
    request_headers: Dict = field(default_factory=dict)
    adaptor_arguments: Dict = field(default_factory=dict)

    @property
    def adaptor(self):
        if self.text:
            return Adaptor(text=self.text, url=self.url, encoding=self.encoding, **self.adaptor_arguments)
        elif self.content:
            return Adaptor(body=self.content, url=self.url, encoding=self.encoding, **self.adaptor_arguments)
        return None

    def __repr__(self):
        return f'<{self.__class__.__name__} [{self.status} {self.reason}]>'


class BaseFetcher:
    def __init__(
            self,
            # Adaptor class parameters
            huge_tree: bool = True,
            keep_comments: Optional[bool] = False,
            auto_match: Optional[bool] = False,
            storage: Any = SQLiteStorageSystem,
            storage_args: Optional[Dict] = None,
            debug: Optional[bool] = True,
    ):
        # I won't validate Adaptor's class parameters here again, I will leave it to be validated later
        self.adaptor_arguments = dict(
            huge_tree=huge_tree,
            keep_comments=keep_comments,
            auto_match=auto_match,
            storage=storage,
            storage_args=storage_args,
            debug=debug,
        )


def check_if_engine_usable(engine):
    # if isinstance(engine, type):
    #     raise TypeError("Expected an engine instance, not a class definition of the engine")

    if hasattr(engine, 'fetch'):
        fetch_function = getattr(engine, "fetch")
        if callable(fetch_function):
            if len(inspect.signature(fetch_function).parameters) > 0:
                return engine
            else:
                # raise TypeError("Engine class instance must have a callable method 'fetch' with the first argument used for the url.")
                raise TypeError("Engine class must have a callable method 'fetch' with the first argument used for the url.")
        else:
            # raise TypeError("Invalid engine instance! Engine class must have a callable method 'fetch'")
            raise TypeError("Invalid engine class! Engine class must have a callable method 'fetch'")
    else:
        # raise TypeError("Invalid engine instance! Engine class must have the method 'fetch'")
        raise TypeError("Invalid engine class! Engine class must have the method 'fetch'")


def get_variable_name(var: Any) -> Optional[str]:
    """Get the name of a variable using global and local scopes.
    :param var: The variable to find the name for
    :return: The name of the variable if found, None otherwise
    """
    for scope in [globals(), locals()]:
        for name, value in scope.items():
            if value is var:
                return name
    return None


def check_type_validity(variable: Any, valid_types: Union[List[Type], None], default_value: Any = None, critical: bool = False, param_name: Optional[str] = None) -> Any:
    """Check if a variable matches the specified type constraints.
    :param variable: The variable to check
    :param valid_types: List of valid types for the variable
    :param default_value: Value to return if type check fails
    :param critical: If True, raises TypeError instead of logging error
    :param param_name: Optional parameter name for error messages
    :return: The original variable if valid, default_value if invalid
    :raise TypeError: If critical=True and type check fails
    """
    # Use provided param_name or try to get it automatically
    var_name = param_name or get_variable_name(variable) or "Unknown"

    # Convert valid_types to a list if None
    valid_types = valid_types or []

    # Handle None value
    if variable is None:
        if type(None) in valid_types:
            return variable
        error_msg = f'Argument "{var_name}" cannot be None'
        if critical:
            raise TypeError(error_msg)
        logging.error(f'[Ignored] {error_msg}')
        return default_value

    # If no valid_types specified and variable has a value, return it
    if not valid_types:
        return variable

    # Check if variable type matches any of the valid types
    if not any(isinstance(variable, t) for t in valid_types):
        type_names = [t.__name__ for t in valid_types]
        error_msg = f'Argument "{var_name}" must be of type {" or ".join(type_names)}'
        if critical:
            raise TypeError(error_msg)
        logging.error(f'[Ignored] {error_msg}')
        return default_value

    return variable


# Pew Pew
def do_nothing(page):
    # Just works as a filler for `page_action` argument in browser engines
    return page