poyee-data-warehouse/dw_base/utils/datetime_utils.py

import time
from datetime import date, timedelta, datetime

from dateutil import parser


def add_days(dt: str, number: int, original_fmt: str = '%Y%m%d', dst_fmt: str = '%Y%m%d'):
    return (parse_datetime(dt, original_fmt) + timedelta(days=number)).strftime(dst_fmt)


def date_to_timestamp(date_str: str, time_unit: int = 1) -> float:
    """
    字符串日期转时间戳
    Args:
        date_str: 字符串的日期
        time_unit: 时间戳单位，1为秒（默认），2为毫秒，3为微秒
    Returns: 时间戳
    """
    return datetime_to_timestamp(parse_datetime(date_str), time_unit)


def datetime_to_timestamp(date_time: datetime, time_unit: int = 1) -> float:
    """
    datetime 转时间戳
    Args:
        date_time: datetime
        time_unit: 时间戳单位，1为秒（默认），2为毫秒，3为微秒
    Returns: 时间戳
    """
    timestamp = date_time.timestamp() * 1000000
    timestamp = float(str(timestamp) + str("%06d" % date_time.microsecond))
    if time_unit == 3:
        return timestamp
    elif time_unit == 2:
        return timestamp / 1000
    else:
        return timestamp / 1000000


def formatted_now(fmt: str = '%Y-%m-%d %H:%M:%S') -> str:
    return time.strftime(fmt, time.localtime())


def get_date_by_priority(cli_dt: str, config_dt: str) -> str:
    # 计算实际日期，优先级: 命令行 > 配置文件 > 默认
    if cli_dt:
        dt = cli_dt
    elif config_dt:
        dt = config_dt
    else:
        dt = get_yesterday()
    return dt


def get_date_range(some_date: str) -> [str]:
    """
    获取日期范围
    Args:
        some_date: 日期参数，有四种表示法 —— 20210801、 20210801-、20210801-20210810、20210801,20210803,20210805
    Returns: 日期范围
    """
    if not some_date:
        return [get_yesterday()]
    date_range = []
    if len(some_date) == 8:
        date_range.append(some_date)
    elif (len(some_date) == 9 or len(some_date) == 17) and some_date[8] == '-':
        date_cursor = some_date[:8]
        if len(some_date) == 9:
            # 20210801-格式
            end_date = get_yesterday()
        else:
            # 20210801-20210810格式
            end_date = some_date[9:]
        while date_cursor <= end_date:
            date_range.append(date_cursor)
            date_cursor = (datetime.strptime(date_cursor, '%Y%m%d') + timedelta(days=1)).strftime('%Y%m%d')
    elif set([len(e) for e in some_date.split(',')]).pop() == 8:
        date_range = some_date.split(',')
    else:
        raise Exception('无效的日期表达形式 %s' % some_date)
    return date_range


def parse_datetime(date_time: str, original_format: str = None) -> datetime:
    """解析日期字符串为 datetime：传 original_format 按它解析，否则用 dateutil 自动识别。

    Args:
        date_time: 日期
        original_format: 原日期格式
    Returns:
        转换后的日期
    """
    if not date_time:
        return None
        # raise ValueError('date_time is null')
    try:
        if original_format:
            parsed_date = datetime.strptime(date_time, original_format)
        else:
            parsed_date = parser.parse(date_time, fuzzy=True)
    except Exception as e:
        raise e
    return parsed_date


def get_today(fmt: str = '%Y%m%d') -> str:
    return date.today().strftime(fmt)


def get_yesterday(fmt: str = '%Y%m%d') -> str:
    return (date.today() - timedelta(days=1)).strftime(fmt)


def local_2_utc(some_date: str) -> datetime:
    # 本地时间转 UTC 时间（ -8:00 ）
    # + ' 00:00:00'
    utc_date = datetime.utcfromtimestamp(date_to_timestamp(some_date))
    return utc_date


if __name__ == '__main__':
    print(parse_datetime('2024年03月10日'.replace('年', '-').replace('月', '-').replace('日', '')))