Importing

class CSVImporter(Importer["CsvImportSession"]):
    @property
    def name(self) -> str:
        return "CSV"

    def suggest(self, input_path: str) -> bool:
        return input_path.endswith(".csv") or input_path == "text/csv"

    def _detect_skip_rows_and_dialect(
        self, input_path: str | BytesIO
    ) -> tuple[int, csv.Dialect]:
        """Detect the number of rows to skip before CSV data begins and the CSV dialect."""
        skip_rows = 0

        try:
            with open(input_path, "r", encoding="utf8") as file:
                MAX_LINES = 50  # check the first 50 lines only
                lines = [line.strip() for i, line in enumerate(file) if i <= MAX_LINES]
                total_lines = len(lines)

                # Only analyze if we have enough lines
                if len(lines) >= 2:
                    # Parse each line and analyze content, keeping track of original line numbers
                    parsed_rows = []
                    line_numbers = []
                    for line_idx, line in enumerate(lines):
                        if not line:  # Skip empty lines
                            continue
                        try:
                            reader = csv.reader([line])
                            row = next(reader)
                            parsed_rows.append(row)
                            line_numbers.append(line_idx)
                        except Exception:
                            parsed_rows.append([line])  # Fallback for problematic lines
                            line_numbers.append(line_idx)

                    if len(parsed_rows) >= 2:
                        # Look for the actual CSV header (column names)
                        for i, row in enumerate(parsed_rows):
                            if self._looks_like_csv_header(row):
                                skip_rows = line_numbers[i]
                                break
                        else:
                            # Fallback: use field count analysis
                            field_counts = [len(row) for row in parsed_rows]
                            from collections import Counter

                            count_frequency = Counter(field_counts)
                            most_common_count = count_frequency.most_common(1)[0][0]

                            # Find first row that matches the most common field count
                            for i, count in enumerate(field_counts):
                                if count == most_common_count:
                                    skip_rows = line_numbers[i]
                                    break

                # Validate skip_rows doesn't exceed available lines
                if skip_rows >= total_lines:
                    skip_rows = 0  # Reset to safe default

                # Now detect dialect from the CSV content (after skip_rows)
                file.seek(0)
                for _ in range(skip_rows):
                    file.readline()

                sample = file.read(65536)
                dialect = Sniffer().sniff(sample)

        except Exception:
            # If anything fails, use defaults and try basic dialect detection
            skip_rows = 0
            try:
                with open(input_path, "r", encoding="utf8") as file:
                    sample = file.read(65536)
                    dialect = Sniffer().sniff(sample)
            except Exception:
                # Create a default dialect if everything fails
                class DefaultDialect:
                    delimiter = ","
                    quotechar = '"'

                dialect = DefaultDialect()

        return skip_rows, dialect

    def _looks_like_csv_header(self, row: list[str]) -> bool:
        """Check if a row looks like a CSV header with column names."""
        if not row or len(row) < 2:
            return False

        # Skip rows where most fields are empty (likely CSV notes with trailing commas)
        non_empty_fields = [field.strip() for field in row if field.strip()]
        if len(non_empty_fields) < len(row) // 2:
            return False

        # Look for typical CSV header characteristics
        header_indicators = 0

        for field in non_empty_fields:
            field = field.lower().strip()

            # Common column name patterns
            if any(
                word in field
                for word in [
                    "id",
                    "name",
                    "date",
                    "time",
                    "user",
                    "tweet",
                    "text",
                    "count",
                    "number",
                    "sent",
                    "screen",
                    "retweeted",
                    "favorited",
                ]
            ):
                header_indicators += 1

            # Short descriptive column names (not long sentences like CSV notes)
            if 3 <= len(field) <= 30 and not field.startswith(
                ("http", "www", "from ", "if you")
            ):
                header_indicators += 1

        # Consider it a CSV header if at least 50% of non-empty fields look like column names
        return header_indicators >= len(non_empty_fields) * 0.5

    def init_session(self, input_path: str | BytesIO):
        skip_rows, dialect = self._detect_skip_rows_and_dialect(input_path)

        return CsvImportSession(
            input_file=input_path,
            separator=dialect.delimiter,
            quote_char=dialect.quotechar,
            has_header=True,
            skip_rows=skip_rows,
        )

class ExcelImporter(Importer["ExcelImportSession"]):
    @property
    def name(self) -> str:
        return "Excel"

    def suggest(self, input_path: str) -> bool:
        return (
            input_path.endswith(".xlsx")
            or input_path
            == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
        )

    def init_session(self, input_path: str | BytesIO):
        """
        Initialize Excel import session.

        For single-sheet files, automatically selects the sheet.
        For multi-sheet files, returns None (caller must provide sheet selection).
        """
        reader = read_excel(
            cast(str, input_path)
            if type(input_path) is not BytesIO
            else input_path.getvalue()
        )
        sheet_names = reader.sheet_names

        if not sheet_names or (len(sheet_names) == 0 or len(sheet_names) > 1):
            return None

        return ExcelImportSession(
            input_file=input_path,
            selected_sheet=sheet_names[0],
            sheet_names=sheet_names,
        )

class Importer[SessionType](ABC):
    @property
    @abstractmethod
    def name(self) -> str:
        """
        The name of the importer. It will be quoted in the UI in texts such as
        "Imported as `name`, so keep it to a format name."
        """
        pass

    @abstractmethod
    def suggest(self, input_path: str) -> bool:
        """
        Check if the importer can handle the given file. This should be fairly
        restrictive based on reasonable assumptions, as it is only used for the
        initial importer suggestion. The user can always override the suggestion.
        """
        pass

    @abstractmethod
    def init_session(self, input_path: str | BytesIO) -> SessionType | None:
        """
        Produces an initial import session object that contains all the configuration
        needed for the import. The user can either accept this configuration or
        customize it.

        Return None here if the importer cannot figure out how to configure the
        import parameters. This doesn't necessarily mean that the file cannot be
        loaded; the UI will force the user to customize the import session if the
        user wants to proceed with this importer.
        """
        pass

class ImporterSession(ABC):
    """
    The ImporterSession interface handles the ongoing configuration of an import.
    It keeps the configuration state and can load a preview of the data from the input file.
    """

    @abstractmethod
    def load_preview(self, n_records: int) -> pl.DataFrame | None:
        """
        Attempt to load a preview of the data from the input file.

        Return None here if it is sure that the file cannot be loaded with the current
        configuration. Only throw an execption in the case of unexpected errors.
        """
        pass

    @abstractmethod
    def import_as_parquet(self, output_path: str) -> None:
        """
        Import the data from the input file to the output file in the Parquet format.
        """
        pass