1. python2 使用MySQLdb
# coding:utf-8
import sys
import MySQLdb
from functools import reduce
import datetime
config = {
'host': '**.100.**.1**',
'port': 3306,
'user': 'root',
'passwd': '123456',
'db': 'bot',
'charset': 'utf8'
}
conn = MySQLdb.connect(**config)
cursor = conn.cursor()
cursor.execute('select uuid from tb where state = 0')
result_values = cursor.fetchall()
try:
for line in result_values:
uuid = line[0]
# 查詢機器人反饋的話術
sql_robot_words = "select b.robot_words,b.corpus_score from tb_mm_record a LEFT JOIN mm_voice b " \
"on a.voice_id = b.id where uuid = '%s'" % uuid
cursor.execute(sql_robot_words)
values_robot_words = cursor.fetchall()
b = list(word[1] for word in values_robot_words if word[0] != None)
if len(b):
customer_score = reduce(lambda x, y: x + y, b)
else:
customer_score = len(b)
state = 1
sql_update = "UPDATE mm_cdr SET customer_score = '%d',state = '%s'" % (customer_score, state) + "WHERE uuid = '%s'" % uuid
cursor.execute(sql_update)
conn.commit()
except:
# Rollback in case there is any error
conn.rollback()
# 關閉數(shù)據(jù)庫連接
cursor.close()
conn.close()
2. python2 使用 mySQL
3. python3 使用pymysql
3.1 首先測試數(shù)據(jù)庫的連接test.py
# coding:utf-8
import pymysql
def connectdb():
print('連接到mysql服務器...')
db = pymysql.connect(
host="localhost",
user="op",
passwd="123456",
port=3306,
db="voic",
charset='utf8',
cursorclass=pymysql.cursors.DictCursor)
print('連接上了!')
return db
if __name__ == '__main__':
connectdb()
輸出：
連接到mysql服務器...
連接上了!
3.2 配置數(shù)據(jù)庫信息
配置文件mysql_configurate
{"passwd": "123456", "charset": "utf8mb4", "port": 3306, "host": "27.140.215.133", "user": "op", "db": "voic"}
py 文件 mysql_configurate.py
# coding:utf-8
from functools import reduce
import datetime
import pymysql
from threading import Timer
def callValueScore():
print('開始吧')
f_read = open('.\mysql_configurate', 'r')
data = f_read.readline()
print(data, type(data))
d = eval(data)
config = {
'host': d['host'],
'port': d['port'],
'user': d['user'],
'passwd': d['passwd'],
'db': d['db'],
'charset': d['charset']
}
conn = pymysql.connect(**config)
cursor = conn.cursor()
cursor.execute('select uuid from tb_cdr where state = 1')
result_values = cursor.fetchall()
print(len(result_values))
try:
for line in result_values:
break
uuid = line[0]
# 查詢機器人反饋的話術
sql_robot_words = "select b.robot_words,b.corpus_score from tb_reply_record a LEFT JOIN tb_voice b " \
"on a.voice_id = b.id where uuid = '%s'" % uuid
cursor.execute(sql_robot_words)
values_robot_words = cursor.fetchall()
b = list(word[1] for word in values_robot_words if word[0] != None)
if len(b):
customer_score = reduce(lambda x, y: x + y, b)
else:
customer_score = len(b)
print(customer_score)
state = 1
# sql_update = "UPDATE tb_cdr SET customer_score = '%d',state = '%s'" % (customer_score, state) + "WHERE uuid = '%s'" % uuid
# cursor.execute(sql_update)
conn.commit()
except:
# Rollback in case there is any error
conn.rollback()
# 關閉數(shù)據(jù)庫連接
cursor.close()
conn.close()
3.3 寫入數(shù)據(jù)庫信息
# coding:utf-8
import json
def testConfigurate():
data = {
'host': '27.110.215.127',
'port': 3306,
'user': 'ops',
'passwd': '123456',
'db': 'voic',
'charset': 'utf8mb4'
}
f = open('.\mysql_configurate', 'w')
data_str = json.dumps(data)
print(data_str)
print(type(data_str))
d = json.loads(data_str)
print(d['db'],type(d['db']))
print(d['port'], type(d['port']))
f.write(data_str)
參考鏈接：
python 更新插入mySQL數(shù)據(jù)庫
安裝
Python連接MySQL數(shù)據(jù)庫
python下的MySQLdb使用
