NaN/NaN/NaN
イベント終了後に500名分の参加証を発行する、オンライン講座の修了者1,000名に修了証を送る、資格試験の合格者への認定証を毎月バッチ発行する――このような規模になると、手動での作業は現実的ではありません。
この記事では、CSVデータから証明書PDFを一括自動発行するシステムをFUNBREW PDF APIを使って構築する方法を解説します。テンプレートHTMLへのデータ差し込み、並列API呼び出しによる高速生成、ZIPでの一括ダウンロード、エラーハンドリングまで、Node.jsとPython両方のコード例を使って実践的に説明します。
証明書テンプレートのデザインについてはHTML証明書テンプレート集を、証明書PDFの基本的な自動化フローについては証明書PDF自動生成ガイドをあわせてご覧ください。大量PDF生成の基礎についてはPDF一括生成ガイドも参考になります。
一括発行が必要になるケース
業界別ユースケース
| 業界 | ユースケース | 発行枚数の目安 |
|---|---|---|
| オンライン教育・eラーニング | コース修了証・学習バッジ | 月100〜5,000枚 |
| 企業研修 | 社内研修修了証・コンプライアンス証明 | 月50〜500枚 |
| カンファレンス・セミナー | 参加証・CPD単位証明 | イベントごとに100〜10,000枚 |
| 資格認定団体 | 資格証明書・更新証 | 月50〜1,000枚 |
| 学校・教育機関 | 卒業証書・成績優秀証 | 年1回に大量発行 |
| スポーツ大会 | 完走証・参加賞 | イベントごとに200〜5,000枚 |
手動発行の限界
10枚なら手作業でも許容範囲です。しかし100枚を超えたあたりから問題が顕在化します。
- 作業時間: 1枚5分の作業でも100枚で8時間以上
- ヒューマンエラー: 名前の誤字、日付のズレ、枚数が増えるほどリスクが上がる
- スケーラビリティ: 毎月の定期発行や急増への対応が困難
- ブランド一貫性: 手動では微妙なレイアウトのばらつきが発生しやすい
100枚以上を扱う場合は、自動化への投資対効果が高くなります。
CSVから証明書を量産する仕組み
一括発行の処理フローは以下のとおりです。
CSV読み込み → データ検証 → テンプレートHTML生成 → API呼び出し（並列） → PDF保存 → ZIP化
CSVの構造例
id,name,email,course,completion_date,score
001,田中 太郎,tanaka@example.com,Python実践講座,2026年4月18日,92
002,鈴木 花子,suzuki@example.com,Python実践講座,2026年4月18日,88
003,佐藤 次郎,sato@example.com,Python実践講座,2026年4月18日,95
テンプレートHTMLの設計
プレースホルダー（{{変数名}}）でデータの差し込み箇所を明示します。詳細なテンプレートデザインはHTML証明書テンプレート集を参照してください。
<!DOCTYPE html>
<html lang="ja">
<head>
<meta charset="UTF-8">
<style>
@page { size: A4 landscape; margin: 0; }
body {
margin: 0; padding: 0;
font-family: "Noto Sans JP", "Hiragino Sans", sans-serif;
background: #fff;
}
.certificate {
width: 297mm; height: 210mm;
display: flex; flex-direction: column;
align-items: center; justify-content: center;
text-align: center; position: relative;
}
.border { position: absolute; top: 10mm; left: 10mm; right: 10mm; bottom: 10mm; border: 3px double #1a365d; }
.inner { position: absolute; top: 15mm; left: 15mm; right: 15mm; bottom: 15mm; border: 1px solid #c4a35a; }
.content { position: relative; z-index: 1; padding: 20mm 30mm; }
.label { font-size: 11pt; color: #888; letter-spacing: 0.4em; margin-bottom: 8mm; }
.title { font-size: 32pt; color: #1a365d; font-weight: bold; margin-bottom: 12mm; letter-spacing: 0.3em; }
.name { font-size: 22pt; color: #333; border-bottom: 2px solid #1a365d; padding-bottom: 3mm; display: inline-block; min-width: 140mm; margin-bottom: 10mm; }
.body-text { font-size: 12pt; color: #555; line-height: 2; margin-bottom: 12mm; }
.meta { display: flex; justify-content: center; gap: 20mm; font-size: 10pt; color: #666; margin-bottom: 10mm; }
.org { font-size: 13pt; color: #1a365d; font-weight: bold; }
.cert-id { position: absolute; bottom: 18mm; right: 20mm; font-size: 8pt; color: #aaa; }
</style>
</head>
<body>
<div class="certificate">
<div class="border"></div>
<div class="inner"></div>
<div class="content">
<div class="label">CERTIFICATE OF COMPLETION</div>
<div class="title">修 了 証</div>
<div class="name">{{name}}</div>
<div class="body-text">
上記の者は「{{course}}」の全課程を優秀な成績で修了したことを証します。
</div>
<div class="meta">
<span>修了日: {{completion_date}}</span>
<span>スコア: {{score}}点</span>
</div>
<div class="org">{{organization}}</div>
</div>
<div class="cert-id">No. {{id}}</div>
</div>
</body>
</html>
実装例
Node.jsによる実装
npm install csv-parse handlebars archiver axios p-limit
// bulk-certificate-generator.js
const fs = require('fs');
const path = require('path');
const { parse } = require('csv-parse/sync');
const Handlebars = require('handlebars');
const archiver = require('archiver');
const axios = require('axios');
const pLimit = require('p-limit');
const API_KEY = process.env.FUNBREW_API_KEY;
const API_URL = 'https://pdf.funbrew.cloud/api/v1/generate';
const CONCURRENCY = 5; // 同時リクエスト数（レート制限に応じて調整）
/**
* CSVを読み込んで受講者リストを返す
*/
function loadRecipients(csvPath) {
const content = fs.readFileSync(csvPath, 'utf-8');
return parse(content, { columns: true, skip_empty_lines: true });
}
/**
* 1件分のPDFをAPIで生成してBufferを返す
*/
async function generateOneCertificate(recipient, template, retries = 3) {
const html = template({
...recipient,
organization: '株式会社テックアカデミー',
});
for (let attempt = 1; attempt <= retries; attempt++) {
try {
const response = await axios.post(
API_URL,
{
html,
options: {
format: 'A4',
landscape: true,
printBackground: true,
margin: { top: '0', right: '0', bottom: '0', left: '0' },
},
},
{
headers: { Authorization: `Bearer ${API_KEY}` },
responseType: 'arraybuffer',
timeout: 60000,
}
);
return { id: recipient.id, name: recipient.name, buffer: Buffer.from(response.data), status: 'success' };
} catch (err) {
if (attempt === retries) {
console.error(`失敗 [${recipient.id}] ${recipient.name}: ${err.message}`);
return { id: recipient.id, name: recipient.name, error: err.message, status: 'failed' };
}
// 指数バックオフ（1秒 → 2秒 → 4秒）
await new Promise((r) => setTimeout(r, 1000 * 2 ** (attempt - 1)));
}
}
}
/**
* 複数のPDFをZIPファイルにまとめる
*/
function archivePdfs(results, outputZipPath) {
return new Promise((resolve, reject) => {
const output = fs.createWriteStream(outputZipPath);
const archive = archiver('zip', { zlib: { level: 6 } });
output.on('close', () => resolve(archive.pointer()));
archive.on('error', reject);
archive.pipe(output);
for (const r of results) {
if (r.status === 'success') {
archive.append(r.buffer, { name: `certificate-${r.id}-${r.name}.pdf` });
}
}
archive.finalize();
});
}
/**
* メイン処理：CSV → 一括PDF生成 → ZIP化
*/
async function bulkGenerateCertificates(csvPath, outputDir = './output') {
fs.mkdirSync(outputDir, { recursive: true });
// テンプレート読み込み
const templateHtml = fs.readFileSync('certificate-template.html', 'utf-8');
const template = Handlebars.compile(templateHtml);
// CSV読み込み
const recipients = loadRecipients(csvPath);
console.log(`${recipients.length}件の証明書を生成します...`);
// 並列数制限付きで一括生成
const limit = pLimit(CONCURRENCY);
const results = await Promise.all(
recipients.map((r) => limit(() => generateOneCertificate(r, template)))
);
// 結果集計
const succeeded = results.filter((r) => r.status === 'success');
const failed = results.filter((r) => r.status === 'failed');
console.log(`完了: ${succeeded.length}件 / 失敗: ${failed.length}件`);
// 失敗リストを記録
if (failed.length > 0) {
fs.writeFileSync(
path.join(outputDir, 'failed.json'),
JSON.stringify(failed, null, 2)
);
console.log(`失敗リスト: ${path.join(outputDir, 'failed.json')}`);
}
// ZIPにまとめる
const zipPath = path.join(outputDir, `certificates-${Date.now()}.zip`);
const zipBytes = await archivePdfs(succeeded, zipPath);
console.log(`ZIP作成完了: ${zipPath} (${(zipBytes / 1024 / 1024).toFixed(1)} MB)`);
return { succeeded: succeeded.length, failed: failed.length, zipPath };
}
// 実行
bulkGenerateCertificates('./recipients.csv', './output')
.then(({ succeeded, failed, zipPath }) => {
console.log(`\n=== 完了 ===`);
console.log(`成功: ${succeeded}件 / 失敗: ${failed}件`);
console.log(`ZIP: ${zipPath}`);
})
.catch(console.error);
Pythonによる実装
pip install aiohttp aiofiles jinja2
# bulk_certificate_generator.py
import asyncio
import csv
import json
import zipfile
import os
from datetime import datetime
from pathlib import Path
import aiohttp
from jinja2 import Template
API_KEY = os.environ["FUNBREW_API_KEY"]
API_URL = "https://pdf.funbrew.cloud/api/v1/generate"
CONCURRENCY = 5 # 同時リクエスト数
def load_recipients(csv_path: str) -> list[dict]:
"""CSVを読み込んで受講者リストを返す"""
with open(csv_path, encoding="utf-8") as f:
return list(csv.DictReader(f))
async def generate_one(
session: aiohttp.ClientSession,
recipient: dict,
template: Template,
semaphore: asyncio.Semaphore,
retries: int = 3,
) -> dict:
"""セマフォ付きで1件の証明書を非同期生成する"""
html = template.render(
**recipient,
organization="株式会社テックアカデミー",
)
async with semaphore:
for attempt in range(1, retries + 1):
try:
async with session.post(
API_URL,
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"html": html,
"options": {
"format": "A4",
"landscape": True,
"printBackground": True,
"margin": {"top": "0", "right": "0", "bottom": "0", "left": "0"},
},
},
timeout=aiohttp.ClientTimeout(total=60),
) as resp:
if resp.status == 200:
pdf_bytes = await resp.read()
return {
"id": recipient["id"],
"name": recipient["name"],
"bytes": pdf_bytes,
"status": "success",
}
error = f"HTTP {resp.status}"
except Exception as e:
error = str(e)
if attempt < retries:
await asyncio.sleep(2 ** (attempt - 1)) # 指数バックオフ
print(f"失敗 [{recipient['id']}] {recipient['name']}: {error}")
return {"id": recipient["id"], "name": recipient["name"], "error": error, "status": "failed"}
def archive_pdfs(results: list[dict], output_zip: Path) -> int:
"""成功したPDFをZIPにまとめて、バイト数を返す"""
with zipfile.ZipFile(output_zip, "w", zipfile.ZIP_DEFLATED) as zf:
for r in results:
if r["status"] == "success":
filename = f"certificate-{r['id']}-{r['name']}.pdf"
zf.writestr(filename, r["bytes"])
return output_zip.stat().st_size
async def bulk_generate(csv_path: str, output_dir: str = "./output") -> dict:
"""CSV → 一括PDF生成 → ZIP化のメイン処理"""
out = Path(output_dir)
out.mkdir(exist_ok=True)
with open("certificate-template.html") as f:
template = Template(f.read())
recipients = load_recipients(csv_path)
print(f"{len(recipients)}件の証明書を生成します...")
semaphore = asyncio.Semaphore(CONCURRENCY)
async with aiohttp.ClientSession() as session:
tasks = [
generate_one(session, r, template, semaphore)
for r in recipients
]
results = await asyncio.gather(*tasks)
succeeded = [r for r in results if r["status"] == "success"]
failed = [r for r in results if r["status"] == "failed"]
print(f"完了: {len(succeeded)}件 / 失敗: {len(failed)}件")
if failed:
failed_path = out / "failed.json"
failed_path.write_text(
json.dumps([{"id": r["id"], "name": r["name"], "error": r["error"]} for r in failed],
ensure_ascii=False, indent=2)
)
print(f"失敗リスト: {failed_path}")
zip_path = out / f"certificates-{datetime.now().strftime('%Y%m%d%H%M%S')}.zip"
zip_bytes = archive_pdfs(results, zip_path)
print(f"ZIP作成完了: {zip_path} ({zip_bytes / 1024 / 1024:.1f} MB)")
return {"succeeded": len(succeeded), "failed": len(failed), "zip_path": str(zip_path)}
if __name__ == "__main__":
result = asyncio.run(bulk_generate("recipients.csv"))
print(f"\n=== 完了 ===")
print(f"成功: {result['succeeded']}件 / 失敗: {result['failed']}件")
print(f"ZIP: {result['zip_path']}")
レート制限対策
FUNBREW PDF APIには1秒あたりのリクエスト数に制限があります。制限を超えると 429 Too Many Requests が返されます。
並列数の目安
| 発行枚数 | 推奨CONCURRENCY | 処理時間の目安 |
|---|---|---|
| 〜100枚 | 5 | 1〜3分 |
| 100〜500枚 | 5〜10 | 5〜15分 |
| 500〜2,000枚 | 10〜20 | 15〜60分 |
| 2,000枚以上 | 非同期キュー推奨 | キュー方式で分割実行 |
429エラーへの対処
上記コードには指数バックオフ（1秒→2秒→4秒）のリトライが組み込まれています。429が頻発する場合は CONCURRENCY を下げるか、以下のように動的スロットリングを追加します。
// 429レスポンス時に自動でスリープしてリトライするAxiosインターセプター
axios.interceptors.response.use(null, async (err) => {
if (err.response?.status === 429) {
const retryAfter = parseInt(err.response.headers['retry-after'] || '5');
console.warn(`レート制限: ${retryAfter}秒待機します`);
await new Promise((r) => setTimeout(r, retryAfter * 1000));
return axios(err.config);
}
return Promise.reject(err);
});
パフォーマンス最適化
バッチサイズの分割
10,000件を一度に処理するのではなく、500件ごとのバッチに分割して順番に処理することで、メモリ消費を抑えられます。
async def bulk_generate_chunked(csv_path: str, chunk_size: int = 500):
recipients = load_recipients(csv_path)
all_results = []
for i in range(0, len(recipients), chunk_size):
chunk = recipients[i:i + chunk_size]
print(f"バッチ {i // chunk_size + 1}: {len(chunk)}件処理中...")
results = await process_chunk(chunk)
all_results.extend(results)
await asyncio.sleep(2) # バッチ間のインターバル
return all_results
非同期キューによる大規模処理
2,000件を超える場合は、Redis + BullMQ（Node.js）や Celery（Python）などのキューシステムと組み合わせることを推奨します。
// BullMQ による証明書生成キュー（概念コード）
const { Queue, Worker } = require('bullmq');
const certQueue = new Queue('certificate-generation');
// キューにジョブを追加
async function enqueueCertificates(recipients) {
const jobs = recipients.map((r) => ({
name: 'generate',
data: r,
opts: { attempts: 3, backoff: { type: 'exponential', delay: 1000 } },
}));
await certQueue.addBulk(jobs);
}
// ワーカーで並列処理
const worker = new Worker('certificate-generation', async (job) => {
const pdf = await generateOneCertificate(job.data);
await saveToS3(job.data.id, pdf);
}, { concurrency: 10 });
発行後の運用
メール自動送信
生成したPDFをそのまま受講者にメール送信するパイプラインを構築できます。
const nodemailer = require('nodemailer');
async function sendCertificateEmail(recipient, pdfBuffer) {
const transporter = nodemailer.createTransport({
host: process.env.SMTP_HOST,
port: 587,
auth: { user: process.env.SMTP_USER, pass: process.env.SMTP_PASS },
});
await transporter.sendMail({
from: process.env.FROM_EMAIL,
to: recipient.email,
subject: `修了証のお知らせ — ${recipient.course}`,
text: `${recipient.name} 様\n\nこの度は「${recipient.course}」を修了されましたことをお祝い申し上げます。\n修了証を添付いたしますので、ご確認ください。`,
attachments: [
{
filename: `certificate-${recipient.id}.pdf`,
content: pdfBuffer,
},
],
});
}
検証用QRコードの埋め込み
各証明書に検証URLを埋め込んだQRコードを追加することで、改ざん防止と真正性確認が可能になります。
<!-- テンプレートHTMLにQRコードを追加 -->
<div class="cert-id">No. {{id}}</div>
<div class="qr-code">
<img src="{{qr_url}}" width="20mm" height="20mm" alt="検証QRコード">
</div>
import qrcode
import io
import base64
def generate_qr_data_url(certificate_id: str, verify_base_url: str) -> str:
"""証明書IDの検証URLをQRコードにしてData URLで返す"""
url = f"{verify_base_url}/verify/{certificate_id}"
qr = qrcode.QRCode(box_size=4, border=1)
qr.add_data(url)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
buf = io.BytesIO()
img.save(buf, format="PNG")
b64 = base64.b64encode(buf.getvalue()).decode()
return f"data:image/png;base64,{b64}"
# 受講者データにQRコードを追加
for recipient in recipients:
recipient["qr_url"] = generate_qr_data_url(
recipient["id"], "https://example.com"
)
改ざん防止：ハッシュ検証
証明書IDにHMACを使ったハッシュを付与することで、IDの改ざんを検知できます。
import hmac
import hashlib
SECRET_KEY = os.environ["CERT_SECRET_KEY"].encode()
def sign_certificate_id(cert_id: str) -> str:
"""証明書IDに署名を付与する"""
sig = hmac.new(SECRET_KEY, cert_id.encode(), hashlib.sha256).hexdigest()[:12]
return f"{cert_id}-{sig}"
def verify_certificate_id(signed_id: str) -> bool:
"""証明書IDの署名を検証する"""
if '-' not in signed_id:
return False
cert_id, sig = signed_id.rsplit('-', 1)
expected = hmac.new(SECRET_KEY, cert_id.encode(), hashlib.sha256).hexdigest()[:12]
return hmac.compare_digest(sig, expected)
エラーハンドリングのベストプラクティス
失敗した証明書のみを再処理する「失敗IDリトライ」パターンが重要です。全件再実行すると重複発行が発生するリスクがあります。
# 前回の失敗リストを読み込んで再処理
async def retry_failed(failed_json_path: str):
with open(failed_json_path) as f:
failed = json.load(f)
print(f"{len(failed)}件をリトライします...")
# 失敗IDのみを対象に再度生成
retry_recipients = [{"id": r["id"], "name": r["name"], ...} for r in failed]
return await bulk_generate_from_list(retry_recipients)
失敗したID一覧は output/failed.json に保存されます。以下のコマンドで再実行できます。
# Node.jsの場合
node bulk-certificate-generator.js --retry ./output/failed.json
# Pythonの場合
python bulk_certificate_generator.py --retry ./output/failed.json
よくある質問
よくある質問とその回答については、この記事の冒頭のFAQセクションも参照してください。その他の疑問点はFUNBREW PDF ドキュメントでご確認いただけます。
まとめ
CSVを使った証明書一括発行の実装ポイントを整理します。
| ポイント | 推奨アプローチ |
|---|---|
| データ読み込み | CSV → 辞書リスト |
| テンプレート | Handlebars（JS）/ Jinja2（Python） |
| 並列制御 | セマフォ（CONCURRENCY = 5〜10） |
| レート制限対策 | 指数バックオフ付きリトライ |
| 出力 | ZIP化して一括ダウンロード |
| 失敗処理 | 失敗IDを記録して再実行 |
| 検証 | QRコード + HMAC署名 |
FUNBREW PDF PlaygroundでテンプレートHTMLの動作を確認してから本番実装に進むことをおすすめします。証明書テンプレートのデザインはHTML証明書テンプレート集、詳細な自動化フローは証明書PDF自動生成ガイドを参照してください。APIキーの取得・認証設定はFUNBREW PDF ドキュメントをご覧ください。
関連記事
- HTML証明書テンプレート集 — 修了証・賞状・ディプロマ・参加証の5テンプレート
- 証明書PDF自動生成ガイド — テンプレート設計からバッチ発行まで
- PDF一括生成ガイド — 大量PDF生成のベストプラクティス
- PDFテンプレートエンジン活用ガイド — Handlebars・Jinja2でデータバインディング
- HTML→PDF CSSスニペット集 — 改ページ・余白・フォント制御
- 本番環境ガイド — 本番運用チェックリスト