82 lines
3.1 KiB
Python
82 lines
3.1 KiB
Python
import json
|
|
import re
|
|
from collections import OrderedDict
|
|
from datetime import datetime
|
|
|
|
from dateutil.tz import gettz
|
|
from src.system_var.constants import (CRM_DATETIME_FORMAT, CSV_FALSE_VALUE,
|
|
CSV_TRUE_VALUE,
|
|
DATE_PATTERN_YYYYMMDDHHMMSSFFF_UTC,
|
|
YYYYMMDDHHMMSS)
|
|
from src.system_var.environments import CONVERT_TZ
|
|
|
|
|
|
class ConvertStrategyFactory:
|
|
def __init__(self) -> None:
|
|
self.__none_value_convert_strategy = NoneValueConvertStrategy()
|
|
self.__boolean_convert_strategy = BooleanConvertStrategy()
|
|
self.__datetime_convert_strategy = DatetimeConvertStrategy()
|
|
self.__int_convert_strategy = IntConvertStrategy()
|
|
self.__string_convert_strategy = StringConvertStrategy()
|
|
self.__dict_convert_strategy = DictConvertStrategy()
|
|
|
|
def create(self, value):
|
|
|
|
if value is None:
|
|
convert_strategy = self.__none_value_convert_strategy
|
|
|
|
elif type(value) == bool:
|
|
convert_strategy = self.__boolean_convert_strategy
|
|
|
|
elif type(value) == str and re.fullmatch(DATE_PATTERN_YYYYMMDDHHMMSSFFF_UTC, value):
|
|
convert_strategy = self.__datetime_convert_strategy
|
|
|
|
elif type(value) == int:
|
|
convert_strategy = self.__int_convert_strategy
|
|
|
|
elif type(value) == dict or type(value) == OrderedDict:
|
|
convert_strategy = self.__dict_convert_strategy
|
|
else:
|
|
convert_strategy = self.__string_convert_strategy
|
|
|
|
return convert_strategy
|
|
|
|
|
|
class NoneValueConvertStrategy:
|
|
def convert_value(self, convert_value: None) -> str:
|
|
"""Noneを''空文字に変換する処理"""
|
|
return ''
|
|
|
|
|
|
class BooleanConvertStrategy:
|
|
def convert_value(self, convert_value: str) -> bool:
|
|
"""booleanを数値に変換する処理"""
|
|
return CSV_TRUE_VALUE if convert_value is True else CSV_FALSE_VALUE
|
|
|
|
|
|
class DatetimeConvertStrategy:
|
|
def convert_value(self, convert_value: str) -> str:
|
|
"""UTCのdatetime文字列をJSTの日時文字列に変換する処理"""
|
|
# データ登録処理がJSTとして登録するため、変換処理内で事前にJSTの日時文字列に変換する
|
|
return datetime.strptime(convert_value, CRM_DATETIME_FORMAT).astimezone(gettz(CONVERT_TZ)).strftime(YYYYMMDDHHMMSS)
|
|
|
|
|
|
class IntConvertStrategy:
|
|
def convert_value(self, convert_value: int):
|
|
"""int型を変換せずに返す処理"""
|
|
# ConvertStrategyFactoryにて型チェックを行っているため値を変換せずに返す
|
|
return convert_value
|
|
|
|
|
|
class StringConvertStrategy:
|
|
def convert_value(self, convert_value: str):
|
|
"""string型を変換せずに返す処理"""
|
|
# ConvertStrategyFactoryにて型チェックを行っているため値を変換せずに返す
|
|
return convert_value
|
|
|
|
|
|
class DictConvertStrategy:
|
|
def convert_value(self, convert_value: dict):
|
|
"""dict型の項目を文字列に変換して返す処理"""
|
|
return json.dumps(convert_value, ensure_ascii=False)
|