はじめに

私は、2026年4月からエンジニア転職をした、駆け出しエンジニアです。
現在自社アプリの開発案件に参画しています。
開発以外のタスクを行うこともあり、その中で、Pythonを使って効率化できそうな部分があったので
自身の学習も兼ねて効率化ツールを作成しました。

なぜこのツールを作成したのか

開発の合間で、特定のユーザーに対してDBにUPDATE文を直接流し込むタスクが定期的にあります。
現状のタスクの手順としては下記のような流れで行っております。

呼び方	内容
作業ファイル	今回UPDATE対象となるユーザー情報が入ったExcelファイル
前回使用したファイル	前回同じタスクで使用した、UPDATE文生成用のExcel関数が入ったファイル(コピー元)

懸念点は、以下のようなことがあります。

Excel関数がずれてしまうと上手く反映されない
合計で2万件以上、シートも10以上あるため、確認作業を含めると1時間半ほどかかる
中に不正な値があったとしても、行数が多いためエラーに気づきにくい

上記の懸念点を解決していきたいと考え効率化ツールを作成しました。

導入効果

このツールを導入することで、以下のような効果が期待できます。

Excel関数の貼り付けミスの削減
不正な値がある行の早期特定
タスクの負担軽減、開発に専念できる環境の実現

なにより、今までタスクが振られた際は、開発の合間で時間が取られてしまっていたので、
タスクの負担を少なくして開発に専念できる環境を整えられることが、一番のメリットだと考えています。

参考として、作業時間の削減を金額換算すると以下のようになります。

※以下の時給は実際の値ではなく、説明のための仮定値です。

前提条件

時給 2,000円
月あたり2回のタスク依頼

ツールの使用	時給	時間(h)	月あたりタスク実施回数	年間コスト
なし	2,000	1.5	2	72,000
あり	2,000	0.25	2	12,000

年間の削減コストとしての金額インパクトは大きくありませんが、上記のような質的なメリットを踏まえると、
導入する価値はあると考えています。

システムの内容

※カラム名や値は実際の物ではなくダミーデータを使用しております。

画面の流れ

起動直後は、ラベルと「Excelファイルの選択」ボタンしかないシンプルな画面です。

Excelファイルを読み込むと、シートを選択できるリストボックスと、出力形式を選ぶ3つのチェックボックス(SQL確認用 Excelの作成 / シート別SQLファイル作成 / すべてのシートを1つのSQLファイルに作成)、実行ボタンが表示されます。

シートを選択し、必要なチェックボックスを選んで「実行」を押すと、以下の処理が行われます。

① データ元Excel(入力)

入力のExcelは、団体コードごとにシートが分かれているのが前提です(例:SAMPLE001シート、SAMPLE002シート)。1シートの中は同じ団体コードの行のみで構成されています。

SAMPLE001シート

団体コード	user_id	user_name	notify_token	platform
SAMPLE001	U0001234	たなか	11111111-aaaa-bbbb-cccc-111111111111	0
SAMPLE001	U0005678	やまだ	22222222-aaaa-bbbb-cccc-222222222222	1

② 生成対象外リスト(元データに不備があった場合のみ自動生成)

選択したシートのデータに、空欄セルや対応表に無い団体コードなど不備がある場合、チェックボックスの選択に関わらず自動で、不備のある行だけをまとめたExcelが生成され、警告が表示されます。

①のデータに、不正な行を2件追加した例です。

団体コード	user_id	user_name	notify_token	platform	sql	メモ
SAMPLE001	U0009999	すずき		1		空のセルが含まれています
SAMPLE999	U0002222	いとう	33333333-aaaa-bbbb-cccc-333333333333	0		対応表に団体コードと一致するIDが含まれていません

③ 各チェックボックスに応じた出力

不備のある行を除いた残りのデータに対して、団体コードに対応するテーブル番号を対応表(JSON)から引いてUPDATE文を組み立てます。団体コードが異なれば、更新先のテーブルも変わります。

対応表(団体コード→テーブル番号)

団体コード	テーブル番号
SAMPLE001	001
SAMPLE002	002

「SQL確認用 Excelの作成」を選んだ場合(sql列付きの確認用Excel)

団体コード	user_id	user_name	notify_token	platform	sql
SAMPLE001	U0001234	たなか	11111111-aaaa-bbbb-cccc-111111111111	0	update sample_db.members_001 set notify_token = '11111111-aaaa-bbbb-cccc-111111111111', user_name = 'たなか', platform = 'android' where user_id = 'U0001234';
SAMPLE002	U0007777	さとう	44444444-aaaa-bbbb-cccc-444444444444	1	update sample_db.members_002 set notify_token = '44444444-aaaa-bbbb-cccc-444444444444', user_name = 'さとう', platform = 'iphone' where user_id = 'U0007777';

「シート別SQLファイル作成」を選んだ場合：シートごとに1つの.sqlファイル

「すべてのシートを1つのSQLファイルに作成」を選んだ場合：全シート分をまとめた1つの.sqlファイル

システムの流れ(フロー図)

ここまでの内容を、処理の流れとして図にまとめると以下のようになります。

「なぜこのツールを作成したのか」で示した手作業のフローと比べると、Excel関数のコピー&ペーストやAutoFillといった手作業が無くなり、バリデーション(必須列チェック・空欄チェック・団体コードの対応チェック)がツール内で自動的に行われる点が大きな違いです。

環境構築

このツールはuvでパッケージ管理を行っています。

Pythonバージョン

Python 3.13

依存パッケージ

pyproject.tomlで管理している主な依存関係は以下の通りです。

パッケージ	用途
pandas	Excelの読み込み・DataFrame操作・SQL生成
openpyxl	pandasがExcelファイルを読み書きする際に使用
pyinstaller(dev)	`.exe`化のためのビルドツール

なお、実行時には団体コードとテーブル番号の対応表(JSONファイル)を、実行ファイルと同じディレクトリに配置する必要があります。

フォルダ構成

push_notifications/
├── main.py                 # エントリーポイント
├── company_id.json         # 団体コード→テーブル番号の対応表(実行時に配置)
├── app/
│   ├── exceptions.py       # 独自例外クラス
│   ├── excel_service.py    # Excel読み込み・SQL生成・出力処理
│   ├── id_manager.py       # company_id.jsonの読み込み
│   ├── logging_config.py   # ログ設定
│   ├── path_utils.py       # 実行時の基準ディレクトリ取得
│   └── ui/
│       └── main_window.py  # GUI(Tkinter)
├── output_excel/           # 確認用Excelの出力先(実行時に自動生成)
├── output_SQL/             # SQLファイルの出力先(実行時に自動生成)
├── abnormal_excel/         # 生成対象外リストの出力先(実行時に自動生成)
└── logs/                   # ログファイルの出力先(実行時に自動生成)

ソースコード

ここでは、main.py, app/exceptions.py, app/ui/main_window.py, app/excel_service.py
のコードを記載します。

※上記、システムの内容のサンプルに合わせて一部コードを変更しています

1. main.py

main.py

from app.logging_config import setup_logging
from app.ui.main_window import MainWindow


def main():
    setup_logging()
    root = MainWindow()
    root.mainloop()

if __name__ == "__main__":
    main()

本アプリのエントリーポイントです。
ここではアプリの起動と、ログの出力設定を行っております。

2. app/exceptions.py

app/exceptions.py

class AppError(Exception):
    """このアプリ内の想定内のエラー"""

class SheetNotSelectedError(AppError):
    pass


class ExcelFileNotLoadedError(AppError):
    pass


class InvalidSheetFormatError(AppError):
    pass


class DataNotFoundError(AppError):
    pass


class WriteFileOpeningError(AppError):
    pass


class ReadFileOpeningError(AppError):
    pass


class IdMappingLoadError(AppError):
    pass

このツールの動作中に起こりえる想定内のエラーを定義しています。
AppErrorを作成して他の例外クラスに継承させることで、エラーハンドリングの際に下記のように
呼び出し元で想定内のエラー時の処理をまとめて記述できます。

try:
    # 処理の関数の呼び出し
except AppError as e:
    # アプリの想定内エラー時の処理
except Exception as e:
    # 想定外のエラー時の処理

3. app/ui/main_window.py

app/ui/main_window.py

import logging
import tkinter as tk
from tkinter import filedialog, messagebox
from pathlib import Path

from app.exceptions import AppError
from app.path_utils import get_base_dir
from app.excel_service import ExcelService


logger = logging.getLogger(__name__)


class MainWindow(tk.Tk):

    def __init__(self):
        super().__init__()
        # タイトルと画面の大きさの初期値を設定
        self.title('サンプルツール')
        self.geometry("400x600")

        # フレームの定義
        self.static_frame = tk.Frame(self)
        self.action_frame = tk.Frame(self)

        # Excel処理, SQL書き出し処理のサービスクラスのインスタンス化
        self.excel_service = ExcelService()
        
        # Excel読み込み後に生成するUIパーツ用の変数を初期化
        self.xlsheets = []
        self.sheets_listbox = None
        self.scrollbar = None
        self.listbox_frame = None
        self.check_xl = None
        self.check_sql = None
        self.check_all_sql = None
        self.run_button = None

        # チェックボックスのデータ型と初期値の設定
        self.xl_action = tk.BooleanVar(value=False)
        self.sql_action = tk.BooleanVar(value=False)
        self.all_sql_action = tk.BooleanVar(value=False)

        # 初期画面の作成と配置
        self.create_widgets()
        self.action_frame.pack()

    def create_widgets(self):
        """起動直後の画面のウィジェット作成"""

        # フレームとラベルを配置
        self.static_frame.pack()
        self.xl_label = tk.Label(self.static_frame, text='未選択')
        self.xl_label.pack(pady=20)

        # Excel読み込みボタンの作成と配置
        self.xl_button = tk.Button(
            self.static_frame,
            text='Excelファイルの選択',
            command=self.load_excel,
        )
        self.xl_button.pack(pady=20)

    def create_add_actions_widget(self):
        """Excel選択状態に応じて、シート選択・実行用ウィジェットの作成(未選択時は既存ウィジェットの削除のみ)"""

        # Excelファイルを再選択した際のため、一度初期化
        self.clear_action_widget()

        if not self.xlsheets:
            return

        # リストボックスのフレームの作成と配置
        self.listbox_frame = tk.Frame(self.action_frame)
        self.listbox_frame.pack(pady=10)

        # リストボックスの設定
        self.sheets_listbox = tk.Listbox(
            self.listbox_frame,
            selectmode=tk.MULTIPLE,
            height=10,
        )

        # スクロールバーの作成とリストボックスへ紐づけ
        self.scrollbar = tk.Scrollbar(self.listbox_frame, orient=tk.VERTICAL, command=self.sheets_listbox.yview)
        self.sheets_listbox.config(yscrollcommand=self.scrollbar.set)

        # 読込んだExcelのシート名リストの内容をリストボックスにセット
        for sheet in self.xlsheets:
            self.sheets_listbox.insert(tk.END, sheet)

        # リストボックスとスクロールバーの配置
        self.sheets_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
        self.scrollbar.pack(side=tk.RIGHT, fill=tk.Y)

        # チェックボックスの設定と配置
        self.check_xl = tk.Checkbutton(self.action_frame, text='SQL確認用 Excelの作成', variable=self.xl_action, anchor='w')
        self.check_sql = tk.Checkbutton(self.action_frame, text='シート別SQLファイル作成', variable=self.sql_action, anchor='w')
        self.check_all_sql = tk.Checkbutton(self.action_frame, text='すべてのシートを1つのSQLファイルに作成', variable=self.all_sql_action, anchor='w')
        self.check_xl.pack(pady=10, fill='x', anchor=tk.W)
        self.check_sql.pack(pady=10, fill='x', anchor=tk.W)
        self.check_all_sql.pack(pady=10, fill='x', anchor=tk.W)

        # 実行ボタンの設定と配置
        self.run_button = tk.Button(self.action_frame, text='実行', command=self.run_apps)
        self.run_button.pack(pady=20)

    def clear_action_widget(self):
        """create_add_actions_widget()で作成するウィジェットの削除と初期化"""
        widgets = [
            self.sheets_listbox,
            self.scrollbar,
            self.listbox_frame,
            self.check_xl,
            self.check_sql,
            self.check_all_sql,
            self.run_button,
        ]

        for widget in widgets:
            if widget is not None:
                widget.destroy()
            
        self.sheets_listbox = None
        self.scrollbar = None
        self.listbox_frame = None
        self.check_xl = None
        self.check_sql = None
        self.check_all_sql = None
        self.run_button = None

    def load_excel(self):
        """Excelを読み込む処理"""

        # 選択されたExcelファイルのパスの取得
        file_type = [('Excelファイル', "*.xlsx *.xls")]
        file_path = filedialog.askopenfilename(
            title='push通知のExcelファイルを選択してください',
            filetypes=file_type,
            initialdir=get_base_dir(),
        )

        # Excelが選択されなかった場合は、初期化をして処理を中断
        if not file_path:
            self.excel_service.drop_excel_file()
            self.xlsheets = []
            self.xl_label.config(text='未選択')
            self.create_add_actions_widget()
            return
        
        try:
            # Excelサービスインスタンスにパスを渡してファイルの読み込みとシート名のリストを取得
            self.excel_service.set_xlfile(file_path)
            self.xlsheets = self.excel_service.get_sheet_names()
        except AppError as e:
            messagebox.showerror('Excelファイル読み込み時エラー', str(e))
            return
        except Exception as e:
            logger.exception('Excelファイルの読み込み中に予期せぬエラーが発生しました')
            messagebox.showerror('エラー', '予期せぬエラーが発生しました')
            return
        
        # ラベルの値を読み込んだExcelのファイル名に変更
        self.xl_label.config(text=Path(file_path).name)
        # アプリ実行用ウィジェットの作成
        self.create_add_actions_widget()

    def run_apps(self):
        """実行ボタンを押した際の処理。DataFrame, Excelファイル, SQLファイルの作成"""
        
        # リストボックスで選択されたシートのインデックスの取得
        listbox_indexes = self.sheets_listbox.curselection()
        if not listbox_indexes:
            messagebox.showwarning('警告', '操作対象のシートが選択されていません')
            return
        
        # チェックボックスがすべて未選択の場合は処理を中断
        if not self.xl_action.get() and not self.sql_action.get() and not self.all_sql_action.get():
            messagebox.showwarning('警告', '実行する処理が選択されていません')
            return
        
        # 選択されたシートのインデックスからシート名のリストを作成
        current_sheets = [
            self.sheets_listbox.get(i)
            for i in listbox_indexes
        ]

        try:
            # 選択されたシートからSQL文を含むDataFrameを作成
            self.excel_service.create_sql_dfs_dict(current_sheets)
        except AppError as e:
            messagebox.showerror('Excelデータ読み込み時エラー', str(e))
            return
        except Exception as e:
            logger.exception('SQLデータの作成中に予期せぬエラーが発生しました')
            messagebox.showerror('エラー', '予期せぬエラーが発生しました')
            return

        try:
            # DataFrameに不正な値があった場合は、そのデータをまとめたExcelの出力
            if self.excel_service.inform_null_by_excel():
                messagebox.showwarning('警告', 'SQL未生成のデータがあります。詳細はabnormal_excel/SQL未生成リスト.xlsxを確認してください')
        except AppError as e:
            messagebox.showerror('異常データ出力時エラー', str(e))
            return
        except Exception as e:
            logger.exception('異常データの出力中に予期せぬエラーが発生しました')
            messagebox.showerror('エラー', '予期せぬエラーが発生しました')
            return

        # Excel出力のチェックボックスが選択されているかの確認
        if self.xl_action.get():
            try:
                # Excelの出力
                self.excel_service.save_to_excel()
                logger.info('Excel出力完了')
            except AppError as e:
                messagebox.showerror('Excel書き込み時エラー', str(e))
                return
            except Exception as e:
                logger.exception('Excel出力中に予期せぬエラーが発生しました')
                messagebox.showerror('エラー', '予期せぬエラーが発生しました')
                return

        # シート別SQLファイル作成のチェックボックスが選択されているかの確認
        if self.sql_action.get():
            try:
                self.excel_service.save_to_sql()
                logger.info('SQLファイル出力完了')
            except AppError as e:
                messagebox.showerror('SQLファイル書き込み時エラー', str(e))
                return
            except Exception as e:
                logger.exception('SQLファイル出力中に予期せぬエラーが発生しました')
                messagebox.showerror('エラー', '予期せぬエラーが発生しました')
                return
        
        # 「すべてのシートを1つのSQLファイルに作成」のチェックボックスが選択されているかの確認
        if self.all_sql_action.get():
            try:
                self.excel_service.save_to_sql_all_in()
                logger.info('SQLファイル出力完了')
            except AppError as e:
                messagebox.showerror('SQLファイル書き込み時エラー', str(e))
                return
            except Exception as e:
                logger.exception('SQLファイル出力中に予期せぬエラーが発生しました')
                messagebox.showerror('エラー', '予期せぬエラーが発生しました')
                return

        messagebox.showinfo('完了', '出力が完了しました')

Tkinterを使用した、UI部分のコードです。
先ほどのexceptions.pyで作成したAppErrorの例外クラスか、それ以外の例外かによって、表示するメッセージボックスの内容を切り替えています。

4. app/excel_service.py

app/excel_service.py

import logging

import pandas as pd

from app import exceptions
from app.id_manager import get_ids_dict
from app.path_utils import get_base_dir


logger = logging.getLogger(__name__)

SQL = "update sample_db.members_{0} set notify_token = '{1}', user_name = '{2}', platform = '{3}' where user_id = '{4}';"
CHECK_COLUMNS = {'団体コード', 'user_id', 'user_name', 'notify_token', 'platform'}
TARGET_EMPTY_CHAR_COLUMNS = ["団体コード", "user_id", "notify_token", "platform"]


class ExcelService:

    def __init__(self):
        self.xl_file = None
        self.dfs_dict = None
        self.null_in_dfs_dict = None
        self.base_dir = get_base_dir()

    def escape_sql(self, value):
        """SQL文に埋め込む値のシングルクォートをエスケープする"""
        return str(value).replace("'", "''")

    def set_xlfile(self, file_path):
        """指定したパスのExcelファイルを読み込み、self.xl_fileにセットする"""
        try:
            self.xl_file = pd.ExcelFile(file_path)
        except FileNotFoundError as e:
            logger.exception('Excelファイルが見つかりません')
            raise exceptions.ExcelFileNotLoadedError('Excelファイルが見つかりません') from e
        except PermissionError as e:
            logger.exception('Excelファイルがすでに開かれており、読み込むことができません')
            raise exceptions.ExcelFileNotLoadedError('Excelファイルがすでに開かれており、読み込むことができません') from e

    def get_sheet_names(self):
        """読み込み済みのExcelファイルからシート名一覧を取得する"""
        if self.xl_file is None:
            logger.error('Excelファイルが見つかりません')
            raise exceptions.ExcelFileNotLoadedError('set_xlfile(file_path) を実行してExcelファイルを読み込んでください')
        return self.xl_file.sheet_names

    def drop_excel_file(self):
        """読み込み済みのExcelファイルと、生成済みのDataFrameを破棄する"""
        if self.xl_file:
            self.xl_file.close()
            self.xl_file = None
            self.dfs_dict = None
            self.null_in_dfs_dict = None

    def create_sql_dfs_dict(self, current_sheets):
        """選択されたシートを検証し、SQL文を含むDataFrameを作成してself.dfs_dictにセットする"""
        if not current_sheets:
            logger.warning('操作対象のシートが選択されていません')
            raise exceptions.SheetNotSelectedError('シートを選択してください')

        if self.xl_file is None:
            logger.warning('対象のExcelファイルが見つかりません')
            raise exceptions.ExcelFileNotLoadedError('Excelファイルが読み込めていません')

        # Excel内の全シート名から、選択されたシート名のみを抽出
        match_sheets = []
        for sheet in self.get_sheet_names():
            if sheet in current_sheets:
                match_sheets.append(sheet)

        # keep_default_na=False: "NA"等の文字列を誤ってNaN扱いしないようにする
        # (本当に空欄かどうかはTARGET_EMPTY_CHAR_COLUMNSに対して後段で別途判定する)
        dfs_dict = self.xl_file.parse(match_sheets, keep_default_na=False)

        # 団体コード→テーブル番号のマッピングを取得
        try:
            id_dict = get_ids_dict()
        except FileNotFoundError as e:
            logger.error('company_id.jsonが見つかりません')
            raise exceptions.IdMappingLoadError('company_id.jsonが読み込めません。実行ファイルと同じディレクトリに配置してください') from e
        except (ValueError, PermissionError) as e:
            logger.exception('company_id.jsonの読み込み中にエラーが発生しました')
            raise exceptions.IdMappingLoadError('company_id.jsonの読み込み中にエラーが発生しました') from e

        try:
            self.null_in_dfs_dict = {}
            for sheetname, df in dfs_dict.items():

                # 必須列(団体コード, user_id, user_name, notify_token, platform)が揃っているか確認
                if not CHECK_COLUMNS <= set(df.columns.values):
                    logger.warning(f"規定の列が含まれていません: {df.columns.values}")
                    raise exceptions.InvalidSheetFormatError(
                        f"{sheetname}:'団体コード', 'user_id', 'user_name', 'notify_token', 'platform'のいずれかの列が含まれていません"
                    )

                is_null_sheet = False
                # 空白のみのセルをNoneに変換し、isnull()で空欄として検知できるようにする
                df[TARGET_EMPTY_CHAR_COLUMNS] = df[TARGET_EMPTY_CHAR_COLUMNS].replace(r"^\s*$", None, regex=True)
                if df.isnull().values.any():
                    # 空欄を含む行は「生成対象外リスト」用に退避し、SQL生成対象からは除外する
                    self.null_in_dfs_dict[sheetname] = df[df.isnull().any(axis=1)].copy()
                    self.null_in_dfs_dict[sheetname]['sql'] = None
                    self.null_in_dfs_dict[sheetname]['メモ'] = '空のセルが含まれています'
                    is_null_sheet = True
                    df = df.dropna()

                # 空欄除去の結果、SQLを生成できる行が1件も残らなかった場合はエラーにする
                if df.empty:
                    logger.error(f"{sheetname}: SQLを生成可能なデータがありません")
                    raise exceptions.InvalidSheetFormatError(
                        f"{sheetname}: SQLを生成可能なデータがありません"
                    )

                # 1行ごとにUPDATE文を組み立てる
                # 団体コードがcompany_id.jsonに存在しない行はSQLを生成せずNoneにする
                df['sql'] = df.apply(lambda row: SQL.format(
                    id_dict.get(row['団体コード']),
                    self.escape_sql(row['notify_token']),
                    self.escape_sql(row['user_name']),
                    'android' if row['platform'] == 0 else 'iphone',
                    self.escape_sql(row['user_id'])
                    )
                    if row['団体コード'] in id_dict
                    else None
                    ,
                    axis=1
                )

                # 団体コードの不一致でsqlがNoneになった行も「生成対象外リスト」に追加する
                if df.isnull().values.any():
                    if is_null_sheet:
                        # 既に空欄セルで退避済みの場合は追記(concat)し、メモが未設定の行にのみ理由を補完する
                        self.null_in_dfs_dict[sheetname] = pd.concat([self.null_in_dfs_dict[sheetname], df[df['sql'].isnull()]])
                        self.null_in_dfs_dict[sheetname]['メモ'] = self.null_in_dfs_dict[sheetname]['メモ'].fillna('company_id.jsonに団体コードと一致するIDが含まれていません')
                    else:
                        self.null_in_dfs_dict[sheetname] = df[df['sql'].isnull()].copy()
                        self.null_in_dfs_dict[sheetname]['メモ'] = 'company_id.jsonに団体コードと一致するIDが含まれていません'

                # SQLを生成できた行だけを最終的なdfs_dictとして保持する
                dfs_dict[sheetname] = df.dropna()

        except KeyError as e:
            logger.exception('形式が不正なシートが選択されています')
            raise exceptions.InvalidSheetFormatError('形式が不正なシートが選択されています') from e

        self.dfs_dict = dfs_dict

    def save_to_excel(self):
        """SQL生成済みのDataFrameを確認用Excelとして出力する"""
        if self.dfs_dict is None:
            logger.warning('Excelに書き込むデータがありません')
            raise exceptions.DataNotFoundError('Excelに書き込むデータがありません')

        xl_dir = self.base_dir / 'output_excel'
        xl_dir.mkdir(exist_ok=True)

        try:
            with pd.ExcelWriter(xl_dir / 'generated_sql.xlsx') as writer:
                for sheet_name, df in self.dfs_dict.items():
                    df.to_excel(writer, sheet_name=sheet_name, index=False)
        except PermissionError as e:
            logger.exception('書き込み先のExcelファイルが開かれており、書き込みができません')
            raise exceptions.WriteFileOpeningError('書き込み先のExcelファイルが開かれており、書き込みができません') from e

    def save_to_sql(self):
        """シートごとに個別の.sqlファイルとしてSQL文を出力する"""
        if self.dfs_dict is None:
            logger.warning('SQLを生成するデータがありません')
            raise exceptions.DataNotFoundError('SQLを生成するデータがありません')

        sql_dir = self.base_dir / 'output_SQL'
        sql_dir.mkdir(exist_ok=True)

        try:
            for sheet_name, df in self.dfs_dict.items():
                with open(f'{sql_dir / sheet_name}.sql', 'w', encoding='utf-8') as f:
                    f.write(f"-- {sheet_name} ------------------------------------------------ \n\n")
                    for sql in df['sql']:
                        # 団体コードの不一致などでSQLを生成できなかった行はスキップする
                        if sql is None:
                            continue
                        f.write(sql + '\n')
        except PermissionError as e:
            logger.exception('書き込み先のSQLファイルが開かれており、書き込みができません')
            raise exceptions.WriteFileOpeningError('書き込み先のSQLファイルが開かれており、書き込みができません') from e

    def inform_null_by_excel(self):
        """SQL未生成の異常データがあれば、まとめてExcelとして出力する"""
        if not self.null_in_dfs_dict:
            return False

        abnormal_xl_dir = self.base_dir / 'abnormal_excel'
        abnormal_xl_dir.mkdir(exist_ok=True)

        try:
            with pd.ExcelWriter(abnormal_xl_dir / 'invalid_data_list.xlsx') as writer:
                for sheet_name, df in self.null_in_dfs_dict.items():
                    df.to_excel(writer, sheet_name=sheet_name, index=False)

            self.null_in_dfs_dict = None
            logger.warning('SQL未生成のデータがあります')
            return True
        except PermissionError as e:
            logger.exception('書き込み先のExcelファイルが開かれており、書き込みができません')
            raise exceptions.WriteFileOpeningError('書き込み先のExcelファイルが開かれており、書き込みができません') from e

    def save_to_sql_all_in(self):
        """全シートのSQL文を1つの.sqlファイルにまとめて出力する"""
        if self.dfs_dict is None:
            logger.warning('SQLを生成するデータがありません')
            raise exceptions.DataNotFoundError('SQLを生成するデータがありません')

        sql_dir = self.base_dir / 'output_SQL'
        sql_dir.mkdir(exist_ok=True)

        try:
            with open(sql_dir / 'all_sheets.sql', 'w', encoding='utf-8') as f:
                for sheet_name, df in self.dfs_dict.items():
                    f.write(f"\n-- {sheet_name} ------------------------------------------------ \n\n")
                    for sql in df['sql']:
                        # 団体コードの不一致などでSQLを生成できなかった行はスキップする
                        if sql is None:
                            continue
                        f.write(sql + '\n')
        except PermissionError as e:
            logger.exception('書き込み先のSQLファイルが開かれており、書き込みができません')
            raise exceptions.WriteFileOpeningError('書き込み先のSQLファイルが開かれており、書き込みができません') from e

本ツールのメインのロジック部分です。
クラスにすることで、インスタンスがDataFrameやFileの状態を保持できます。
そのため呼び出し元(app/ui/main_window.py)では、ほとんどの場合で引数を渡さずに呼び出すことができ、引数の管理を簡素化できています。

開発で学んだこと

pandasの`ExcelFile`クラスについて

今回の開発では、Excelファイル内のシート名をUIに表示し、ユーザーが選択した任意のシートだけを処理する必要がありました。

そのため、シート名の一覧を取得できるExcelFileクラスを使用しました。

通常、読み込むシートがあらかじめ決まっている場合はread_excel()が手軽です。一方、今回のようにシート名を確認してから読み込むシートを選択する場合は、ExcelFileが適しています。

import pandas as pd

excel_filepath = "sample.xlsx"

xl_file = pd.ExcelFile(excel_filepath)

# シート名の一覧を取得
sheet_names = xl_file.sheet_names

取得したsheet_namesをListboxなどのUIに表示し、ユーザーが選択したシート名をparse()に渡します。

selected_sheets = ["Sheet1", "Sheet2"]

dfs = xl_file.parse(selected_sheets)

複数のシート名を指定すると、シート名をキー、DataFrameを値とする辞書が返されます。

{
    "Sheet1": DataFrame,
    "Sheet2": DataFrame,
}

今回のような、次の処理と相性がよいと感じました。

Excelファイルを選択する
シート名を取得してUIに表示する
ユーザーが任意のシートを選択する
選択されたシートをDataFrameの辞書として取得する
シートごとに処理を行う

Excelセルの値が`'NA'`だと、pandasに`NaN`として読み飛ばされてしまった点

開発中、user_name列に実際にNAという文字列が入っている行があったのですが、ExcelFile.parse()で読み込むと、その値が**NaN(欠損値)に化けてしまう**という問題に気づきました。

団体コード	user_id	user_name	notify_token	platform
SAMPLE001	U0003333	NA	55555555-...	0

このデータを読み込んだDataFrameを確認すると、user_name列は"NA"という文字列ではなくNaNになっていました。

原因: pandasはデフォルトで、特定の文字列を自動的に`NaN`に変換する

pandasはread_excel()やExcelFile.parse()で読み込む際、デフォルト(keep_default_na=True)で、""(空文字列)はもちろん、"NA", "NULL", "None", "n/a"などあらかじめ決められた文字列を、値の内容に関わらず自動的にNaNとして扱います。

今回はたまたまuser_nameが"NA"という、この判定対象の文字列と一致してしまったため、本来は正常なデータだったにもかかわらず欠損値として扱われてしまったわけです。エラーも警告も出ないため、実際にデータを見比べるまで気づけませんでした。

対策：`keep_default_na=False`と独自の空欄判定

そこで、pandasによる自動的な欠損値への変換を無効にし、空欄かどうかを自分で判定するようにしました。

import pandas as pd

excel_file_path = "sample.xlsx"

# 読み込むシート
target_sheet_names = ["Sheet1", "Sheet2"]

# 空欄をチェックする列
TARGET_EMPTY_CHAR_COLUMNS = [
    "団体コード",
    "user_id",
    "notify_token",
    "platform",
]

with pd.ExcelFile(excel_file_path) as excel_file:
    # 「NA」や「N/A」などをNaNへ自動変換しない
    dataframes = excel_file.parse(
        target_sheet_names,
        keep_default_na=False,
    )

for sheet_name, df in dataframes.items():
    # 空文字や空白だけの文字列をNoneへ変換する
    df[TARGET_EMPTY_CHAR_COLUMNS] = df[
        TARGET_EMPTY_CHAR_COLUMNS
    ].replace(
        r"^\s*$",
        None,
        regex=True,
    )

    # 対象列に空欄を含む行を取得する
    empty_rows = df[
        df[TARGET_EMPTY_CHAR_COLUMNS].isnull().any(axis=1)
    ]

keep_default_na=Falseを指定すると、"NA"や"N/A"などの文字列が、自動的にNaNへ変換されなくなります。

一方で、Excel上の空欄も空文字列""として読み込まれるため、そのままではisnull()で検出できません。

そこで、replace()と正規表現のr"^\s*$"を使用し、次のような値をNoneへ変換しています。

""       # 空文字列
"   "    # スペースだけの文字列

Noneへ変換することで、isnull()を使って空欄を検出できるようになります。

今後の展望

exeファイルをチームで共有

タスク時間の短縮、Excel関数の貼り付けミスの防止、不正な値の早期発見といった効果を踏まえて、
まずはプロジェクトリーダーにこのツールを提案し、実務で活用してもらえるようにしていきたいです。
提案していく中でフィードバックを頂きながら、よりプロジェクトに沿ったツールにしていきたいです。

プロジェクト内で案件化

このツールで行っている処理を、最終的には実際のアプリの管理画面から実行できるようにしたいと考えています。
この機能追加を社内の案件として提案し、私自身がその開発を担当できるように動いていきたいです。