はじめに
前回は、複数テーブルのリレーション・JOIN・集計クエリを使った研修管理を実装しました。
前回の記事：Python × Dify × RAGで学ぶ業務システム開発入門【第6回 研修管理編】
第7回では、Excel業務の自動化を実装します。
人事・研修業務で最も時間を取られる作業の1つが「Excelとの格闘」です。
現状の課題
→ 社員台帳Excelを手作業でシステムに入力している
→ 研修結果Excelから1行ずつコピペしている
→ 評価シートを1人ずつExcelで手作業作成している
本回で実現すること
→ Excelアップロード → pandas で読込 → DBに一括登録
→ DBのデータ → openpyxl で整形 → Excelファイルとしてダウンロード
本連載は「作って終わり」ではなく、なぜそう設計するのかを重視しています。
設計の意図を理解することで、他のシステムにも応用できる力が身につきます。
本記事で学ぶこと
| 項目 | 内容 |
|---|---|
| pandas でのExcel読込 |
pd.read_excel でデータをDataFrameに変換する |
| データのバリデーション | Excel内のデータに不整合がないか検証する |
| DBへの一括登録 | DataFrameのレコードをループでINSERTする |
| openpyxl での書き出し | DBのデータをExcelファイルとして出力する |
| ファイルアップロード | Flaskでファイルを受け取り安全に保存する |
| ファイルダウンロード | 生成したExcelをブラウザからダウンロードさせる |
pandas と openpyxl の役割分担
Excel処理にはpandasとopenpyxlの2つのライブラリを使います。
それぞれ得意分野が異なります。
| ライブラリ | 得意なこと | 使う場面 |
|---|---|---|
| pandas | データの読込・変換・集計 | Excelインポート（読込→DB登録） |
| openpyxl | セルの書式設定・レイアウト調整 | Excel出力（DB→帳票作成） |
Excelインポート（取込）
Excel → pandas.read_excel → DataFrame → DB登録
pandasが得意：列名マッピング・型変換・欠損値処理
Excelエクスポート（出力）
DB → データ取得 → openpyxl で整形 → Excelダウンロード
openpyxlが得意：ヘッダー装飾・列幅調整・罫線・シート名設定
Excelインポートの全体フロー
利用者がExcelファイルをアップロード
↓
Flask がファイルを受け取り、uploads/ に保存
↓
拡張子チェック（.xlsx / .csv のみ許可）
↓
pandas でExcelを読み込み → DataFrame
↓
列名のマッピング（日本語ヘッダー → 英語カラム名）
↓
行ごとにバリデーション
↓ エラーあり
エラー行と理由を一覧表示。DB登録はしない
↓ エラーなし
DBにINSERT（重複はスキップ）
↓
結果メッセージ表示（登録件数 / スキップ件数）
実装：excel_service.py（インポート）
# services/excel_service.py
import os
import pandas as pd
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
from config import Config
from database import get_db
def allowed_file(filename: str) -> bool:
"""アップロードファイルの拡張子チェック"""
return (
"." in filename
and filename.rsplit(".", 1)[1].lower() in Config.ALLOWED_EXTENSIONS
)
# ============================================================
# 社員台帳インポート
# ============================================================
def import_employee_excel(file_path: str) -> dict:
"""
社員台帳Excelを読み込んでDBに一括登録する
想定するExcelの列構成:
A列: 社員番号
B列: 氏名
C列: 部署
D列: 役職
E列: 入社日
Returns:
{
"success_count": 登録成功件数,
"skip_count": スキップ件数（重複など）,
"errors": [エラーメッセージのリスト]
}
"""
result = {"success_count": 0, "skip_count": 0, "errors": []}
try:
# pandasでExcel読込
df = pd.read_excel(file_path, header=0)
except Exception as e:
result["errors"].append(f"Excelファイルの読込に失敗しました: {str(e)}")
return result
# 列数チェック
if len(df.columns) < 5:
result["errors"].append(
f"列数が不足しています（5列必要ですが{len(df.columns)}列しかありません）"
)
return result
# 列名を英語にマッピング
df.columns = ["employee_no", "name", "department", "position", "hire_date"]
# 入社日を文字列に変換（NaT対策）
df["hire_date"] = pd.to_datetime(df["hire_date"], errors="coerce")
df["hire_date"] = df["hire_date"].apply(
lambda x: x.strftime("%Y-%m-%d") if pd.notna(x) else None
)
conn = get_db()
for idx, row in df.iterrows():
row_num = idx + 2 # Excelの行番号（ヘッダー=1行目）
# 必須チェック
employee_no = str(row["employee_no"]).strip() if pd.notna(row["employee_no"]) else ""
name = str(row["name"]).strip() if pd.notna(row["name"]) else ""
if not employee_no:
result["errors"].append(f"{row_num}行目: 社員番号が空です")
continue
if not name:
result["errors"].append(f"{row_num}行目: 氏名が空です")
continue
# 重複チェック
existing = conn.execute(
"SELECT id FROM employee WHERE employee_no = ?",
(employee_no,)
).fetchone()
if existing:
result["skip_count"] += 1
continue
# DB登録
conn.execute(
"""
INSERT INTO employee (employee_no, name, department, position, hire_date)
VALUES (?, ?, ?, ?, ?)
""",
(
employee_no,
name,
str(row["department"]).strip() if pd.notna(row["department"]) else None,
str(row["position"]).strip() if pd.notna(row["position"]) else None,
row["hire_date"],
)
)
result["success_count"] += 1
conn.commit()
conn.close()
return result
# ============================================================
# 研修結果インポート
# ============================================================
def import_training_excel(file_path: str, training_id: int) -> dict:
"""
研修結果Excelを読み込んでDBに一括登録する
想定するExcelの列構成:
A列: 社員番号
B列: 氏名（参考表示用・DB登録には使わない）
C列: 出席率
D列: 理解度
E列: 課題スコア
Returns:
{
"success_count": 登録成功件数,
"skip_count": スキップ件数,
"errors": [エラーメッセージのリスト]
}
"""
result = {"success_count": 0, "skip_count": 0, "errors": []}
try:
df = pd.read_excel(file_path, header=0)
except Exception as e:
result["errors"].append(f"Excelファイルの読込に失敗しました: {str(e)}")
return result
if len(df.columns) < 5:
result["errors"].append(
f"列数が不足しています（5列必要ですが{len(df.columns)}列しかありません）"
)
return result
df.columns = ["employee_no", "name", "attendance_rate",
"understanding_level", "report_score"]
conn = get_db()
for idx, row in df.iterrows():
row_num = idx + 2
employee_no = str(row["employee_no"]).strip() if pd.notna(row["employee_no"]) else ""
if not employee_no:
result["errors"].append(f"{row_num}行目: 社員番号が空です")
continue
# 社員IDを取得
emp = conn.execute(
"SELECT id FROM employee WHERE employee_no = ?",
(employee_no,)
).fetchone()
if emp is None:
result["errors"].append(
f"{row_num}行目: 社員番号 {employee_no} はシステムに登録されていません"
)
continue
employee_id = emp["id"]
# 数値バリデーション
try:
attendance = float(row["attendance_rate"]) if pd.notna(row["attendance_rate"]) else 0.0
understanding = int(row["understanding_level"]) if pd.notna(row["understanding_level"]) else 0
report = int(row["report_score"]) if pd.notna(row["report_score"]) else 0
except (ValueError, TypeError):
result["errors"].append(f"{row_num}行目: 数値が不正です")
continue
# 既存レコードがあれば更新、なければ登録
existing = conn.execute(
"""
SELECT id FROM training_history
WHERE employee_id = ? AND training_id = ?
""",
(employee_id, training_id)
).fetchone()
if existing:
conn.execute(
"""
UPDATE training_history
SET attendance_rate = ?, understanding_level = ?, report_score = ?
WHERE id = ?
""",
(attendance, understanding, report, existing["id"])
)
else:
conn.execute(
"""
INSERT INTO training_history
(employee_id, training_id, attendance_rate,
understanding_level, report_score)
VALUES (?, ?, ?, ?, ?)
""",
(employee_id, training_id, attendance, understanding, report)
)
result["success_count"] += 1
conn.commit()
conn.close()
return result
実装：excel_service.py（エクスポート）
# ============================================================
# 評価シート出力
# ============================================================
def export_evaluation_excel(training_id: int) -> str:
"""
研修の評価データをExcelファイルに出力する
出力内容:
社員番号 | 氏名 | 部署 | 出席率 | 理解度 | 課題スコア | 総合評価
Returns:
生成したExcelファイルのパス
"""
conn = get_db()
rows = conn.execute(
"""
SELECT
e.employee_no,
e.name,
e.department,
th.attendance_rate,
th.understanding_level,
th.report_score
FROM training_history th
JOIN employee e ON th.employee_id = e.id
WHERE th.training_id = ?
ORDER BY e.employee_no
""",
(training_id,)
).fetchall()
training = conn.execute(
"SELECT training_name FROM training WHERE id = ?",
(training_id,)
).fetchone()
conn.close()
training_name = training["training_name"] if training else "研修"
# openpyxl でワークブック作成
wb = Workbook()
ws = wb.active
ws.title = "評価シート"
# --- ヘッダーのスタイル定義 ---
header_font = Font(bold=True, size=11)
header_fill = PatternFill(
start_color="4472C4", end_color="4472C4", fill_type="solid"
)
header_font_white = Font(bold=True, size=11, color="FFFFFF")
thin_border = Border(
left=Side(style="thin"),
right=Side(style="thin"),
top=Side(style="thin"),
bottom=Side(style="thin")
)
# --- タイトル行 ---
ws.merge_cells("A1:G1")
ws["A1"] = f"評価シート：{training_name}"
ws["A1"].font = Font(bold=True, size=14)
# --- ヘッダー行 ---
headers = ["社員番号", "氏名", "部署", "出席率(%)",
"理解度", "課題スコア", "総合評価"]
for col, header in enumerate(headers, start=1):
cell = ws.cell(row=3, column=col, value=header)
cell.font = header_font_white
cell.fill = header_fill
cell.border = thin_border
cell.alignment = Alignment(horizontal="center")
# --- データ行 ---
for row_idx, row in enumerate(rows, start=4):
data = dict(row)
# 総合評価 = (出席率 + 理解度 + 課題スコア) / 3
total = round(
(data["attendance_rate"] + data["understanding_level"]
+ data["report_score"]) / 3, 1
)
values = [
data["employee_no"],
data["name"],
data["department"] or "―",
data["attendance_rate"],
data["understanding_level"],
data["report_score"],
total,
]
for col, value in enumerate(values, start=1):
cell = ws.cell(row=row_idx, column=col, value=value)
cell.border = thin_border
if col >= 4: # 数値列は右寄せ
cell.alignment = Alignment(horizontal="right")
# --- 列幅調整 ---
column_widths = [12, 14, 12, 12, 10, 12, 10]
for col, width in enumerate(column_widths, start=1):
ws.column_dimensions[chr(64 + col)].width = width
# ファイル保存
output_path = os.path.join(
Config.OUTPUT_FOLDER,
f"evaluation_{training_id}.xlsx"
)
wb.save(output_path)
return output_path
実装：app.py（インポート・エクスポートルーティング）
# app.py（Excel インポート・エクスポート部分）
import os
from flask import send_file
from werkzeug.utils import secure_filename
from services.excel_service import (
allowed_file, import_employee_excel,
import_training_excel, export_evaluation_excel
)
from services.training_service import get_all_trainings
@app.route("/import", methods=["GET", "POST"])
@login_required
@roles_required("admin", "staff")
def import_excel():
"""Excelインポート画面"""
trainings = get_all_trainings() # 研修結果インポート用のプルダウン
if request.method == "POST":
# ファイルの存在チェック
if "file" not in request.files:
flash("ファイルが選択されていません", "error")
return render_template("import.html", trainings=trainings)
file = request.files["file"]
if file.filename == "":
flash("ファイルが選択されていません", "error")
return render_template("import.html", trainings=trainings)
# 拡張子チェック
if not allowed_file(file.filename):
flash("許可されていないファイル形式です（.xlsx または .csv のみ）", "error")
return render_template("import.html", trainings=trainings)
# ファイルを安全に保存
filename = secure_filename(file.filename)
file_path = os.path.join(app.config["UPLOAD_FOLDER"], filename)
file.save(file_path)
# インポート種別に応じて処理を分岐
import_type = request.form.get("import_type", "employee")
if import_type == "employee":
result = import_employee_excel(file_path)
elif import_type == "training":
training_id = request.form.get("training_id")
if not training_id:
flash("研修を選択してください", "error")
return render_template("import.html", trainings=trainings)
result = import_training_excel(file_path, int(training_id))
else:
flash("不正なインポート種別です", "error")
return render_template("import.html", trainings=trainings)
# アップロードファイルを削除（一時ファイルのため）
os.remove(file_path)
# 結果メッセージ
if result["errors"]:
for err in result["errors"]:
flash(err, "error")
if result["success_count"] > 0:
flash(f"{result['success_count']}件を登録しました", "success")
if result["skip_count"] > 0:
flash(f"{result['skip_count']}件をスキップしました（重複）", "info")
return render_template("import.html", trainings=trainings)
return render_template("import.html", trainings=trainings)
@app.route("/export/<int:training_id>")
@login_required
@roles_required("admin", "staff")
def export_excel(training_id):
"""評価シートExcelダウンロード"""
output_path = export_evaluation_excel(training_id)
return send_file(
output_path,
as_attachment=True,
download_name=f"evaluation_{training_id}.xlsx",
mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
)
secure_filename はファイル名から危険な文字（../ など）を除去します。
ユーザーが送信したファイル名をそのまま使うと、ディレクトリトラバーサル攻撃を受ける可能性があります。
実装：テンプレート（import.html）
{% extends "base.html" %}
{% block title %}Excelインポート - AI人事・研修システム{% endblock %}
{% block content %}
<div class="page-header">
<h1>📥 Excelインポート</h1>
</div>
<div class="form-card">
<form method="POST" enctype="multipart/form-data">
<!-- インポート種別 -->
<div class="form-group">
<label>インポート種別 <span class="required">*</span></label>
<div class="radio-group">
<label>
<input type="radio" name="import_type" value="employee" checked
onchange="document.getElementById('training-select').style.display='none'">
社員台帳
</label>
<label>
<input type="radio" name="import_type" value="training"
onchange="document.getElementById('training-select').style.display='block'">
研修結果
</label>
</div>
</div>
<!-- 研修選択（研修結果の場合のみ表示） -->
<div class="form-group" id="training-select" style="display:none;">
<label for="training_id">対象研修</label>
<select name="training_id" id="training_id">
<option value="">-- 研修を選択 --</option>
{% for tr in trainings %}
<option value="{{ tr.id }}">{{ tr.training_name }}</option>
{% endfor %}
</select>
</div>
<!-- ファイル選択 -->
<div class="form-group">
<label for="file">Excelファイル <span class="required">*</span></label>
<input type="file" id="file" name="file"
accept=".xlsx,.csv" required>
<small class="text-muted">
対応形式：.xlsx / .csv（最大16MB）
</small>
</div>
<button type="submit" class="btn btn-primary">インポート実行</button>
</form>
</div>
<!-- Excelフォーマット説明 -->
<div class="info-card">
<h3>📋 社員台帳Excelのフォーマット</h3>
<table class="table table-compact">
<thead>
<tr><th>A列</th><th>B列</th><th>C列</th><th>D列</th><th>E列</th></tr>
</thead>
<tbody>
<tr><td>社員番号</td><td>氏名</td><td>部署</td><td>役職</td><td>入社日</td></tr>
<tr><td>EMP001</td><td>山田太郎</td><td>人事部</td><td>主任</td><td>2020/4/1</td></tr>
</tbody>
</table>
<h3>📋 研修結果Excelのフォーマット</h3>
<table class="table table-compact">
<thead>
<tr><th>A列</th><th>B列</th><th>C列</th><th>D列</th><th>E列</th></tr>
</thead>
<tbody>
<tr><td>社員番号</td><td>氏名</td><td>出席率</td><td>理解度</td><td>課題スコア</td></tr>
<tr><td>EMP001</td><td>山田太郎</td><td>90</td><td>85</td><td>95</td></tr>
</tbody>
</table>
</div>
{% endblock %}
ファイルアップロードのセキュリティ
ファイルアップロード機能にはセキュリティリスクが伴います。
本システムでは以下の対策を行っています。
| 対策 | 実装方法 | 防ぐ攻撃 |
|---|---|---|
| 拡張子チェック |
allowed_file() で .xlsx / .csv のみ許可 |
悪意あるファイルの実行 |
| ファイル名のサニタイズ |
secure_filename() で危険な文字を除去 |
ディレクトリトラバーサル |
| ファイルサイズ制限 | MAX_CONTENT_LENGTH = 16MB |
DoS攻撃（巨大ファイル送信） |
| 一時ファイルの削除 | 処理完了後に os.remove()
|
ディスク圧迫・情報漏洩 |
enctype="multipart/form-data" をフォームタグに付け忘れると、ファイルが送信されません。
ファイルアップロードフォームでは必ず設定してください。
動作確認
手順
# Flaskアプリを起動
python app.py
# ブラウザで http://localhost:5000/import にアクセス
テスト用Excelファイルの作成
動作確認用のExcelファイルを用意します。
# create_test_excel.py
"""テスト用Excelファイルを作成するスクリプト"""
import pandas as pd
# 社員台帳テスト
employee_data = {
"社員番号": ["EMP001", "EMP002", "EMP003"],
"氏名": ["山田太郎", "鈴木花子", "田中一郎"],
"部署": ["人事部", "研修部", "営業部"],
"役職": ["主任", "担当", "係長"],
"入社日": ["2020-04-01", "2021-04-01", "2019-04-01"],
}
df_emp = pd.DataFrame(employee_data)
df_emp.to_excel("test_employee.xlsx", index=False)
print("test_employee.xlsx を作成しました")
# 研修結果テスト
training_data = {
"社員番号": ["EMP001", "EMP002", "EMP003"],
"氏名": ["山田太郎", "鈴木花子", "田中一郎"],
"出席率": [100, 90, 85],
"理解度": [85, 70, 75],
"課題スコア": [95, 80, 70],
}
df_tr = pd.DataFrame(training_data)
df_tr.to_excel("test_training_result.xlsx", index=False)
print("test_training_result.xlsx を作成しました")
確認ポイント
| 操作 | 期待する動作 |
|---|---|
| 社員台帳Excelをアップロード | 「3件を登録しました」と表示される |
| 同じExcelを再アップロード | 「3件をスキップしました（重複）」と表示される |
| 社員番号が空の行を含むExcel | 「X行目: 社員番号が空です」エラーが表示される |
| .txt ファイルをアップロード | 「許可されていないファイル形式です」と表示される |
| 研修結果Excelをアップロード | 受講履歴が登録・更新される |
| 評価シートExcel出力リンクを押す | Excelがダウンロードされる |
| ダウンロードしたExcelを開く | ヘッダーに色・罫線がついた帳票が表示される |
今後の連載予定
| 回 | タイトル | 主な内容 |
|---|---|---|
| 第1回 | 業務システム全体設計編 | システム概要・技術選定・アーキテクチャ |
| 第2回 | 要求定義・要件定義編 | 業務分析・機能要件・非機能要件 |
| 第3回 | ER図・画面設計編 | テーブル設計・画面遷移図・ワイヤーフレーム |
| 第4回 | Flaskログイン機能編 | セッション・パスワードハッシュ・認証ミドルウェア |
| 第5回 | 社員管理CRUD編 | 一覧・登録・更新・削除・バリデーション |
| 第6回 | 研修管理編 | リレーション・集計・出席率自動計算 |
| 第7回 | Excel業務自動化編（本記事） | pandas取込・openpyxl出力・テンプレート活用 |
| 第8回 | Dify API連携編 | プロンプト設計・API呼び出し・エラーハンドリング |
| 第9回 | RAG構築編 | Knowledgeへの登録・Embedding・検索精度改善 |
| 第10回 | FAQチャットボット編 | チャットUI・会話履歴・ストリーミングレスポンス |
| 第11回 | GitHubチーム開発編 | ブランチ戦略・Pull Request・コードレビュー |
| 第12回 | テスト・発表編 | pytest・テスト設計・デモ・振り返り |
おわりに
第7回では、pandasによるExcelインポートとopenpyxlによるExcelエクスポートを実装しました。
ポイントを振り返ります。
-
pandas読込：
pd.read_excelでDataFrameに変換し、列名マッピング・型変換を行った - 行単位のバリデーション：必須チェック・重複チェック・数値チェックを行単位で実施し、エラー行を特定した
- UPSERT処理：研修結果インポートでは既存データがあれば更新、なければ新規登録するパターンを実装した
- openpyxl出力：ヘッダー装飾・罫線・列幅調整を施した実務レベルの帳票を生成した
- セキュリティ：拡張子チェック・ファイル名サニタイズ・サイズ制限・一時ファイル削除を実施した
次回は**「Dify API連携編」**として、受講者の評価データをDify APIに送信し、AI評価コメントを自動生成する機能を実装します。
次回 : Python × Dify × RAGで学ぶ業務システム開発入門【第8回 Dify API連携編】