業務で毎月発生する「データ集計 → 整形 → 出力(Excel)」を手作業でやっていませんか?
この記事では、Python(pandas)で集計処理を行い、OpenPyXLで見栄え良く書き出して「月次報告書(Excelファイル)」を自動生成する実践的ワークフローを段階的に解説します。中級者向けに、実務で使えるエラーハンドリング・ログ出力・スケジューリングまで含めています。
想定する業務フロー(例)
- 各部門からCSV/Excelでデータがアップロードされる(
/data/incoming/
)。 - スクリプトがファイルを読み込み、不要列を削除、集計(合計・件数・平均など)を行う。
- ピボットや集計表を作成。
- OpenPyXLでフォーマット(見出し、罫線、数値書式、グラフ)して最終Excelを出力(
/reports/YYYY-MM.xlsx
)。 - 必要ならメール送付または共有フォルダへ保存。
(この流れをWindowsタスクスケジューラやcronで定期実行)
必要な環境とライブラリ
- Python 3.9+(3.8 でも可)
- pandas
- openpyxl
- numpy(任意)
- python-dateutil(pandasに同梱される場合あり)
- optional: matplotlib(グラフを画像で貼る場合)、xlsxwriter(別の書式エンジン)
インストール例:
python -m pip install pandas openpyxl numpy
# グラフを作るなら
python -m pip install matplotlib
サンプルデータの想定フォーマット
/data/incoming/sales_2025-08.csv
の一例
date,department,product,units,unit_price
2025-08-01,営業A,Product X,3,1200
2025-08-01,営業B,Product Y,1,2300
2025-08-02,営業A,Product Z,2,500
...
date
は ISO 形式(YYYY-MM-DD)- 数値列は
int/float
フルワークフローのサンプルスクリプト
下記は実務でそのまま使えるサンプルです。
- 入力フォルダからCSV/XLSXを検出して結合
- 集計(部署別売上、日別トレンド、商品別トップN)
- OpenPyXLでの整形(ヘッダスタイル、列幅、自動書式、罫線)
- 最終ファイル保存(
reports/YYYY-MM月次報告.xlsx
)
# monthly_report.py
import os
from pathlib import Path
import glob
import logging
from datetime import datetime
import pandas as pd
import numpy as np
from openpyxl import load_workbook
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side, NamedStyle
from openpyxl.utils.dataframe import dataframe_to_rows
# --- 設定 ---
BASE_DIR = Path(__file__).resolve().parent
IN_DIR = BASE_DIR / "data" / "incoming"
OUT_DIR = BASE_DIR / "reports"
OUT_DIR.mkdir(parents=True, exist_ok=True)
LOG_FILE = BASE_DIR / "logs" / "monthly_report.log"
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
# --- ログ ---
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
handlers=[
logging.FileHandler(LOG_FILE, encoding="utf-8"),
logging.StreamHandler()
]
)
# --- ヘルパー ---
def read_input_files(input_dir: Path):
logging.info(f"Reading files from {input_dir}")
files = list(input_dir.glob("*.csv")) + list(input_dir.glob("*.xlsx"))
dfs = []
for f in files:
try:
if f.suffix.lower() == ".csv":
df = pd.read_csv(f)
else:
df = pd.read_excel(f)
logging.info(f"Loaded {f.name} ({len(df)} rows)")
dfs.append(df)
except Exception as e:
logging.exception(f"Failed to read {f}: {e}")
if not dfs:
logging.warning("No input files found.")
return pd.DataFrame()
return pd.concat(dfs, ignore_index=True, sort=False)
def ensure_columns(df: pd.DataFrame, required_cols):
missing = [c for c in required_cols if c not in df.columns]
if missing:
raise ValueError(f"Missing required columns: {missing}")
# --- 集計処理 ---
def process_data(df: pd.DataFrame, target_month: str):
# target_month e.g. '2025-08'
if df.empty:
return {}
# 基本整形
df['date'] = pd.to_datetime(df['date'], errors='coerce')
df = df.dropna(subset=['date'])
# フィルタ:対象月のみ
df = df[df['date'].dt.strftime("%Y-%m") == target_month].copy()
if df.empty:
logging.warning("No data for target month.")
return {}
# 数値列の変換
df['units'] = pd.to_numeric(df.get('units', 0), errors='coerce').fillna(0).astype(int)
df['unit_price'] = pd.to_numeric(df.get('unit_price', 0), errors='coerce').fillna(0.0).astype(float)
df['amount'] = df['units'] * df['unit_price']
# 部署別集計
dept = df.groupby('department').agg(
total_amount=('amount', 'sum'),
total_units=('units', 'sum'),
avg_unit_price=('unit_price', 'mean'),
transactions=('amount', 'count')
).reset_index().sort_values('total_amount', ascending=False)
# 日次トレンド
daily = df.groupby(df['date'].dt.date).agg(
daily_amount=('amount', 'sum'),
daily_units=('units', 'sum'),
).reset_index().rename(columns={'date': 'day'})
# 商品別トップN
product = df.groupby('product').agg(
total_amount=('amount', 'sum'),
total_units=('units', 'sum')
).reset_index().sort_values('total_amount', ascending=False)
top_products = product.head(10)
# その他必要な表を返す
return {
'raw': df,
'department': dept,
'daily': daily,
'top_products': top_products
}
# --- Excel出力(OpenPyXL) ---
def style_sheet(ws):
# 行ヘッダー用スタイル
header_font = Font(bold=True)
for cell in list(ws[1]):
cell.font = header_font
cell.alignment = Alignment(horizontal='center', vertical='center')
cell.fill = PatternFill(start_color="FFEECC", end_color="FFEECC", fill_type="solid")
# 列幅自動調整(シンプル版)
for column_cells in ws.columns:
lengths = [len(str(cell.value)) if cell.value is not None else 0 for cell in column_cells]
maxlen = max(lengths) + 2
col_letter = column_cells[0].column_letter
ws.column_dimensions[col_letter].width = min(maxlen, 50)
def write_report(out_path: Path, data_dict: dict, report_title: str):
from openpyxl import Workbook
wb = Workbook()
# 各表を別シートで書き出し
for sheet_name, df in [('Summary', data_dict.get('department')),
('Daily', data_dict.get('daily')),
('TopProducts', data_dict.get('top_products'))]:
if df is None or df.empty:
continue
ws = wb.create_sheet(title=sheet_name)
rows = dataframe_to_rows(df, index=False, header=True)
for r_idx, row in enumerate(rows, 1):
ws.append(row)
style_sheet(ws)
# raw データシート(非表示にすることも可能)
raw = data_dict.get('raw')
if raw is not None and not raw.empty:
ws_raw = wb.create_sheet(title='RawData')
for r in dataframe_to_rows(raw, index=False, header=True):
ws_raw.append(r)
ws_raw.sheet_state = 'hidden'
# Remove default sheet if empty
if 'Sheet' in wb.sheetnames and wb['Sheet'].max_row == 1:
wb.remove(wb['Sheet'])
# 保存
wb.save(out_path)
logging.info(f"Report saved to {out_path}")
# --- メイン ---
def main(target_month=None):
if target_month is None:
# 前月をデフォルトターゲットにする例
today = datetime.today()
first = today.replace(day=1)
prev_month_last = first - pd.Timedelta(days=1)
target_month = prev_month_last.strftime("%Y-%m")
logging.info(f"Generating report for {target_month}")
df_all = read_input_files(IN_DIR)
if df_all.empty:
logging.error("No data loaded. Exiting.")
return
required = ['date', 'department', 'product', 'units', 'unit_price']
try:
ensure_columns(df_all, required)
except ValueError as e:
logging.error(e)
return
results = process_data(df_all, target_month)
if not results:
logging.error("No results to write. Exiting.")
return
out_file = OUT_DIR / f"{target_month}_月次報告.xlsx"
write_report(out_file, results, report_title=f"{target_month} 月次報告")
logging.info("Done.")
if __name__ == "__main__":
main()
補足と実務Tips(詳細)
1) 入力データのバリデーション
date
のパースに失敗した行はログに残してスキップする方が安全。- カラム名のバリエーション(例:
unit price
,Unit_Price
)が混在する場合は読み込んだあとにカラム正規化(df.columns = df.columns.str.strip().str.lower().str.replace(' ', '_')
)を検討。
2) 大量データ対応
- 数十万〜百万行規模なら
chunksize
読みや、データベース(Postgres, BigQuery)にアップして集計する方が安定。 - pandasでメモリ不足が起きる場合は
dask.dataframe
を検討。
3) フォーマット強化:OpenPyXLでの体裁
- 列ごとの数値書式を設定:
cell.number_format = '#,##0'
や#,##0.00
。 - 罫線(
Border(Side(...))
)を引くと見やすい。 - 重要項目は色付きセルで強調。
- グラフを挿入する場合は
openpyxl.chart
を使用してExcel内グラフを作成可能(ただし少しコードが複雑)。
4) Excel内でのピボットは不要に
- 集計はPython側で済ませ、出力結果をそのまま配布すればユーザはそのまま使えます。ピボットテーブルをユーザに渡したい場合は
openpyxl
でピボット作成も可能ですが、互換性が面倒なので事前に検討。
5) 自動実行(スケジューリング)
- Windows: タスクスケジューラで Python スクリプトを指定(引数で
YYYY-MM
を渡せるように)。 - Linux/macOS:
cron
で実行。ログ出力をファイルに残すこと。
例(cron: 毎月1日 6:00 に実行)
0 6 1 * * /usr/bin/python3 /path/to/monthly_report.py >> /path/to/logs/cron_monthly.log 2>&1
6) 完成物の配布(オプション)
- 共有フォルダ(社内ネットワーク)へ保存。
- メール送信(
smtplib
)で自動送付。 - Slackへファイルアップロード(Slack APIを利用)。
よくある落とし穴
- Excelのローカル日付形式(例えば Excelのシリアル日付)を直接読み書きすると日付ズレが発生することがある → pandasの
to_datetime
を必ず使う。 - 小数点や通貨の丸めルールを事前に合意しておく(四捨五入、切り捨て等)。
- 入力ファイルのカラム名が変わるとスクリプトが壊れる → カラム名マッピングや意図的に変換ロジックを入れる。
拡張案(将来的に追加したい機能)
- ダッシュボード出力(Excel内部のグラフを増やす/Power BI, Looker などへ出力)
- データソースを Google Sheets / Google Drive / AWS S3 に対応
- テンプレートExcelに動的に値を埋め込む(社内テンプレートを保持しておき、指定セルに集計値を埋める)
- 単体テスト(pytest)で集計ロジックの検証を自動化
最後に:導入チェックリスト
- 入力フォルダのパスと権限を確認
- 必要なライブラリをインストール(pandas, openpyxl)
- サンプルファイルでスクリプトを実行して期待通りのExcelが出力されるか確認
- スケジューリング(タスクスケジューラ / cron)を設定
- ログとエラーハンドリングの運用ルールを決定