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

"""
时区工具模块
提供东八区（Asia/Shanghai）时间处理功能
"""

from datetime import datetime

try:
    # Python 3.9+
    from zoneinfo import ZoneInfo
except ImportError:
    # Python 3.8 使用 backports
    from backports.zoneinfo import ZoneInfo

# 东八区时区
CHINA_TZ = ZoneInfo("Asia/Shanghai")


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:
        # 如果是naive datetime，假设它是UTC时间
        dt = dt.replace(tzinfo=ZoneInfo("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=ZoneInfo("UTC"))
    return dt.astimezone(CHINA_TZ).replace(tzinfo=None)