document-management/backend/dms/common/time_range.py

"""DMS查询接口统一的带时区ISO 8601时间范围解析。"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Mapping

from dms.common.errors import InvalidArgumentError


def parse_datetime(value: str | None, *, name: str) -> datetime | None:
    """解析带明确时区的ISO 8601时间并转换为UTC无时区值。"""
    if value is None:
        return None
    if not isinstance(value, str) or not value.strip():
        raise InvalidArgumentError(f"{name}必须是包含时区的ISO 8601时间")
    normalized = value.strip()
    if normalized.endswith(("Z", "z")):
        normalized = normalized[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(normalized)
    except ValueError as exc:
        raise InvalidArgumentError(
            f"{name}必须是包含时区的ISO 8601时间"
        ) from exc
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        raise InvalidArgumentError(f"{name}必须包含明确时区")
    return parsed.astimezone(timezone.utc).replace(tzinfo=None)


def parse_updated_range(
    params: Mapping[str, str],
) -> tuple[datetime | None, datetime | None]:
    """解析updatedFrom/updatedTo并校验闭区间方向。"""
    updated_from = parse_datetime(params.get("updatedFrom"), name="updatedFrom")
    updated_to = parse_datetime(params.get("updatedTo"), name="updatedTo")
    if (
        updated_from is not None
        and updated_to is not None
        and updated_from > updated_to
    ):
        raise InvalidArgumentError("updatedFrom不得晚于updatedTo")
    return updated_from, updated_to


__all__ = ["parse_datetime", "parse_updated_range"]