76 lines
2.8 KiB
Python
76 lines
2.8 KiB
Python
import csv
|
|
import io
|
|
|
|
from src.config.objects import TargetObject
|
|
from src.converter.convert_strategy import ConvertStrategyFactory
|
|
|
|
|
|
class CSVStringConverter:
|
|
def __init__(self, target_object: TargetObject, sf_object_jsons: dict) -> None:
|
|
self.__target_object = target_object
|
|
self.__sf_object_jsons = sf_object_jsons
|
|
self.__convert_strategy_factory = ConvertStrategyFactory()
|
|
|
|
def convert(self) -> str:
|
|
extracted_sf_object_jsons = self.__extract_sf_object_jsons()
|
|
csv_data = self.__convert_to_csv(extracted_sf_object_jsons)
|
|
csv_string = self.__write_csv_string(csv_data)
|
|
return csv_string
|
|
|
|
def __extract_sf_object_jsons(self) -> list:
|
|
try:
|
|
extracted_sf_object_jsons = []
|
|
for sf_object_json in self.__sf_object_jsons:
|
|
extracted_sf_object_jsons.append(
|
|
self.__extract_necessary_props_from(sf_object_json))
|
|
|
|
return extracted_sf_object_jsons
|
|
|
|
except Exception as e:
|
|
raise Exception('必要なjsonのデータ成形に失敗しました', e)
|
|
|
|
def __extract_necessary_props_from(self, sf_object_json) -> dict:
|
|
clone_sf_object = {**sf_object_json}
|
|
|
|
del clone_sf_object['attributes']
|
|
|
|
uppercase_key_sf_object = {
|
|
k.upper(): v for k, v in clone_sf_object.items()}
|
|
|
|
return uppercase_key_sf_object
|
|
|
|
def __convert_to_csv(self, extracted_sf_object_jsons) -> list:
|
|
try:
|
|
columns = self.__target_object.columns
|
|
csv_data = []
|
|
for i, json_object in enumerate(extracted_sf_object_jsons, 1):
|
|
csv_row = []
|
|
for column in columns:
|
|
v = json_object[column.upper()]
|
|
|
|
convert_strategy = self.__convert_strategy_factory.create(v)
|
|
converted_value = convert_strategy.convert_value(v)
|
|
|
|
csv_row.append(converted_value)
|
|
|
|
csv_data.append(csv_row)
|
|
return csv_data
|
|
|
|
except Exception as e:
|
|
raise Exception(
|
|
f'CSV変換に失敗しました カラム名：[{column}] 行番号: [{i}] エラー内容:[{e}]')
|
|
|
|
def __write_csv_string(self, csv_data) -> str:
|
|
try:
|
|
with io.StringIO(newline='') as string_stream:
|
|
writer = csv.writer(string_stream, delimiter=',', lineterminator='\r\n',
|
|
doublequote=True, quotechar='"', quoting=csv.QUOTE_ALL, strict=True)
|
|
writer.writerow(self.__target_object.columns)
|
|
writer.writerows(csv_data)
|
|
csv_value = string_stream.getvalue()
|
|
|
|
return csv_value
|
|
|
|
except Exception as e:
|
|
raise Exception('CSVデータの出力に失敗しました', e)
|