[NOTE] Excelデータベース更新プログラム

Excelのデータベースを更新する際、複数の任意の列が一致する場合、マスターファイルの行を上書き、そうでない場合はマスターファイルの行を追記するプログラム

"""
from openpyxl import load_workbook
from copy import copy

def update_excel_database(
path_db,
path_import,
key_columns,
sheet_name=None,
):
"""
ExcelマスターDBをインポートファイルで更新する。

Parameters
----------
path_db : str
    マスターとなるExcelファイル
path_import : str
    読み込み用Excelファイル
key_columns : list[str]
    一致判定に使用する列名
    例: ["案件番号", "顧客名"]
sheet_name : str | None
    対象シート名。
    Noneの場合は各Excelのアクティブシートを使用
"""

# ==========================================================
# 1. Excel読み込み
# ==========================================================

wb_db = load_workbook(path_db)
wb_import = load_workbook(path_import, data_only=False)

# シート取得
if sheet_name is None:
    ws_db = wb_db.active
    ws_import = wb_import.active
else:
    ws_db = wb_db[sheet_name]
    ws_import = wb_import[sheet_name]

# ==========================================================
# 2. ヘッダー取得
# ==========================================================

db_headers = {
    cell.value: cell.column
    for cell in ws_db[1]
    if cell.value is not None
}

import_headers = {
    cell.value: cell.column
    for cell in ws_import[1]
    if cell.value is not None
}

# キー列の存在確認
for key in key_columns:

    if key not in db_headers:
        raise ValueError(
            f"マスター側にキー列「{key}」がありません。"
        )

    if key not in import_headers:
        raise ValueError(
            f"読み込み側にキー列「{key}」がありません。"
        )

# ==========================================================
# 3. import側にしか存在しない列をマスターに追加
# ==========================================================

for header in import_headers:

    if header not in db_headers:

        new_col = ws_db.max_column + 1

        ws_db.cell(
            row=1,
            column=new_col,
            value=header
        )

        db_headers[header] = new_col

# ==========================================================
# 4. マスター側のキー → 行番号 の辞書を作成
# ==========================================================

db_index = {}

for row in range(2, ws_db.max_row + 1):

    key = tuple(
        ws_db.cell(
            row=row,
            column=db_headers[col]
        ).value
        for col in key_columns
    )

    db_index[key] = row

# ==========================================================
# 5. import側を1行ずつ処理
# ==========================================================

update_count = 0
append_count = 0

for import_row in range(2, ws_import.max_row + 1):

    # ----------------------------------------------
    # キーを作成
    # ----------------------------------------------

    key = tuple(
        ws_import.cell(
            row=import_row,
            column=import_headers[col]
        ).value
        for col in key_columns
    )

    # ----------------------------------------------
    # 空キーはスキップ
    # ----------------------------------------------

    if all(
        value is None or value == ""
        for value in key
    ):
        continue

    # ==================================================
    # 6. 一致する行が存在する場合
    # ==================================================

    if key in db_index:

        db_row = db_index[key]

        # import側の全列をコピー
        for header, import_col in import_headers.items():

            db_col = db_headers[header]

            source_cell = ws_import.cell(
                row=import_row,
                column=import_col
            )

            target_cell = ws_db.cell(
                row=db_row,
                column=db_col
            )

            # 値をコピー
            target_cell.value = source_cell.value

            # ハイパーリンク
            if source_cell.hyperlink:

                target_cell._hyperlink = copy(
                    source_cell.hyperlink
                )
            else:

                target_cell._hyperlink = None

            # 書式
            if source_cell.has_style:
                target_cell._style = copy(
                    source_cell._style
                )

            # number format
            target_cell.number_format = source_cell.number_format

            # コメント
            if source_cell.comment:
                target_cell.comment = copy(
                    source_cell.comment
                )
            else:
                target_cell.comment = None

        update_count += 1

    # ==================================================
    # 7. 一致する行が存在しない場合
    # ==================================================

    else:

        new_row = ws_db.max_row + 1

        for header, import_col in import_headers.items():

            db_col = db_headers[header]

            source_cell = ws_import.cell(
                row=import_row,
                column=import_col
            )

            target_cell = ws_db.cell(
                row=new_row,
                column=db_col
            )

            # 値
            target_cell.value = source_cell.value

            # ハイパーリンク
            if source_cell.hyperlink:

                target_cell._hyperlink = copy(
                    source_cell.hyperlink
                )

            # 書式
            if source_cell.has_style:
                target_cell._style = copy(
                    source_cell._style
                )

            # number format
            target_cell.number_format = source_cell.number_format

            # コメント
            if source_cell.comment:
                target_cell.comment = copy(
                    source_cell.comment
                )

        # 作成したキーを辞書にも追加
        db_index[key] = new_row

        append_count += 1

# ==========================================================
# 8. マスターExcelを保存
# ==========================================================

wb_db.save(path_db)

# ==========================================================
# 9. 結果
# ==========================================================

print("処理完了")
print(f"更新件数 : {update_count}")
print(f"追加件数 : {append_count}")