-
Notifications
You must be signed in to change notification settings - Fork 1
/
scraping_rakuten_point.py
219 lines (169 loc) · 6.42 KB
/
scraping_rakuten_point.py
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
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
import pandas as pd
from configparser import ConfigParser
import time
import csv
import os
# 楽天ポイントクラブ・ポイント実績URL
RAKUTEN_POINT_URL = 'https://point.rakuten.co.jp/history/?l-id=point_top_history_pc'
# 設定ファイル
CONFIG_FILE = '.config'
# 出力ファイルのエンコーディング
ENCODING = 'utf-8-sig'
# pythonファイルと同じディレクトリにchromeriver.exeがある場合引数は空でOK
driver = webdriver.Chrome()
# 楽天へのログイン
def login_rakuten():
# 設定ファイルからパスワードを取得
config = ConfigParser()
config.read(CONFIG_FILE)
section = 'RAKUTEN'
print("ログインページ呼び出し")
driver.get(RAKUTEN_POINT_URL)
print("ログインページ表示")
# chromedriver起動待ちを3秒似設定
time.sleep(3)
try:
print("ログイン処理開始")
# 楽天ポイントのログイン画面からIDフィールドを検索、IDを入力
id = driver.find_element(By.ID, "loginInner_u")
print("ID欄検索")
# ↓自分のIDを入力
id.send_keys(config.get(section, 'userid'))
print("IDセット")
# 同じくパスワードフィールドを検索、IDを入力
password = driver.find_element(By.ID, "loginInner_p")
print("パスワード欄検索")
# ↓自分のパスワードを入力
password.send_keys(config.get(section, 'password'))
print("パスワードセット")
time.sleep(1)
# ログインボタンを検索、ログインボタンをクリック
login_button = driver.find_element(By.NAME, "submit")
login_button.click()
print("ログイン実行")
except Exception as e:
print('Login Error')
print('type:' + str(type(e)))
print('args:' + str(e.args))
print('message:' + e.message)
print('e自身:' + str(e))
return False
print("ログイン処理完了")
return True
### ポイント1件分の処理 ###
def get_point_info(pd, df, columns, post):
try:
# 日付
date = post.find_element(By.CSS_SELECTOR, 'td.detail .date').text
date = date.replace("[", "")
date = date.replace("]", "")
# サービス
service = post.find_element(By.CSS_SELECTOR, '.service a').text
# 内容
detail = post.find_element(By.CSS_SELECTOR, 'td.detail').text
# ランクアップ
try:
rankup = post.find_element(By.CSS_SELECTOR, '.label-rankup').text
detail = detail.replace('ランクアップ対象', '').strip()
except:
rankup = "-"
# アクション、期間限定
action = post.find_element(By.CSS_SELECTOR, '.action').text
limited = "-"
if action == "獲得\n期間限定":
action = "獲得"
limited = "期間限定"
# ポイント
point = post.find_element(By.CSS_SELECTOR, '.point').text
point = point.replace(',','')
# point = int(point.replace(',','')) * -1
# 備考
note = post.find_element(By.CSS_SELECTOR, '.note').text
if len(note) < 1:
note = "-"
# 連番列を設定
no = len(df.index) + 1
# スクレイピングした情報をリストに追加
se = pd.Series([no, date, service, detail, rankup, action, limited, point, note], columns)
df = pd.concat([df, pd.DataFrame([se])], ignore_index=True)
print("1件完了(" + str(no) + "):" + date + ": " + point)
except Exception as e:
print('Proc Error')
print('type:' + str(type(e)))
print('args:' + str(e.args))
print('message:' + e.message)
print('e自身:' + str(e))
return df
### 次ページを取得 ###
def get_next_page(page):
try:
# 次のページに進むためのURLを取得
confirm_page = driver.find_element(By.CSS_SELECTOR, "ul.pagination li:last-child a").text
url = driver.find_element(By.CSS_SELECTOR, "ul.pagination li:last-child a").get_attribute("href")
time.sleep(3)
print(str(page) + "ページを取得中...")
driver.get(url)
time.sleep(1)
except:
print("最終ページ")
return confirm_page
### ポイント情報の取得 ###
def get_r_info():
# 表示ページの定義
page = 1
confirm_page = 'NEXT'
# リストを作成
columns = ['No', '日付', 'サービス', '内容', 'ランクアップ対象', '利用/獲得', '期間限定', 'ポイント', '備考']
# 配列名を指定する
df = pd.DataFrame(columns=columns)
# 実行
while(str(confirm_page) == 'NEXT'):
# ポイントの獲得情報を取得
print("ポイント獲得情報取得")
posts = driver.find_elements(By.CSS_SELECTOR, '.get')
for post in posts:
df = get_point_info(pd, df, columns, post)
print("獲得情報群取得完了")
# ポイントの利用情報を取得
print("ポイント利用情報取得")
posts = driver.find_elements(By.CSS_SELECTOR, '.use')
for post in posts:
df = get_point_info(pd, df, columns, post)
print("利用情報取得完了")
print("ページ遷移処理開始")
# ページ数を1増やす
page += 1
# デバッグモードでは2ページで強制終了
global debug_mode
if debug_mode:
if page > 2:
print("デバッグモードのため強制終了")
break
# 次ページを取得
confirm_page = get_next_page(page)
return df
### データの書き出し ###
def write_point_data(df):
# 最後に得たデータをCSVにして保存
filename = "rakuten_point.csv"
df.to_csv("csv/" + filename, encoding=ENCODING, index=False)
print("終了しました!!")
### メイン ###
# デフォルト値
default_configs = {'DEBUG': { 'debug' : 'no'}}
# DEBUGモード
debug_mode = False
config = ConfigParser()
config.read_dict(default_configs)
config.read(CONFIG_FILE)
# メイン処理
if config.get('DEBUG', 'debug') == 'yes':
print("DEBUGモード")
debug_mode = True
if login_rakuten():
df = get_r_info()
if not df.empty:
write_point_data(df)