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:
    """
    先识别日期，再按照指定格式输出
    %y 两位数的年份表示（00-99）
    %Y 四位数的年份表示（0000-9999）
    %m 月份（01-12）
    %d 月内中的一天（0-31）
    %H 24小时制小时数（0-23）
    %I 12小时制小时数（01-12）
    %M 分钟数（00-59）
    %S 秒（00-59）
    %a 本地简化星期名称
    %A 本地完整星期名称
    %b 本地简化的月份名称
    %B 本地完整的月份名称
    %c 本地相应的日期表示和时间表示
    %j 年内的一天（001-366）
    %p 本地A.M.或P.M.的等价符
    %U 一年中的星期数（00-53）星期天为星期的开始
    %w 星期（0-6），星期天为星期的开始
    %W 一年中的星期数（00-53）星期一为星期的开始
    %x 本地相应的日期表示
    %X 本地相应的时间表示
    %Z 当前时区的名称
    %% %号本身
    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('日', '')))