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

classDBHelper():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

defconnectMysql(self):

conn=pymysql.connect(host=self.host,

port=self.port,

user=self.user,

passwd=self.passwd,

charset='utf8')returnconn#连接数据库

defconnectDatabase(self):

conn=pymysql.connect(host=self.host,

port=self.port,

user=self.user,

passwd=self.passwd,

db=self.db,

charset='utf8')returnconn#创建数据库

defcreateDatabase(self):

conn=self.connectMysql()

sql="create database if not exists"+self.db

cur=conn.cursor()

cur.execute(sql)

cur.close()

conn.close()#创建数据表

defcreateTable(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()#测试数据库操作

classTestDBHelper():def __init__(self):

self.dbHelper=DBHelper()deftestCreateDatebase(self):

self.dbHelper.createDatabase()deftestCreateTable(self):

sql="create table testtable(id int primary key auto_increment,name varchar(50),url varchar(200))"self.dbHelper.createTable(sql)deftestInsert(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)deftestUpdate(self):

sql="update testtable set name=%s,url=%s where id=%s"params=("update","update","1")

self.dbHelper.update(sql,*params)deftestDelete(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() #