import sys
from functools import partial

from remake.errors import SplitRuntimeError
from remake.operators.base import BaseOperator
from remake.utils import get_info
import pandas as pd


class SplitOperator(BaseOperator):

    def __init__(self, name, transform, named=False):
        super().__init__(name, [], priority="LAZY", order=2, type="SEQUENTIAL")
        self.transform = transform
        self.named = named

    def __call__(self, df: pd.DataFrame):
        try:
            result = self.transform(df)
        except:
            exc_type, exc_value, exc_traceback = sys.exc_info()
            msg = f"Failed to execute split '{self.name}'. Reason: {exc_type(exc_value)}"
            raise SplitRuntimeError(msg).with_traceback(exc_traceback)

        base_msg = f"Invalid return-type from split '{self.name}'."

        if not result:
            raise SplitRuntimeError(base_msg + " The return result was empty")

        if not self.named:
            if not (isinstance(result, (list, tuple)) and all([isinstance(el, pd.DataFrame) for el in result])):
                raise SplitRuntimeError(
                    base_msg + f" expected 'List[pd.DataFrame]' was {type(result)}[{type(result[0])}]")
            else:
                return list(result)
        else:
            if not (isinstance(result, dict) and all([isinstance(el, pd.DataFrame) for el in result.values()])):
                raise SplitRuntimeError(
                    base_msg + f" expected 'Dict[Any, pd.DataFrame]' was {type(result)}[Any, {type(result[0])}]")
            else:
                return result


def split(function):
    """
    A split takes a dataframe and converts it into either a list of dataframes.
    """

    from remake.schematic.cache import SchematicCache

    schematic, name, _ = get_info(function)

    SchematicCache.add_split(schematic, SplitOperator(name, function, named=False))


def named_split(function):
    """
    A split takes a dataframe and converts it into a dictionary with dataframes as values.
    """

    from remake.schematic.cache import SchematicCache

    schematic, name, _ = get_info(function)

    SchematicCache.add_split(schematic, SplitOperator(name, function, named=True))
import sys
from functools import partial

from remake.errors import SplitRuntimeError
from remake.operators.base import BaseOperator
from remake.utils import get_info
import pandas as pd


class SplitOperator(BaseOperator):

    def __init__(self, name, transform, named=False):
        super().__init__(name, [], priority="LAZY", order=2, type="SEQUENTIAL")
        self.transform = transform
        self.named = named

    def __call__(self, df: pd.DataFrame):
        try:
            result = self.transform(df)
        except:
            exc_type, exc_value, exc_traceback = sys.exc_info()
            msg = f"Failed to execute split '{self.name}'. Reason: {exc_type(exc_value)}"
            raise SplitRuntimeError(msg).with_traceback(exc_traceback)

        base_msg = f"Invalid return-type from split '{self.name}'."

        if not result:
            raise SplitRuntimeError(base_msg + " The return result was empty")

        if not self.named:
            if not (isinstance(result, (list, tuple)) and all([isinstance(el, pd.DataFrame) for el in result])):
                raise SplitRuntimeError(
                    base_msg + f" expected 'List[pd.DataFrame]' was {type(result)}[{type(result[0])}]")
            else:
                return list(result)
        else:
            if not (isinstance(result, dict) and all([isinstance(el, pd.DataFrame) for el in result.values()])):
                raise SplitRuntimeError(
                    base_msg + f" expected 'Dict[Any, pd.DataFrame]' was {type(result)}[Any, {type(result[0])}]")
            else:
                return result


def split(function):
    """
    A split takes a dataframe and converts it into either a list of dataframes.
    """

    from remake.schematic.cache import SchematicCache

    schematic, name, _ = get_info(function)

    SchematicCache.add_split(schematic, SplitOperator(name, function, named=False))


def named_split(function):
    """
    A split takes a dataframe and converts it into a dictionary with dataframes as values.
    """

    from remake.schematic.cache import SchematicCache

    schematic, name, _ = get_info(function)

    SchematicCache.add_split(schematic, SplitOperator(name, function, named=True))
A split takes a dataframe and converts it into either a list of dataframes. A split takes a dataframe and converts it into a dictionary with dataframes as values.