マルチインデックスをもつエクセルファイルの差分をセル単位で抽出するツール

はじめに

こんにちは、品質管理部の沖です。

OS のバージョンの違いにおける製品への影響を調査・検証する業務において、複数の観点による比較のために行と列が多重になった表（マルチインデックスをもつ表）を用いることがあります。資料 [1] では、マルチインデックスをもつエクセルファイルの表を効率的に作成するツールを紹介しました。

本記事では、上記のツールで作成した表に値を書き込んだ2つのエクセルファイルの差分をセル単位で抽出する方法を紹介します。エクセルファイルの内容を日々更新した際に、日付単位でファイルを保存して本ツールを使用することで、「いつ」「どのセル」を変更したかをトレースできます。

例

図1 のエクセルファイルと図 2 のエクセルファイルを比較し、赤い文字がセル単位の差分です。これを抽出します。

図 1. エクセルファイル1

図 2. エクセルファイル2

図 3. Slack によるエクセルファイルのセル単位の差分通知です。行インデックスと列インデックスのラベルのリストによるセル指定と差分がある値が確認できます。

課題

エクセルファイルを CSV にエクスポートして diff ファイルで差分を抽出することは可能ですが、行単位の比較になり、セル単位の比較ができません。
WinMerge という既存のツールにより、エクセルファイルの表を画像比較してセル単位の差分を目視で確認できますが、差分があるセルの値とインデックスを文字列としてプログラムに取り込むことが困難です。
エクセルファイルがマルチインデックスをもつ場合、差分があるセルとそのインデックスを特定することが困難です。

次に記載のような、マルチインデックスの具体例を考えます。

行インデックス: 「制御 1～5」の 5 種類、「制御 A～E」の 5 種類としたとき、5 * 5 = 25 通り。
列インデックス: 「機種種別α」が 5 種類、「機種種別β」が 5 種類、「製品の種別」が 3 種類、「OS 種別」が 2 種類としたとき、5 * 5 * 3 * 2 = 150 通り。

合計 25 * 150 = 3750 個のセルをもつ 2 個のエクセルファイルに対して、10 個のセル単位の差分があると仮定します。このとき、目視でそのセルとマルチインデックスの特定を試行すると、早くて約 1 日がかりの作業になります。

ツールによる解決

マルチインデックスをもつエクセルファイルの差分をセル単位で抽出するツールを作成しました。具体的な行インデックスと列インデックスを示し、値の差分を表示します。
さらに、差分比較の結果を Slack に通知する仕組みを追加しました。これにより、差分の結果を即時に周知できます。

Python のテストフレームワーク pytest に基づき作成しました。実際の値と期待値を比較して結果を出力します。

課題1と課題2は、pytestを用いてセル単位の妥当性を検証することで解決しました。課題3は、Slackの通知を加工して差分があるセルの具体的な行インデックスと列インデックスを表示することで解決しました。

今後の応用的な展開としては、テスト自動化における実際の値と期待値との突合、その結果の通知に本ツールの一部を改修して使用予定です。

ソースコード

MOTEX 社内の AI サービスと相談しながら作成しました。所要時間は生成と微調整で30分ほどです。また、生成の際に資料 [1] のツールに記載のコードもプロンプトに含めました。本ツールで作成したエクセルファイルに基づくという条件を付与するためです。

差分抽出用の excel_validation.py と Slack 通知用の conftest.py の 2 種のソースコードからなります。

pytest excel_validation.py

excel_validation.py

Slack 通知の事前準備として、Slack Apps を作成し、その中で Incoming Webhook 機能を利用可能にすることが必要です。

次に記載のソースコードの https://hooks.slack.com/services/xxxを適切なものに置き換えるとご利用いただけます。

import pandas as pd
import pytest
import requests

def auto_detect_levels(filename, max_levels=5):
    """
    streamlit アプリで出力された Excel ファイルにおいて、
    行インデックスのレベル数（ROW_INDEX_LEVELS）および
    列インデックスのレベル数（COL_INDEX_LEVELS）を自動検出します。
    
    Excel ファイルから header=None で読み込んだ後、可能な組み合わせ (row_levels, col_levels)
    について pd.read_excel を試し、インデックス名がそれぞれ "A…"（行側）および "B…"（列側）
    で始まるものを採用します。max_levels はレベル数の上限です。
    
    見つかった組み合わせと DataFrame を返します。
    """
    for row_levels in range(1, max_levels+1):
        for col_levels in range(1, max_levels+1):
            try:
                df_try = pd.read_excel(
                    filename, 
                    header=list(range(col_levels)),
                    index_col=list(range(row_levels)),
                    engine="openpyxl"
                )
                # MultiIndex では index.names / columns.names がタプルまたはリストとして取得される
                idx_names = df_try.index.names
                col_names = df_try.columns.names
                # streamlit アプリでは、行側インデックス名は "A1", "A2", …、列側は "B1", "B2", … と設定されている前提です。
                if (all(isinstance(name, str) and name.startswith("A") for name in idx_names)
                    and all(isinstance(name, str) and name.startswith("B") for name in col_names)):
                    return row_levels, col_levels, df_try
            except Exception:
                continue
    raise ValueError("Excel ファイルから MultiIndex のレベル数を自動検出できませんでした。")

def read_excel(filename):
    """
    Excel ファイルを読み込み、DataFrame（MultiIndex付き）として返す関数です。
    マルチインデックスのレベル数は、Excel ファイル内のインデックス名（"A…" / "B…"）
    を手がかりに自動検出されます。
    
    この Excel ファイルは、streamlit アプリで作成されたものを前提としています。
    """
    _, _, df = auto_detect_levels(filename)
    return df

# テスト対象の Excel ファイルのパス（適宜変更してください）
EXCEL1_FILENAME = 'excel1.xlsx'
EXCEL2_FILENAME = 'excel2.xlsx'

# モジュールレベルで Excel ファイルを読み込み
df1 = read_excel(EXCEL1_FILENAME)
df2 = read_excel(EXCEL2_FILENAME)

# 形状（行数・列数）が一致しているかチェック
assert df1.shape == df2.shape, (
    f"エクセルファイルの形状が一致していません。 ファイル1: {df1.shape}, ファイル2: {df2.shape}"
)

def format_index(idx):
    """
    マルチインデックス（または単層インデックス）を文字列に変換します。
    例：タプルの場合は (A2,A5) 形式の文字列にします。
    """
    if isinstance(idx, tuple):
        return "(" + ",".join(map(str, idx)) + ")"
    else:
        return f"({idx})"

# 各セルごとにテスト対象とするため、df1.index と df1.columns の直積で parameterize します
@pytest.mark.parametrize("row, col", [
    (row, col)
    for row in df1.index
    for col in df1.columns
])
def test_excel_cell(row, col):
    """
    エクセルファイル1 と エクセルファイル2 の各セル（マルチインデックス）の値が一致するか
    を検証するテストです。不一致の場合、行・列それぞれのマルチインデックスは
    (A2,A5) や (B1,B4) の形式で表示されます。
    """
    cell1 = df1.loc[row, col]
    cell2 = df2.loc[row, col]
    row_str = format_index(row)
    col_str = format_index(col)
    assert cell1 == cell2, (
        f"セルの不一致発生: 行 {row_str} 列 {col_str} で、"
        f"エクセルファイル1 の値は '{cell1}'、エクセルファイル2 の値は '{cell2}' です。"
    )

# Slack 通知用の関数
def slack_notify(text):
    """
    引数 text の内容を Slack の Webhook 経由で通知する関数です。
    ※ Webhook URL は環境に合わせて適宜変更してください。
    """
    url = "https://hooks.slack.com/services/xxx"  # 実際の URL に変更すること
    payload = {"text": text}
    headers = {"Content-type": "application/json"}
    try:
        response = requests.post(url, json=payload, headers=headers)
        if response.status_code != 200:
            print("Slack 通知に失敗しました。ステータスコード:", response.status_code)
            print("Response:", response.text)
    except Exception as e:
        print("Slack 通知中に例外発生:", e)

if __name__ == "__main__":
    pytest.main()

conftest.py

from excel_validation import slack_notify

def pytest_terminal_summary(terminalreporter, exitstatus, config):
    """
    不一致のセルについて Slack に通知します。
    """
    result_lines = ["セルの差分の抽出結果:"]
    total = sum(len(terminalreporter.stats.get(status, []))
                for status in ('passed', 'failed'))
    result_lines.append(f"総数: {total}")
    
    passed = len(terminalreporter.stats.get('passed', []))
    failed = len(terminalreporter.stats.get('failed', []))
    result_lines.append(f"　一致: {passed} 件")
    result_lines.append(f"　不一致: {failed} 件")
    result_lines.append("")
    
    # セルの値に不一致があった場合、"E" で始まる行のみを抜粋
    if failed:
        result_lines.append("【不一致だったセルの詳細】")
        for status in ('failed', 'error'):
            for report in terminalreporter.stats.get(status, []):
                # report.longreprtext があれば利用、なければ str(report.longrepr) を利用
                detailed_error = getattr(report, "longreprtext", str(report.longrepr)).splitlines()
                error_line = next((line for line in detailed_error if line.startswith("E")), detailed_error[0])
                result_lines.append(error_line.replace("E       AssertionError: ", ""))
    
    message_text = "\n".join(result_lines)
    slack_notify(message_text)