Python | 自動化：Web API 集計 BOT

Python
スポンサーリンク

概要（Web API 集計 BOT は「データを取って、すぐに意味のある形にするロボット」）

Web API 集計 BOT は、単なる「データ収集 BOT」から一歩進んで、「取ってきたデータをその場で集計して、使える数字にしてくれるロボット」です。
例えば、毎時間アクセス数を返す API を叩いて、日ごと・サービスごとに集計する。
毎日売上明細を返す API を叩いて、カテゴリ別売上をまとめる。
こうしたことを自動でやらせるのが Web API 集計 BOT の役割です。

大事なのは、単に JSON を保存するのではなく、「どんな集計をしたいか」を最初に決めておき、そのロジックをコードで固定することです。

全体像（Web API 集計 BOT を分解すると何をしているのか）

処理の流れを頭の中で整理する

Web API 集計 BOT の中身は、次のようなステップに分解できます。

まず、どの API を、どんなパラメータで叩くかを決めます。
次に、返ってきた JSON のどのキーを使って、どんな「表形式」に変換するかを決めます。
そのうえで、pandas などを使って、日別、ユーザー別、カテゴリ別などの集計を行います。
最後に、集計結果を CSV や Excel、データベースなどに保存します。

この「API → JSON → DataFrame → 集計 → 保存」という線を一本通してしまうと、あとは「いつ実行するか」を決めるだけで BOT になります。

例題の設定（アクセスログ API を集計する BOT を作るイメージ）

想定する Web API の仕様を決める

ここでは、すごく簡略化した「アクセスログ API」を想定します。
例えば、次のような API があるとします。

URL は https://api.example.com/v1/access_logs。
パラメータで date=YYYY-MM-DD を渡すと、その日のアクセス明細が返ってくる。
レスポンスは JSON で、ざっくり次のような構造になっている。

{
  "date": "2025-01-01",
  "logs": [
    {"timestamp": "2025-01-01T09:01:23Z", "user_id": 1, "service": "A"},
    {"timestamp": "2025-01-01T09:02:10Z", "user_id": 2, "service": "B"},
    {"timestamp": "2025-01-01T09:02:55Z", "user_id": 1, "service": "A"}
  ]
}

この API からデータを取ってきて、「日ごとのサービス別アクセス数」を集計し、CSV に保存する BOT を作ります。

1回分のデータ取得と DataFrame への変換

requests で API を叩いて JSON を取る

まずは、2025-01-01 のデータを 1 回だけ取得してみます。

import requests
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
DATA_DIR.mkdir(exist_ok=True)

API_URL = "https://api.example.com/v1/access_logs"

def fetch_logs_for_date(date_str: str):
    params = {"date": date_str}
    resp = requests.get(API_URL, params=params, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    return data
Python

この段階では、まだ Python の dict として JSON を受け取っているだけです。
ここから、「集計しやすい形」に変換していきます。

JSON の配列部分（logs）を pandas に食べさせる

pandas を使うと、JSON の「配列部分」をそのまま DataFrame に変換できます。

import pandas as pd

def logs_to_dataframe(data: dict) -> pd.DataFrame:
    logs = data["logs"]              # リスト of dict
    df = pd.DataFrame(logs)          # そのまま表にする
    df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce")
    return df
Python

ここで重要なのは、timestamp を必ず datetime 型にしておくことです。
後で「日ごと」「時間ごと」に集計するとき、文字列のままだととても扱いにくくなります。

集計処理（サービス別・日別のアクセス数をまとめる）

1日分の logs から「サービス別アクセス数」を集計する

DataFrame になってしまえば、あとは groupby で自由に集計できます。

def aggregate_service_counts(df: pd.DataFrame) -> pd.DataFrame:
    result = (
        df.groupby("service", as_index=False)
          .size()
          .rename(columns={"size": "access_count"})
    )
    return result
Python

これで、「サービス A のアクセス数」「サービス B のアクセス数」といった集計表が作れます。

日付列を持たせて「日別集計」に育てる

日別に並べたい場合は、日付列を追加しておくと便利です。

from datetime import datetime

def aggregate_service_counts_with_date(df: pd.DataFrame, date_str: str) -> pd.DataFrame:
    result = aggregate_service_counts(df)
    result["date"] = pd.to_datetime(date_str)
    cols = ["date", "service", "access_count"]
    return result[cols]
Python

このようにしておくと、複数日分を縦に結合したときに、サービス別×日別の集計表として扱いやすくなります。

複数日をループして「期間集計 BOT」にする

期間をループして API を連続で叩く

例えば、「直近 7 日分のログを集計する BOT」を作りたいとします。
datetime と timedelta を使って、日付をループします。

from datetime import datetime, timedelta

def date_range(start_date: str, end_date: str):
    start = datetime.strptime(start_date, "%Y-%m-%d").date()
    end = datetime.strptime(end_date, "%Y-%m-%d").date()
    current = start
    while current <= end:
        yield current.strftime("%Y-%m-%d")
        current += timedelta(days=1)
Python

この date_range を使って、毎日分を取得・集計し、最後にひとつの DataFrame にまとめます。

def collect_and_aggregate_range(start_date: str, end_date: str) -> pd.DataFrame:
    all_list = []

    for date_str in date_range(start_date, end_date):
        print(f"{date_str} を取得中…")
        data = fetch_logs_for_date(date_str)
        df_logs = logs_to_dataframe(data)
        df_agg = aggregate_service_counts_with_date(df_logs, date_str)
        all_list.append(df_agg)

    if not all_list:
        return pd.DataFrame(columns=["date", "service", "access_count"])

    all_data = pd.concat(all_list, ignore_index=True)
    return all_data
Python

ここでは単純に print していますが、実運用では logging に置き換えると良いです。

期間集計結果を CSV に保存する

集計結果は、日×サービスの表として CSV に出しておきます。

def save_aggregated_csv(df: pd.DataFrame, start_date: str, end_date: str):
    DATA_DIR.mkdir(exist_ok=True)
    out_name = f"service_access_{start_date}_to_{end_date}.csv"
    out_path = DATA_DIR / out_name
    df.to_csv(out_path, index=False, encoding="utf-8-sig")
    print(f"保存しました: {out_path}")
Python

最後に、run_bot 的な関数で全体をつなぎます。

def run_bot():
    start_date = "2025-01-01"
    end_date   = "2025-01-07"
    df = collect_and_aggregate_range(start_date, end_date)
    print(f"集計行数: {len(df)}")
    save_aggregated_csv(df, start_date, end_date)

if __name__ == "__main__":
    run_bot()
Python

設計の重要ポイントを深掘りする

パラメータをハードコーディングしない（柔軟な BOT にする）

最初は start_date や end_date をコードに直接書いて構いませんが、
実運用では「昨日分だけ」「先月分」など、動的に変えたいケースがほとんどです。

例えば、「昨日分だけを取る」関数を作っておくと便利です。

def get_yesterday_str():
    today = datetime.today().date()
    yesterday = today - timedelta(days=1)
    return yesterday.strftime("%Y-%m-%d")
Python

これを使って、日次 BOT 用の run_bot_daily を作れます。

def run_bot_daily():
    date_str = get_yesterday_str()
    df = collect_and_aggregate_range(date_str, date_str)
    save_aggregated_csv(df, date_str, date_str)
Python

こうしておけば、cron やタスクスケジューラから毎日実行するだけで、
「昨日分のサービス別アクセス数 CSV」が毎日増えていきます。

API エラーや一時的な失敗に備える（リトライの考え方）

Web API は、必ずいつか失敗します。
ネットワークが不安定だったり、API サーバーの負荷が高かったりするとき、一発での取得に失敗することがあります。

一時的なエラーに対しては、「少し待ってからもう一度やってみる」リトライ処理が実務ではよく使われます。

簡単なリトライ関数のイメージはこうです。

import time
import requests

def fetch_with_retry(params, retries=3, delay=5):
    for i in range(1, retries + 1):
        try:
            resp = requests.get(API_URL, params=params, timeout=10)
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as e:
            print(f"API エラー ({i}/{retries}): {e}")
            if i == retries:
                raise
            time.sleep(delay)
Python

そして fetch_logs_for_date の中でこれを使います。

def fetch_logs_for_date(date_str: str):
    params = {"date": date_str}
    data = fetch_with_retry(params)
    return data
Python

これで、一時的なエラーで BOT 全体が落ちてしまう可能性を減らせます。

冪等性（同じ範囲を取り直してもおかしくならない）

集計 BOT は、途中で失敗したときに「もう一度同じ範囲を取り直したい」ことがあります。
そのときに、同じ結果を上書きしても問題ない設計にしておくと運用が楽です。

例えば、期間集計のファイル名を
service_access_2025-01-01_to_2025-01-07.csv
のようにしておけば、同じ期間で再実行すると上書きになります。
これを意図的な仕様としておけば、「取り直し」が怖くありません。

逆に、「履歴として全部取っておきたい」場合は、ファイル名にタイムスタンプも含めるなどして区別します。
ここは用途によって変えるところなので、「再実行時どうしたいか」を先に決めてから設計するとスムーズです。

API キーやトークンの扱い

多くの Web API には API キーやアクセストークンが必要です。
ソースコードに直書きするのは危険なので、環境変数などから読み込む形にします。

import os

API_KEY = os.environ.get("MY_API_KEY")

def fetch_with_auth(params):
    headers = {"Authorization": f"Bearer {API_KEY}"}
    resp = requests.get(API_URL, headers=headers, params=params, timeout=10)
    resp.raise_for_status()
    return resp.json()
Python

API キーの管理は情報漏洩リスクに直結するので、「コード・Git には絶対に埋め込まない」という意識を早めに持っておくと良いです。

自動化との接続（本当に「BOT」にしていく）

Python 側は「1回分をきっちり」、時間管理は OS に任せる

ここまでで、「run_bot_daily」を呼べば「昨日分の集計ができる」状態が作れたとします。
この時点ではまだ手動実行ですが、ここからは時間の話になります。

Linux や Mac なら cron、Windows ならタスクスケジューラを使って、
毎朝 9:00 に python webapi_aggregate_bot.py を実行する設定を入れます。

Python 側には「時間のループ」は書かず、「呼ばれたら 1 回分だけ実行する」形にしておくのがポイントです。
時間管理は OS に任せ、Python はビジネスロジックに集中させる方が、シンプルで壊れにくい構成になります。

まとめ（「API → DataFrame → 集計 → 保存」を一本のレシピにする）

Web API 集計 BOT の本質は、難しい技を使うことではありません。
やるべきことをきちんと分解して、一本のレシピとして組み立てることです。

具体的には、
どの API をどう叩くかを決める。
レスポンス JSON から必要な部分だけを取り出し、DataFrame にする。
pandas の groupby などで、日別、サービス別、ユーザー別などの集計を行う。
結果を CSV などに保存し、再実行時の挙動（上書きか履歴か）も含めて設計する。
1 回分の処理を関数にし、cron / タスクスケジューラで定期実行させる。

この型を一度自分の手で作ってみると、「売上 API 集計 BOT」「KPI API 集計 BOT」「アクセスログ API 集計 BOT」など、同じ考え方でどんどん応用が効くようになります。

スポンサーリンク
タイトルとURLをコピーしました