core/engine/src/primitives/_time.py - main - alexefan136/flowstack

core/engine/src/primitives/_time.py
267 строк8 KB
Alexander Efanov
upd fix
31 июл 2026, 19:17
31 июл 2026, 19:17d146d86
"""
Утилиты времени для FlowStack Engine.

Единая точка для работы с временем во всех primitives:
- Timezone-aware UTC (замена deprecated datetime.utcnow())
- Monotonic-время для точного измерения duration
- Безопасный парсинг/сериализация ISO 8601
- Проверки истечения срока (deadline, trial)
- Относительное время для UI ("5 мин назад")

Использование:
    from src.primitives._time import utc_now, parse_datetime, to_iso, monotonic, elapsed_ms

    # Timestamps (сериализуемые)
    created_at = utc_now()
    restored = parse_datetime(data.get("created_at"))
    serialized = to_iso(created_at)

    # Duration (точное, monotonic)
    start = monotonic()
    ... do work ...
    duration_ms = elapsed_ms(start)

    # Проверки
    if is_expired(task.deadline):
        ...
"""

from __future__ import annotations

import time
from datetime import UTC, datetime

# ============================================================================
# Constants (для расчёта относительного времени)
# ============================================================================

_SECONDS_PER_MINUTE = 60
_MINUTES_PER_HOUR = 60
_HOURS_PER_DAY = 24
_DAYS_PER_WEEK = 7
_DAYS_PER_MONTH = 30
_DAYS_PER_YEAR = 365


# ============================================================================
# UTC Now (замена deprecated datetime.utcnow())
# ============================================================================


def utc_now() -> datetime:
    """
    Текущее время в UTC (timezone-aware).

    Замена deprecated ``datetime.utcnow()`` (Python 3.12+).
    Возвращает aware datetime с ``tzinfo=UTC``.

    Использовать для всех timestamp-полей (created_at, updated_at, deadline...).
    """
    return datetime.now(UTC)


def utc_from_timestamp(ts: float) -> datetime:
    """Создать aware UTC datetime из UNIX timestamp."""
    return datetime.fromtimestamp(ts, tz=UTC)


def to_timestamp(dt: datetime) -> float:
    """Получить UNIX timestamp из datetime (aware или naive-UTC)."""
    return ensure_aware(dt).timestamp()


# ============================================================================
# Monotonic time (для измерения duration)
# ============================================================================


def monotonic() -> float:
    """
    Monotonic-время в секундах (для измерения длительности).

    В отличие от ``time.time()``, не зависит от изменений системных часов
    (NTP sync, смена часового пояса, летнее время). Идеально для duration.

    НЕ сериализуется (это относительное время) — только для runtime-замеров.
    """
    return time.monotonic()


def monotonic_ms() -> float:
    """Monotonic-время в миллисекундах."""
    return time.monotonic() * 1000


def elapsed_ms(start: float) -> float:
    """
    Длительность в мс от ``start`` (monotonic) до текущего момента.

    Usage:
        start = monotonic()
        ... do work ...
        duration = elapsed_ms(start)
    """
    return (time.monotonic() - start) * 1000


# ============================================================================
# Serialization / Parsing
# ============================================================================


def to_iso(dt: datetime | None) -> str | None:
    """
    Сериализовать datetime в ISO 8601 строку.

    Naive datetime считается UTC (добавляет ``+00:00``).
    None → None.
    """
    if dt is None:
        return None
    return ensure_aware(dt).isoformat()


def parse_datetime(value: str | datetime | None) -> datetime | None:
    """
    Безопасно распарсить datetime из ISO строки.

    - ``None`` → ``None``
    - ``datetime`` → as-is (naive → UTC)
    - ``str`` → ``fromisoformat`` (naive → UTC, поддерживает суффикс ``Z``)
    - Ошибка парсинга → ``None`` (не бросает исключение)

    Использовать в ``from_dict()`` для восстановления datetime-полей.
    """
    if value is None:
        return None
    if isinstance(value, datetime):
        return ensure_aware(value)
    if not isinstance(value, str) or not value:
        return None
    try:
        # Python 3.10 fromisoformat не парсит суффикс 'Z' — нормализуем
        normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
        return ensure_aware(datetime.fromisoformat(normalized))
    except (ValueError, TypeError):
        return None


def ensure_aware(dt: datetime) -> datetime:
    """
    Сделать datetime timezone-aware.

    Naive datetime считается UTC (добавляет ``tzinfo=UTC``).
    Aware datetime возвращается как есть.
    """
    if dt.tzinfo is None:
        return dt.replace(tzinfo=UTC)
    return dt


# ============================================================================
# Expiry / Comparison
# ============================================================================


def is_expired(deadline: datetime | None, now: datetime | None = None) -> bool:
    """
    Проверить, истёк ли срок.

    Args:
        deadline: дедлайн (``None`` = нет дедлайна → не истёк)
        now: текущее время (по умолчанию ``utc_now()``)

    Returns:
        True если дедлайн прошёл.
    """
    if deadline is None:
        return False
    current = now or utc_now()
    return ensure_aware(deadline) < ensure_aware(current)


def is_overdue(
    deadline: datetime | None,
    is_done: bool = False,
    now: datetime | None = None,
) -> bool:
    """
    Проверить, просрочена ли задача.

    Задача просрочена, если дедлайн прошёл И она не завершена.

    Args:
        deadline: дедлайн задачи
        is_done: завершена ли задача
        now: текущее время
    """
    if is_done:
        return False
    return is_expired(deadline, now)


# ============================================================================
# Relative time (для UI)
# ============================================================================


def format_relative_time(dt: datetime | None, now: datetime | None = None) -> str:
    """
    Относительное время для отображения в UI.

    Примеры: "только что", "5 мин назад", "2 ч назад", "вчера", "3 дн назад".
    Для будущего времени — "только что".

    Args:
        dt: целевое время
        now: текущее время (по умолчанию ``utc_now()``)
    """
    if dt is None:
        return ""

    current = now or utc_now()
    diff_seconds = (ensure_aware(current) - ensure_aware(dt)).total_seconds()

    if diff_seconds < 0:
        return "только что"  # будущее время

    seconds = int(diff_seconds)
    minutes = seconds // _SECONDS_PER_MINUTE
    hours = minutes // _MINUTES_PER_HOUR
    days = hours // _HOURS_PER_DAY

    if seconds < _SECONDS_PER_MINUTE:
        return "только что"
    if minutes < _MINUTES_PER_HOUR:
        return f"{minutes} мин назад"
    if hours < _HOURS_PER_DAY:
        return f"{hours} ч назад"
    if days == 1:
        return "вчера"
    if days < _DAYS_PER_WEEK:
        return f"{days} дн назад"
    if days < _DAYS_PER_MONTH:
        return f"{days // _DAYS_PER_WEEK} нед назад"
    if days < _DAYS_PER_YEAR:
        return f"{days // _DAYS_PER_MONTH} мес назад"
    return f"{days // _DAYS_PER_YEAR} г назад"


# ============================================================================
# Exports
# ============================================================================

__all__ = [
    "elapsed_ms",
    "ensure_aware",
    "format_relative_time",
    "is_expired",
    "is_overdue",
    "monotonic",
    "monotonic_ms",
    "parse_datetime",
    "to_iso",
    "to_timestamp",
    "utc_from_timestamp",
    "utc_now",
]