68 lines
1.9 KiB
Python
68 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
|
|
TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
|
|
DEFAULT_TIMEZONE = "Asia/Shanghai"
|
|
|
|
|
|
def format_beijing_time(
|
|
epoch_seconds: float | int | str | None,
|
|
*,
|
|
offset_seconds: float | int = 0,
|
|
timezone_name: str = DEFAULT_TIMEZONE,
|
|
) -> str | None:
|
|
epoch = _optional_float(epoch_seconds)
|
|
if epoch is None:
|
|
return None
|
|
zone = _zone(timezone_name)
|
|
timestamp = epoch + float(offset_seconds)
|
|
return datetime.fromtimestamp(timestamp, tz=timezone.utc).astimezone(zone).strftime(
|
|
TIME_FORMAT
|
|
)
|
|
|
|
|
|
def derive_time_from_reference(
|
|
reference_time: str | None,
|
|
*,
|
|
reference_offset_seconds: float | int | None,
|
|
target_offset_seconds: float | int | None,
|
|
) -> str | None:
|
|
if not reference_time:
|
|
return None
|
|
reference_offset = _optional_float(reference_offset_seconds)
|
|
target_offset = _optional_float(target_offset_seconds)
|
|
if reference_offset is None or target_offset is None:
|
|
return None
|
|
try:
|
|
reference = datetime.strptime(reference_time, TIME_FORMAT)
|
|
except ValueError:
|
|
return None
|
|
return (reference + timedelta(seconds=target_offset - reference_offset)).strftime(
|
|
TIME_FORMAT
|
|
)
|
|
|
|
|
|
def timeline_start_epoch(record: dict[str, Any]) -> float | None:
|
|
for key in ("actual_begin", "requested_begin"):
|
|
value = _optional_float(record.get(key))
|
|
if value is not None:
|
|
return value
|
|
return None
|
|
|
|
|
|
def _zone(timezone_name: str) -> ZoneInfo:
|
|
try:
|
|
return ZoneInfo(timezone_name)
|
|
except ZoneInfoNotFoundError:
|
|
return ZoneInfo(DEFAULT_TIMEZONE)
|
|
|
|
|
|
def _optional_float(value: Any) -> float | None:
|
|
if value is None or value == "":
|
|
return None
|
|
return float(value)
|