import json
import openpyxl as pyxl
from openpyxl.workbook.workbook import Workbook
# from openpyxl.worksheet.worksheet import Worksheet
from openpyxl.utils import range_to_tuple
from openpyxl.utils import column_index_from_string
def MakeDDL(wb: Workbook, scanArea: dict[str, str]):
# 走査範囲からシート名、セル範囲を取得
sheetCellRef = scanArea["sheetCellRef"]
(wsName, (minCol, minRow, maxCol, maxRow)) = range_to_tuple(sheetCellRef)
# シート取得
# ws = wb.worksheets[0]
ws = wb[wsName]
if minCol is None:
print(f"EXCELシートの走査範囲不正です。\nJSONファイル内「sheetCellRef」キーを確認して下さい: {sheetCellRef}")
return
else:
# セル番地(列[アルファベット])のインデックス[数値]変換
T = column_index_from_string(scanArea["lColName"]) - minCol
AB = column_index_from_string(scanArea["pColName"]) - minCol
AJ = column_index_from_string(scanArea["dataType"]) - minCol
AN = column_index_from_string(scanArea["wLength"]) - minCol
AP = column_index_from_string(scanArea["fLength"]) - minCol
AR = column_index_from_string(scanArea["pk"]) - minCol
AU = column_index_from_string(scanArea["notNull"]) - minCol
AX = column_index_from_string(scanArea["id"]) - minCol
BA = column_index_from_string(scanArea["defVal"]) - minCol
# スキーマ名、テーブル名
schemaName = str(ws[scanArea["schemaName"]].value)
logicTabName = str(ws[scanArea["lTabName"]].value)
physTabName = str(ws[scanArea["pTabName"]].value)
bldr1: list[str] = []
bldr2: list[str] = []
pks: list[str] = []
execStr = "EXEC sp_addextendedproperty @name = N'MS_Description', @value = N'{}', @level0type = N'SCHEMA', @level0name = N'{}', @level1type = N'TABLE', @level1name = N'{}'"
# 行ループ処理
for row in ws.iter_rows(min_row=minRow, max_row=maxRow, min_col=minCol, max_col=maxCol):
# 行内にデータありセルがない場合、処理をスキップ
if len([cell for cell in row if cell.value is not None and str(cell.value).strip() != ""]) == 0:
continue
# for cell in row:
# if isinstance(cell, pyxl.cell.cell.MergedCell):
# continue
# print(f"{cell.coordinate}[{cell.column}]: {cell.value}")
# print(str(row[T].value)) # logicColName
# print(str(row[AB].value)) # physColName
# print(str(row[AJ].value)) # dataType
# print(str(row[AN].value)) # wholeLength
# print(str(row[AP].value)) # fractionLength
# print(str(row[AR].value)) # primaryKey
# print(str(row[AU].value)) # notNull
# print(str(row[AX].value)) # identity
# print(str(row[BA].value)) # defaultVal
# カラム名
logicColName = str(row[T].value)
physColName = str(row[AB].value)
# 桁数(小数)
# if row[AP].value is None or str(row[AP].value).strip() == "" or row[AP].value == 0:
# fractionLength = ""
# else:
# fractionLength = f", {row[AP].value}"
if str(row[AP].value).strip().isdigit() is True:
fractionLength = f", {row[AP].value}"
else:
fractionLength = ""
# データ型(+桁数)
# wholeLength = row[AN].value
# if wholeLength is None or str(wholeLength).strip() == "":
# dataType = f"{row[AJ].value} "
# else:
# dataType = f"{row[AJ].value}({wholeLength}{fractionLength}) "
wholeLength = row[AN].value
if str(wholeLength).strip().isdigit() is True:
dataType = f"{row[AJ].value}({wholeLength}{fractionLength}) "
else:
dataType = f"{row[AJ].value} "
# 主キー
if str(row[AR].value) == "○":
pks.append(f"[{physColName}]")
# NOT NULL制約
if str(row[AU].value) == "○":
notNull = "NOT NULL "
else:
notNull = ""
# IDENTITY制約
if str(row[AX].value) == "○":
identity = "IDENTITY(1, 1) "
else:
identity = ""
# デフォルト値
if row[BA].value is not None and str(row[BA].value).strip() != "":
defaultVal = f"DEFAULT {row[BA].value}"
else:
defaultVal = ""
# CREATE TABLE文のカラム定義部
# カラム論理名の設定文の作成
bldr1.append(f"[{physColName}] {dataType}{identity}{notNull}{defaultVal}")
bldr2.append(f"{execStr.format(logicColName, schemaName, physTabName)}, @level2type = N'COLUMN', @level2name = N'{physColName}';")
# DDL作成、表示
ctStr = "\n , ".join(bldr1)
pkStr = ", ".join(pks)
cmtStr = "\n".join(bldr2)
print(f"---------- 【DDL作成】テーブル名: [{schemaName}].[{physTabName}] ----------\n")
print(f"-- シート名: {ws.title}")
print(f"-- 行数: {ws.max_row}, 列数: {ws.max_column}\n")
print(f"DROP TABLE IF EXISTS [{schemaName}].[{physTabName}];\n")
print(f"CREATE TABLE [{schemaName}].[{physTabName}] (\n {ctStr}\n);\n")
print(f"ALTER TABLE [{schemaName}].[{physTabName}] ADD CONSTRAINT [PK_{physTabName}] PRIMARY KEY ({pkStr});\n")
print(f"{execStr.format(logicTabName, schemaName, physTabName)};\n{cmtStr}\n")
# print(f"--------------------------------------------------")
if __name__ == "__main__":
# ユーザ入力
jsonFile = input("JSONファイル: ")
# JSONファイルの読み込み
with open(jsonFile, "r", encoding="UTF-8") as f:
data = json.load(f)
# EXCELファイル読み込み
wb = pyxl.load_workbook(data["excelFile"])
scanAreas: list[dict[str, str]] = data["scanAreas"]
# シート単位の走査範囲毎に処理を実行
for sArea in scanAreas:
MakeDDL(wb, sArea)
{
"excelFile": "./EXCEL/DbSpec.xlsx",
"scanAreas": [
{
"sheetCellRef": "ユーザ情報テーブル!T6:BI15",
"schemaName": "P2",
"lTabName": "D6",
"pTabName": "L6",
"lColName": "T",
"pColName": "AB",
"dataType": "AJ",
"wLength": "AN",
"fLength": "AP",
"pk": "AR",
"notNull": "AU",
"id": "AX",
"defVal": "BA"
},
{
"sheetCellRef": "bk_ユーザ情報テーブル!W8:BL17",
"schemaName": "P2",
"lTabName": "G8",
"pTabName": "O8",
"lColName": "W",
"pColName": "AE",
"dataType": "AM",
"wLength": "AQ",
"fLength": "AS",
"pk": "AU",
"notNull": "AX",
"id": "BA",
"defVal": "BD"
}
]
}