1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
| import os
import json
import csv
import datetime
from serpapi import GoogleSearch
import mylib.common as common
SERPAPI_KEY = "[APIキー]"
DOMAIN = "[ドメイン]"
DATA_DIR = '[出力先ディレクトリ]'
SEARCH_WORD_FILE = '[検索ワードファイル]'
def makeDir(path):
"""ディレクトリを作成する
Args:
path (str): 作成するディレクトリのパス
Returns:
None
Note:
既に存在するディレクトリの場合、何もしない
"""
if not os.path.isdir(path):
os.mkdir(path)
def sanitize_filename(s: str) -> str:
"""ファイル名として使えない文字を安全な文字に置換する
Args:
s (str): オリジナルのファイル名（キーワードなど）
Returns:
str: サニタイズされたファイル名
Note:
置換対象：/, \\, :, *, ", <, >, | → _
連続したアンダースコアは1つに統一
先頭・末尾の空白とアンダースコアは削除
"""
import re
s = re.sub(r'[/\\:*?"<>|]', '_', s)
s = re.sub(r'_+', '_', s)
return s.strip('_ ')
def save_results_as_csv(keyword: str, organic_results: list):
"""検索結果を CSV ファイルに保存する
SerpAPI から取得した検索結果を CSV ファイルに保存。
各検索キーワードに対して1つのCSVファイルが生成される。
Args:
keyword (str): 検索キーワード（ファイル名に使用）
organic_results (list): オーガニック検索結果のリスト
各要素は {'title': str, 'link': str, 'snippet': str, ...} の形式
Returns:
None
Note:
ファイル名はsanitize_filename()で安全化される
自サイト判定：DOMAIN がURLに含まれていれば '✓' マーク
"""
makeDir(DATA_DIR)
csv_dir = os.path.join(DATA_DIR, 'csv')
makeDir(csv_dir)
safe_keyword = sanitize_filename(keyword)
csv_path = os.path.join(csv_dir, f'{safe_keyword}.csv')
try:
with open(csv_path, 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = ['順位', 'タイトル', 'URL', '説明', '自サイト']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for idx, result in enumerate(organic_results, 1):
title = result.get('title', '')
link = result.get('link', '')
snippet = result.get('snippet', '')
is_own_site = '✓' if DOMAIN in link else ''
writer.writerow({
'順位': idx,
'タイトル': title,
'URL': link,
'説明': snippet,
'自サイト': is_own_site
})
except Exception as e:
print(f"[ERROR] CSV save failed: {e}")
def save_summary_csv(summary_data: list):
"""複数キーワードの検索結果をサマリー CSV に保存する
全キーワードの検索結果（検出/未検出、順位）を1つの CSV にまとめる。
Args:
summary_data (list): サマリーデータのリスト
各要素は {'キーワード': str, '検索結果': str, '順位': int or ''} の形式
Returns:
None
Note:
ファイル名には実行日の日付が自動付与される
検出時：「検出（n位）」、未検出時：「未検出」
"""
makeDir(DATA_DIR)
today = datetime.datetime.today().strftime("%Y%m%d")
summary_path = os.path.join(DATA_DIR, f'{today}_summary.csv')
try:
with open(summary_path, 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = ['キーワード', '検索結果', '順位']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for entry in summary_data:
writer.writerow(entry)
except Exception as e:
print(f"[ERROR] Summary save failed: {e}")
def check_site_in_results(keyword):
"""キーワード検索を実行し、自サイトが検索結果に含まれているか確認する
SerpAPI を使用して Google 検索を実行。取得した検索結果から
自サイト（DOMAIN 変数で指定）が含まれているか確認し、
順位を取得する。結果は JSON と CSV の両形式で保存する。
Args:
keyword (str): 検索キーワード（例：「湊線で行く磯崎海岸...」）
Returns:
tuple: (found: bool, rank: int or None)
found (bool):
True - 自サイトが検索結果の1～10位に見つかった
False - 見つからなかった
rank (int or None):
自サイトの順位（1～10）
未検出の場合は None
Note:
- 検索は日本向け（gl=jp, hl=ja）で実行
- 上位10件の検索結果を取得
- 全結果は JSON で保存、整形結果は CSV で保存
"""
try:
params = {
"q": keyword,
"api_key": SERPAPI_KEY,
"gl": "jp",
"hl": "ja",
"num": 10
}
search = GoogleSearch(params)
results = search.get_dict()
makeDir(DATA_DIR)
save_dir = os.path.join(DATA_DIR, 'response')
makeDir(save_dir)
with open(os.path.join(save_dir, f'response_{keyword}.json'), 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
organic_results = results.get('organic_results', [])
found = False
rank = None
for idx, result in enumerate(organic_results, 1):
link = result.get('link', '')
title = result.get('title', '')
if DOMAIN in link:
found = True
rank = idx
break
if not found:
print(f"✗ '{keyword}' - サイトが見つかりません")
save_results_as_csv(keyword, organic_results)
return found, rank
except Exception as e:
print(f"[ERROR] {e}")
return False, None
def main():
"""メイン処理：url.txt から読み込んだキーワードで一括検索を実行
処理フロー：
1. url.txt からキーワードリストを読み込み
2. 各キーワードについて check_site_in_results() を呼び出し
3. 結果をサマリーに集約
4. 最終結果を20260215_summary.csv のような
日付付きサマリーファイルに出力
Returns:
None
Note:
- url.txt がない or 空の場合は処理終了
- キーワード毎に print で進捗表示
- エラーは個別キーワードでキャッチされ、全体処理は続行
"""
lists = common.read_file(SEARCH_WORD_FILE)
if not lists:
print("no urls to process")
return
summary_data = []
for i, keyword in enumerate(lists, start=1):
print(f"\n[{i}/{len(lists)}] start: {keyword}")
found, rank = check_site_in_results(keyword)
status = f"検出" if found else "未検出"
summary_data.append({
'検索結果': status,
'順位': rank if rank else '',
'キーワード': keyword,
})
print(f"\n[{i}/{len(lists)}] end: {keyword}")
save_summary_csv(summary_data)
if __name__ == "__main__":
main()
|
