62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""数据类型解析器"""
|
||
from datetime import datetime
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
from dateutil import parser as dtparser
|
||
from zoneinfo import ZoneInfo
|
||
|
||
class TypeParser:
|
||
"""类型解析工具"""
|
||
|
||
@staticmethod
|
||
def parse_timestamp(s: str, tz: ZoneInfo) -> datetime | None:
|
||
"""解析时间戳"""
|
||
if s is None:
|
||
return None
|
||
try:
|
||
# 区分 null 与 0：0 视为 Unix 时间戳，不当作空值。
|
||
if isinstance(s, (int, float)) and not isinstance(s, bool):
|
||
ts = float(s)
|
||
if abs(ts) >= 1_000_000_000_000:
|
||
ts = ts / 1000.0
|
||
return datetime.fromtimestamp(ts, tz=tz)
|
||
|
||
text = str(s).strip()
|
||
if text == "":
|
||
return None
|
||
|
||
dt = dtparser.parse(text)
|
||
if dt.tzinfo is None:
|
||
return dt.replace(tzinfo=tz)
|
||
return dt.astimezone(tz)
|
||
except Exception:
|
||
return None
|
||
|
||
@staticmethod
|
||
def parse_decimal(value, scale: int = 2) -> Decimal | None:
|
||
"""解析金额"""
|
||
if value is None:
|
||
return None
|
||
try:
|
||
d = Decimal(str(value))
|
||
return d.quantize(Decimal(10) ** -scale, rounding=ROUND_HALF_UP)
|
||
except Exception:
|
||
return None
|
||
|
||
@staticmethod
|
||
def parse_int(value) -> int | None:
|
||
"""解析整数"""
|
||
if value is None:
|
||
return None
|
||
try:
|
||
return int(value)
|
||
except Exception:
|
||
return None
|
||
|
||
@staticmethod
|
||
def format_timestamp(dt: datetime | None, tz: ZoneInfo) -> str | None:
|
||
"""格式化时间戳"""
|
||
if not dt:
|
||
return None
|
||
return dt.astimezone(tz).strftime("%Y-%m-%d %H:%M:%S")
|