|
| 1 | +#!/usr/bin/env python3 |
| 2 | +# -*- coding: utf-8 -*- |
| 3 | +import datetime |
| 4 | +import pytz |
| 5 | + |
| 6 | +from backend.app.core.conf import settings |
| 7 | + |
| 8 | + |
| 9 | +class DateTimeUtils: |
| 10 | + def __init__(self, timezone_str=settings.DATETIME_TIMEZONE): |
| 11 | + """ |
| 12 | + 初始化函数，设置时区 |
| 13 | +
|
| 14 | + :param timezone_str: 时区字符串，默认为 UTC |
| 15 | + """ |
| 16 | + self.timezone_str = timezone_str |
| 17 | + self.timezone = pytz.timezone(self.timezone_str) |
| 18 | + |
| 19 | + def get_current_time(self) -> datetime.datetime: |
| 20 | + """ |
| 21 | + 获取当前时间 |
| 22 | +
|
| 23 | + :return: 当前时间的 datetime 对象 |
| 24 | + """ |
| 25 | + return datetime.datetime.now(self.timezone) |
| 26 | + |
| 27 | + @staticmethod |
| 28 | + def get_current_timestamp() -> int: |
| 29 | + """ |
| 30 | + 获取当前时间戳 (秒) |
| 31 | +
|
| 32 | + :return: 当前时间戳 (秒) |
| 33 | + """ |
| 34 | + return int(datetime.datetime.now().timestamp()) |
| 35 | + |
| 36 | + @staticmethod |
| 37 | + def get_current_milliseconds() -> int: |
| 38 | + """ |
| 39 | + 获取当前时间戳 (毫秒) |
| 40 | +
|
| 41 | + :return: 当前时间戳 (毫秒) |
| 42 | + """ |
| 43 | + return int(datetime.datetime.now().timestamp() * 1000) |
| 44 | + |
| 45 | + def timestamp_to_datetime(self, timestamp: int) -> datetime.datetime: |
| 46 | + """ |
| 47 | + 时间戳转 datetime 对象 |
| 48 | +
|
| 49 | + :param timestamp: 时间戳 (秒) |
| 50 | + :return: datetime 对象 |
| 51 | + """ |
| 52 | + return datetime.datetime.utcfromtimestamp(timestamp).replace(tzinfo=self.timezone) |
| 53 | + |
| 54 | + def datetime_to_timestamp(self, dt: datetime.datetime) -> int: |
| 55 | + """ |
| 56 | + datetime 对象转时间戳（秒） |
| 57 | +
|
| 58 | + :param dt: datetime 对象 |
| 59 | + :return: 时间戳 (秒) |
| 60 | + """ |
| 61 | + return int(dt.astimezone(self.timezone).timestamp()) |
| 62 | + |
| 63 | + def datetime_to_milliseconds(self, dt: datetime.datetime) -> int: |
| 64 | + """ |
| 65 | + datetime 对象转时间戳（毫秒） |
| 66 | +
|
| 67 | + :param dt: datetime 对象 |
| 68 | + :return: 时间戳 (毫秒) |
| 69 | + """ |
| 70 | + return int(dt.astimezone(self.timezone).timestamp() * 1000) |
| 71 | + |
| 72 | + def str_to_datetime(self, time_str: str, format_str: str = settings.DATETIME_FORMAT) -> datetime.datetime: |
| 73 | + """ |
| 74 | + 时间字符串转 datetime 对象 |
| 75 | +
|
| 76 | + :param time_str: 时间字符串 |
| 77 | + :param format_str: 时间字符串的格式，默认为 '%Y-%m-%d %H:%M:%S' |
| 78 | + :return: datetime 对象 |
| 79 | + """ |
| 80 | + return datetime.datetime.strptime(time_str, format_str).replace(tzinfo=self.timezone) |
| 81 | + |
| 82 | + def datetime_to_str(self, dt: datetime.datetime, format_str: str = settings.DATETIME_FORMAT) -> str: |
| 83 | + """ |
| 84 | + datetime 对象转时间字符串 |
| 85 | +
|
| 86 | + :param dt: datetime 对象 |
| 87 | + :param format_str: 时间字符串的格式，默认为 '%Y-%m-%d %H:%M:%S' |
| 88 | + :return: 时间字符串 |
| 89 | + """ |
| 90 | + return dt.astimezone(self.timezone).strftime(format_str) |
| 91 | + |
| 92 | + @staticmethod |
| 93 | + def get_timezone(timezone_str: str) -> pytz.timezone: |
| 94 | + """ |
| 95 | + 获取指定时区的 pytz.timezone 对象 |
| 96 | +
|
| 97 | + :param timezone_str: 时区字符串 |
| 98 | + :return: pytz.timezone 对象 |
| 99 | + """ |
| 100 | + return pytz.timezone(timezone_str) |
| 101 | + |
| 102 | + def get_timezone_time(self, timezone_str: str) -> datetime.datetime: |
| 103 | + """ |
| 104 | + 获取指定时区的当前时间 |
| 105 | +
|
| 106 | + :param timezone_str: 时区字符串 |
| 107 | + :return: 当前时间的 datetime 对象 |
| 108 | + """ |
| 109 | + timezone = self.get_timezone(timezone_str) |
| 110 | + return datetime.datetime.now(timezone) |
| 111 | + |
| 112 | + def datetime_to_timezone(self, dt: datetime.datetime, timezone_str: str) -> datetime.datetime: |
| 113 | + """ |
| 114 | + 将 datetime 对象转换为指定时区的 datetime 对象 |
| 115 | +
|
| 116 | + :param dt: datetime 对象 |
| 117 | + :param timezone_str: 目标时区字符串 |
| 118 | + :return: 目标时区的 datetime 对象 |
| 119 | + """ |
| 120 | + timezone = self.get_timezone(timezone_str) |
| 121 | + return dt.astimezone(timezone) |
| 122 | + |
| 123 | + def datetime_to_timezone_str( |
| 124 | + self, dt: datetime.datetime, timezone_str: str, format_str: str = settings.DATETIME_FORMAT |
| 125 | + ) -> str: |
| 126 | + """ |
| 127 | + 将 datetime 对象转换为指定时区的时间字符串 |
| 128 | +
|
| 129 | + :param dt: datetime 对象 |
| 130 | + :param timezone_str: 目标时区字符串 |
| 131 | + :param format_str: 时间字符串的格式，默认为 '%Y-%m-%d %H:%M:%S' |
| 132 | + :return: 目标时区的时间字符串 |
| 133 | + """ |
| 134 | + dt_timezone = self.datetime_to_timezone(dt, timezone_str) |
| 135 | + return dt_timezone.strftime(format_str) |
| 136 | + |
| 137 | + def str_to_timezone( |
| 138 | + self, time_str: str, timezone_str: str, format_str: str = settings.DATETIME_FORMAT |
| 139 | + ) -> datetime.datetime: |
| 140 | + """ |
| 141 | + 将指定时区的时间字符串转换为 datetime 对象 |
| 142 | +
|
| 143 | + :param time_str: 指定时区的时间字符串 |
| 144 | + :param timezone_str: 指定时区字符串 |
| 145 | + :param format_str: 时间字符串的格式，默认为 '%Y-%m-%d %H:%M:%S' |
| 146 | + :return: datetime 对象 |
| 147 | + """ |
| 148 | + dt = datetime.datetime.strptime(time_str, format_str).replace(tzinfo=self.timezone) |
| 149 | + return self.datetime_to_timezone(dt, timezone_str) |
| 150 | + |
| 151 | + @staticmethod |
| 152 | + def datetime_to_utc(dt: datetime.datetime) -> datetime.datetime: |
| 153 | + """ |
| 154 | + 将 datetime 对象转换为 UTC 时间 |
| 155 | +
|
| 156 | + :param dt: datetime 对象 |
| 157 | + :return: UTC 时间的 datetime 对象 |
| 158 | + """ |
| 159 | + return dt.astimezone(pytz.utc) |
| 160 | + |
| 161 | + def str_to_utc(self, time_str: str, format_str: str = settings.DATETIME_FORMAT) -> datetime.datetime: |
| 162 | + """ |
| 163 | + 将时间字符串转换为 UTC 时间的 datetime 对象 |
| 164 | +
|
| 165 | + :param time_str: 时间字符串 |
| 166 | + :param format_str: 时间字符串的格式，默认为 '%Y-%m-%d %H:%M:%S' |
| 167 | + :return: UTC 时间的 datetime 对象 |
| 168 | + """ |
| 169 | + dt = datetime.datetime.strptime(time_str, format_str).replace(tzinfo=self.timezone) |
| 170 | + return self.datetime_to_utc(dt) |
| 171 | + |
| 172 | + def utc_to_datetime(self, utc_time: datetime.datetime) -> datetime.datetime: |
| 173 | + """ |
| 174 | + 将 UTC 时间的 datetime 对象转换为指定时区的 datetime 对象 |
| 175 | +
|
| 176 | + :param utc_time: UTC 时间的 datetime 对象 |
| 177 | + :return: 目标时区的 datetime 对象 |
| 178 | + """ |
| 179 | + return utc_time.replace(tzinfo=pytz.utc).astimezone(self.timezone).replace(tzinfo=None) |
| 180 | + |
| 181 | + def get_expire_time(self, expires_delta: datetime.timedelta) -> datetime: |
| 182 | + """ |
| 183 | + 获取过期时间 |
| 184 | +
|
| 185 | + :param expires_delta: 时间间隔对象 |
| 186 | + :return: 过期时间的 datetime 对象 |
| 187 | + """ |
| 188 | + return self.get_current_time() + expires_delta |
| 189 | + |
| 190 | + @staticmethod |
| 191 | + def get_expire_time_from_datetime(expire_time: datetime, seconds: int) -> datetime: |
| 192 | + """ |
| 193 | + 获取从指定时间开始一定时间后的过期时间 |
| 194 | +
|
| 195 | + :param expire_time: 指定时间的 datetime 对象 |
| 196 | + :param seconds: 时间间隔（秒） |
| 197 | + :return: 过期时间的 datetime 对象 |
| 198 | + """ |
| 199 | + return expire_time + datetime.timedelta(seconds=seconds) |
| 200 | + |
| 201 | + @staticmethod |
| 202 | + def get_expire_seconds(expires_delta: datetime.timedelta) -> int: |
| 203 | + """ |
| 204 | + 获取过期时间（秒） |
| 205 | +
|
| 206 | + :param expires_delta: 时间间隔对象 |
| 207 | + :return: 过期时间（秒） |
| 208 | + """ |
| 209 | + return int(expires_delta.total_seconds()) |
| 210 | + |
| 211 | + def get_expire_seconds_from_datetime(self, expire_datetime: datetime) -> int: |
| 212 | + """ |
| 213 | + 获取从指定时间开始到当前时间的时间间隔（秒） |
| 214 | +
|
| 215 | + :param expire_datetime: 指定时间的 datetime 对象 |
| 216 | + :return: 时间间隔（秒） |
| 217 | + """ |
| 218 | + current_time = self.get_current_time() |
| 219 | + if expire_datetime < current_time: |
| 220 | + return 0 |
| 221 | + return int((expire_datetime - current_time).total_seconds()) |
| 222 | + |
| 223 | + |
| 224 | +datetime_utils = DateTimeUtils() |
0 commit comments