Day 60：Web/APIプロジェクト ― 「データ収集・分析ツール」を一本の流れで作る
Day 60では、ここまで学んできた Web・API・保存・集計の知識をまとめて、 「データ収集・分析ツール」を一本のプロジェクトとして組み立てる ことを目標にします。
流れはこうです。
API ↓ データ取得 ↓ データ加工 ↓ CSV / Excel保存 ↓ 集計 ↓ レポート作成
今日は、この一本のパイプラインを ステップごとに分解しながら、実際に動く形をイメージできるところまで 一緒にたどっていきます。
プロジェクトの全体像を先に描く
どんなツールを作るかイメージする
まずは、ツールのイメージを決めます。
例として、次のような「APIベースのデータ収集・分析ツール」を考えます。
- あるキーワードで検索APIを叩く
- 検索結果（タイトル・URL・説明文・スコアなど）を取得する
- 取得したデータを整形してCSVに保存する
- CSVをもとに簡単な集計（件数・平均スコアなど）を行う
- 集計結果をテキストレポートとして出力する
ここでは「検索APIっぽいモック」を使いながら、 この流れをコードで組み立てていきます。
ステップ1：APIからデータ取得 ― 「外の世界」から情報を持ってくる
APIクライアントの基本形を作る
まずは、APIからデータを取得する部分です。 requests を使って、検索APIっぽいエンドポイントを叩くクライアントクラスを用意します。
# day60_api_client.py
import requests
class SearchApiClient:
"""
検索APIを呼び出すためのクライアントクラスです。
API呼び出しのロジックをひとつの場所にまとめておくことで、
プロジェクト全体の見通しが良くなります。
"""
def __init__(self, base_url: str):
# ベースとなるAPIのURLを保持します。
self.base_url = base_url
def search(self, keyword: str, limit: int = 20) -> dict:
"""
検索APIを呼び出して結果を返すメソッドです。
実際には GET /search?q=keyword&limit=20 のようなイメージです。
"""
params = {"q": keyword, "limit": limit}
print(f"[API] 検索APIを呼び出します: {self.base_url}, params={params}")
# timeout を付けて、いつまでも待ち続けないようにします。
response = requests.get(self.base_url, params=params, timeout=10)
# HTTPエラーなら例外を投げます。
response.raise_for_status()
# JSONレスポンスを辞書として返します。
return response.json()
重要ポイントの深掘り：
- API呼び出しをクラスにまとめることで、「認証」「ヘッダー」「別エンドポイント」などを後から追加しやすくなります。
response.raise_for_status()は、HTTPステータスコードがエラーのときに例外を投げてくれる便利なメソッドです。limitのようなパラメータをメソッド引数にしておくと、ツール側から柔軟に制御できます。
ステップ2：データ加工 ― レスポンスJSONを「分析しやすい形」に整える
レスポンスから必要な項目を抜き出す
次に、APIレスポンスを「分析しやすい形」に整えます。
ここでは、レスポンスが次のような構造だと仮定します。
{
"items": [
{
"title": "Python入門",
"link": "https://example.com/python",
"snippet": "Pythonの基本を学ぶための入門記事です。",
"score": 0.92
},
...
]
}
これを、Python側では次のような形に変換します。
titleurlsnippetscore
# day60_transformer.py
def transform_search_results(raw_response: dict) -> list[dict]:
"""
検索APIの生のレスポンスから、
分析しやすい形に整形する関数です。
変換後の形:
- title: 結果のタイトル
- url: リンク先URL
- snippet: 簡単な説明文
- score: 検索スコア（なければ None）
"""
print("[TRANSFORM] 検索結果を整形します。")
items = raw_response.get("items", [])
results: list[dict] = []
for item in items:
title = item.get("title", "")
url = item.get("link", "")
snippet = item.get("snippet", "")
score = item.get("score", None)
results.append(
{
"title": title,
"url": url,
"snippet": snippet,
"score": score,
}
)
print(f"[TRANSFORM] {len(results)} 件の結果を整形しました。")
return results
重要ポイントの深掘り：
- 「生のレスポンス」と「分析用データ」を分けることで、後から集計ロジックを書くときに迷いが減ります。
scoreのような数値項目は、後の集計で使うことを意識して残しておきます。.get()にデフォルト値を指定しておくと、レスポンス仕様が少し変わっても壊れにくくなります。
ステップ3：CSV / Excel保存 ― 「分析の土台」をファイルとして残す
CSVに保存して、Excelで開ける形にする
次に、整形したデータを CSVとして保存 します。 CSVは、Excelやスプレッドシートで開けるので、 「プログラム＋人間の両方が扱いやすい形式」です。
# day60_storage.py
import csv
from pathlib import Path
def save_results_to_csv(results: list[dict], filename: str) -> Path:
"""
検索結果をCSVファイルに保存する関数です。
- results: title, url, snippet, score を持つ辞書のリスト
- filename: 保存するファイル名
"""
output_path = Path(filename)
fieldnames = ["title", "url", "snippet", "score"]
print(f"[SAVE] 結果をCSVに保存します: {output_path}")
with output_path.open(mode="w", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for item in results:
writer.writerow(item)
print(f"[SAVE] 保存完了: {output_path}")
return output_path
Excelとのつながり：
- このCSVファイルは、そのままExcelで開けます。
- Excel側でフィルタや並べ替え、グラフ作成などを行うこともできます。
- Python側で集計した結果と、Excelでの目視確認を組み合わせると、分析の精度が上がります。
ステップ4：集計 ― 「数字で全体を眺める」ための処理を書く
スコアの平均・最大・件数などを計算する
次に、保存したデータをもとに簡単な集計を行います。 ここでは、整形済みの results をそのまま使って集計します。
# day60_aggregator.py
from typing import Optional
def aggregate_results(results: list[dict]) -> dict:
"""
検索結果のリストから、簡単な集計情報を作る関数です。
集計項目の例:
- count: 件数
- score_avg: スコアの平均
- score_max: スコアの最大値
- score_min: スコアの最小値
"""
print("[AGGREGATE] 検索結果を集計します。")
count = len(results)
# スコアが None でないものだけを集計対象にします。
scores: list[float] = []
for item in results:
score = item.get("score")
if isinstance(score, (int, float)):
scores.append(float(score))
if scores:
score_avg = sum(scores) / len(scores)
score_max = max(scores)
score_min = min(scores)
else:
score_avg = None
score_max = None
score_min = None
summary = {
"count": count,
"score_avg": score_avg,
"score_max": score_max,
"score_min": score_min,
}
print("[AGGREGATE] 集計完了。")
return summary
重要ポイントの深掘り：
- 「件数」と「スコアの統計」を出すだけでも、データの傾向がかなり見えてきます。
Noneを許容することで、「スコアがない結果」が混ざっていても壊れないようにしています。- 集計ロジックをひとつの関数にまとめておくと、後から「中央値」「標準偏差」「カテゴリ別集計」などを追加しやすくなります。
ステップ5：レポート作成 ― 集計結果を「読みやすい文章」にする
テキストレポートを生成する
最後に、集計結果をもとに テキストレポート を作ります。 ここでは、コンソール出力とテキストファイル保存の両方を行う形にします。
# day60_reporter.py
from pathlib import Path
def generate_report(summary: dict, output_file: str = "report.txt") -> Path:
"""
集計結果からテキストレポートを生成し、ファイルに保存する関数です。
"""
path = Path(output_file)
count = summary.get("count")
score_avg = summary.get("score_avg")
score_max = summary.get("score_max")
score_min = summary.get("score_min")
lines: list[str] = []
lines.append("=== 検索結果レポート ===")
lines.append("")
lines.append(f"総件数: {count}")
if score_avg is not None:
lines.append(f"スコア平均: {score_avg:.3f}")
lines.append(f"スコア最大: {score_max:.3f}")
lines.append(f"スコア最小: {score_min:.3f}")
else:
lines.append("スコア情報: なし（スコアが付与されていない結果のみ）")
lines.append("")
lines.append("このレポートは、APIから取得した検索結果をもとに自動生成されました。")
# コンソールにも表示します。
print("\n".join(lines))
# ファイルにも保存します。
path.write_text("\n".join(lines), encoding="utf-8")
print(f"[REPORT] レポートを保存しました: {path}")
return path
重要ポイントの深掘り：
- 「数字だけ」ではなく、「文章としてまとめる」ことで、他の人にも共有しやすくなります。
- レポート生成を関数にしておくと、将来的に「HTMLレポート」「PDFレポート」などに拡張しやすくなります。
- コンソール出力とファイル保存を両方行うことで、開発中の確認と運用時の記録を両立できます。
ステップ6：すべてをつなぐメインスクリプト
API → データ取得 → 加工 → 保存 → 集計 → レポート を一本にする
ここまでの部品を、ひとつのメインスクリプトでつなぎます。
# day60_main.py
from day60_api_client import SearchApiClient
from day60_transformer import transform_search_results
from day60_storage import save_results_to_csv
from day60_aggregator import aggregate_results
from day60_reporter import generate_report
def run_data_collection_and_analysis(keyword: str) -> None:
"""
データ収集・分析ツールのメイン処理です。
流れ:
1. APIからデータ取得
2. データ加工
3. CSV保存
4. 集計
5. レポート作成
"""
print("=== データ収集・分析ツールを開始します ===")
# 1. APIからデータ取得
base_url = "https://example.com/api/search" # 実際のAPI URLに置き換えます。
client = SearchApiClient(base_url=base_url)
raw_response = client.search(keyword, limit=50)
# 2. データ加工
results = transform_search_results(raw_response)
if not results:
print("検索結果が0件でした。処理を終了します。")
return
# 3. CSV保存
csv_path = save_results_to_csv(results, "search_results.csv")
# 4. 集計
summary = aggregate_results(results)
# 5. レポート作成
report_path = generate_report(summary, output_file="search_report.txt")
print("=== データ収集・分析ツールを終了します ===")
print(f"CSVファイル: {csv_path}")
print(f"レポートファイル: {report_path}")
if __name__ == "__main__":
# 実際には、コマンドライン引数や設定ファイルからキーワードを受け取ることが多いです。
run_data_collection_and_analysis(keyword="python 入門")
重要ポイントの深掘り：
- メイン処理は「流れ」を読みやすく書くことが大事です。
- 各ステップを関数・モジュールとして分けておくことで、 「APIを変える」「保存形式を変える」「集計項目を増やす」といった変更に強くなります。
- ここに Day 59 で学んだ「エラー処理」「ログ」を組み込むと、より実践的なツールになります。
Day 60ミニテンプレート：コンパクト版データ収集・分析ツール
最後に、Day 60の内容をぎゅっとまとめた コンパクト版テンプレート を載せておきます。
# day60_compact_template.py
import requests
import csv
from pathlib import Path
def fetch_data(keyword: str) -> dict:
base_url = "https://example.com/api/search"
params = {"q": keyword, "limit": 50}
resp = requests.get(base_url, params=params, timeout=10)
resp.raise_for_status()
return resp.json()
def transform(raw: dict) -> list[dict]:
items = raw.get("items", [])
results: list[dict] = []
for item in items:
results.append(
{
"title": item.get("title", ""),
"url": item.get("link", ""),
"snippet": item.get("snippet", ""),
"score": item.get("score", None),
}
)
return results
def save_csv(results: list[dict], filename: str) -> Path:
path = Path(filename)
fieldnames = ["title", "url", "snippet", "score"]
with path.open(mode="w", encoding="utf-8", newline="") as f:
w = csv.DictWriter(f, fieldnames=fieldnames)
w.writeheader()
for r in results:
w.writerow(r)
return path
def aggregate(results: list[dict]) -> dict:
count = len(results)
scores = [
float(r["score"])
for r in results
if isinstance(r.get("score"), (int, float))
]
if scores:
avg = sum(scores) / len(scores)
return {"count": count, "score_avg": avg}
else:
return {"count": count, "score_avg": None}
def report(summary: dict) -> None:
print("=== レポート ===")
print(f"総件数: {summary['count']}")
if summary["score_avg"] is not None:
print(f"スコア平均: {summary['score_avg']:.3f}")
else:
print("スコア平均: なし")
def main():
keyword = "python 入門"
raw = fetch_data(keyword)
results = transform(raw)
if not results:
print("結果0件。終了します。")
return
csv_path = save_csv(results, "search_results.csv")
summary = aggregate(results)
report(summary)
print(f"CSV保存先: {csv_path}")
if __name__ == "__main__":
main()
Day 60のまとめ
Day 60では、Web/APIプロジェクトとしての「データ収集・分析ツール」を題材に、
- APIからデータを取得するクライアントを用意し
- レスポンスJSONを分析しやすい形に加工し
- CSVとして保存して、Excelとも連携できる形にし
- スコアなどの数値を集計して全体の傾向をつかみ
- 集計結果をテキストレポートとしてまとめる
という一本の流れを、コードとともにステップバイステップでたどりました。
ここまで来ると、 「APIでデータを集めて、少し分析して、レポートを出す」という 実務にかなり近い形のツールを、自分の手で組み立てられるようになっています。
