实验三 Python 常用类库与数据库访问

实验三 Python 常用类库与数据库访问

一、实验基本原理

运用 Anaconda 搭建的 Jupyter notebook 平台编写实例 Python 程序。

二、实验目的：

1、熟悉在 Jupyter notebook 平台下与 MySQL 数据库建立联系。

2、通过实际案例编程，掌握 Jupyter notebook 平台下调用 MySQL 的基础语法。

3、熟悉 panadas、pymysql 等库的使用方法。

三、具体要求：

1、使用 pymysql 与 MySQL 建立连接。

2、将数据写入和读出 MySQL 数据库中。

3、用 Pandas 读取 MYSQL 数据库中的数据。

四、实验环境：

1、Windows 10 电脑一台。

2、Anaconda、Python、Jupyter notebook 平台、MySQL 数据库。

五、 实验内容：

准备工作：安装 MySQL 数据库。

MySQL的安装流程:

①打开电脑路径：”C:\\ProgramData”                      ——-路径

②将U盘内压缩包“mysql-8.0.30-winx64.zip”拷贝到路径下，右键‘解压文件…’，得到“mysql-8.0.30-winx64“文件夹，同时删除“mysql-8.0.30-winx64.zip”；

注意：

如果此时解压出来的“mysql-8.0.30-winx64“文件夹下还有同名子文件夹，注意删除一层“mysql-8.0.30-winx64“文件夹，保证路径中只出现一次“mysql-8.0.30-winx64“

③鼠标右击桌面左下角Windows图标，弹出菜单栏中选择“命令提示符(管理员)”或“Windows PowerShell(管理员)”，进入系统命令窗口；

④命令窗口中输入

“ cd C:\\ProgramData\\mysql-8.0.30-winx64\\bin ”

(即cd到“mysql-8.0.30-winx64”文件夹中bin子文件夹的路径)

   继续输入：

        “ .\\mysqld –initialize –console ”   //注意是两个前缀符都是双横杠

        注意输出内容的最后字符串（随机密码）

   继续输入：

        “ .\\mysqld -install ”  //安装MySQL

   继续输入：

        “ net start mysql ”   //启动MySQL

   继续输入：

       “ .\\mysql -u root -p ”  //登录MySQL，手动输入上面的随机密码，不支持复制

   MySQL内输入：

“ alter user'root'@'localhost'identified by '123456'; ”//更新密码

   MySQL内输入：

       “ commit; ”

   MySQL内输入：

       “ create DATABASE test01; ”  //创建数据库，

如果显示“Query OK, 1 row affected (*** sec)”即安装成

安装完成后，启动步骤如下：

（1）打开 Windows PowerShell,在命令窗口输入

“ cd C:\\ProgramData\\mysql-8.0.30-winx64\\bin ”

(即 cd 到“mysql-8.0.30-winx64”文件夹中 bin 子文件夹的路径)

（2）继续输入

“ net start mysql ” //启动 MySQL

（3）继续输入

“ .\\mysql -u root -p ” //登录 MySQ

（4）输入密码(默认密码为 123456)，输入密码后输入

”commit;”

（5）继续输入

“ create DATABASE test01; ” //创建数据库

其中 test01 为所创建数据库名称，可以按照实验要求更改。

        至此，MySQL 成功开启并成功创建所需 DATABASE，接下来的实验在 jupyter 中进行。

(实验过程中请勿关闭 Windows PowerShell 以及 Anaconda 命令窗口)

实例 1：创建名单数据库

        将学生名单存入数据库中。将实验二产生的 Excel 文档“student_ XYY.xlsx”中点名用 的数据存入 MySQL 数据库。数据库的名字命名规则：student_XYY，X 是你的 1 位班号，YY 是你的两位学号，以后同此。提示：先在 MySQL 中创建此数据库。

要求：

（1）使用 pymysql 建立数据库连接，并创建名为 student_XYY 的数据库。

（2）使用 SQL 语句创建 mingdan 表。

（3）通过字典输出 student_XYY 中学生的信息。

（4）将学生信息插入 MySQL 所创建好的数据库中。

以下为步骤提示。

#在建好的数据库 student_ XYY 中建立表 mingdan

import pymysql

db = pymysql.connect(host='localhost'

user='root'

password='123456'

database='student_XYY') #建立数据库连接

#使用 cursor()方法创建一个游标对象 cursor

#使用 SQL 语句 CREATE TABLE mingdan 创建表

#使用 execute()方法执行 SQL，创建表

#建立工作簿

from openpyxl import load_workbook

#用字典读出数据。

#将字典里的数据使用 INSERT 语句插入到所建的数据库。

#实例 1：创建名单数据库
import pymysql
from openpyxl import load_workbook
import os

# 数据库配置
DB_HOST = 'localhost'
DB_USER = 'root'
DB_PASS = '123456'
DB_NAME = 'student_553'

FILE_PATH = r"E:\\人工智能与大数据\\实验\\shiyan2\\student_101.xlsx"

def instance1_import_data():
print(f"导入数据到 {DB_NAME}")

# 检查文件
if not os.path.exists(FILE_PATH):
print(f"错误：找不到文件，请检查路径：\\n{FILE_PATH}")
return

# 数据库初始化 (如果不存在则创建)
try:
conn_init = pymysql.connect(host=DB_HOST, user=DB_USER, password=DB_PASS)
cursor_init = conn_init.cursor()
cursor_init.execute(f"CREATE DATABASE IF NOT EXISTS {DB_NAME} DEFAULT CHARSET=utf8mb4")
conn_init.close()
except Exception as e:
print(f"数据库初始化失败: {e}")
return

# 连接数据库并操作
try:
db = pymysql.connect(host=DB_HOST, user=DB_USER, password=DB_PASS, database=DB_NAME)
cursor = db.cursor()

# 创建表
sql_create_table = """
CREATE TABLE IF NOT EXISTS mingdan (
sno VARCHAR(50) NOT NULL,
sname VARCHAR(50) NOT NULL,
PRIMARY KEY (sno)
)
"""
cursor.execute(sql_create_table)

# 读取Excel并插入数据
wb = load_workbook(FILE_PATH)
sheet = wb.active

sql_insert = "INSERT INTO mingdan (sno, sname) VALUES (%s, %s)"
count = 0

for row in sheet.iter_rows(min_row=2, values_only=True):
if row[0] is None: continue
s_no = str(row[0]).strip()
s_name = str(row[1]).strip()

try:
cursor.execute(sql_insert, (s_no, s_name))
count += 1
except pymysql.err.IntegrityError:
pass # 跳过重复
except Exception:
pass # 跳过错误

db.commit()
print(f"数据导入完成。")

print(f"\\n输出结果")

# 查询所有数据
cursor.execute("SELECT * FROM mingdan")
results = cursor.fetchall()

# 构建字典 {学号: 姓名}
student_dict = {}
for row in results:
# row[0] 是学号, row[1] 是姓名
student_dict[row[0]] = row[1]

# 打印字典
print(student_dict)

except Exception as e:
print(f"运行出错: {e}")
finally:
if 'cursor' in locals(): cursor.close()
if 'db' in locals(): db.close()

if __name__ == "__main__":
instance1_import_data()

实例 2：读出名单数据库

        将上题创建的数据库名单读出。打开上题生成的含学生学号及姓名的数据库 student_XYY，并将学号和姓名读取到字典里。

要求：

（1）使用 pymysql 建立与 MYSQL 数据库的连接。

（2）通过字典输出 student_XYY 中学生的信息。

#实例 2：读出名单数据库
import pymysql

def instance2_read_data():
print(f"从 {DB_NAME} 读取名单")

try:
# 连接数据库 student_553
db = pymysql.connect(host='localhost', user='root', password='123456', database='student_553')
cursor = db.cursor()

# 查询数据
sql = "SELECT sno, sname FROM mingdan"
cursor.execute(sql)
results = cursor.fetchall()

# 转换为字典并输出
student_dict = {}
for row in results:
student_dict[row[0]] = row[1] # 学号为Key，姓名为Value

print("学生信息字典:")
print(student_dict)

except Exception as e:
print(f"读取失败: {e}")
finally:
if 'cursor' in locals(): cursor.close()
if 'db' in locals(): db.close()

if __name__ == "__main__":
instance2_read_data()

实例 3：成绩录入和读出

        四名同学的语文、数学、英语、计算机成绩如下所示，要求在 MYSQL 数据库中创建成绩表， 存入姓名和成绩。并从数据库中读取和输出姓名和语文成绩。

要求：用 MYSQL 数据库进行成绩处理。

#实例 3：成绩录入和读出
import pymysql
import pandas as pd

# 数据库配置
DB_HOST = 'localhost'
DB_USER = 'root'
DB_PASS = '123456'
DB_NAME = 'student_553'

def instance3_scores_pandas():
print(f"成绩录入与读出")

conn = pymysql.connect(host=DB_HOST, user=DB_USER, password=DB_PASS, database=DB_NAME)

try:
cursor = conn.cursor()
cursor.execute("DROP TABLE IF EXISTS scores")
cursor.execute("CREATE TABLE scores (name VARCHAR(20), chinese FLOAT, math FLOAT, english FLOAT, computer FLOAT)")
data = [("张三", 88, 90, 98, 95), ("李四", 85, 92, 95, 98), ("王五", 89, 89, 90, 92), ("丁六", 82, 86, 89, 90)]
cursor.executemany("INSERT INTO scores VALUES (%s, %s, %s, %s, %s)", data)
conn.commit()
cursor.close()

sql_query = "SELECT name AS '姓名', chinese AS '语文成绩' FROM scores"

df = pd.read_sql(sql_query, conn)

print("\\n[学生成绩单]")
print(df.to_string(index=False, justify='center'))

finally:
conn.close()

if __name__ == "__main__":
instance3_scores_pandas()

六：思考题

思考 1：成绩处理 (panadas 库)

用 Pandas 库进行成绩处理。用 Pandas 将上面实例 3 生成的 MySQL 数据库成绩表里的数据

（姓名和成绩）读出，计算每名学生的平均成绩和最高分及每门课程的最高分，再将这些平均

分、最高分存入数据库的表中，最后将数据库的成绩表中所有数据显示输出。结果如下所示:要求：

（1）使用 pymysql 建立于 MYSQL 数据库的连接。

（2）用 Pandas 读出 MYSQL 数据库中的数据到字典中。

（3）将字典里的数据使用 INSERT 语句插入到所建的数据库。

#思考题：Pandas 成绩处理
import pandas as pd
import pymysql

def get_display_len(s):
"""计算字符串显示宽度 (中文=2, 英文/数字=1)"""
length = 0
for char in str(s):
if '\\u4e00' <= char <= '\\u9fa5': # 判断是否为汉字
length += 2
else:
length += 1
return length

def print_clean_table(df, title):
"""打印无边框、对齐的表格"""
print(f"\\n[{title}]")

# 计算每一列需要的最大宽度
col_widths = []
for col in df.columns:
# 获取表头的宽度
max_w = get_display_len(col)
# 获取该列所有数据的最大宽度
for val in df[col]:
# 格式化浮点数
if isinstance(val, float):
val_str = f"{val:.2f}"
else:
val_str = str(val)
max_w = max(max_w, get_display_len(val_str))
col_widths.append(max_w + 4)

# 打印表头 (居中对齐，无边框)
header_row = ""
for col, w in zip(df.columns, col_widths):
s = str(col)
real_len = get_display_len(s)
padding = w – real_len
pad_l = padding // 2
pad_r = padding – pad_l
header_row += " " * pad_l + s + " " * pad_r
print(header_row)

# 打印数据行 (居中对齐，无边框)
for _, row in df.iterrows():
data_row = ""
for val, w in zip(row, col_widths):
# 处理数据格式
if isinstance(val, float):
s = f"{val:.2f}"
else:
s = str(val)

real_len = get_display_len(s)
padding = w – real_len
pad_l = padding // 2
pad_r = padding – pad_l
data_row += " " * pad_l + s + " " * pad_r
print(data_row)

# 数据库配置
conn = pymysql.connect(host='localhost', user='root', password='123456', database='student_553')

try:
# 读取数据
sql = "SELECT * FROM scores"
df = pd.read_sql(sql, conn)

df_display_raw = df.copy()
df_display_raw.columns = ['姓名', '语文', '数学', '英语', '计算机']

# — 输出原始数据 —
print_clean_table(df_display_raw, "原始数据")

# 数据处理：计算平均分和最高分
df['平均分'] = df[['chinese', 'math', 'english', 'computer']].mean(axis=1)
df['最高分'] = df[['chinese', 'math', 'english', 'computer']].max(axis=1)

# 计算汇总行 (每列的最高分)
col_max = df[['chinese', 'math', 'english', 'computer', '平均分', '最高分']].max(axis=0)
max_row = pd.DataFrame(col_max).T
max_row['name'] = '最高分'

# 拼接与排序
df_final = pd.concat([df, max_row], ignore_index=True)
cols = ['name', 'chinese', 'math', 'english', 'computer', '平均分', '最高分']
df_final = df_final[cols]

# 输出最终表格
df_display_final = df_final.copy()
df_display_final.columns = ['姓名', '语文', '数学', '英语', '计算机', '平均分', '最高分']

print_clean_table(df_display_final, "处理后的最终表格")

# 写回数据库
# 为了不破坏原表，创建一个新表 scores_final
cursor = conn.cursor()
cursor.execute("DROP TABLE IF EXISTS scores_final")
create_final_sql = """
CREATE TABLE scores_final (
name VARCHAR(20),
chinese FLOAT,
math FLOAT,
english FLOAT,
computer FLOAT,
avg_score FLOAT,
max_score FLOAT
)
"""
cursor.execute(create_final_sql)

insert_sql = """
INSERT INTO scores_final (name, chinese, math, english, computer, avg_score, max_score)
VALUES (%s, %s, %s, %s, %s, %s, %s)
"""

for index, row in df_final.iterrows():
cursor.execute(insert_sql, (
row['name'],
row['chinese'],
row['math'],
row['english'],
row['computer'],
row['平均分'],
row['最高分']
))

conn.commit()
print("\\n统计结果已成功存入数据库表 'scores_final'。")

cursor.close()

finally:
conn.close()