155 lines
5.0 KiB
Python
155 lines
5.0 KiB
Python
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_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()
|