1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160
|
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
from datetime import date, datetime
from .._common import strip_ansi_escape
from .._const import DefaultValue, ParamKey
from ..error import TypeConversionError
from ._interface import AbstractValueConverter
class DateTimeConverter(AbstractValueConverter):
__DAYS_TO_SECONDS_COEF = 60**2 * 24
__MICROSECONDS_TO_SECONDS_COEF = 1000**2
__COMMON_DST_TIMEZONE_TABLE = {
-36000: "America/Adak", # -1000
-32400: "US/Alaska", # -0900
-28800: "US/Pacific", # -0800
-25200: "US/Mountain", # -0700
-21600: "US/Central", # -0600
-18000: "US/Eastern", # -0500
-14400: "America/Halifax", # -0400
-12600: "America/St_Johns", # -0330
-10800: "America/Miquelon", # -0300
7200: "Africa/Tripoli", # 0200
}
def __init__(self, value, params):
super().__init__(value, params)
self.__datetime = None
self.__timezone = self._params.get(ParamKey.TIMEZONE)
def force_convert(self):
self.__datetime = self.__from_datetime()
if self.__datetime:
return self.__datetime
self.__datetime = self.__from_timestamp()
if self.__datetime:
return self.__datetime
return self.__from_datetime_string()
def __from_datetime(self):
if not isinstance(self._value, (date, datetime)):
return None
if isinstance(self._value, datetime):
self.__datetime = self._value
elif isinstance(self._value, date):
self.__datetime = datetime(
year=self._value.year, month=self._value.month, day=self._value.day
)
if self.__timezone:
if self.__datetime.tzinfo is None:
self.__datetime = self.__timezone.localize(self.__datetime)
else:
self.__datetime = datetime.fromtimestamp(
self.__datetime.timestamp(), tz=self.__timezone
)
return self.__datetime
def __from_timestamp(self):
from ..type._integer import Integer
from ..type._realnumber import RealNumber
conv_error = TypeConversionError(
"timestamp is out of the range of values supported by the platform"
)
timestamp = Integer(self._value, strict_level=1).try_convert()
if timestamp:
try:
self.__datetime = datetime.fromtimestamp(timestamp, self.__timezone)
except (ValueError, OSError, OverflowError):
raise conv_error
return self.__datetime
timestamp = RealNumber(self._value, strict_level=1).try_convert()
if timestamp:
try:
self.__datetime = datetime.fromtimestamp(int(timestamp), self.__timezone).replace(
microsecond=int((timestamp - int(timestamp)) * 1000000)
)
except (ValueError, OSError, OverflowError):
raise conv_error
return self.__datetime
return None
def __from_datetime_string(self):
import dateutil.parser
import pytz
self.__validate_datetime_string()
try:
self.__datetime = dateutil.parser.parse(self._value)
except (AttributeError, ValueError, OverflowError):
if self._params.get(ParamKey.STRIP_ANSI_ESCAPE, DefaultValue.STRIP_ANSI_ESCAPE):
try:
self.__datetime = dateutil.parser.parse(strip_ansi_escape(self._value))
except (AttributeError, ValueError, OverflowError):
pass
if self.__datetime is None:
raise TypeConversionError(f"failed to parse as a datetime: type={type(self._value)}")
if self.__timezone:
pytz_timezone = self.__timezone
else:
try:
dst_timezone_name = self.__get_dst_timezone_name(self.__get_timedelta_sec())
except (AttributeError, KeyError):
return self.__datetime
pytz_timezone = pytz.timezone(dst_timezone_name)
self.__datetime = self.__datetime.replace(tzinfo=None)
self.__datetime = pytz_timezone.localize(self.__datetime)
return self.__datetime
def __get_timedelta_sec(self):
dt = self.__datetime.utcoffset()
return int(
dt.days * self.__DAYS_TO_SECONDS_COEF
+ float(dt.seconds)
+ dt.microseconds / self.__MICROSECONDS_TO_SECONDS_COEF
)
def __get_dst_timezone_name(self, offset):
return self.__COMMON_DST_TIMEZONE_TABLE[offset]
def __validate_datetime_string(self):
"""
This will require validating version string (such as "3.3.5").
A version string could be converted to a datetime value if this
validation is not executed.
"""
from packaging.version import InvalidVersion, Version
try:
try:
Version(self._value)
raise TypeConversionError(
f"invalid datetime string: version string found {self._value}"
)
except InvalidVersion:
pass
except TypeError:
raise TypeConversionError(f"invalid datetime string: type={type(self._value)}")
|