"""Utility functions for file handling."""

import gzip
import hashlib
import io
import os
import tempfile
import uuid
from itertools import islice
from pathlib import Path

import pyarrow as pa
import pyarrow.csv as pcsv
import pyarrow.parquet as pq
from abacus_common_logic.utils.formatting import format_bytes
from charset_normalizer import from_bytes

from src.enums import FileType
from src.errors import InvalidFileTypeError, TransientError
from src.infra.log import logger
from src.schemas import FileMetadata


def create_temp_file(
    ref_path: str | Path | None = None, suffix: str | None = None
) -> Path:
    """Create a new secure temporary file with proper permissions.

    Creates a temporary file with restricted permissions (mode 0600) and returns
    its path. The file is created but must be explicitly deleted by the caller
    when no longer needed.

    Args:
        ref_path: Optional reference path to use for directory.
            If provided, the new file will be created in the same directory.
        suffix: An optional file path suffix.

    Returns:
        Path: A new unique file path with secure permissions (mode 0600).
            Caller is responsible for deleting this file after use.

    Raises:
        OSError: If unable to create temporary file or access file system.
    """
    from config import config

    # Create a secure temporary file with mode 0600, using context manager
    # to ensure file descriptor is properly closed.
    dir_path = get_dir_path(ref_path) if ref_path else config.storage.TEMP_DIR
    with tempfile.NamedTemporaryFile(
        dir=dir_path, delete=False, suffix=suffix
    ) as tmp_file:
        return Path(tmp_file.name)


def gen_id() -> str:
    """Generate a random alphanumeric ID."""
    return uuid.uuid4().hex


def get_dir_path(file_path: str | Path) -> str:
    """Get the directory path from a file or directory path.

    If the input is a directory, returns it as-is.
    If the input is a file, returns the directory containing the file.

    Args:
        file_path: Path to a file or directory.

    Returns:
        str: The directory path.

    Raises:
        OSError: If unable to access file system.
    """
    real_path = os.path.realpath(file_path)
    return real_path if os.path.isdir(real_path) else os.path.dirname(real_path)


def get_available_cpus() -> int:
    """Get the physical CPU limit.

    This acts as the base core count for all thread scaling.

    Returns:
        The number of available CPUs.
    """
    from config import config

    try:
        # Get CPUs actually assigned to the process
        cpu_count = len(os.sched_getaffinity(0))  # type: ignore
    except AttributeError:
        # Fallback for local dev
        cpu_count = os.cpu_count() or config.exec.DEFAULT_CPU_COUNT

    # Apply limit
    if config.exec.MAX_CPU_COUNT:
        cpu_count = min(config.exec.MAX_CPU_COUNT, cpu_count)

    return max(1, cpu_count)


def get_num_cpu_workers() -> int:
    """CPU-bound tasks: Use (Avail - 1) to leave room for orchestration.

    Returns:
        The number of CPU workers.
    """
    return max(1, get_available_cpus() - 1)


def get_num_io_workers() -> int:
    """I/O-bound tasks: Based on Avail to maximize network throughput.

    Returns:
        The number of I/O workers.
    """
    from config import config

    return get_available_cpus() * config.exec.IO_THREADS_PER_VCPU


def get_file_type(
    file_path: str | Path, encoding: str, gzipped: bool = False
) -> FileType:
    """Determine file type from content.

    Args:
        file_path: Path to the file.
        encoding: The file's character encoding.
        gzipped: Whether the file is gzip compressed.

    Returns:
        FileType: The detected file type.

    Raises:
        InvalidFileTypeError: If the file type is not supported.
    """
    if is_xlsx_file(file_path, gzipped):
        return FileType.XLSX

    if is_csv_file(file_path, encoding, gzipped):
        return FileType.CSV

    if is_parquet_file(file_path, gzipped):
        return FileType.PQT

    raise InvalidFileTypeError('Unsupported file type.')


def is_csv_file(file_path: str | Path, encoding: str, gzipped: bool = False) -> bool:
    """Check if file is a CSV by attempting to parse it with pyarrow.

    Args:
        file_path: Path to the file to check.
        encoding: The file's encoding.
        gzipped: Whether the file is gzip compressed.

    Returns:
        bool: True if file is a valid CSV, False otherwise.
    """
    from config import config

    try:
        file_bytes = get_first_n_lines(
            file_path, config.encoding.CSV_SAMPLE_ROWS, gzipped
        )
        input_stream = io.BytesIO(file_bytes)
        read_options = pcsv.ReadOptions(encoding=encoding)
        pcsv.read_csv(input_stream, read_options=read_options)
        return True
    except (pa.ArrowInvalid, OSError, UnicodeDecodeError, gzip.BadGzipFile):
        # If pyarrow can't read it or file access fails, it's not a valid CSV
        return False


def is_gzip_file(file_path: str | Path) -> bool:
    """Check if file is gzipped by inspecting magic numbers.

    Args:
        file_path: Path to the file to check.

    Returns:
        bool: True if file has gzip signature, False otherwise.
    """
    with open(file_path, 'rb') as f:
        header = f.read(2)
    # Check for gzip magic numbers (0x1f 0x8b)
    return header == b'\x1f\x8b'


def is_parquet_file(file_path: str | Path, gzipped: bool = False) -> bool:
    """Check if file is a parquet file.

    Args:
        file_path: Path to the file to check.
        gzipped: Whether the file is gzip compressed.

    Returns:
        bool: True if file is a valid parquet file, False otherwise.
    """
    try:
        if gzipped:
            with gzip.open(file_path, 'rb') as gz:
                pq.read_metadata(gz)
        else:
            pq.read_metadata(file_path)
        return True
    except (pa.ArrowInvalid, OSError, gzip.BadGzipFile):
        return False


def is_xlsx_file(file_path: str | Path, gzipped: bool = False) -> bool:
    """Check if file is an XLSX by inspecting magic numbers.

    Args:
        file_path: Path to the file to check.
        gzipped: Whether the file is gzip compressed.

    Returns:
        bool: True if file has XLSX signature, False otherwise.
    """
    try:
        if gzipped:
            with gzip.open(file_path, 'rb') as f:
                header = f.read(4)
        else:
            with open(file_path, 'rb') as f:
                header = f.read(4)
        # Check for ZIP signature (PK\x03\x04) used by .xlsx
        return header == b'\x50\x4b\x03\x04'
    except gzip.BadGzipFile:
        return False


def detect_encoding(file_path: str | Path, gzipped: bool = False) -> str | None:
    """Detect the encoding of a file.

    Args:
        file_path: Path to the file.
        gzipped: Whether the file is gzipped.

    Returns:
        str: Detected encoding (e.g., 'utf-8', 'latin-1'), normalized for DuckDB.
    """
    from config import config

    open_func = gzip.open if gzipped else open
    with open_func(file_path, 'rb') as f:
        data = f.read(config.encoding.SAMPLE_BYTES)
    best_match = from_bytes(data, chunk_size=config.encoding.SAMPLE_CHUNK_BYTES).best()
    return best_match.encoding if best_match else None


def get_file_metadata(file_path: str | Path) -> FileMetadata:
    """Get comprehensive metadata about a file.

    Detects file type, encoding, and compression status.

    Args:
        file_path: Path to the file to analyze.

    Returns:
        FileMetadata: Object containing file path, encoding, type, and gzip status.

    Raises:
        InvalidFileTypeError: If the file type is not supported.
    """
    from config import config

    gzipped = is_gzip_file(file_path)
    logger.info(f'- gzipped={gzipped}')

    encoding = detect_encoding(file_path, gzipped) or config.encoding.DEFAULT
    logger.info(f'- encoding={encoding}')

    file_type = get_file_type(file_path, encoding, gzipped)
    logger.info(f'- file_type={file_type.value}')

    return FileMetadata(
        file_path=Path(file_path),
        encoding=encoding,
        file_type=file_type,
        gzipped=gzipped,
    )


def get_first_n_lines(
    file_path: str | Path, num_lines: int, gzipped: bool = False
) -> bytes:
    """Get the first given lines of a file as bytes.

    Args:
        file_path: The file path.
        num_lines: The number of lines to read.
        gzipped: Whether the file is gzipped.

    Returns:
        bytes: The content of the first n lines.

    Raises:
        ValueError: If num_lines is not positive.
    """
    if num_lines <= 0:
        raise ValueError(f'n must be positive, got {num_lines}')

    open_func = gzip.open if gzipped else open
    with open_func(file_path, 'rb') as f:
        lines = list(islice(f, num_lines))
    return b''.join(lines)


def get_free_disk_space(path: str | Path) -> int:
    """Get free disk space.

    Args:
        path: Path to check for free space.

    Returns:
        int: Free disk space in bytes.
    """
    st = os.statvfs(path)
    # Total disk space: st.f_blocks * st.f_frsize
    # Used disk space: (st.f_blocks - st.f_bfree) * st.f_frsize
    return st.f_bavail * st.f_frsize


def try_disk_space(required_bytes: int, path: str) -> None:
    """Check if sufficient disk space is available at the specified path.

    Validates that the filesystem at the given path has enough free space
    before attempting file operations. Raises TransientError to allow retry
    if space is insufficient.

    Args:
        required_bytes: Minimum bytes required (e.g., 1024 for 1 KB).
        path: Path to check (default: /tmp for Lambda environment).

    Raises:
        TransientError: If insufficient disk space is available or
        if the disk space check fails due to OS errors.
    """
    try:
        free_bytes = get_free_disk_space(path)

        if free_bytes < required_bytes:
            raise TransientError(
                f'Insufficient disk space in {path}. '
                f'{format_bytes(free_bytes)} available, '
                f'{format_bytes(required_bytes)} required'
            )
    except OSError as e:
        logger.error(f'Unable to check disk space for {path}: {e}')
        raise TransientError(
            f'Failed to check disk space availability in {path}: {e}'
        ) from e


def compute_md5(file_path: str | Path) -> str:
    """Compute MD5 checksum of a file.

    Reads file in chunks to handle large files efficiently without
    loading entire file into memory.

    Args:
        file_path: Path to the file.

    Returns:
        Hexadecimal MD5 checksum.

    Raises:
        OSError: If file cannot be read.
    """
    from config import config

    md5_hash = hashlib.md5()
    with open(file_path, 'rb') as f:
        chunk_size = config.encoding.MD5_CHUNK_BYTES
        for chunk in iter(lambda: f.read(chunk_size), b''):
            md5_hash.update(chunk)
    return md5_hash.hexdigest()


def is_simple_etag(etag: str) -> bool:
    """Check if ETag is a simple MD5 hash (not multipart upload).

    S3 multipart uploads have ETags in format: {hash}-{part_count}
    Single-part uploads have simple MD5 hash as ETag.

    Args:
        etag: S3 ETag value (without surrounding quotes).

    Returns:
        bool: True if ETag is a simple MD5, False if multipart.
    """
    return '-' not in etag