aimbat.utils

def delete_sampledata() -> None:
    """Delete sample data."""

    logger.info(f"Deleting sample data in {settings.sampledata_dir}.")

    shutil.rmtree(settings.sampledata_dir)

def download_sampledata(force: bool = False) -> None:
    """Download sample data."""

    logger.info(
        f"Downloading sample data from {settings.sampledata_src} to {settings.sampledata_dir}."
    )

    if (
        settings.sampledata_dir.exists()
        and len(os.listdir(settings.sampledata_dir)) != 0
    ):
        if force is True:
            delete_sampledata()
        else:
            raise FileExistsError(
                f"The directory {settings.sampledata_dir} already exists and is non-empty."
            )

    with urlopen(settings.sampledata_src) as zipresp:
        logger.debug(f"Extracting sample data to {settings.sampledata_dir}.")
        with ZipFile(BytesIO(zipresp.read())) as zfile:
            zfile.extractall(settings.sampledata_dir)

    logger.info("Sample data downloaded and extracted successfully.")

@lru_cache(maxsize=None)
def get_title_map(model_class: type[BaseModel]) -> dict[str, str]:
    """Creates a mapping from field names to their 'title' metadata."""
    mapping: dict[str, str] = {}

    for name, info in model_class.model_fields.items():
        mapping[name] = info.title or name.replace("_", " ")

    computed_fields: dict[str, Any] = getattr(
        model_class, "__pydantic_computed_fields__", {}
    )
    for name, info in computed_fields.items():
        title = getattr(info, "title", None)
        mapping[name] = title or name.replace("_", " ")

    return mapping

def mean_and_sem(
    data: Sequence[float | None],
) -> tuple[float | None, float | None]:
    """Return the mean and standard error of the mean (SEM) for a list of numeric values, ignoring None values.

    Args:
        data: List of numeric values (float or int) or None.

    Returns:
        A tuple containing the mean and SEM of the input data, both as floats or None if not computable.
    """
    return _mean_and_sem_tuple(tuple(data))

def mean_and_sem_timedelta(
    values: Sequence[pd.Timedelta],
) -> tuple[pd.Timedelta | None, pd.Timedelta | None]:
    """Return (mean, sem) for a list of pd.Timedelta values.

    Args:
        values: List of pd.Timedelta values.

    Returns:
        `(None, None)` when empty. SEM is `None` for fewer than two values.
    """
    if not values:
        return None, None
    ns_vals = [float(td.value) for td in values]
    mean_ns, sem_ns = mean_and_sem(ns_vals)
    return (
        pd.Timedelta(int(mean_ns), unit="ns") if mean_ns is not None else None,
        pd.Timedelta(int(sem_ns), unit="ns") if sem_ns is not None else None,
    )

def rel(attr: Any) -> QueryableAttribute[Any]:
    """Cast a SQLModel relationship attribute to `QueryableAttribute` for use with `selectinload`.

    SQLModel types relationship fields as their Python collection type (e.g. `list[Foo]`),
    but SQLAlchemy's `selectinload` expects a `QueryableAttribute`. This helper performs
    the cast to satisfy mypy without requiring per-call `# type: ignore` comments.
    """
    return cast(QueryableAttribute[Any], attr)

def string_to_uuid(
    session: Session,
    id: str,
    aimbat_class: type[AimbatTypes],
    custom_error: str | None = None,
) -> UUID:
    """Determine a UUID from a string containing the first few characters.

    Args:
        session: Database session.
        id: Input string to find UUID for.
        aimbat_class: Aimbat class to use to find UUID.
        custom_error: Overrides the default error message.

    Returns:
        The full UUID.

    Raises:
        ValueError: If the UUID could not be determined.
    """
    statement = select(aimbat_class.id).where(
        func.replace(cast(aimbat_class.id, String), "-", "").like(
            f"{id.replace('-', '')}%"
        )
    )
    uuid_set = set(session.exec(statement).all())
    if len(uuid_set) == 1:
        resolved = uuid_set.pop()
        logger.debug(f"Resolved {id} to UUID: {resolved}")
        return resolved
    if len(uuid_set) == 0:
        raise ValueError(
            custom_error or f"Unable to find {aimbat_class.__name__} using id: {id}."
        )
    raise ValueError(f"Found more than one {aimbat_class.__name__} using id: {id}")

def uuid_shortener[T: AimbatTypes](
    session: Session,
    aimbat_obj: T | type[T],
    min_length: int = 2,
    str_uuid: str | None = None,
) -> str:
    """Calculates the shortest unique prefix for a UUID, returning with dashes.

    Args:
        session: An active SQLModel/SQLAlchemy session.
        aimbat_obj: Either an instance of a SQLModel or the SQLModel class itself.
        min_length: The starting character length for the shortened ID.
        str_uuid: The full UUID string. Required only if `aimbat_obj` is a class.

    Returns:
        str: The shortest unique prefix string, including hyphens where applicable.
    """

    if isinstance(aimbat_obj, type):
        model_class = aimbat_obj
        if str_uuid is None:
            raise ValueError("str_uuid must be provided when aimbat_obj is a class.")
        target_full = str(UUID(str_uuid))
    else:
        model_class = type(aimbat_obj)
        target_full = str(aimbat_obj.id)

    prefix_clean = target_full.replace("-", "")[:min_length]

    # select with a WHERE clause that removes dashes and compares the cleaned prefix
    statement = select(model_class.id).where(
        func.replace(cast(model_class.id, String), "-", "").like(f"{prefix_clean}%")
    )

    # Store results as standard hyphenated strings
    results = session.exec(statement).all()
    relevant_pool = [str(uid) for uid in results]

    if target_full not in relevant_pool:
        raise ValueError(f"ID {target_full} not found in table {model_class.__name__}")

    current_length = min_length
    while current_length < len(target_full):
        candidate = target_full[:current_length]
        if candidate.endswith("-"):
            current_length += 1
            candidate = target_full[:current_length]

        matches = [u for u in relevant_pool if u.startswith(candidate)]
        if len(matches) == 1:
            logger.debug(f"Shortened {target_full} to: {candidate}")
            return candidate
        current_length += 1

    return target_full