JAVA高级面试进阶训练营视频教程

admin

1、简单的在pipelines.py上添加如下代码：

import pymysql

class ScrapyTextMysqlPipeLine(object):
    #__init__函数里面初始化就是连接数据库，便于实现增删改查
    def __init__(self):
        # connection database
        self.connect = pymysql.connect('localhost', 'root', '', 'test')# 后面三个依次是数据库连接名、数据库密码、数据库名称
        # get cursor
        self.cursor = self.connect.cursor()
        print("连接数据库成功")

    def process_item(self, item, spider):
        print("开始输入数据")
        print(item['申报要素'])
        try:
            self.cursor.execute("insert into test(申报要素, hscode, 申报名称, 参考均价, 参考最高价, 参考最低价) values (%s, %s, %s, %s, %s, %s)", (item['申报要素'], item['hscode'], item['申报名称'], item['参考均价'], item['参考最高价'], item['参考最低价']))
            self.connect.commit()
        except Exception as error:
            #print error
            print(error)
        return item

问题：需要删除到爬取数据的第一个元组，解决方法是使用if语句去除第一个元组（其中的美元符号需要除去，其中的一些数据类型如下）如下：

        for i in range(0, len(hscodes)):
            #设置if语句去除不必要的标题类数据，如“申报要素”等
            if i != 0:
                item['申报要素'] = sbElements
                item['hscode'] = hscodes
                item['申报名称'] = sbNames
                item['参考均价'] = ckAvgPrices.lstrip("$")
                item['参考最高价'] = ckMaxPrices.lstrip("$")
                item['参考最低价'] = ckMinPrice.lstrip("$")
                yield item
                print(item)
            else:
                print("已经删除多余值")

 3、保存数据到数据库却不能使得数据重复，解决方法如下：

            yield scrapy.Request(third_page, callback=self.details_parse2, dont_filter=False)

 4、创建一个与spiders的同级的文件夹db，在其下创建DBHelper.py来实现爬取数据在数据库的增删查改，在下面的代码虽然使用了insert语句，但是其他的也应该是一样的，然后在settings.py中配置MySQL数据库的信息，最后只要在pipelines.py中调用即可。

dbHelper.py:

 

import pymysql
from scrapy.utils.project import get_project_settings#引入settings配置

class DBHelper():

    def __init__(self):
        self.settings=get_project_settings()#获取settings配置数据

        self.host=self.settings['MYSQL_HOST']
        self.port=self.settings['MYSQL_PORT']
        self.user=self.settings['MYSQL_USER']
        self.passwd=self.settings['MYSQL_PASSWD']
        self.db=self.settings['MYSQL_DBNAME']
    #连接mysql
    def connectMysql(self):
        conn=pymysql.connect(host=self.host,
                             port=self.port,
                             user=self.user,
                             passwd=self.passwd,
                             charset='utf8')
        return conn
    #连接数据库
    def connectDatabase(self):
        conn=pymysql.connect(host=self.host,
                             port=self.port,
                             user=self.user,
                             passwd=self.passwd,
                             db=self.db,
                             charset='utf8')
        return conn

    #创建数据库
    def createDatabase(self):
        conn=self.connectMysql()

        sql="create database if not exists "+self.db
        cur=conn.cursor()
        cur.execute(sql)
        cur.close()
        conn.close()

    #创建数据表
    def createTable(self,sql):
        conn=self.connectDatabase()

        cur=conn.cursor()
        cur.execute(sql)
        cur.close()
        conn.close()

    #插入数据
    def insert(self,sql,*params):
        conn=self.connectDatabase()

        cur=conn.cursor()
        cur.execute(sql,params)
        conn.commit()
        cur.close()
        conn.close()

    #更新数据
    def update(self,sql,*params):
        conn=self.connectDatabase()

        cur=conn.cursor()
        cur.execute(sql,params)
        conn.commit()
        cur.close()
        conn.close()

    #删除数据
    def delete(self,sql,*params):
        conn=self.connectDatabase()

        cur=conn.cursor()
        cur.execute(sql,params)
        conn.commit()
        cur.close()
        conn.close()


#测试数据库操作
class TestDBHelper():

    def __init__(self):
        self.dbHelper=DBHelper()

    def testCreateDatebase(self):
        self.dbHelper.createDatabase()

    def testCreateTable(self):
        sql="create table testtable(id int primary key auto_increment,name varchar(50),url varchar(200))"
        self.dbHelper.createTable(sql)

    def testInsert(self, item):
        sql="insert into example(hscode, 申报名称, 申报要素, 参考均价, 参考最高价, 参考最低价) values (%s, %s, %s, %s, %s, %s)"
        params=(item['hscode'], item['申报名称'], item['申报名称'], item['参考均价'], item['参考最高价'], item['参考最低价'])
        self.dbHelper.insert(sql,*params)
    def testUpdate(self):
        sql="update testtable set name=%s,url=%s where id=%s"
        params=("update","update","1")
        self.dbHelper.update(sql,*params)

    def testDelete(self):
        sql="delete from testtable where id=%s"
        params=("1")
        self.dbHelper.delete(sql,*params)

if __name__=="__main__":
    testDBHelper=TestDBHelper()
    #testDBHelper.testCreateDatebase()  #
    #testDBHelper.testCreateTable()     #
    testDBHelper.testInsert()          #
    #testDBHelper.testUpdate()          #
    #testDBHelper.testDelete()          #

 settings.py:

#Mysql数据库的配置信息
MYSQL_HOST = '127.0.0.1'
MYSQL_DBNAME = 'test'    #数据库名字
MYSQL_USER = 'root'      #数据库账号
MYSQL_PASSWORD = ''      #数据库密码

MYSQL_PORT = 3306        #数据库端口，在dbHelper中使用

pipelines.py:

class ScrapyTextManyMysqlPipeLine(object):
    def __init__(self):

        from ScrapyTest.db.dbHelper import TestDBHelper
        self.db = TestDBHelper()

    def process_item(self, item, spider):

        self.db.testInsert(item)
        return item