DataOps-platform/app/core/common/timezone_utils.py

"""
时区工具模块
提供东八区（Asia/Shanghai，UTC+8，无夏令时）时间处理功能
"""

from datetime import datetime, timedelta, timezone

# 东八区：Asia/Shanghai 不使用夏令时，固定 UTC+8 即可
CHINA_TZ = timezone(timedelta(hours=8))
UTC = timezone.utc


def now_china() -> datetime:
    """
    获取当前东八区时间（带时区信息）

    Returns:
        datetime: 当前东八区时间
    """
    return datetime.now(CHINA_TZ)


def now_china_naive() -> datetime:
    """
    获取当前东八区时间（不带时区信息，用于数据库存储）

    Returns:
        datetime: 当前东八区时间（naive datetime）
    """
    return datetime.now(CHINA_TZ).replace(tzinfo=None)


def to_china_time(dt: datetime) -> datetime:
    """
    将任意时区的时间转换为东八区时间

    Args:
        dt: 输入的datetime对象

    Returns:
        datetime: 转换后的东八区时间
    """
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=UTC)
    return dt.astimezone(CHINA_TZ)


def utc_to_china_naive(dt: datetime) -> datetime:
    """
    将UTC时间转换为东八区时间（不带时区信息）

    Args:
        dt: UTC时间

    Returns:
        datetime: 东八区时间（naive datetime）
    """
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=UTC)
    return dt.astimezone(CHINA_TZ).replace(tzinfo=None)