Day 44：業務自動化設計の「考え方」をコードに落とし込む
Day 44では、これまで学んできたCSV・Excel・PDF・メール・定期実行などを、 1本の「業務自動化システム」として設計する視点 を身につけていきます。
キーワードは次の7つです。
- 入力
- 処理
- 出力
- ログ
- エラー処理
- 設定
- 再実行
ここでは、初心者の方にも分かりやすいように、 「売上CSVを毎日集計して結果を出力する」というシンプルな業務を題材に、 設計の考え方をステップバイステップでコードに落とし込んでいきます。
業務自動化の全体像を「箱」と「矢印」でイメージする
入力 → 処理 → 出力＋ログ・エラー・設定・再実行
業務自動化をざっくり図にすると、次のようなイメージになります。
- 入力：
- CSV・Excel・API・フォルダ内ファイルなどからデータを読み込む箱
- 処理：
- データ加工・集計・判定・レポート生成などを行う箱
- 出力：
- 新しいCSV・Excel・PDF・メール送信などで結果を出す箱
- ログ：
- 「いつ」「何が」行われたかを記録する箱
- エラー処理：
- 失敗したときに落ちず、原因を残す仕組み
- 設定：
- 入力パス・出力パス・対象期間・メール宛先などを外出しする箱
- 再実行：
- 失敗してもやり直せるようにする考え方
これを「1つのPythonスクリプトの中でどう構造化するか」がDay 44のテーマです。
入力・処理・出力を関数に分ける（骨格づくり）
まずは「役割を分ける」ことから始める
いきなり全部盛り込もうとせず、 まずは 入力・処理・出力 の3つに分けた「骨格」を作ります。
# day44_pipeline_basic.py
import pandas as pd
def load_input(input_path: str) -> pd.DataFrame:
"""入力：CSVファイルを読み込む関数です。"""
print(f"[INPUT] {input_path} を読み込みます。")
df = pd.read_csv(input_path) # CSVをDataFrameとして読み込みます。
return df
def process_data(df: pd.DataFrame) -> pd.DataFrame:
"""処理：データを集計する関数です。"""
print("[PROCESS] 店舗ごとの売上合計を集計します。")
# 店舗ごとに売上を合計します。
grouped = df.groupby("店舗")["売上"].sum().reset_index()
grouped.rename(columns={"売上": "売上合計"}, inplace=True)
return grouped
def save_output(df_result: pd.DataFrame, output_path: str) -> None:
"""出力：集計結果をCSVとして保存する関数です。"""
print(f"[OUTPUT] 集計結果を {output_path} に保存します。")
df_result.to_csv(output_path, index=False, encoding="utf-8")
def main():
input_path = "sales.csv"
output_path = "sales_aggregated.csv"
# 入力 → 処理 → 出力の流れをつなげます。
df_input = load_input(input_path)
df_result = process_data(df_input)
save_output(df_result, output_path)
if __name__ == "__main__":
main()
ここでの重要ポイントは、
- 関数ごとに「1つの責務」に絞る ことです。
- 後からログ・エラー処理・設定を追加するときに、どこに書けばよいかが明確になります。
ログ設計：あとから「何が起きたか」を追えるようにする
なぜログが必要か
業務自動化は、毎日・毎週・毎月と「長く動き続ける」前提です。 その中で、
- ある日だけ入力ファイルが壊れていた
- ある店舗のデータが欠損していた
- 処理が途中で止まっていた
といったことが起きたとき、 「何が起きたか」を後から確認できるかどうか が非常に重要になります。
そのために、ログを残します。
loggingモジュールでログをファイルに出す
# day44_logging_setup.py
import logging
def setup_logger(log_path: str = "job.log"):
"""ログ設定を行う関数です。"""
logging.basicConfig(
filename=log_path, # ログを書き出すファイル名
level=logging.INFO, # INFO以上のログを記録します
format="%(asctime)s [%(levelname)s] %(message)s"
# 日時・ログレベル・メッセージを記録するフォーマットです
)
logging.info("=== ログ開始 ===")
ログを組み込んだ入力・処理・出力
# day44_pipeline_with_logging.py
import logging
import pandas as pd
def load_input(input_path: str) -> pd.DataFrame:
logging.info(f"[INPUT] {input_path} を読み込みます。")
df = pd.read_csv(input_path)
logging.info("[INPUT] 読み込み成功")
return df
def process_data(df: pd.DataFrame) -> pd.DataFrame:
logging.info("[PROCESS] 店舗ごとの売上合計を集計します。")
grouped = df.groupby("店舗")["売上"].sum().reset_index()
grouped.rename(columns={"売上": "売上合計"}, inplace=True)
logging.info("[PROCESS] 集計成功")
return grouped
def save_output(df_result: pd.DataFrame, output_path: str) -> None:
logging.info(f"[OUTPUT] 集計結果を {output_path} に保存します。")
df_result.to_csv(output_path, index=False, encoding="utf-8")
logging.info("[OUTPUT] 保存成功")
ポイント：
print()は「その場で見るため」、loggingは「後から振り返るため」と考えると分かりやすいです。- ログファイル（例：
job.log）を見れば、「いつ」「どのステップまで進んだか」が分かります。
エラー処理：落ちない・原因が分かる・再実行できる
try / except で「落ちない」ようにする
業務自動化では、エラーが起きたときに、
- スクリプトが静かに落ちてしまう
- 原因が分からない
- 再実行してよいのか分からない
という状態が一番困ります。
そこで、エラーを捕まえてログに残す ことが大事になります。
# day44_error_handling_input.py
import logging
import pandas as pd
def load_input_safe(input_path: str) -> pd.DataFrame:
"""エラー処理付きの入力関数です。"""
logging.info(f"[INPUT] {input_path} を読み込みます。")
try:
df = pd.read_csv(input_path)
logging.info("[INPUT] 読み込み成功")
return df
except FileNotFoundError:
logging.error(f"[INPUT] ファイルが見つかりません: {input_path}")
# ここでraiseすることで、上位の処理に「失敗した」ことを伝えます。
raise
except Exception as e:
logging.error(f"[INPUT] 不明なエラーが発生しました: {e}")
raise
ジョブ全体をラップする
# day44_error_handling_job.py
import logging
import pandas as pd
def run_job():
"""ジョブ全体をエラー処理付きで実行する関数です。"""
logging.info("=== ジョブ開始 ===")
try:
df_input = load_input_safe("sales.csv")
df_result = process_data(df_input)
save_output(df_result, "sales_aggregated.csv")
logging.info("=== ジョブ正常終了 ===")
except Exception as e:
logging.error(f"ジョブがエラーで終了しました: {e}")
logging.info("=== ジョブ異常終了 ===")
# ここでメール通知などを行うこともできます。
def process_data(df: pd.DataFrame) -> pd.DataFrame:
logging.info("[PROCESS] 店舗ごとの売上合計を集計します。")
grouped = df.groupby("店舗")["売上"].sum().reset_index()
grouped.rename(columns={"売上": "売上合計"}, inplace=True)
logging.info("[PROCESS] 集計成功")
return grouped
def save_output(df_result: pd.DataFrame, output_path: str) -> None:
logging.info(f"[OUTPUT] 集計結果を {output_path} に保存します。")
df_result.to_csv(output_path, index=False, encoding="utf-8")
logging.info("[OUTPUT] 保存成功")
重要ポイント：
- エラーを握りつぶさず、ログに残してから再度投げる（raise）かどうかを決める ことが大事です。
- 「どこで」「何が原因で」止まったかが分かれば、再実行しやすくなります。
設定ファイル：コードと設定を分離する
設定をコードにベタ書きすると何が困るか
- 入力ファイル名が変わるたびにコードを書き換える必要がある
- テスト環境と本番環境でパスが違うときに面倒
- メール宛先などを変えるたびにコードを編集するのは危険
そこで、設定ファイル を使って、 「変わりやすい値」をコードから外に出します。
JSON形式の設定ファイル例
config.json：
{
"input_path": "sales.csv",
"output_path": "sales_aggregated.csv",
"log_path": "job.log",
"mail_to": "target@example.com"
}
設定ファイルを読み込むコード
# day44_config_load.py
import json
def load_config(config_path: str = "config.json") -> dict:
"""設定ファイル（JSON）を読み込む関数です。"""
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
return config
def main():
config = load_config()
print("=== 設定内容 ===")
print(config)
print("入力ファイル:", config["input_path"])
print("出力ファイル:", config["output_path"])
if __name__ == "__main__":
main()
ポイント：
- 設定値をJSONにまとめることで、コードを触らずに設定だけ変えられる ようになります。
- テスト用の
config_test.jsonと本番用のconfig_prod.jsonを用意して切り替える、という使い方も簡単です。
再実行の考え方：失敗しても「やり直せる」設計
再実行しやすい設計のポイント
- 入力ファイルを壊さない
- 原則として「読み込み専用」にする
- 出力ファイルは再生成可能なものにする
- 出力は「結果」であり、何度でも作り直せるようにする
- ログで失敗箇所が分かるようにする
- どのステップで止まったかが分かれば、そこから再実行しやすい
- 処理をステップに分けておく
- 「入力→加工→集計→出力」の各ステップを関数に分けておく
再実行を意識したジョブテンプレート
# day44_job_template.py
import logging
import json
import pandas as pd
def load_config(config_path: str = "config.json") -> dict:
"""設定ファイルを読み込む関数です。"""
with open(config_path, "r", encoding="utf-8") as f:
return json.load(f)
def setup_logger(log_path: str):
"""ログ設定を行う関数です。"""
logging.basicConfig(
filename=log_path,
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
logging.info("=== ログ開始 ===")
def load_input(input_path: str) -> pd.DataFrame:
"""入力：CSVを読み込む関数です。"""
logging.info(f"[INPUT] {input_path} を読み込みます。")
df = pd.read_csv(input_path)
logging.info("[INPUT] 読み込み成功")
return df
def process_data(df: pd.DataFrame) -> pd.DataFrame:
"""処理：店舗ごとの売上合計を集計する関数です。"""
logging.info("[PROCESS] 店舗ごとの売上合計を集計します。")
grouped = df.groupby("店舗")["売上"].sum().reset_index()
grouped.rename(columns={"売上": "売上合計"}, inplace=True)
logging.info("[PROCESS] 集計成功")
return grouped
def save_output(df_result: pd.DataFrame, output_path: str):
"""出力：集計結果をCSVとして保存する関数です。"""
logging.info(f"[OUTPUT] 集計結果を {output_path} に保存します。")
df_result.to_csv(output_path, index=False, encoding="utf-8")
logging.info("[OUTPUT] 保存成功")
def run_job(config_path: str = "config.json"):
"""ジョブ全体を実行する関数です（設定・ログ・エラー処理込み）。"""
config = load_config(config_path)
setup_logger(config["log_path"])
logging.info("=== ジョブ開始 ===")
try:
df_input = load_input(config["input_path"])
df_result = process_data(df_input)
save_output(df_result, config["output_path"])
logging.info("=== ジョブ正常終了 ===")
except Exception as e:
logging.error(f"ジョブがエラーで終了しました: {e}")
logging.info("=== ジョブ異常終了 ===")
# ここでメール通知などを行うこともできます。
def main():
run_job("config.json")
if __name__ == "__main__":
main()
このテンプレートは、
- 入力・処理・出力・ログ・設定・エラー処理 を一通り含んだ「業務自動化の型」になっています。
- 失敗してもログを見て原因を特定し、設定を変えたり入力ファイルを修正したりして、再実行しやすい構造です。
Day 44のまとめ
Day 44では、業務自動化設計として、
- 入力・処理・出力を関数に分けて「パイプライン」として捉える考え方
- loggingモジュールを使って「何が起きたか」を後から追えるようにするログ設計
- try / except とログを組み合わせたエラー処理の基本
- 設定ファイル（JSON）でコードと設定を分離し、変更に強い設計にする方法
- 再実行しやすいように「入力を壊さない」「出力は再生成可能」「処理をステップに分ける」ことを意識するポイント
- これらをまとめた「業務自動化ジョブのテンプレートコード」
をステップバイステップで整理しました。
ここまで来ると、 「とりあえず動くスクリプト」から一歩進んで、「業務で安心して使える自動化システム」を設計する視点 が身についているはずです。
このDay 44の考え方をベースに、 これまで学んだCSV・Excel・PDF・pandas・メール・定期実行を組み合わせて、 自分の現場に合わせた業務自動化フローを設計・構築していくことができます。