python使用adbapi實現(xiàn)MySQL數(shù)據(jù)庫的異步存儲

# -*- coding: utf-8 -*-
import io
import sys
import scrapy
import re
import datetime
from scrapy.http import Request
from urllib import parse
from ArticleSpider.items import JobboleArticleItem, ArticleItemLoader
from scrapy.loader import ItemLoader
sys.stdout = io.TextIOWrapper(sys.stdout.buffer,encoding='utf-8')
 
class JobboleSpider(scrapy.Spider):
 """docstring for JobboleSpider"""
 name = "jobbole"
 allowed_domain = ["blog.jobbole.com"]
 start_urls = ['http://blog.jobbole.com/all-posts/']
 
 def parse(self, response):
 """
 1.獲取列表頁中的文章url
 """
 # 解析列表匯中所有文章url并交給scrapy下載器并進行解析
 post_nodes = response.css("#archive .floated-thumb .post-thumb a")
 for post_node in post_nodes:
 image_url = post_node.css("img::attr(src)").extract_first("")# 這里取出每篇文章的封面圖，并作為meta傳入Request
 post_url = post_node.css("::attr(href)").extract_first("")
 yield Request(url = parse.urljoin(response.url, post_url), meta = {"front_image_url":image_url}, callback = self.parse_detail)
 
 def parse_detail(self, response):
 article_item = JobboleArticleItem()
 # 通過ItemLoader加載Item
 # 通過add_css后的返回值都是list型，所有我們再items.py要進行處理
 item_loader = ArticleItemLoader(item = JobboleArticleItem(), response = response)
 item_loader.add_css("title", ".entry-header h2::text")
 item_loader.add_value("url", response.url)
 # item_loader.add_value("url_object_id", get_md5(response.url))
 item_loader.add_value("url_object_id", response.url)
 item_loader.add_css("create_date", "p.entry-meta-hide-on-mobile::text")
 item_loader.add_value("front_image_url", [front_image_url])
 item_loader.add_css("praise_nums", ".vote-post-up h20::text")
 item_loader.add_css("comment_nums", "a[href='#article-comment'] span::text")
 item_loader.add_css("fav_nums", ".bookmark-btn::text")
 item_loader.add_css("tags", "p.entry-meta-hide-on-mobile a::text")
 item_loader.add_css("content", "div.entry")
 
 article_item = item_loader.load_item()
 print(article_item["tags"])
 
 yield article_item
 pass

# -*- coding: utf-8 -*-
 
# Define here the models for your scraped items
#
# See documentation in:
# https://doc.scrapy.org/en/latest/topics/items.html
import datetime
import re
 
 
import scrapy
from scrapy.loader import ItemLoader
from scrapy.loader.processors import MapCompose, TakeFirst,Join
from ArticleSpider.utils.common import get_md5
 
 
def convert_date(value):
 try:
 create_date = datetime.datetime.strptime(create_date, "%Y/%m/%d").date()
 except Exception as e:
 create_date = datetime.datetime.now().date()
 return create_date
 
def get_nums(value):
 match_re = re.match(".*?(\d+).*", value)
 if match_re:
 nums = int(match_re.group(1))
 else:
 nums = 0
 
 return nums
 
def remove_comment_tags(value):
 # 去掉tags中的評論內(nèi)容
 if "評論" in value:
 # 這里做了修改，如果返回""，則在list中仍然會占位，會變成類似于["程序員",,"解鎖"]這樣
 # return ""
 return None 
 else:
 return value
 
def return_value(value):
 return 
 
class ArticleItemLoader(ItemLoader):
 """docstring for AriticleItemLoader"""
 # 自定義ItemLoader
 default_output_processor = TakeFirst()
 
class ArticlespiderItem(scrapy.Item):
 # define the fields for your item here like:
 # name = scrapy.Field()
 pass
 
class JobboleArticleItem(scrapy.Item):
 """docstring for ArticlespiderItem"""
 title = scrapy.Field()
 create_date = scrapy.Field(
 input_processor = MapCompose(convert_date)
 )
 url = scrapy.Field()
 url_object_id = scrapy.Field(
 output_processor = MapCompose(get_md5)
 )
 # 這里注意front_image_url還是一個list，在進行sql語句時還需要處理
 front_image_url = scrapy.Field(
 output_processor = MapCompose(return_value)
 )
 front_image_path = scrapy.Field()
 praise_nums = scrapy.Field(
 input_processor = MapCompose(get_nums)
 )
 comment_nums = scrapy.Field(
 input_processor = MapCompose(get_nums)
 )
 fav_nums = scrapy.Field(
 input_processor = MapCompose(get_nums)
 )
 # tags要做另行處理，因為tags我們需要的就是list
 tags = scrapy.Field(
 input_processor = MapCompose(remove_comment_tags),
 output_processor = Join(",")
 )
 content = scrapy.Field()

# -*- coding: utf-8 -*-
 
# Define your item pipelines here
#
# Don't forget to add your pipeline to the ITEM_PIPELINES setting
# See: https://doc.scrapy.org/en/latest/topics/item-pipeline.html
import codecs
import json
from twisted.enterprise import adbapi
import MySQLdb
import MySQLdb.cursors
 
 
class MysqlTwistedPipeline(object):
 """docstring for MysqlTwistedPipeline"""
 #采用異步的機制寫入mysql
 def __init__(self, dbpool):
 self.dbpool = dbpool
 
 @classmethod
 def from_settings(cls, settings):
 dbparms = dict(
 host = settings["MYSQL_HOST"],
 db = settings["MYSQL_DBNAME"],
 user = settings["MYSQL_USER"],
 passwd = settings["MYSQL_PASSWORD"],
 charset='utf8',
 cursorclass=MySQLdb.cursors.DictCursor,
 use_unicode=True,
 )
 dbpool = adbapi.ConnectionPool("MySQLdb", **dbparms)
 
 return cls(dbpool)
 
 def process_item(self, item, spider):
 #使用twisted將mysql插入變成異步執(zhí)行
 query = self.dbpool.runInteraction(self.do_insert, item)
 query.addErrback(self.handle_error, item, spider) #處理異常
 return item
 
 def handle_error(self, failure, item, spider):
 # 處理異步插入的異常
 print (failure)
 
 def do_insert(self, cursor, item):
 #執(zhí)行具體的插入
 #根據(jù)不同的item 構(gòu)建不同的sql語句并插入到mysql中
 # insert_sql, params = item.get_insert_sql()
 # print (insert_sql, params)
 # cursor.execute(insert_sql, params)
 insert_sql = """
 insert into jobbole_article(title, url, create_date, fav_nums, url_object_id)
 VALUES (%s, %s, %s, %s, %s)
 """
 # 可以只使用execute，而不需要再使用commit函數(shù)
 cursor.execute(insert_sql, (item["title"], item["url"], item["create_date"], item["fav_nums"], item["url_object_id"]))

MYSQL_HOST = "localhost"
MYSQL_DBNAME = "article_wilson"
MYSQL_USER = "root"
MYSQL_PASSWORD = "root"

ITEM_PIPELINES = {
 # 'ArticleSpider.pipelines.ArticlespiderPipeline': 300,
 # 'ArticleSpider.pipelines.JsonWithEncodingPipeline': 1
 
 # # 'scrapy.pipelines.images.ImagePipeline': 1,
 # 'ArticleSpider.pipelines.JsonExporterPipleline': 1
 # 'ArticleSpider.pipelines.ArticleImagePipeline': 2
 # 'ArticleSpider.pipelines.MysqlPipeline': 1
 'ArticleSpider.pipelines.MysqlTwistedPipeline': 1
}