「毎日のキーワード調査、もううんざり…」
そんなあなたの心の声が聞こえてきます。SEO担当者やブロガーなら誰もが経験する、あの果てしないキーワード調査の苦労。手作業で検索ボリュームを調べ、競合度を確認し、CSVファイルを作成する…。一体どれだけの時間を費やしているでしょうか？
でも、もしその作業が全て自動化できたら？
今回は、Python初心者でも理解できるよう、Googleキーワードプランナーの自動化方法を丁寧に解説します。この記事を読み終える頃には、あなたの貴重な時間が劇的に短縮され、より戦略的な業務に集中できるようになるでしょう。
なぜキーワードプランナーの自動化が必要なのか？
手作業の限界とストレス
現代のSEO業界では、キーワード調査は避けて通れない作業です。しかし：
- 時間の浪費: 一つ一つのキーワードを手動で調査するのに何時間も費やす
- 人的ミス: 大量のデータを扱う際の入力ミスや見落とし
- スケーラビリティの欠如: 案件が増えるほど作業量が比例して増加
- 精神的負担: 単調で反復的な作業によるストレス
自動化がもたらす革命的変化
自動化を導入すると：
- 時間短縮: 手作業で1日かかっていた作業が数分で完了
- 精度向上: 人的ミスの排除と一貫したデータ取得
- スケーラビリティ: 大量のキーワードも一括処理可能
- 戦略性の向上: 単純作業から解放され、分析や戦略立案に集中
事前準備：必要なツールとアカウント設定
1. Google Ads APIの設定
キーワードプランナーを自動化するには、Google Ads APIの利用が必須です。
必要なもの：
- Googleアカウント
- Google Adsアカウント（広告出稿の必要なし）
- Google Cloud Platformアカウント
設定手順：
- Google Cloud Consoleにアクセス
- Google Cloud Consoleにログイン
- 新しいプロジェクトを作成
- Google Ads APIを有効化
- APIとサービス → ライブラリ
- 「Google Ads API」を検索して有効化
- 認証情報の作成
- 認証情報 → 認証情報を作成 → サービスアカウント
- JSONキーファイルをダウンロード
2. 必要なPythonライブラリのインストール
pip install google-ads google-auth google-auth-oauthlib pandas openpyxl
各ライブラリの役割：
google-ads: Google Ads APIとの通信google-auth: Google認証の処理pandas: データ操作と分析openpyxl: Excelファイルの読み書き
基本的な実装：ステップバイステップガイド
ステップ1: 認証設定ファイルの作成
まず、APIとの通信に必要な設定ファイルを作成します。
# config.py
import os
from google.ads.googleads.client import GoogleAdsClient
class GoogleAdsConfig:
def __init__(self):
# 環境変数から認証情報を取得
self.client = GoogleAdsClient.load_from_storage()
def get_client(self):
return self.client
ステップ2: キーワードプランナークラスの作成
# keyword_planner.py
import pandas as pd
from google.ads.googleads.client import GoogleAdsClient
from google.ads.googleads.v14.services.services.keyword_plan_idea_service import KeywordPlanIdeaServiceClient
from google.ads.googleads.v14.common.types.keyword_plan_common import KeywordPlanNetwork
class KeywordPlannerAutomation:
def __init__(self, client):
self.client = client
self.keyword_plan_idea_service = client.get_service("KeywordPlanIdeaService")
def get_keyword_ideas(self, seed_keywords, location_id="2392", language_id="1005"):
"""
キーワードアイデアを取得
Args:
seed_keywords (list): シードキーワードのリスト
location_id (str): 地域ID（デフォルト：日本）
language_id (str): 言語ID（デフォルト：日本語）
Returns:
pandas.DataFrame: キーワードデータ
"""
# リクエストの作成
request = self.keyword_plan_idea_service.generate_keyword_ideas_request()
request.customer_id = self._get_customer_id()
# キーワードの設定
request.seed_keywords.extend(seed_keywords)
# 地域と言語の設定
request.geo_target_constants.append(f"geoTargetConstants/{location_id}")
request.language = f"languageConstants/{language_id}"
# ネットワーク設定
request.keyword_plan_network = KeywordPlanNetwork.GOOGLE_SEARCH
try:
# APIリクエストの実行
response = self.keyword_plan_idea_service.generate_keyword_ideas(request=request)
# レスポンスをDataFrameに変換
return self._parse_response(response)
except Exception as e:
print(f"エラーが発生しました: {e}")
return pd.DataFrame()
def _parse_response(self, response):
"""
APIレスポンスをDataFrameに変換
"""
data = []
for idea in response:
keyword_text = idea.text
# 検索ボリューム情報
if idea.keyword_idea_metrics:
avg_monthly_searches = idea.keyword_idea_metrics.avg_monthly_searches
competition = idea.keyword_idea_metrics.competition.name
low_top_of_page_bid = idea.keyword_idea_metrics.low_top_of_page_bid_micros / 1000000
high_top_of_page_bid = idea.keyword_idea_metrics.high_top_of_page_bid_micros / 1000000
else:
avg_monthly_searches = 0
competition = "UNKNOWN"
low_top_of_page_bid = 0
high_top_of_page_bid = 0
data.append({
'キーワード': keyword_text,
'月間検索ボリューム': avg_monthly_searches,
'競合度': competition,
'推奨入札価格（下限）': low_top_of_page_bid,
'推奨入札価格（上限）': high_top_of_page_bid
})
return pd.DataFrame(data)
def _get_customer_id(self):
"""
顧客IDを取得（実際の実装では環境変数等から取得）
"""
# 実際の顧客IDに置き換えてください
return "1234567890"
ステップ3: メイン実行スクリプト
# main.py
from config import GoogleAdsConfig
from keyword_planner import KeywordPlannerAutomation
import pandas as pd
from datetime import datetime
def main():
"""
メイン実行関数
"""
print("🚀 キーワードプランナー自動化を開始します...")
try:
# 設定の初期化
config = GoogleAdsConfig()
client = config.get_client()
# キーワードプランナーインスタンスの作成
kp_automation = KeywordPlannerAutomation(client)
# 調査したいシードキーワードのリスト
seed_keywords = [
"Python プログラミング",
"SEO 対策",
"キーワード調査",
"自動化 ツール",
"データ分析"
]
print(f"📊 {len(seed_keywords)}個のシードキーワードを調査中...")
# キーワードアイデアの取得
results_df = kp_automation.get_keyword_ideas(seed_keywords)
if not results_df.empty:
# 結果の表示
print(f"✅ {len(results_df)}個のキーワードを取得しました！")
print("\n=== 取得結果サンプル ===")
print(results_df.head(10).to_string(index=False))
# CSVファイルとして保存
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
csv_filename = f"keyword_research_{timestamp}.csv"
results_df.to_csv(csv_filename, index=False, encoding='utf-8-sig')
# Excelファイルとして保存
excel_filename = f"keyword_research_{timestamp}.xlsx"
results_df.to_excel(excel_filename, index=False)
print(f"📁 結果を以下のファイルに保存しました:")
print(f" - {csv_filename}")
print(f" - {excel_filename}")
else:
print("❌ データの取得に失敗しました。設定を確認してください。")
except Exception as e:
print(f"💥 エラーが発生しました: {e}")
print("設定ファイルやAPI認証情報を確認してください。")
if __name__ == "__main__":
main()
高度な機能の実装
1. 複数地域での調査機能
def multi_region_research(self, seed_keywords, regions):
"""
複数地域でのキーワード調査
Args:
seed_keywords (list): シードキーワード
regions (dict): 地域ID辞書 {'地域名': 'location_id'}
Returns:
dict: 地域別結果辞書
"""
results = {}
for region_name, location_id in regions.items():
print(f"🌍 {region_name}でのキーワード調査中...")
df = self.get_keyword_ideas(
seed_keywords=seed_keywords,
location_id=location_id
)
df['地域'] = region_name
results[region_name] = df
return results
# 使用例
regions = {
'日本': '2392',
'アメリカ': '2840',
'イギリス': '2826',
'オーストラリア': '2036'
}
multi_results = kp_automation.multi_region_research(seed_keywords, regions)
2. 競合分析機能
def competitor_analysis(self, competitor_domains, max_keywords=100):
"""
競合サイトのキーワード分析
Args:
competitor_domains (list): 競合ドメインのリスト
max_keywords (int): 取得する最大キーワード数
Returns:
pandas.DataFrame: 競合キーワード分析結果
"""
all_competitor_data = []
for domain in competitor_domains:
print(f"🔍 {domain}の分析中...")
# ドメインベースのキーワード調査
request = self.keyword_plan_idea_service.generate_keyword_ideas_request()
request.customer_id = self._get_customer_id()
request.url_seed.url = f"https://{domain}"
try:
response = self.keyword_plan_idea_service.generate_keyword_ideas(request=request)
df = self._parse_response(response)
df['競合サイト'] = domain
df = df.head(max_keywords)
all_competitor_data.append(df)
except Exception as e:
print(f"❌ {domain}の分析でエラー: {e}")
if all_competitor_data:
return pd.concat(all_competitor_data, ignore_index=True)
else:
return pd.DataFrame()
3. 季節性分析機能
def seasonal_analysis(self, keywords, months_back=12):
"""
キーワードの季節性分析
Args:
keywords (list): 分析対象キーワード
months_back (int): 過去何ヶ月分を分析するか
Returns:
pandas.DataFrame: 季節性分析結果
"""
seasonal_data = []
for keyword in keywords:
print(f"📈 '{keyword}'の季節性分析中...")
# 過去の検索トレンドデータを取得
request = self.keyword_plan_idea_service.generate_keyword_historical_metrics_request()
request.customer_id = self._get_customer_id()
request.keywords = [keyword]
try:
response = self.keyword_plan_idea_service.generate_keyword_historical_metrics(request=request)
for result in response:
for monthly_metric in result.keyword_metrics.monthly_search_volumes:
seasonal_data.append({
'キーワード': keyword,
'年': monthly_metric.year,
'月': monthly_metric.month,
'検索ボリューム': monthly_metric.monthly_searches
})
except Exception as e:
print(f"❌ {keyword}の季節性分析でエラー: {e}")
return pd.DataFrame(seasonal_data)
データの可視化と分析
1. 検索ボリューム分析
import matplotlib.pyplot as plt
import seaborn as sns
def visualize_search_volume(df):
"""
検索ボリュームの可視化
"""
plt.figure(figsize=(12, 8))
# 検索ボリューム上位20キーワードのバープロット
top_keywords = df.nlargest(20, '月間検索ボリューム')
sns.barplot(data=top_keywords,
y='キーワード',
x='月間検索ボリューム',
palette='viridis')
plt.title('検索ボリューム上位20キーワード', fontsize=16, fontweight='bold')
plt.xlabel('月間検索ボリューム', fontsize=12)
plt.ylabel('キーワード', fontsize=12)
plt.tight_layout()
plt.show()
def competition_analysis(df):
"""
競合度分析
"""
plt.figure(figsize=(10, 6))
# 競合度別キーワード数
competition_counts = df['競合度'].value_counts()
colors = ['#2ecc71', '#f39c12', '#e74c3c']
plt.pie(competition_counts.values,
labels=competition_counts.index,
autopct='%1.1f%%',
colors=colors,
startangle=90)
plt.title('競合度別キーワード分布', fontsize=16, fontweight='bold')
plt.axis('equal')
plt.show()
2. キーワード難易度スコアの算出
def calculate_keyword_difficulty(df):
"""
キーワード難易度スコアの算出
"""
df_scored = df.copy()
# 正規化関数
def normalize_score(series, reverse=False):
min_val = series.min()
max_val = series.max()
if max_val == min_val:
return pd.Series([0.5] * len(series))
normalized = (series - min_val) / (max_val - min_val)
return 1 - normalized if reverse else normalized
# 各要素のスコア算出（0-1の範囲）
volume_score = normalize_score(df_scored['月間検索ボリューム'])
competition_map = {'LOW': 0.3, 'MEDIUM': 0.6, 'HIGH': 0.9}
competition_score = df_scored['競合度'].map(competition_map).fillna(0.5)
bid_score = normalize_score(df_scored['推奨入札価格（上限）'])
# 難易度スコアの算出（重み付き平均）
difficulty_score = (
volume_score * 0.4 + # 検索ボリューム（40%）
competition_score * 0.4 + # 競合度（40%）
bid_score * 0.2 # 入札価格（20%）
) * 100
df_scored['難易度スコア'] = difficulty_score.round(1)
# 難易度レベルの分類
def classify_difficulty(score):
if score >= 80:
return '非常に困難'
elif score >= 60:
return '困難'
elif score >= 40:
return '普通'
elif score >= 20:
return '易しい'
else:
return '非常に易しい'
df_scored['難易度レベル'] = df_scored['難易度スコア'].apply(classify_difficulty)
return df_scored
エラーハンドリングとトラブルシューティング
よくあるエラーと解決方法
1. 認証エラー
def handle_auth_errors():
"""
認証エラーのハンドリング
"""
try:
client = GoogleAdsClient.load_from_storage()
return client
except FileNotFoundError:
print("❌ google-ads.yamlファイルが見つかりません")
print("💡 解決方法:")
print(" 1. Google Ads APIの設定を確認")
print(" 2. 認証情報ファイルの場所を確認")
return None
except Exception as e:
print(f"❌ 認証エラー: {e}")
print("💡 Google Cloud Consoleで認証設定を確認してください")
return None
2. API制限エラー
import time
from google.ads.googleads.errors import GoogleAdsException
def safe_api_call(func, max_retries=3, delay=1):
"""
APIリクエストの安全な実行（リトライ機能付き）
"""
for attempt in range(max_retries):
try:
return func()
except GoogleAdsException as ex:
if "RATE_EXCEEDED" in str(ex):
print(f"⏳ API制限に達しました。{delay}秒待機します...")
time.sleep(delay * (2 ** attempt)) # 指数的バックオフ
continue
else:
raise ex
except Exception as e:
if attempt == max_retries - 1:
raise e
time.sleep(delay)
return None
3. データ品質チェック
def validate_keyword_data(df):
"""
取得したキーワードデータの品質チェック
"""
issues = []
# 必須カラムの存在確認
required_columns = ['キーワード', '月間検索ボリューム', '競合度']
missing_columns = [col for col in required_columns if col not in df.columns]
if missing_columns:
issues.append(f"必須カラムが不足: {missing_columns}")
# データの妥当性チェック
if 'キーワード' in df.columns:
empty_keywords = df['キーワード'].isna().sum()
if empty_keywords > 0:
issues.append(f"空のキーワードが{empty_keywords}件あります")
if '月間検索ボリューム' in df.columns:
invalid_volumes = df['月間検索ボリューム'].apply(lambda x: x < 0 if pd.notna(x) else False).sum()
if invalid_volumes > 0:
issues.append(f"無効な検索ボリュームが{invalid_volumes}件あります")
# レポート出力
if issues:
print("⚠️ データ品質に関する警告:")
for issue in issues:
print(f" - {issue}")
else:
print("✅ データ品質チェック: 問題なし")
return len(issues) == 0
実践的な活用例
1. コンテンツ戦略の立案
def content_strategy_analysis(df):
"""
コンテンツ戦略立案のための分析
"""
# 検索ボリューム別にキーワードを分類
high_volume = df[df['月間検索ボリューム'] >= 10000]
medium_volume = df[(df['月間検索ボリューム'] >= 1000) & (df['月間検索ボリューム'] < 10000)]
low_volume = df[df['月間検索ボリューム'] < 1000]
# 競合度と検索ボリュームの関係分析
strategy_recommendations = []
for _, keyword_data in df.iterrows():
keyword = keyword_data['キーワード']
volume = keyword_data['月間検索ボリューム']
competition = keyword_data['競合度']
if volume >= 5000 and competition == 'LOW':
strategy_recommendations.append({
'キーワード': keyword,
'戦略': '最優先ターゲット',
'理由': '高検索ボリューム・低競合',
'優先度': 'A'
})
elif volume >= 1000 and competition in ['LOW', 'MEDIUM']:
strategy_recommendations.append({
'キーワード': keyword,
'戦略': '中期ターゲット',
'理由': '適度な検索ボリューム・実現可能な競合度',
'優先度': 'B'
})
elif competition == 'LOW':
strategy_recommendations.append({
'キーワード': keyword,
'戦略': 'ロングテール戦略',
'理由': '低競合でニッチな需要',
'優先度': 'C'
})
return pd.DataFrame(strategy_recommendations)
2. 自動レポート生成
def generate_automated_report(df, output_path="keyword_report.html"):
"""
HTML形式の自動レポート生成
"""
html_template = """
<!DOCTYPE html>
<html>
<head>
<title>キーワード調査レポート</title>
<meta charset="utf-8">
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
.header {{ background-color: #2c3e50; color: white; padding: 20px; text-align: center; }}
.summary {{ background-color: #ecf0f1; padding: 15px; margin: 20px 0; }}
.metric {{ display: inline-block; margin: 10px; text-align: center; }}
.metric-value {{ font-size: 2em; color: #3498db; font-weight: bold; }}
.metric-label {{ color: #7f8c8d; }}
table {{ border-collapse: collapse; width: 100%; margin: 20px 0; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background-color: #3498db; color: white; }}
.high-priority {{ background-color: #e8f5e8; }}
.medium-priority {{ background-color: #fff3e0; }}
.low-priority {{ background-color: #fce4ec; }}
</style>
</head>
<body>
<div class="header">
<h1>キーワード調査レポート</h1>
<p>生成日時: {timestamp}</p>
</div>
<div class="summary">
<h2>調査サマリー</h2>
<div class="metric">
<div class="metric-value">{total_keywords}</div>
<div class="metric-label">総キーワード数</div>
</div>
<div class="metric">
<div class="metric-value">{avg_volume}</div>
<div class="metric-label">平均検索ボリューム</div>
</div>
<div class="metric">
<div class="metric-value">{high_volume_count}</div>
<div class="metric-label">高ボリューム<br>キーワード数</div>
</div>
</div>
<h2>キーワード詳細</h2>
{keywords_table}
</body>
</html>
"""
# 統計情報の計算
total_keywords = len(df)
avg_volume = int(df['月間検索ボリューム'].mean())
high_volume_count = len(df[df['月間検索ボリューム'] >= 5000])
# HTMLテーブルの生成
keywords_table = df.head(50).to_html(
classes='keywords-table',
table_id='keywords',
escape=False,
index=False
)
# HTMLレポートの生成
html_content = html_template.format(
timestamp=datetime.now().strftime("%Y年%m月%d日 %H:%M:%S"),
total_keywords=total_keywords,
avg_volume=avg_volume,
high_volume_count=high_volume_count,
keywords_table=keywords_table
)
# ファイルに保存
with open(output_path, 'w', encoding='utf-8') as f:
f.write(html_content)
print(f"📊 レポートを {output_path} に生成しました")
return output_path
3. 定期実行の自動化
import schedule
import time
def automated_keyword_research():
"""
定期的なキーワード調査の実行
"""
def run_research():
print("🔄 定期キーワード調査を開始...")
# メイン処理の実行
config = GoogleAdsConfig()
client = config.get_client()
kp_automation = KeywordPlannerAutomation(client)
# 複数のシードキーワードセットで調査
keyword_sets = {
'SEO関連': ['SEO', 'キーワード調査', '検索エンジン最適化'],
'Python関連': ['Python', 'プログラミング', 'データ分析'],
'マーケティング関連': ['デジタルマーケティング', 'コンテンツマーケティング', 'SNSマーケティング']
}
all_results = []
for category, keywords in keyword_sets.items():
df = kp_automation.get_keyword_ideas(keywords)
df['カテゴリ'] = category
all_results.append(df)
# 結果を統合
final_df = pd.concat(all_results, ignore_index=True)
# レポート生成
timestamp = datetime.now().strftime("%Y%m%d")
generate_automated_report(final_df, f"daily_report_{timestamp}.html")
print("✅ 定期キーワード調査完了")
# 毎日午前9時に実行
schedule.every().day.at("09:00").do(run_research)
# 毎週月曜日に実行
schedule.every().monday.at("10:00").do(run_research)
print("⏰ 定期実行スケジュールを設定しました")
while True:
schedule.run_pending()
time.sleep(60) # 1分ごとにスケジュールをチェック
パフォーマンス最適化のテクニック
1. 並列処理による高速化
import concurrent.futures
import threading
class OptimizedKeywordPlanner(KeywordPlannerAutomation):
def __init__(self, client, max_workers=5):
super().__init__(client)
self.max_workers = max_workers
self.request_lock = threading.Lock()
def batch_keyword_research(self, keyword_batches):
"""
バッチ処理による並列キーワード調査
Args:
keyword_batches (list): キーワードバッチのリスト
Returns:
pandas.DataFrame: 統合結果
"""
all_results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
# 並列でバッチ処理を実行
future_to_batch = {
executor.submit(self._process_batch, batch): batch
for batch in keyword_batches
}
for future in concurrent.futures.as_completed(future_to_batch):
batch = future_to_batch[future]
try:
result = future.result()
if not result.empty:
all_results.append(result)
print(f"✅ バッチ処理完了: {len(batch)}キーワード")
except Exception as e:
print(f"❌ バッチ処理エラー: {e}")
return pd.concat(all_results, ignore_index=True) if all_results else pd.DataFrame()
def _process_batch(self, keyword_batch):
"""
個別バッチの処理
"""
with self.request_lock: # API制限を考慮したリクエスト制御
time.sleep(0.1) # レート制限対策
return self.get_keyword_ideas(keyword_batch)
2. キャッシュ機能の実装
import pickle
import hashlib
from functools import wraps
def cache_results(cache_duration_hours=24):
"""
結果をキャッシュするデコレータ
"""
def decorator(func):
@wraps(func)
def wrapper(self, *args, **kwargs):
# キャッシュキーの生成
cache_key = hashlib.md5(
str(args).encode() + str(kwargs).encode()
).hexdigest()
cache_file = f"cache_{cache_key}.pkl"
# キャッシュファイルの確認
if os.path.exists(cache_file):
file_age = time.time() - os.path.getmtime(cache_file)
if file_age < cache_duration_hours * 3600: # 24時間以内
print("📦 キャッシュからデータを読み込みました")
with open(cache_file, 'rb') as f:
return pickle.load(f)
# 新しいデータを取得
result = func(self, *args, **kwargs)
# キャッシュに保存
with open(cache_file, 'wb') as f:
pickle.dump(result, f)
return result
return wrapper
return decorator
# 使用例
class CachedKeywordPlanner(KeywordPlannerAutomation):
@cache_results(cache_duration_hours=24)
def get_keyword_ideas(self, seed_keywords, location_id="2392", language_id="1005"):
return super().get_keyword_ideas(seed_keywords, location_id, language_id)
3. メモリ最適化
def memory_efficient_processing(large_keyword_list, batch_size=100):
"""
大量のキーワードをメモリ効率よく処理
"""
def process_chunk(chunk):
# チャンク単位での処理
kp_automation = KeywordPlannerAutomation(client)
result = kp_automation.get_keyword_ideas(chunk)
# 必要な列のみを保持
essential_columns = ['キーワード', '月間検索ボリューム', '競合度']
return result[essential_columns]
# ジェネレータを使用してメモリ使用量を削減
for i in range(0, len(large_keyword_list), batch_size):
chunk = large_keyword_list[i:i + batch_size]
yield process_chunk(chunk)
# ガベージコレクションを強制実行
import gc
gc.collect()
セキュリティとベストプラクティス
1. 認証情報の安全な管理
import os
from cryptography.fernet import Fernet
class SecureConfig:
def __init__(self):
self.encryption_key = self._get_or_create_key()
self.cipher_suite = Fernet(self.encryption_key)
def _get_or_create_key(self):
"""
暗号化キーの取得または生成
"""
key_file = '.secret_key'
if os.path.exists(key_file):
with open(key_file, 'rb') as f:
return f.read()
else:
key = Fernet.generate_key()
with open(key_file, 'wb') as f:
f.write(key)
return key
def encrypt_credentials(self, credentials_dict):
"""
認証情報の暗号化
"""
json_data = json.dumps(credentials_dict).encode()
encrypted_data = self.cipher_suite.encrypt(json_data)
with open('.encrypted_credentials', 'wb') as f:
f.write(encrypted_data)
def decrypt_credentials(self):
"""
認証情報の復号化
"""
with open('.encrypted_credentials', 'rb') as f:
encrypted_data = f.read()
decrypted_data = self.cipher_suite.decrypt(encrypted_data)
return json.loads(decrypted_data.decode())
2. ログ機能の実装
import logging
from logging.handlers import RotatingFileHandler
def setup_logging():
"""
ログ設定の初期化
"""
# ログフォーマットの設定
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# ファイルハンドラー（回転ログ）
file_handler = RotatingFileHandler(
'keyword_planner.log',
maxBytes=10*1024*1024, # 10MB
backupCount=5
)
file_handler.setFormatter(formatter)
file_handler.setLevel(logging.INFO)
# コンソールハンドラー
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
console_handler.setLevel(logging.WARNING)
# ロガーの設定
logger = logging.getLogger('KeywordPlanner')
logger.setLevel(logging.INFO)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
# 使用例
logger = setup_logging()
class LoggingKeywordPlanner(KeywordPlannerAutomation):
def __init__(self, client):
super().__init__(client)
self.logger = logger
def get_keyword_ideas(self, seed_keywords, location_id="2392", language_id="1005"):
self.logger.info(f"キーワード調査開始: {seed_keywords}")
try:
result = super().get_keyword_ideas(seed_keywords, location_id, language_id)
self.logger.info(f"調査完了: {len(result)}件のキーワードを取得")
return result
except Exception as e:
self.logger.error(f"キーワード調査エラー: {e}")
raise
3. API使用量の監視
class APIUsageMonitor:
def __init__(self, daily_limit=10000):
self.daily_limit = daily_limit
self.usage_file = '.api_usage.json'
self.usage_data = self._load_usage_data()
def _load_usage_data(self):
"""
使用量データの読み込み
"""
if os.path.exists(self.usage_file):
with open(self.usage_file, 'r') as f:
return json.load(f)
return {}
def _save_usage_data(self):
"""
使用量データの保存
"""
with open(self.usage_file, 'w') as f:
json.dump(self.usage_data, f)
def record_api_call(self, endpoint, cost=1):
"""
APIコールの記録
"""
today = datetime.now().strftime("%Y-%m-%d")
if today not in self.usage_data:
self.usage_data[today] = {}
if endpoint not in self.usage_data[today]:
self.usage_data[today][endpoint] = 0
self.usage_data[today][endpoint] += cost
self._save_usage_data()
# 使用量チェック
daily_total = sum(self.usage_data[today].values())
if daily_total >= self.daily_limit * 0.9: # 90%に達したら警告
logger.warning(f"API使用量が制限の90%に達しました: {daily_total}/{self.daily_limit}")
return daily_total < self.daily_limit
def get_usage_report(self):
"""
使用量レポートの生成
"""
today = datetime.now().strftime("%Y-%m-%d")
if today in self.usage_data:
daily_total = sum(self.usage_data[today].values())
remaining = self.daily_limit - daily_total
print(f"📊 本日のAPI使用量: {daily_total}/{self.daily_limit}")
print(f"📊 残り使用可能量: {remaining}")
return {
'used': daily_total,
'limit': self.daily_limit,
'remaining': remaining,
'percentage': (daily_total / self.daily_limit) * 100
}
else:
return {'used': 0, 'limit': self.daily_limit, 'remaining': self.daily_limit, 'percentage': 0}
まとめ：自動化がもたらす未来
あなたが手に入れるもの
この記事で紹介したキーワードプランナー自動化を実装することで、あなたは以下を手に入れることができます：
時間の自由
- 手作業で数時間かかっていた作業が数分で完了
- 浮いた時間をより戦略的な業務に活用可能
- ワークライフバランスの大幅な改善
データの精度向上
- 人的ミスの完全排除
- 一貫したデータ取得プロセス
- 大量データの効率的な処理
スケーラビリティの獲得
- 案件数が増えても作業時間は変わらず
- 複数のプロジェクトを並行処理可能
- チーム全体の生産性向上
競争優位性の確立
- 他社より迅速な市場分析
- データドリブンな意思決定
- クライアントへの付加価値提供
次のステップ
1. 今すぐ始めよう
# 今すぐこのコマンドを実行してください
pip install google-ads google-auth pandas
2. 段階的な実装
- まずは基本機能から実装
- 徐々に高度な機能を追加
- チームメンバーと知識を共有
3. 継続的な改善
- 定期的な機能追加
- パフォーマンスの監視と最適化
- 新しいAPIアップデートへの対応
最後のメッセージ
「自動化は魔法ではありません。それは、あなたの時間と労力を本当に価値のあることに向ける道具なのです。」
今日から始めれば、来週にはすでに数時間の時間を節約できているでしょう。来月には、手作業では不可能だった大規模な分析が当たり前になっているはずです。
そして1年後、あなたは振り返ってこう思うでしょう： 「なぜもっと早く自動化しなかったのだろう？」
その未来を今から始めませんか？
※本記事で紹介したコードは教育目的のものです。実際の運用では、各種利用規約を遵守し、適切なテストを行った上でご利用ください。
コメント