Day 45：Excel業務自動化システムを「一本のプロジェクト」として組み上げる
Day 45では、これまで学んできた内容をまとめて、 「Excel業務自動化システム」を1つのプロジェクトとして設計・実装する ことを目指します。
機能の流れは次の通りです。
- Excel読み込み
- データチェック
- データ加工
- 集計
- レポート作成
- Excel出力
- ログ保存
これを、初心者の方にも分かりやすいように、 ステップバイステップで「考え方 → コード →テンプレート」として説明していきます。
全体設計：処理の流れを「ステップ」に分解する
処理フローを言葉で整理する
今回のExcel業務自動化システムは、 例えば「店舗別売上のExcelを読み込んで、チェック・加工・集計・レポート出力を自動で行う」 というイメージで考えます。
フローは次のようになります。
- Excel読み込み：
- 元データのExcelファイルを読み込んで、DataFrameに変換します。
- データチェック：
- 必須列があるか、欠損値がないか、型がおかしくないかなどを確認します。
- データ加工：
- 日付の整形、数値型への変換、不要列の削除などを行います。
- 集計：
- 店舗別・日別・商品別などの集計を行います。
- レポート作成：
- 集計結果をレポート用の形式（表）に整えます。
- Excel出力：
- レポートを新しいExcelファイルとして保存します。
- ログ保存：
- 処理の開始・終了・エラーなどをログファイルに記録します。
この「流れ」をそのままコードの構造に落とし込んでいきます。
ログの準備：何が起きたかを残す土台を作る
loggingの基本設定
まずは、プロジェクト全体で使うログ設定を用意します。
# day45_excel_system.py
import logging
def setup_logger(log_path: str = "excel_job.log"):
"""ログ設定を行う関数です。"""
logging.basicConfig(
filename=log_path, # ログを書き出すファイル名
level=logging.INFO, # INFO以上のログを記録します
format="%(asctime)s [%(levelname)s] %(message)s"
# 日時・ログレベル・メッセージを記録するフォーマットです
)
logging.info("=== Excel業務自動化システム ログ開始 ===")
ポイント：
- ログは「あとから何が起きたかを追うための記録」です。
- システムの開始時に
setup_logger()を呼んでおくことで、 以降の処理でlogging.info()やlogging.error()を使えるようになります。
Excel読み込み：pandasでDataFrameにする
Excelファイルを読み込む関数
ここでは、pandas の read_excel() を使ってExcelを読み込みます。
import pandas as pd
import logging
def load_excel(input_path: str) -> pd.DataFrame:
"""Excelファイルを読み込んでDataFrameを返す関数です。"""
logging.info(f"[INPUT] Excelファイルを読み込みます: {input_path}")
try:
df = pd.read_excel(input_path) # シート指定が必要ならsheet_name引数を使います。
logging.info("[INPUT] Excel読み込み成功")
return df
except FileNotFoundError:
logging.error(f"[INPUT] ファイルが見つかりません: {input_path}")
raise
except Exception as e:
logging.error(f"[INPUT] Excel読み込み中にエラーが発生しました: {e}")
raise
ポイント：
read_excel()でExcelをDataFrameとして扱えるようになります。- エラーが起きた場合はログに残しつつ
raiseして、上位の処理に「失敗した」ことを伝えます。
データチェック：壊れたデータを早めに見つける
必須列のチェック・欠損値チェック
業務データでは、「列が足りない」「売上が空欄」などがよく起こります。 これを早めに検出しておくことで、後続の処理での謎エラーを防ぎます。
import logging
import pandas as pd
def check_data(df: pd.DataFrame) -> None:
"""データチェックを行う関数です。必須列・欠損値などを確認します。"""
logging.info("[CHECK] データチェックを開始します。")
# 必須列の一覧（例として「店舗」「日付」「売上」を必須とします）
required_columns = ["店舗", "日付", "売上"]
# 必須列がすべて存在するか確認します。
for col in required_columns:
if col not in df.columns:
logging.error(f"[CHECK] 必須列が存在しません: {col}")
raise ValueError(f"必須列が存在しません: {col}")
logging.info("[CHECK] 必須列の存在を確認しました。")
# 欠損値のチェック（売上が欠損していないか）
if df["売上"].isnull().any():
logging.warning("[CHECK] 売上列に欠損値があります。0として扱います。")
# 欠損値を0で埋めるなどの対応をここで行います。
df["売上"] = df["売上"].fillna(0)
logging.info("[CHECK] データチェック完了。")
重要ポイント：
- 必須列のチェック は、業務自動化システムの「入り口の守り」です。
- 欠損値は「エラーにするか」「補正するか」を業務ルールに合わせて決めます。
- ログに「何をどう扱ったか」を残しておくと、後から確認しやすくなります。
データ加工：型変換・日付整形・不要列削除
日付をdatetime型に変換し、売上を数値型に揃える
import logging
import pandas as pd
def transform_data(df: pd.DataFrame) -> pd.DataFrame:
"""データ加工を行う関数です。型変換や不要列削除などを行います。"""
logging.info("[TRANSFORM] データ加工を開始します。")
# 日付列をdatetime型に変換します。
logging.info("[TRANSFORM] 日付列をdatetime型に変換します。")
df["日付"] = pd.to_datetime(df["日付"], errors="coerce")
# 変換に失敗した日付がないかチェックします。
if df["日付"].isnull().any():
logging.warning("[TRANSFORM] 日付の変換に失敗した行があります。")
# 業務ルールに応じて、該当行を削除するなどの対応を行います。
df = df.dropna(subset=["日付"])
# 売上を数値型に変換します。
logging.info("[TRANSFORM] 売上列を数値型に変換します。")
df["売上"] = pd.to_numeric(df["売上"], errors="coerce")
if df["売上"].isnull().any():
logging.warning("[TRANSFORM] 売上の変換に失敗した行があります。0として扱います。")
df["売上"] = df["売上"].fillna(0)
# 不要な列があれば削除します（例として「メモ」列を削除）
if "メモ" in df.columns:
logging.info("[TRANSFORM] 不要列「メモ」を削除します。")
df = df.drop(columns=["メモ"])
logging.info("[TRANSFORM] データ加工完了。")
return df
ポイント：
- 日付・数値の型を揃えることで、後の集計処理が安定します。
- 「変換に失敗した行」をどう扱うかは、業務ルールに合わせて決めます。
- ここでもログに「何をどうしたか」を残すことが重要です。
集計：店舗別・日別などの集計を行う
店舗別売上合計の集計例
import logging
import pandas as pd
def aggregate_data(df: pd.DataFrame) -> pd.DataFrame:
"""集計処理を行う関数です。店舗ごとの売上合計を計算します。"""
logging.info("[AGGREGATE] 集計処理を開始します。")
# 店舗ごとの売上合計を計算します。
grouped = df.groupby("店舗")["売上"].sum().reset_index()
grouped.rename(columns={"売上": "売上合計"}, inplace=True)
logging.info("[AGGREGATE] 集計処理完了。")
return grouped
日付×店舗の集計（必要に応じて）
def aggregate_by_date_and_store(df: pd.DataFrame) -> pd.DataFrame:
"""日付×店舗ごとの売上集計を行う例です。"""
logging.info("[AGGREGATE] 日付×店舗の集計処理を開始します。")
grouped = df.groupby(["日付", "店舗"])["売上"].sum().reset_index()
grouped.rename(columns={"売上": "売上合計"}, inplace=True)
logging.info("[AGGREGATE] 日付×店舗の集計処理完了。")
return grouped
ポイント：
- 集計は「業務で見たい単位」に合わせて設計します。
- 店舗別・日別・商品別など、複数の集計結果を作ることもよくあります。
レポート作成：集計結果を「見やすい表」に整える
レポート用DataFrameを作る
ここでは、店舗別売上合計を「レポート用の表」として整える例を示します。
import logging
import pandas as pd
def create_report(df_agg: pd.DataFrame) -> pd.DataFrame:
"""レポート用のDataFrameを作成する関数です。"""
logging.info("[REPORT] レポート作成を開始します。")
# 売上合計の降順で並べ替えます（売上が多い店舗順）
df_report = df_agg.sort_values(by="売上合計", ascending=False).reset_index(drop=True)
# レポート用に順位列を追加します。
df_report["順位"] = df_report.index + 1
logging.info("[REPORT] レポート作成完了。")
return df_report
ポイント：
- レポートは「人が見るための形式」に整えるステップです。
- 並べ替え・順位付け・見出し列の追加などをここで行います。
Excel出力：レポートをExcelファイルとして保存する
pandasでExcel出力する基本
import logging
import pandas as pd
def export_to_excel(df_report: pd.DataFrame, output_path: str) -> None:
"""レポートをExcelファイルとして出力する関数です。"""
logging.info(f"[OUTPUT] レポートをExcelとして出力します: {output_path}")
# Excelファイルとして保存します。
with pd.ExcelWriter(output_path, engine="openpyxl") as writer:
df_report.to_excel(writer, sheet_name="レポート", index=False)
logging.info("[OUTPUT] Excel出力完了。")
ポイント：
ExcelWriterを使うことで、シート名を指定して出力できます。- 複数のシートに複数のDataFrameを出力することも可能です。
全体をつなぐ：Excel業務自動化システムのメイン関数
一連の流れを1本の関数にまとめる
ここまで作った関数をつないで、 「Excel読み込み → データチェック → データ加工 → 集計 → レポート作成 → Excel出力 → ログ保存」 の流れを1本のジョブとして実行します。
# day45_excel_system.py
import logging
import pandas as pd
def setup_logger(log_path: str = "excel_job.log"):
logging.basicConfig(
filename=log_path,
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
logging.info("=== Excel業務自動化システム ログ開始 ===")
def load_excel(input_path: str) -> pd.DataFrame:
logging.info(f"[INPUT] Excelファイルを読み込みます: {input_path}")
df = pd.read_excel(input_path)
logging.info("[INPUT] Excel読み込み成功")
return df
def check_data(df: pd.DataFrame) -> None:
logging.info("[CHECK] データチェックを開始します。")
required_columns = ["店舗", "日付", "売上"]
for col in required_columns:
if col not in df.columns:
logging.error(f"[CHECK] 必須列が存在しません: {col}")
raise ValueError(f"必須列が存在しません: {col}")
if df["売上"].isnull().any():
logging.warning("[CHECK] 売上列に欠損値があります。0として扱います。")
df["売上"] = df["売上"].fillna(0)
logging.info("[CHECK] データチェック完了。")
def transform_data(df: pd.DataFrame) -> pd.DataFrame:
logging.info("[TRANSFORM] データ加工を開始します。")
df["日付"] = pd.to_datetime(df["日付"], errors="coerce")
if df["日付"].isnull().any():
logging.warning("[TRANSFORM] 日付の変換に失敗した行があります。削除します。")
df = df.dropna(subset=["日付"])
df["売上"] = pd.to_numeric(df["売上"], errors="coerce")
if df["売上"].isnull().any():
logging.warning("[TRANSFORM] 売上の変換に失敗した行があります。0として扱います。")
df["売上"] = df["売上"].fillna(0)
logging.info("[TRANSFORM] データ加工完了。")
return df
def aggregate_data(df: pd.DataFrame) -> pd.DataFrame:
logging.info("[AGGREGATE] 集計処理を開始します。")
grouped = df.groupby("店舗")["売上"].sum().reset_index()
grouped.rename(columns={"売上": "売上合計"}, inplace=True)
logging.info("[AGGREGATE] 集計処理完了。")
return grouped
def create_report(df_agg: pd.DataFrame) -> pd.DataFrame:
logging.info("[REPORT] レポート作成を開始します。")
df_report = df_agg.sort_values(by="売上合計", ascending=False).reset_index(drop=True)
df_report["順位"] = df_report.index + 1
logging.info("[REPORT] レポート作成完了。")
return df_report
def export_to_excel(df_report: pd.DataFrame, output_path: str) -> None:
logging.info(f"[OUTPUT] レポートをExcelとして出力します: {output_path}")
with pd.ExcelWriter(output_path, engine="openpyxl") as writer:
df_report.to_excel(writer, sheet_name="レポート", index=False)
logging.info("[OUTPUT] Excel出力完了。")
def run_excel_job(input_path: str, output_path: str, log_path: str = "excel_job.log"):
"""Excel業務自動化システムのメイン関数です。"""
setup_logger(log_path)
logging.info("=== ジョブ開始 ===")
try:
df_input = load_excel(input_path)
check_data(df_input)
df_transformed = transform_data(df_input)
df_agg = aggregate_data(df_transformed)
df_report = create_report(df_agg)
export_to_excel(df_report, output_path)
logging.info("=== ジョブ正常終了 ===")
except Exception as e:
logging.error(f"ジョブがエラーで終了しました: {e}")
logging.info("=== ジョブ異常終了 ===")
def main():
input_path = "sales.xlsx" # 元のExcelファイル
output_path = "sales_report.xlsx" # レポートExcelファイル
log_path = "excel_job.log" # ログファイル
run_excel_job(input_path, output_path, log_path)
if __name__ == "__main__":
main()
このコードが、Day 45で目指す 「Excel業務自動化システム」の基本テンプレート です。
Day 45のまとめ
Day 45では、業務自動化プロジェクトとして、
- Excel読み込み（pandasでDataFrame化）
- データチェック（必須列・欠損値・型の確認）
- データ加工（日付・数値の型変換、不要列削除）
- 集計（店舗別売上合計など）
- レポート作成（並べ替え・順位付けなど）
- Excel出力（レポートを新しいExcelとして保存）
- ログ保存（処理の開始・終了・警告・エラーを記録）
を、1本の「Excel業務自動化システム」として組み上げました。
ここまで来ると、 「単発のスクリプト」ではなく、「業務で使える自動化システム」を設計・実装する感覚 がかなり具体的になっているはずです。
このテンプレートをベースに、
- 集計軸を増やす（商品別・日別など）
- レポートを複数シートに出力する
- メール送信（Day 43）や定期実行（Day 42）と組み合わせる
ことで、より実務に近い「Excel業務自動化プロジェクト」を育てていくことができます。
