TJWaterServerBinary/app/domain/time.py

from datetime import date, datetime, time, timedelta, timezone

from dateutil import parser, tz

'''
2025-02-09T15:45:00+00:00 采用的是 ISO 8601 国际标准日期时间格式，具体特点如下：

日期部分：YYYY-MM-DD（年-月-日），例如2025-02-09表示2025年2月9日。
时间部分：HH:mm:ss（时:分:秒），例如15:45:00表示下午3点45分0秒。
分隔符：日期与时间之间用字母T连接，表明这是一个完整的时间点。
时区偏移：末尾的+00:00表示该时间基于协调世界时（UTC），即零时区。若使用Z替代+00:00（如2025-02-09T15:45:00Z），也符合ISO 8601标准，两者等价

北京时间格式
    2025-02-09T15:45:00+08:00 

'''
BG_TZ = tz.gettz("Asia/Shanghai")
UTC_TZ = timezone.utc

TIMEZONE_REQUIRED_MESSAGE = (
    "Datetime values must include an explicit timezone offset, for example "
    "'2025-02-09T15:45:00Z' or '2025-02-09T23:45:00+08:00'."
)


def parse_aware_time(query_time: datetime | str, field_name: str = "datetime") -> datetime:
    """
    解析时间并确保结果带有时区信息。
    """
    dt = parser.parse(query_time) if isinstance(query_time, str) else query_time
    if dt.tzinfo is None:
        raise ValueError(f"{field_name} is missing timezone information. {TIMEZONE_REQUIRED_MESSAGE}")
    return dt


def extract_date(value: date | datetime | str, field_name: str = "date") -> date:
    """
    提取日期部分，但保留调用方原始时区语义，不强制转换到 UTC。
    """
    if isinstance(value, date) and not isinstance(value, datetime):
        return value
    return parse_aware_time(value, field_name=field_name).date()


def utc_now() -> datetime:
    """
    返回带 UTC 时区的当前时间。
    """
    return datetime.now(UTC_TZ)


def parse_utc_time(query_time: datetime | str, field_name: str = "datetime") -> datetime:
    '''
    接受带时区的时间字符串/对象，并统一转换成 UTC 时间。
    '''
    return parse_aware_time(query_time, field_name=field_name).astimezone(UTC_TZ)
    

def parse_beijing_time(query_time: datetime | str, field_name: str = "datetime") -> datetime:
    '''
    接受带时区的时间字符串/对象，并统一转换成北京时间。
    '''
    return parse_aware_time(query_time, field_name=field_name).astimezone(tz=BG_TZ)


def to_utc_time(dt: datetime | str, field_name: str = "datetime") -> datetime:
    '''
    将一个带时区的时间点，转换成 UTC。
    '''
    return parse_aware_time(dt, field_name=field_name).astimezone(UTC_TZ)


def to_beijing_time(dt: datetime | str, field_name: str = "datetime") -> datetime:
    '''
    将一个带时区的时间点，转换成北京时间。
    '''
    return parse_aware_time(dt, field_name=field_name).astimezone(tz=BG_TZ)

    
def to_time_range(dt: datetime, delta: float) -> tuple[datetime, datetime]:
    '''
    将一个时间点，转换成 start/end 时间段
    有些查询按照一个时间点查不到，用时间段保证能成功
    
    delta 单位是秒
    '''
    start_time = dt - timedelta(seconds=delta)
    end_time = dt + timedelta(seconds=delta)

    return (start_time, end_time)


def parse_clock_duration_seconds(clock: str, field_name: str = "duration") -> int:
    """
    Parse EPANET-style clock durations into seconds.

    Accepted formats include H:MM, HH:MM, H:MM:SS, and HH:MM:SS.
    """
    if not isinstance(clock, str):
        raise ValueError(f"{field_name} must be a string clock duration.")

    parts = clock.strip().split(":")
    if len(parts) not in (2, 3):
        raise ValueError(
            f"{field_name} must use H:MM or H:MM:SS format, got {clock!r}."
        )

    try:
        values = [int(part) for part in parts]
    except ValueError as exc:
        raise ValueError(
            f"{field_name} must contain numeric clock parts, got {clock!r}."
        ) from exc

    if any(value < 0 for value in values):
        raise ValueError(f"{field_name} must not contain negative values.")

    hours, minutes = values[0], values[1]
    seconds = values[2] if len(values) == 3 else 0
    if minutes >= 60 or seconds >= 60:
        raise ValueError(
            f"{field_name} minutes and seconds must be less than 60, got {clock!r}."
        )

    return hours * 3600 + minutes * 60 + seconds

def parse_beijing_date_range(query_date: str) -> tuple[datetime, datetime]:
    '''
    将一个日期字符串，转换成 start/end 时间段，传进来的日期被认为是北京时间
    日期字符串格式：YYYY-MM-DD
    '''
    target_date = date.fromisoformat(query_date)
    start_time = datetime.combine(target_date, time.min, BG_TZ)
    end_time = start_time + timedelta(days=1)

    return (start_time, end_time)


def get_day_start(dt: datetime.date) -> datetime:
    '''
    获取 某一天的 00:00:00
    这一天可以是北京时间，也可以是 utc 时间
    '''
    return dt.replace(hour=0, minute=0, second=0, microsecond=0)


def get_day_end(dt: datetime.date) -> datetime:
    '''
    获取 某一天的 23:59:59
    这一天可以是北京时间，也可以是 utc 时间
    '''
    return dt.replace(hour=23, minute=59, second=59, microsecond=0)

def get_date_from_time(time: str) -> str:
    '''
    将一个时间点，转换成日期
    '''
    dt = parse_beijing_time(time, field_name="time")
    return str(dt.date())


def is_today(query_date: str) -> bool:
    '''
    判断一个日期是否是今天
    '''
    dt = parse_beijing_time(query_date, field_name="query_date")
    return dt.date() == datetime.now(BG_TZ).date()


def is_yesterday(query_date: str) -> bool:
    '''
    判断一个日期是否是昨天
    '''
    dt = parse_beijing_time(query_date, field_name="query_date")
    return dt.date() == (datetime.now(BG_TZ).date() - timedelta(days=1))

def is_tomorrow(query_date: str) -> bool:
    '''
    判断一个日期是否是明天
    '''
    dt = parse_beijing_time(query_date, field_name="query_date")
    return dt.date() == (datetime.now(BG_TZ).date() + timedelta(days=1))

def is_today_or_future(query_date: str) -> bool:
    '''
    判断一个日期是否是今天或未来
    '''
    dt = parse_beijing_time(query_date, field_name="query_date")
    return dt.date() >= datetime.now(BG_TZ).date()