56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
import pymysql
|
|
from db.connection import MySQLPool
|
|
import bcrypt
|
|
|
|
class DatabaseManager:
|
|
def __init__(self):
|
|
# 使用MySQLPool初始化数据库连接池
|
|
self.pool = MySQLPool()
|
|
|
|
def fetch(self, query, params=None):
|
|
conn = self.pool.get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(query, params or ())
|
|
result = cursor.fetchall()
|
|
return result
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
def execute(self, query, params=None):
|
|
conn = self.pool.get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(query, params or ())
|
|
conn.commit()
|
|
return cursor.rowcount
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
def user_exists(self, phone_number):
|
|
sql = "SELECT 1 FROM user WHERE phone_number=%s LIMIT 1"
|
|
result = self.fetch(sql, (phone_number,))
|
|
return len(result) > 0
|
|
|
|
def insert_user(self, user):
|
|
sql = """
|
|
INSERT INTO user (nickname, phone_number, password, identity, is_active)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
"""
|
|
data = (user.nickname, user.phone_number, user.password, user.identity, user.is_active)
|
|
# print(data)
|
|
return self.execute(sql, data)
|
|
|
|
def valid_login(self, phone_number, password_attempt):
|
|
# SQL查询获取用户的哈希密码
|
|
sql = "SELECT password FROM user WHERE phone_number=%s LIMIT 1"
|
|
result = self.fetch(sql, (phone_number,))
|
|
if result:
|
|
stored_hash = result[0]['password'] # 假设结果是密码字段
|
|
# 使用bcrypt进行密码验证
|
|
if bcrypt.checkpw(password_attempt.encode('utf-8'), stored_hash.encode('utf-8')):
|
|
return True # 密码匹配，登录成功
|
|
return False # 密码不匹配或用户不存在，登录失败
|