93 lines
1.8 KiB
Python
93 lines
1.8 KiB
Python
import time
|
|
from datetime import datetime
|
|
from datetime import timedelta
|
|
from ipaddress import IPv4Address
|
|
|
|
|
|
def is_valid_date(v, **kwargs):
|
|
try:
|
|
date = datetime.strptime(v, "%Y-%m-%d %H:%M:%S")
|
|
zone_offset = kwargs.get('#zone_offset', 8)
|
|
return (date - timedelta(hours=zone_offset)).strftime("%Y-%m-%d %H:%M:%S")
|
|
except:
|
|
return None
|
|
|
|
|
|
def is_valid_int(v, **kwargs):
|
|
try:
|
|
return int(v)
|
|
except:
|
|
return None
|
|
|
|
|
|
def is_valid_srt(v, **kwargs):
|
|
if isinstance(v, str):
|
|
return v
|
|
return None
|
|
|
|
|
|
def is_valid_float(v, **kwargs):
|
|
try:
|
|
return float(v)
|
|
except:
|
|
return None
|
|
|
|
|
|
def is_valid_bool(v, **kwargs):
|
|
if isinstance(v, bool):
|
|
return v
|
|
return None
|
|
|
|
|
|
def is_valid_array(v, **kwargs):
|
|
if isinstance(v, list):
|
|
return [str(i) for i in v]
|
|
return None
|
|
|
|
|
|
def is_valid_ipv4(v, **kwargs):
|
|
try:
|
|
return str(IPv4Address(v))
|
|
except:
|
|
return None
|
|
|
|
|
|
TYPE_CK2PY = {
|
|
"DateTime('UTC')": is_valid_date,
|
|
"Nullable(DateTime('UTC'))": is_valid_date,
|
|
"DateTime()": is_valid_date,
|
|
|
|
"Nullable(IPv4)": is_valid_ipv4,
|
|
"IPv4": is_valid_ipv4,
|
|
|
|
"String": is_valid_srt,
|
|
"Nullable(String)": is_valid_srt,
|
|
|
|
"Nullable(UInt8)": is_valid_int,
|
|
"UInt8": is_valid_srt,
|
|
|
|
"Nullable(Int8)": is_valid_int,
|
|
"Int8": is_valid_srt,
|
|
|
|
"Nullable(UInt16)": is_valid_int,
|
|
"UInt16": is_valid_srt,
|
|
|
|
"Nullable(Int16)": is_valid_int,
|
|
"Int16": is_valid_srt,
|
|
|
|
"Nullable(UInt32)": is_valid_int,
|
|
"UInt32": is_valid_srt,
|
|
|
|
"Nullable(UInt64)": is_valid_int,
|
|
"UInt64": is_valid_srt,
|
|
|
|
"Nullable(Int64)": is_valid_int,
|
|
"Int64": is_valid_srt,
|
|
|
|
"Array(String)": is_valid_array,
|
|
|
|
"Nullable(Float)": is_valid_float,
|
|
"Float": is_valid_float,
|
|
|
|
}
|