楽天店舗をPythonスクレイピングして商品一覧をリサーチ
プログラムの力を使えば、時間のかかる面倒な作業も自動で処理できる。私自身、ネットショップの運営に携わったことがあり、その時は人手が足らず全てが後手で大変だった。
今はプログラムが書けるので、定期処理などは自動化できる。
ネットショップのバックオフィス業務で、自動化できることの一つに競合チェックがある。楽天では（というか大抵のモールでは）、同一商品の最安価格が簡単にわかる機能がある。
利用者はワンクリックで商品の安い順で一覧を見ることができる。当然ネットショップとしては、お客さんの目につく場所に自分の店が表示されるように、最安価格を維持したいわけだ。
かくしてネットショップ店長やら従業員は、人手をかけて商品の最安価格をリストアップするのだが、こういったことは今やプログラムで簡単に自動処理できる。
また、最安価格以外にも、競合店舗がどんな商品を扱っているのか、その店舗の取り扱い商品を一覧にして、商品名、価格、画像などをデータセットにしたいという要望もあるだろう。
人手をかけるよりプログラムにやらせる
これも人手をかけてする場合は、商品一覧から商品詳細ページにクリックで遷移し、1ページずつ商品名や価格をコピペするという気の遠くなるような作業となる。商品数が多い場合は真面目にやっているとバカみたいに時間がかかる。中には短期のアルバイトを雇ったり、クラウドソーシングで作業依頼したりすることもあるだろう。
しかし、これもプログラムの力を使えば、ネットショップ店長が寝ている間にコンピュータが勝手に情報を収集してくれるからありがたい。
インターネットに公開されている情報を収集する方法を一般に「スクレイピング」という。人がブラウザを操作するのではなく、コンピュータに自動でブラウザを操作させるような感じだ。アクセスしたページの特定の情報だけを抜き出してエクセルやスプレッドシートにするなんてこともできる。
以下に楽天の指定の店舗から商品一覧を生成するサンプルのソースコードを掲載する。もちろんサーバーの過負荷にならない配慮をしている。
import mechanicalsoup
import requests
import time
import datetime
import selenium
import random
from bs4 import BeautifulSoup
import os
import sys
import re
from requests.exceptions import Timeout
import csv
today = str(datetime.date.today())
class make_promotion_list:
"""
店舗リスト作成用のスクレイプ
"""
def __init__(self, url):
self.base_url = url
self.shopcode = None
self.master_list = []
self.item_page_list = []
def simple_request(self, url, first):
"""
指定URLから一意なaタグをリスト化
first:初回リクエストかどうか bool
"""
print(url, 'を実行 ', len(self.master_list), '個のリスト')
atag_list = []
#urlにshopcodeが含まれない場合は対象外とする
if self.shopcode not in url:
return False
try:
r = requests.get(url, timeout=(30.0, 47.5))
except (requests.exceptions.MissingSchema) as e:
return False
if r.status_code == 200 and 'text/html' in r.headers['Content-type']:
soup = BeautifulSoup(r.content, 'html.parser')
inquiry_anchor_list = soup.find_all('a')
if first == True: #初回リクエスト
for row in inquiry_anchor_list:
if row.get('href') not in self.master_list:
self.master_list.append(row.get('href'))
if row.get('href') not in atag_list:
atag_list.append(row.get('href'))
return atag_list
elif first == False: #2回目以降のリクエストはmasterに追加するだけ
for row in inquiry_anchor_list:
if row.get('href') not in self.master_list:
self.master_list.append(row.get('href'))
return len(self.master_list)
def get_rakuten_shopcode(self, url):
"""
urlから楽天のshopcodeを取得
https://www.rakuten.co.jp/shopcode/
"""
pattern = 'https://www.rakuten.co.jp/.+/'
if re.match(pattern, url) != None:
self.shopcode = url[26:].replace('/', '')
return self.shopcode
def get_item_url(self):
"""
完成したmaster_listから個別商品のurlのみをピックアップ
https://item.rakuten.co.jp/shopcode/10000004/
楽天の場合はこういうurl
"""
pattern = 'https://item.rakuten.co.jp/[a-z\-]+/[0-9]+/'
for row in self.master_list:
try:
if re.match(pattern, row) != None:
self.item_page_list.append(row)
except TypeError:
pass
return self.item_page_list
def get_item_info(self, url):
"""
楽天の詳細ページから商品情報を取得
"""
try:
r = requests.get(url, timeout=(30.0, 47.5))
print(r.status_code)
except (requests.exceptions.MissingSchema) as e:
return False
i = 0
while r.status_code == 503:
try:
r = requests.get(url, timeout=(30.0, 47.5))
print(r.status_code)
except (Timeout, requests.exceptions.MissingSchema) as e:
print('timeoutで取得できず')
time.sleep(10)
i += 1
if i > 5: #5回503が続いたら次にいく
break
if r.status_code == 200:
break
tmp_item_data = {}
if r.status_code == 200 and 'text/html' in r.headers['Content-type']:
soup = BeautifulSoup(r.content, 'html.parser')
item_name = soup.select_one('.item_name')
item_price = soup.select_one('.price2')
tmp_item_data['url'] = url
item_number_pattern = '[0-9]+'
match_result = re.search(item_number_pattern, url)
if match_result != None:
tmp_item_data['item_id'] = match_result.group()
else:
tmp_item_data['item_id'] = None
"""
try:
tmp_item_data['item_id'] = item_id.text
except AttributeError:
tmp_item_data['item_id'] = None
"""
try:
tmp_item_data['item_name'] = item_name.text
except AttributeError:
tmp_item_data['item_name'] = None
try:
tmp_item_data['item_price'] = item_price.text
except AttributeError:
tmp_item_data['item_price'] = None
elif r.status_code == 503:
tmp_item_data['url'] = url
tmp_item_data['item_id'] = "楽天のサーバー遅延などで取得できず"
tmp_item_data['item_name'] = None
tmp_item_data['item_price'] = None
return tmp_item_data
if __name__ == "__main__":
while True:
print('調査したい楽天店舗のショップトップページを入力してください')
url = input('url:')
#urlのvalidation
pattern = 'https://www.rakuten.co.jp/[a-z\-]+/?'
result = re.match(pattern, url)
if result == None:
print('urlが不正です。https://www.rakuten.co.jp/shopcode/ のように入力してください')
elif result != None:
print('スクレイピングを開始します。商品数によっては数時間〜1日かかることもあります。')
break
request_instance = make_promotion_list(url)
shopcode = request_instance.get_rakuten_shopcode(url)
#初回リクエストでリンクリスト作成
url_list = request_instance.simple_request(url, True)
if url_list == False:
print('処理を終了')
exit()
#2回目以降
for url in url_list:
request_instance.simple_request(url, False)
session_interval_time = random.uniform(1, 1.7)
time.sleep(session_interval_time) #何秒おきにリクエストするか1〜1.7秒の中でランダムで設定
else:
count = request_instance.simple_request(url, False)
print('数:', count)
#アイテムページのみをピックアップ
item_page_list = request_instance.get_item_url()
print('アイテム数: ', len(item_page_list))
#詳細ページurlから商品情報を取得
item_list = []
for row in item_page_list:
print(row, 'を処理')
item_list.append(request_instance.get_item_info(row))
time.sleep(1)
#csv保存
filepath = "./csv_data/"
filename = today + '-' + str(shopcode) + '.csv'
with open(filepath + filename, 'w') as f:
writer = csv.writer(f)
for row in item_list:
writer.writerow([row['url'], row['item_id'], row['item_name'], row['item_price']])
