[Backend] Flask + MongoDB + Scrapy 各種串接
由於之前工作都是接觸 SQL 資料庫較多，這次藉 side project 的機會來碰碰 NoSQL。另外沒有接觸過爬蟲相關的框架，於是想了一個小 project 來做做看。以下為該專案的架構圖，後端機器包含 Flask & Scrapy 框架，DB 則是使用 mongoDB。
Dockerfile
version: '3.7'
services:
backend:
build:
context: .
dockerfile: backend/Dockerfile
container_name: backend-container
image: localhost/backend-image
ports:
- 5000:5000
volumes:
- ./backend:/backend-code
env_file:
- .env
depends_on:
- db
db:
container_name: mongodb
image: mongo:latest
environment:
- MONGO_INITDB_DATABASE=mongodb
- MONGO_INITDB_ROOT_USERNAME=${MONGO_INITDB_ROOT_USERNAME}
- MONGO_INITDB_ROOT_PASSWORD=${MONGO_INITDB_ROOT_PASSWORD}
volumes:
- ./mongo_db:/data/db
- ./backend/mongo-init.js:/docker-entrypoint-initdb.d/init-db.js:ro
ports:
- 27018:27017
Backend — Flask
- Authenticated ( handling twitter sigend-in）
- MongoDB (pymongo)
- Form (Flask-WTF)
本專案需要紀錄登入者 id，這裡用 twitter 當作第三方社群媒體。原本使用 Flask-Dance 套件，它結合 flask blueprint 的路徑導向，可以直接轉址至各方社群媒體的認證頁，而且也提供 SQLAlchemy 將取得的 token 存入 db。可惜呀，現在該套件已經不支援 twitter ，所以參考此文章寫一支轉址的 api。
1. Twitter develop console：在 twitter develop console 建立一個專案，並設定 callback 路徑，也就是認證成功後要 twiiter 導向的地方，這裡是用當地開發端的首頁。
2. 複製 API key & API token
3. 建立 twitter client
oauth = Client(os.getenv('TWITTER_API_KEY'), client_secret=os.getenv('TWITTER_API_SECRET'))4. 新增登入api
def twitter_login():
# 利用上面的 client 來取得 request token
uri, headers, body = oauth.sign('https://twitter.com/oauth/request_token')
res = requests.get(uri, headers=headers, data=body)
# 若 app 認證成功就可以將擷取 access token
res_split = res.text.split('&')
oauth_token = res_split[0].split('=')[1]
# 並用 token 來轉址到使用者登入頁面
return redirect('https://api.twitter.com/oauth/authenticate?oauth_token=' + oauth_token, 302)5. 將 twitter 取得的 token 當作該網站的 user id
Twitter 一但認證使用者成功後，就會在轉址的 url 後方加上 oauth_verifier 的 token，但此時是轉回首頁（也就是 http://127.0.0.1:5000），因此要在載入頁面的 api 加上 session 儲存 token 的程式。
@app.route('/')
def index():
# 非 twitter 用戶的登入
if session.get('user_id') is not None:
session_user_id = json.loads(session.get('user_id'))
if isinstance(session_user_id, str):
g.user_id = session_user_id
elif isinstance(session_user_id, dict):
g.user_id = session_user_id['$oid']
# twitter 用戶的登入
else:
oauth_verifier = request.args.get('oauth_verifier')
if oauth_verifier is not None:
# 將拿到的 oauth_verifier 存在 session 中
session['user_id'] = json_util.dumps(oauth_verifier)
g.user_id = oauth_verifier
return render_template("home.html")當然，如果要再加上非第三方社群軟體的登入管道，可以使用下一段所介紹的 mongoDB 來當作例子。
pymongo
使用 python 來操作 MongoDB 的方法有很多，這裡使用 pymongo 作為連接 flask 和 mongoDB 的橋樑。首先要在 docker 建立一個映像檔為 mongo 的 container 並定義其相關設定的檔案位置，像是 mongo-init.js。
- Dockerfile
...
db:
container_name: mongodb
image: mongo:latest
environment:
- MONGO_INITDB_DATABASE=mongodb
- MONGO_INITDB_ROOT_USERNAME=${MONGO_INITDB_ROOT_USERNAME}
- MONGO_INITDB_ROOT_PASSWORD=${MONGO_INITDB_ROOT_PASSWORD}
volumes:
- ./mongo_db:/data/db
- ./backend/mongo-init.js:/docker-entrypoint-initdb.d/init-db.js:ro
ports:
- 27018:270172. mongo-init.js：若是有什麼初始的資料像是使用者名單，可以在這裡先存入。
# mongo-init.js
// pre-populate the data at the time of the creation of MongoDB
db = db.getSiblingDB("mongodb");
db.scrapy_tb.drop();
db.scrapy_tb.insertMany([
{
"url": "test_url",
"detail": "test_detail",
"website_name": "test_name"
},
{
"url": "test_url_2",
"detail": "test_detail_2",
"website_name": "test_name_2"
},
]);3. Flask 設定：flask 呼叫 mongoDB 不必像其他 SQL DB 需要在啟動 flask app 時就連線，他可以在需要 DB query 時呼叫 mongo client 就好。為了統一管理，在 db.py 檔建立 pymongo 的 mongo client 並定義是要抓取哪一個 database。
# db.py
import os
from pymongo import MongoClient
db_client = MongoClient(
'db', 27017,
username=os.environ.get("MONGO_INITDB_ROOT_USERNAME"),
password=os.environ.get("MONGO_INITDB_ROOT_PASSWORD")
)
scrapy_db = db_client['scrapy']4. MongoDB query：這裡用使用者註冊為例子，若使用者不想用 twitter 登入也可以用 username & password 註冊新使用者。
from db import scrapy_db
bp = Blueprint('auth', __name__, url_prefix='/auth')
@bp.route('/register', methods=('GET', 'POST'))
def register():
if request.method == 'POST':
username = request.form['username']
password = request.form['password']
error = None
if not username:
error = 'Username is required.'
elif not password:
error = 'Password is required.'
if error is None:
user_db = scrapy_db['user_tb']
# 檢查是否有一樣的使用者名稱
result = user_db.find_one({"user_name":username})
if result is not None:
error = 'Username is already existed'
else:
try:
user_db.insert_one(
{
'user_name': username,
'password': generate_password_hash(password)
}
)
except:
error = 'MongoDB error'
else:
return redirect(url_for("auth.login"))
flash(error)
return render_template('auth/register.html')在 register API 中，透過前端 template 的表格 (form) 來取得使用者名稱與密碼。記錄使用者的表格 (collection) 為 user_tb，利用 find_one 來看是否有存在的使用者，若沒有就可以插入這筆新資料。
另外，這裡簡單記錄 docker 環境中要如何使用 mongodb shell。如果想要直接進到 mongoDB 查看資料，可以用以下指令進入。
docker-compose exec db mongosh -u "root" -p "pass"MongoDB shell 常用指令
- show dbs : 列出目前存在的 database，預設的 database 有：admin, config 及 local
- use <db_name> : 指定 database，若不存在則創建新 database
- show collections 列出所有 collections
- db.getCollectionNames() : 列出目前 database 裡的 collection
- db.getName() : 顯示當前位置
- db.form_tb.find() 等於在 form_tb table 下 SELECT*FROMForm (Flask-WTF)
除了使用者註冊及登入有使用到 form，該專案也有新增一個液面讓使用者能夠自由紀錄網站路徑，這邊就是使用 Flask-WTF 來做欄位控制。
首先，可以定義一個 FlaskForm，當然，可以依照需求來新增欄位。
class ScrapeForm(FlaskForm):
scrape_url = StringField('scrape_url', validators=[DataRequired(), url_check])在 validators 的欄位加入檢查的變數，也可以自己定義檢查的 function，例入下面是規定只能輸入包含 bookwalker 字樣的 url。
def url_check(form, field):
if 'bookwalker' not in field.data:
raise ValidationError('Field must be bookwalker url')再來，產生 form 頁面的 API。透過 session 取得目前使用者的 user_id，再根據 POST 或 GET method 決定要做 mongoDB 的哪一步驟。若是 POST，就會進行 form 檢查以及檢查該使用者是否有存過相同的 url。若是 GET，就會進行 query 該使用者所有的 url 記錄，並吐給前端頁面呈現。
@fm.route('/submit', methods=('GET', 'POST'))
def submit():
form = ScrapeForm()
error = None
user_id = session.get('user_id')
# 先檢查是否有使用者 id
if user_id is None:
error = 'Please login first'
flash(error)
return redirect('/form/submit')
if request.method == 'POST':
if form.validate_on_submit():
scrape_url = request.form['scrape_url']
form_tb = scrapy_db['form_tb']
result = form_tb.find_one({"user_id":user_id, 'scrape_url':scrape_url})
if result is not None:
error = 'Record is already existed'
else:
# 在 form_tb collection 中存入該筆資料
try:
form_tb.insert_one({'user_id': user_id,'scrape_url': scrape_url})
except:
error = 'MongoDB error'
flash(error)
return redirect('/form/submit')
form_tb = scrapy_db['form_tb']
# 取得該使用者所有資料
stored_data = form_tb.find({'user_id':user_id})
json_data = loads(dumps(stored_data))
render_list = []
if len(json_data):
render_list = [element['scrape_url'] for element in json_data]
return render_template('form.html', form=form, render_list=render_list)最後是前端頁面（form.html），利用 form 來產生上述表格，也利用 for loop 來呈現該使用者的過往儲存記錄。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<p>Store your favorite comics from bookwalker</p>
<form method="POST">
{{ form.csrf_token }}
{{ form.scrape_url.label }} {{ form.scrape_url(size=50) }}
<input type="submit" value="Go">
</form>
{% if form.scrape_url.errors %}
<ul class="errors">
{% for error in form.scrape_url.errors %}
<li>{{ error }}</li>
{% endfor %}
</ul>
{% endif %}
<h3>History</h3>
{% if render_list|length > 0 %}
{% for element in render_list %}
<li>
<a href="{{ element }}">{{ element }}</a>
</li>
{% endfor %}
{% endif %}
</body>
</html>Backend — Scrapy
在 backend 資料夾中，所有 scrapy 相關的程式都會放在 realscrapy 資料夾中。從頭初始的步驟如下：
- pip install Scrapy
- scrapy startproject realscrapy ：初始 scrapy 的資料夾，隨意取名 (realscrapy)。注意！此時會產生一個資料夾稱為 realscrapy 其中又含有一個資料夾也稱為 realscrapy ，不要隨意更改資料夾名稱。
- scrapy genspider example https://www.bookwalker.com.tw/ ：會在 spiders 資料夾建立名為 bookwalker 的 spider，主要是要爬 https://www.bookwalker.com.tw/ 這個網站。接下來就可以定義這隻 spider 各個爬蟲設定。這裡會自動建立一個 spider class (subclass of CrawlSpider which is based on a pre-defined template crawl)。
因為我只是要簡單抓取書籍名稱及封面圖，所以欄位並不多(BookWalkerClass)。
class BookWalkerClass(scrapy.Item):
url = scrapy.Field()
title = scrapy.Field()
img_src = scrapy.Field()接下來就是處理爬蟲資料的位置以及如何將資料帶回。這裡有多做一個 is_next_page 的判別，來控制是否要自動去爬下一頁的資料。
# spiders/bookwalker.py
class BookwalkerSpider(scrapy.Spider):
name = "bookwalker"
allowed_domains = ["www.bookwalker.com.tw"]
start_urls = []
def __init__(self, url_to_parse, is_next_page=False):
self.start_urls.append(url_to_parse)
self.is_next_page = is_next_page
def start_requests(self):
for url in self.start_urls:
yield scrapy.Request(url=url, callback=self.parse)
def parse(self, response):
# 定義爬取欄位的位置
title = response.xpath('//h4[@class="bookname"]/text()').extract()
img_src = response.xpath('//div[@class="bwbookitem"]//img/@data-src').extract()
data = zip(title,img_src)
# bookwalker_tb = scrapy_db['bookwalker_tb']
for value in data:
yield BookWalkerClass(
url = response.url,
title = value[0],
img_src = value[1]
)
# 若是 is_next_page 為真，就會去抓下一頁的 url 並進行爬蟲
if self.is_next_page == True:
next_page_url = response.xpath('//ul[@class="bw_pagination text-center"]//li/a/@href').get()
yield scrapy.Request(url=next_page_url, callback=self.parse)
簡單說明爬蟲位置的規則，詳細請見說明。
//h4[@class=”bookname”]/text() 代表 class 名為 bookname 的 h4 文字。
//div[@class=”bwbookitem”]//img/@data-src 代表 class 名為 bwbookitem 的 div tag，其底層的 img tag，抓取該 img tag 裡的 data-src 變數。
到這裡為止都還是建立 scrapy spider，接下來才是如何在 flask 中呼叫這隻 spider 工作。在開始之前，需要先搞清楚 python 中 yield & return 的差別，這關係到爬蟲在抓取資料時如何把一比一比的資料傳回到 flask api。簡單來說，yield 是運算時一邊運行一邊把結果傳出來，但 return 是把所有運算做完後才把所有的結果傳出來。在我們這邊的例子，是需要用 yield 把每一個爬到的執傳回到 flask api，因此在上面的 code 中可以看到，parse function 中最後是利用 yield 來返回每一個 BookWalkerClass。
另外，還要再弄清楚 scrapy 和 WSGI 的關係。在這個 repo 有談到，scrapy 之所以無法直接在 Flask 這類用 WSGI 伺服器的框架中運行，是因為 scrapy 的底層有個 Twisted 引擊，此引擊並不像 Flask 使用的 gunicorn 可以重啟，他一旦暫停就無法重啟。因此，我們需要使用 crochet 套件讓 scrapy 能運行於非同步是的框架。
# snstwitter/bookwalker.py
import crochet
from scrapy import signals
from scrapy.crawler import CrawlerRunner
from scrapy.signalmanager import dispatcher
crochet.setup()
crawl_runner = CrawlerRunner()
class BookWalkerScraper:
output_data = []
@crochet.run_in_reactor
def scrape_with_crochet(self, baseURL, is_next_page):
# This will connect to the dispatcher that will kind of loop the code between these two functions.
dispatcher.connect(self._crawler_result, signal=signals.item_scraped)
# 連接到上面 scrapy.py 的 BookwalkerSpider，並且在每個值 yield 後
# 傳遞給crawler_result函數
eventual = crawl_runner.crawl(BookwalkerSpider, url_to_parse=baseURL, is_next_page = is_next_page)
return self.output_data
# 將爬蟲傳回來的每一比資料記下來
def _crawler_result(self, signal, sender, item, response, spider):
self.output_data.append(dict(item))以上 BookWalkerScraper 主要是調度 Bookwalker 爬蟲，讓他可以持續地把資料傳過來。這時就可以寫一隻 API，將爬到的資料呈現在前端。下面的範例還使用了 Flask-WTF 的功能，將爬到的資料當作使用者能勾選的欄位。
# snstwitter/bookwalker.py
from flask_wtf import Form, FlaskForm
from wtforms import BooleanField, widgets, SelectMultipleField
import time, json, sys
@sc.route("/bookwalker", methods=['post','get'])
def scrape():
is_next_page = False
# 呼叫上述的 BookWalkerScraper
scraper_int = BookWalkerScraper()
# 定義要抓取的頁面及是否要抓取下一頁
scraper_int.scrape_with_crochet(
baseURL='https://www.bookwalker.com.tw/search?m=2&s=23&restricted=2&series_display=1&order=sell_desc&page=1',
is_next_page=is_next_page
)
# 這裡會讓該 API 休息 10 秒等待爬蟲爬完資料
# 重要！根據資料量的多寡決定要休息多久，若時間不夠，爬蟲還沒爬完 API 就會先 reponse 空資料
time.sleep(10)
result_list = json.dumps(scraper_int.output_data)
groups_list=[(i['title'], i['title']) for i in json.loads(result_list)]
# 將爬蟲的資料當作 choice 選項
class SimpleForm(FlaskForm):
example = MultiCheckboxField('Label', choices=groups_list)
form = SimpleForm()
if request.method == 'POST':
# do your logic with the submitted form data here
return redirect('/')
return render_template('bookwalker.html', form=form)做到這裡，大部分想要練習的功能都完成了。若真的要使用爬蟲，不建議按照我目前的架構，因為每一次載入頁面就爬蟲的方式太吃效能，若是每幾秒就變動的資料網站還說得過去，若是要爬一個資料來源較固定的網站，建議用以下架構。
另外開一個 container 專心做爬蟲，把爬蟲回來的資料存入 DB 供後端直接 query。這裡提供以 scrapy 為映像檔的 dockerfile：
scrapy:
build:
context: .
dockerfile: scrapy/Dockerfile
container_name: scrapy-container
image: localhost/scrapy-image
volumes:
- ./scrapy:/scrapy-code
ports:
- 8000:8000附上該專案的 github，有任何問題或是說明有霧的地方不吝指教，感謝。