refactor: ファイル展開処理を共通化するにあたり、定数の重複をcommonに一本化。

This commit is contained in:
shimoda.m@nds-tyo.co.jp 2025-06-10 14:10:11 +09:00
parent 2091623df3
commit 6d7917a7f7
3 changed files with 131 additions and 131 deletions

View File

@ -2,57 +2,16 @@ import csv
import io import io
import os import os
import sys import sys
import zipfile
from datetime import datetime from datetime import datetime
import boto3 import boto3
from common import convert_quotechar, debug_log from common import (ERROR, INFO, LINE_FEED_CODE, SETTINGS_ITEM,
convert_quotechar, debug_log, uncompress_zip)
from end import end from end import end
from error import error from error import error
# 定数 # 定数
DIRECTORY_WORK = '/work/' DIRECTORY_WORK = '/work/'
LOCAL_DIRECTORY_TMP = '/tmp'
# チェック処理で解凍した圧縮ファイルの中身を格納するフォルダ
LOCAL_TEMPORARY_FILE_PATH = f'{LOCAL_DIRECTORY_TMP}/temporary_file.dat'
LOG_LEVEL = {'i': 'Info', 'e': 'Error', 'w': "Warning"}
SETTINGS_ITEM = {
'dataSource': 0,
'delimiter': 1,
'charCode': 2,
'quotechar': 3,
'lineFeedCode': 4,
'headerFlag': 5,
'csvNumItems': 6,
'csvNameItems': 7,
'dbColumuName': 8,
'storageSchemaName': 9,
'loadSchemaName': 10,
'exSqlFileName': 11,
'commaReplaceColumns': 12,
'importManner': 13,
'bulkImportFlag': 14,
'compressedFlag': 15,
'compression': 16,
'reserved1': 17,
'reserved2': 18,
'reserved3': 19,
'reserved4': 20,
'reserved5': 21,
'reserved6': 22,
'reserved7': 23,
'reserved8': 24
}
LINE_FEED_CODE = {
'CR': '\r',
'LF': '\n',
'CRLF': '\r\n',
}
INFO = LOG_LEVEL["i"]
ERROR = LOG_LEVEL["e"]
WARNING = LOG_LEVEL["w"]
# クラス変数 # クラス変数
s3_client = boto3.client('s3') s3_client = boto3.client('s3')
@ -236,50 +195,19 @@ def uncompress_file(work_data_file: bytes, settings_list: list, log_info) -> byt
print( print(
f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {INFO} I-CHK-11 - 投入ファイルが圧縮されているため、展開処理を行います。圧縮形式:{compression}') f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {INFO} I-CHK-11 - 投入ファイルが圧縮されているため、展開処理を行います。圧縮形式:{compression}')
file_bytes: bytes = None
if compression == 'zip': if compression == 'zip':
file_bytes = None # zipファイルを展開し、ファイルを書き出す。
with zipfile.ZipFile(work_data_file, 'r') as zip_ref: file_bytes = uncompress_zip()
# 昇順でソートする。
file_list: list[str] = sorted(zip_ref.namelist())
if len(file_list) > 1:
# 圧縮ファイル内に複数ファイルが存在する場合、warningログを出力する。
print(
f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {WARNING} W-CHK-01 - 圧縮データ内に複数ファイルが存在したため、{file_list[0]}のみ登録を行います。')
target_file_name = file_list[0]
# 末尾が「/」で終わるのはフォルダ
if target_file_name.endswith('/'):
# 圧縮ファイル内の先頭がフォルダの場合、エラー処理を行う。
raise Exception(
f'展開したデータはファイルではありません。ファイルパス: {target_file_name}')
# zipファイル内には1ファイルのみ
with zip_ref.open(target_file_name) as file:
file_bytes = file.read()
# ファイルを一時ディレクトリに書き出す。
encoding = settings_list[SETTINGS_ITEM["charCode"]]
line_feed = LINE_FEED_CODE[settings_list[SETTINGS_ITEM["lineFeedCode"]]]
delimiter = settings_list[SETTINGS_ITEM["delimiter"]]
quote_char = convert_quotechar(
settings_list[SETTINGS_ITEM["quotechar"]])
uncompressed_file = io.TextIOWrapper(io.BytesIO(file_bytes), encoding=encoding,
newline=line_feed)
csv_reader = csv.reader(uncompressed_file, quotechar=convert_quotechar(settings_list[SETTINGS_ITEM["quotechar"]]),
delimiter=delimiter)
with open(LOCAL_TEMPORARY_FILE_PATH, 'w', encoding=encoding, newline='') as csvfile:
csv_writer = csv.writer(csvfile, quotechar=quote_char,
delimiter=delimiter)
for row in csv_reader:
csv_writer.writerow(row)
print(
f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {INFO} I-CHK-12 - 投入ファイルの展開が正常終了しました')
return file_bytes
# MEMO: zip以外の圧縮形式に対応する際に追記すること # MEMO: zip以外の圧縮形式に対応する際に追記すること
else: else:
raise ValueError("圧縮形式が一致しません") raise ValueError("圧縮形式が一致しません")
print(
f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {INFO} I-CHK-12 - 投入ファイルの展開が正常終了しました')
return file_bytes
def reverse_readline_stream(f: io.BytesIO, line_feed: str, chunk_size=4096): def reverse_readline_stream(f: io.BytesIO, line_feed: str, chunk_size=4096):
"""バイトモードのファイルストリームを後ろから読み、1行ずつ返すジェネレータ。 """バイトモードのファイルストリームを後ろから読み、1行ずつ返すジェネレータ。

View File

@ -1,16 +1,80 @@
import csv
import io
import zipfile
from datetime import datetime from datetime import datetime
# 定数 # 定数
LOG_LEVEL = {"d": 'Debug'} LOG_LEVEL = {
"d": 'Debug',
'i': 'Info',
'e': 'Error',
'w': "Warning"
}
INFO = LOG_LEVEL["i"]
ERROR = LOG_LEVEL["e"]
WARNING = LOG_LEVEL["w"]
LINE_FEED_CODE = {
'CR': '\r',
'LF': '\n',
'CRLF': '\r\n',
}
# LOAD DATA文で文字コードを指定するために、個別設定ファイルの文字コード指定をMySQLの文字コード表記に変換する
MYSQL_CHARSET_CODE = {
'utf-8': 'utf8mb4',
'utf8': 'utf8mb4',
'utf-8-sig': 'utf8mb4',
'shift_jis': 'cp932',
'cp932': 'cp932',
}
MODE_TYPE = { MODE_TYPE = {
'n': 'normal', 'n': 'normal',
'd': 'debug', 'd': 'debug',
} }
# 設定ファイルのの項目行数のマップ
SETTINGS_ITEM = {
'dataSource': 0,
'delimiter': 1,
'charCode': 2,
'quotechar': 3,
'lineFeedCode': 4,
'headerFlag': 5,
'csvNumItems': 6,
'csvNameItems': 7,
'dbColumuName': 8,
'storageSchemaName': 9,
'loadSchemaName': 10,
'exSqlFileName': 11,
'commaReplaceColumns': 12,
'importManner': 13,
'bulkImportFlag': 14,
'compressedFlag': 15,
'compression': 16,
'reserved1': 17,
'reserved2': 18,
'reserved3': 19,
'reserved4': 20,
'reserved5': 21,
'reserved6': 22,
'reserved7': 23,
'reserved8': 24
}
LOCAL_DIRECTORY_TMP = '/tmp'
# チェック処理で解凍した圧縮ファイルの中身を格納するフォルダ
LOCAL_TEMPORARY_FILE_PATH = f'{LOCAL_DIRECTORY_TMP}/temporary_file.dat'
def debug_log(log, log_info, mode): def debug_log(log, log_info, mode):
if MODE_TYPE['d'] == mode: if MODE_TYPE['d'] == mode:
print(f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["d"]} {log}') print(
f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {LOG_LEVEL["d"]} {log}')
def convert_quotechar(quotechar): def convert_quotechar(quotechar):
"""csvモジュールの囲い文字を変換する """csvモジュールの囲い文字を変換する
@ -26,3 +90,57 @@ def convert_quotechar(quotechar):
return None return None
return quotechar return quotechar
def uncompress_zip(work_data_file: bytes, settings_list: list, log_info) -> bytes:
"""zip圧縮されているファイルを展開する
Args:
work_data_file (bytes): 展開対象のバイナリ
settings_list (list): 個別設定ファイルの情報
log_info (_type_): ログ情報
Raises:
Exception: 展開したものがファイルではなかった場合
Returns:
bytes: 展開後ファイルのバイナリ
"""
file_bytes = None
with zipfile.ZipFile(work_data_file, 'r') as zip_ref:
# 昇順でソートする。
file_list: list[str] = sorted(zip_ref.namelist())
if len(file_list) > 1:
# 圧縮ファイル内に複数ファイルが存在する場合、warningログを出力する。
print(
f'{datetime.now():%Y-%m-%d %H:%M:%S} {log_info} {WARNING}' +
f'W-CHK-01 - 圧縮データ内に複数ファイルが存在したため、{file_list[0]}のみ登録を行います。'
)
target_file_name = file_list[0]
# 末尾が「/」で終わるのはフォルダ
if target_file_name.endswith('/'):
# 圧縮ファイル内の先頭がフォルダの場合、エラー処理を行う。
raise Exception(
f'展開したデータはファイルではありません。ファイルパス: {target_file_name}')
# zipファイル内には1ファイルのみ
with zip_ref.open(target_file_name) as file:
file_bytes = file.read()
# ファイルを一時ディレクトリに書き出す。
encoding = settings_list[SETTINGS_ITEM["charCode"]]
line_feed = LINE_FEED_CODE[settings_list[SETTINGS_ITEM["lineFeedCode"]]]
delimiter = settings_list[SETTINGS_ITEM["delimiter"]]
quote_char = convert_quotechar(
settings_list[SETTINGS_ITEM["quotechar"]])
uncompressed_file = io.TextIOWrapper(io.BytesIO(file_bytes), encoding=encoding,
newline=line_feed)
csv_reader = csv.reader(uncompressed_file, quotechar=convert_quotechar(settings_list[SETTINGS_ITEM["quotechar"]]),
delimiter=delimiter)
with open(LOCAL_TEMPORARY_FILE_PATH, 'w', encoding=encoding, newline='') as csvfile:
csv_writer = csv.writer(csvfile, quotechar=quote_char,
delimiter=delimiter)
for row in csv_reader:
csv_writer.writerow(row)

View File

@ -5,65 +5,19 @@ from datetime import datetime
import boto3 import boto3
import pymysql import pymysql
from chk import LOCAL_TEMPORARY_FILE_PATH from chk import LOCAL_TEMPORARY_FILE_PATH
from common import convert_quotechar, debug_log from common import (ERROR, INFO, LINE_FEED_CODE, MYSQL_CHARSET_CODE,
SETTINGS_ITEM, WARNING, convert_quotechar, debug_log)
from error import error from error import error
from pymysql.constants import CLIENT from pymysql.constants import CLIENT
# 定数 # 定数
DIRECTORY_WORK = '/work/' DIRECTORY_WORK = '/work/'
LOG_LEVEL = {"i": 'Info', "e": 'Error', "w": 'Warning'}
SETTINGS_ITEM = {
'dataSource': 0,
'delimiter': 1,
'charCode': 2,
'quotechar': 3,
'lineFeedCode': 4,
'headerFlag': 5,
'csvNumItems': 6,
'csvNameItems': 7,
'dbColumuName': 8,
'storageSchemaName': 9,
'loadSchemaName': 10,
'exSqlFileName': 11,
'commaReplaceColumns': 12,
'importManner': 13,
'bulkImportFlag': 14,
'compressedFlag': 15,
'compression': 16,
'reserved1': 17,
'reserved2': 18,
'reserved3': 19,
'reserved4': 20,
'reserved5': 21,
'reserved6': 22,
'reserved7': 23,
'reserved8': 24
}
LINE_FEED_CODE = {
'CR': '\r',
'LF': '\n',
'CRLF': '\r\n',
}
# LOAD DATA文で文字コードを指定するために、個別設定ファイルの文字コード指定をMySQLの文字コード表記に変換する
MYSQL_CHARSET_CODE = {
'utf-8': 'utf8mb4',
'utf8': 'utf8mb4',
'utf-8-sig': 'utf8mb4',
'shift_jis': 'cp932',
'cp932': 'cp932',
}
DIRECTORY_SETTINGS = '/settings/' DIRECTORY_SETTINGS = '/settings/'
TRUNCATE_SRC_TABLE_SYMBOL = 'truncate_src_table:' TRUNCATE_SRC_TABLE_SYMBOL = 'truncate_src_table:'
TRUNCATE_SRC_TABLE_IDENTIFY_SYMBOL_FORMAT = f'{TRUNCATE_SRC_TABLE_SYMBOL}[蓄積スキーマのテーブル名]' TRUNCATE_SRC_TABLE_IDENTIFY_SYMBOL_FORMAT = f'{TRUNCATE_SRC_TABLE_SYMBOL}[蓄積スキーマのテーブル名]'
INVALID_CONFIG_EXCEPTION_MESSAGE = f'個別設定ファイルのインポート方法に不備がありました。 インポート方法は "{TRUNCATE_SRC_TABLE_IDENTIFY_SYMBOL_FORMAT}" のように設定してください' INVALID_CONFIG_EXCEPTION_MESSAGE = f'個別設定ファイルのインポート方法に不備がありました。 インポート方法は "{TRUNCATE_SRC_TABLE_IDENTIFY_SYMBOL_FORMAT}" のように設定してください'
INFO = LOG_LEVEL['i']
WARNING = LOG_LEVEL['w']
ERROR = LOG_LEVEL['e']
# クラス変数 # クラス変数
s3_client = boto3.client('s3') s3_client = boto3.client('s3')