aimbat.utils

def check_field_name_flags(
    by_alias: bool, by_title: bool, from_read_model: bool | None = None
) -> None:
    """Check the field-naming flags the `dump_*_table` functions take.

    Args:
        by_alias: Whether serialisation aliases are to be used for the field names.
        by_title: Whether field title metadata is to be used for the field names.
        from_read_model: Whether the caller is dumping from its read model, for
            the functions that offer the choice. `None` for the ones that don't.

    Raises:
        ValueError: If both `by_alias` and `by_title` are True.
        ValueError: If `by_title` is True but `from_read_model` is False.
    """
    if by_alias and by_title:
        raise ValueError("Arguments 'by_alias' and 'by_title' are mutually exclusive.")

    if from_read_model is False and by_title:
        raise ValueError("'by_title' is only supported when 'from_read_model' is True.")

def delete_sampledata() -> None:
    """Delete the sample data directory (`Settings.sampledata_dir`) and its contents.

    Raises:
        ValueError: If `Settings.sampledata_dir` looks unsafe to delete (see
            `_check_safe_to_delete`).
    """

    _check_safe_to_delete(settings.sampledata_dir)

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

    shutil.rmtree(settings.sampledata_dir)

def download_sampledata(force: bool = False) -> None:
    """Download and extract the AIMBAT sample dataset.

    Downloads the sample data archive and extracts it into
    `Settings.sampledata_dir`.

    Args:
        force: Delete and replace an existing non-empty sample data
            directory instead of raising an error.

    Raises:
        FileExistsError: If the sample data directory already exists and is
            non-empty, and `force` is False.
    """

    logger.info(
        f"Downloading sample data from {_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(_SAMPLEDATA_SRC, timeout=_DOWNLOAD_TIMEOUT_SECONDS) as zipresp:
        with tempfile.NamedTemporaryFile() as tmpfile:
            shutil.copyfileobj(zipresp, tmpfile)
            tmpfile.flush()
            tmpfile.seek(0)
            logger.debug(f"Extracting sample data to {settings.sampledata_dir}.")
            with ZipFile(tmpfile) as zfile:
                _extract_regular_files(zfile, settings.sampledata_dir)

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

def dump_models[T: BaseModel](
    records: Sequence[T],
    model_class: type[T],
    *,
    by_alias: bool = False,
    by_title: bool = False,
    exclude: set[str] | None = None,
) -> list[dict[str, Any]]:
    """Serialise records to a JSON-compatible list of dicts.

    The shared tail of the `core.dump_*_table` functions: each builds its own
    query, and its read models where it has them, then hands the result here
    so field naming and exclusions are applied the same way everywhere.

    Args:
        records: Records to serialise, all instances of `model_class`.
        model_class: Class of the records, used for both the type adapter and
            the title map.
        by_alias: Whether to use serialisation aliases for the field names.
        by_title: Whether to use the field title metadata for the field names.
            Mutually exclusive with `by_alias`.
        exclude: Field names to exclude from every record.

    Returns:
        One dict per record.

    Raises:
        ValueError: If both `by_alias` and `by_title` are True.
    """
    check_field_name_flags(by_alias, by_title)

    exclude_spec: dict[str, set[str]] | None = {"__all__": exclude} if exclude else None

    adapter: TypeAdapter[Sequence[T]] = TypeAdapter(Sequence[model_class])  # type: ignore[valid-type]
    data: list[dict[str, Any]] = adapter.dump_python(
        records, mode="json", by_alias=by_alias, exclude=exclude_spec
    )

    if by_title:
        title_map = get_title_map(model_class)
        return [{title_map.get(k, k): v for k, v in row.items()} for row in data]

    return data

def exception_message(exc: BaseException) -> str:
    """Return an exception's message together with any notes attached to it.

    Code that knows context the raising site does not - which data source was
    being read, say - attaches it with `BaseException.add_note` instead of
    re-raising a different exception type. Notes only show up in a traceback,
    so anything that reports an exception without one has to add them back.

    Args:
        exc: The exception to describe.

    Returns:
        The exception's message, followed by one line per note.
    """
    return "\n".join([str(exc), *getattr(exc, "__notes__", [])])

def format_validation_error(exc: ValidationError) -> str:
    """Format a `ValidationError` as a short, semicolon-joined message.

    Strips the redundant `"Value error, "` prefix Pydantic adds to custom
    validator messages, so the result is fit for display without leaking
    implementation detail.

    Args:
        exc: The validation error to format.

    Returns:
        A single-line, semicolon-joined summary of all validation errors.
    """
    return "; ".join(e["msg"].removeprefix("Value error, ") for e in exc.errors())

@cache
def get_title_map(model_class: type[BaseModel]) -> dict[str, str]:
    """Build a mapping from field names to their display titles.

    Includes both regular and computed fields. A field without a `title` set
    falls back to its name with underscores replaced by spaces.

    Args:
        model_class: Pydantic model class to inspect.

    Returns:
        Mapping of field name to display title.
    """
    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 no record matches `id`, or more than one record does.
    """
    statement = select(aimbat_class.id).where(
        func.replace(cast(aimbat_class.id, String), "-", "").like(
            f"{_escape_like(id.replace('-', ''))}%", escape=_LIKE_ESCAPE_CHAR
        )
    )
    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 | None = None,
    str_uuid: str | None = None,
) -> str:
    """Return the shortest unique prefix for a UUID, formatted with dashes.

    The table's ids are fetched once per session and cached there, so calling
    this once per row (or once per cell) stays a single query - see
    `_id_pool` for what that caching does and does not guarantee.

    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.
            Defaults to `Settings.min_id_length`.
        str_uuid: The full UUID string. Required only if `aimbat_obj` is a class.

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

    Raises:
        ValueError: If `aimbat_obj` is a class and `str_uuid` is not
            provided, or if the resolved UUID has no matching record in the
            table.
    """

    if min_length is None:
        min_length = settings.min_id_length

    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)

    # Compare on the dash-free form throughout.
    target_clean = target_full.replace("-", "")

    pool = _id_pool(session, model_class, target_clean)
    if not _contains(pool, target_clean):
        raise ValueError(f"ID {target_full} not found in table {model_class.__name__}")

    length = min(
        max(min_length, _unique_prefix_length(pool, target_clean)), len(target_clean)
    )
    candidate = _dashed_prefix(target_full, length)
    logger.debug(f"Shortened {target_full} to: {candidate}")
    return candidate