MySQL（11） - Python+MySQL开发新闻管理系统

1.前置知识及条件

1.1.数据库脚本

SET FOREIGN_KEY_CHECKS=0;

-- ----------------------------
-- Table structure for t_news
-- ----------------------------
DROP TABLE IF EXISTS `t_news`;
CREATE TABLE `t_news` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT,
`title` varchar(40) NOT NULL,
`editor_id` int(10) unsigned NOT NULL,
`type_id` int(10) unsigned NOT NULL,
`content_id` char(12) NOT NULL,
`is_top` tinyint(3) unsigned NOT NULL,
`create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
`update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
`state` enum('草稿','待审批','已审批','隐藏') NOT NULL,
PRIMARY KEY (`id`),
KEY `editor_id` (`editor_id`),
KEY `type_id` (`type_id`),
KEY `state` (`state`),
KEY `create_time` (`create_time`),
KEY `is_top` (`is_top`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of t_news
-- ----------------------------
INSERT INTO `t_news` VALUES ('1', '新闻标题1', '2', '1', '1', '1', '2018-11-22 18:55:56', '2018-11-22 18:55:56', '待审批');

-- ----------------------------
-- Table structure for t_role
-- ----------------------------
DROP TABLE IF EXISTS `t_role`;
CREATE TABLE `t_role` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT,
`role` varchar(20) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `role` (`role`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of t_role
-- ----------------------------
INSERT INTO `t_role` VALUES ('2', '新闻编辑');
INSERT INTO `t_role` VALUES ('1', '管理员');

-- ----------------------------
-- Table structure for t_type
-- ----------------------------
DROP TABLE IF EXISTS `t_type`;
CREATE TABLE `t_type` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT,
`type` varchar(20) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `type` (`type`)
) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of t_type
-- ----------------------------
INSERT INTO `t_type` VALUES ('2', '体育');
INSERT INTO `t_type` VALUES ('5', '历史');
INSERT INTO `t_type` VALUES ('4', '娱乐');
INSERT INTO `t_type` VALUES ('3', '科技');
INSERT INTO `t_type` VALUES ('1', '要闻');

-- ----------------------------
-- Table structure for t_user
-- ----------------------------
DROP TABLE IF EXISTS `t_user`;
CREATE TABLE `t_user` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT,
`username` varchar(20) NOT NULL,
`password` varchar(500) NOT NULL,
`email` varchar(100) NOT NULL,
`role_id` int(10) unsigned NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `username` (`username`),
KEY `username_2` (`username`)
) ENGINE=InnoDB AUTO_INCREMENT=14 DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of t_user
-- ----------------------------
INSERT INTO `t_user` VALUES ('1', 'admin', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'admin@163.com', '1');
INSERT INTO `t_user` VALUES ('2', 'scott', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'scott@163.com', '1');
INSERT INTO `t_user` VALUES ('3', 'test_1', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_1@163.com', '2');
INSERT INTO `t_user` VALUES ('4', 'test_2', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_2@163.com', '2');
INSERT INTO `t_user` VALUES ('5', 'test_3', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_3@163.com', '2');
INSERT INTO `t_user` VALUES ('6', 'test_4', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_4@163.com', '2');
INSERT INTO `t_user` VALUES ('7', 'test_5', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_5@163.com', '2');
INSERT INTO `t_user` VALUES ('8', 'test_6', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_6@163.com', '2');
INSERT INTO `t_user` VALUES ('9', 'test_7', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_7@163.com', '2');
INSERT INTO `t_user` VALUES ('10', 'test_8', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_8@163.com', '2');
INSERT INTO `t_user` VALUES ('11', 'test_9', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_9@163.com', '2');
INSERT INTO `t_user` VALUES ('12', 'test_10', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_10@163.com', '2');
INSERT INTO `t_user` VALUES ('13', 'test_11', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_11@163.com', '2');

1.2.python相关组件安装

1.2.1.更新pip

python -m pip install --upgrade pip

注意：如果Python环境安装在C盘，必须用管理员身份打开终端窗口才能升级pip，以及安装模块

1.2.2.安装Colorama模块

安装：

Python程序向控制台输出彩色文字，先要安装colorama模块

pip install colorama

基础使用：

1.3.项目描述

1.3.1.项目结构描述

项目采用的是DAO设计模式，封装数据层的调用，封装数据库的操作，如果后续表格发生变动，只需要修改dao对应的文件即可，不需要全部修改。

什么是DAO，说白了就是封装数据库操作，链接跳转了解:https://blog.csdn.net/zxy144/article/details/109279693

• db：存放于数据库相关的python类，操作数据表的代码
• service：业务类代码
• 核心代码：app.py

1.3.2.项目功能/需求描述

2.Code

2.1.db目录

2.1.1.mysql_db.py

import mysql.connector.pooling

__config={
"host":"localhost",
"port":3306,
"user":"root",
"password":"123456",
"database":"vega"
}

try:
pool=mysql.connector.pooling.MySQLConnectionPool(
**__config,
pool_size=10
)
except Exception as e:
print(e)

2.1.2.news_dao.py

from db.mysql_db import pool

class NewsDao:
#查询待审批新闻列表
def search_unreview_list(self,page):
try:
con=pool.get_connection()
cursor=con.cursor()
sql="SELECT n.id,n.title,t.type,u.username " \
"FROM t_news n JOIN t_type t ON n.type_id=t.id " \
"JOIN t_user u ON n.editor_id=u.id " \
"WHERE n.state=%s " \
"ORDER BY n.create_time DESC " \
"LIMIT %s,%s"
cursor.execute(sql,("待审批",(page-1)*10,10))
result=cursor.fetchall()
return result
except Exception as e:
print(e)
finally:
if "con" in dir():
con.close()

# 查询待审批新闻的总页数
def search_unreview_count_page(self):
try:
con=pool.get_connection()
cursor=con.cursor()
sql="SELECT CEIL(COUNT(*)/10) FROM t_news WHERE state=%s"
cursor.execute(sql,["待审批"])
count_page=cursor.fetchone()[0]
return count_page
except Exception as e:
print(e)
finally:
if "con" in dir():
con.close()

#审批新闻
def update_unreview_news(self,id):
try:
con = pool.get_connection()
con.start_transaction()
cursor=con.cursor()
sql="UPDATE t_news SET state=%s WHERE id=%s"
cursor.execute(sql,("已审批",id))
con.commit()
except Exception as e:
if "con" in dir():
con.rollback()
print(e)
finally:
if "con" in dir():
con.close()

#查询新闻列表
def search_list(self,page):
try:
con=pool.get_connection()
cursor=con.cursor()
sql="SELECT n.id,n.title,t.type,u.username " \
"FROM t_news n JOIN t_type t ON n.type_id=t.id " \
"JOIN t_user u ON n.editor_id=u.id " \
"ORDER BY n.create_time DESC " \
"LIMIT %s,%s"
cursor.execute(sql,((page-1)*10,10))
result=cursor.fetchall()
return result
except Exception as e:
print(e)
finally:
if "con" in dir():
con.close()

#查询新闻总页数
def search_count_page(self):
try:
con=pool.get_connection()
cursor=con.cursor()
sql="SELECT CEIL(COUNT(*)/10) FROM t_news"
cursor.execute(sql)
count_page=cursor.fetchone()[0]
return count_page
except Exception as e:
print(e)
finally:
if "con" in dir():
con.close()

#删除新闻
def delete_by_id(self,id):
try:
con = pool.get_connection()
con.start_transaction()
cursor=con.cursor()
sql="DELETE FROM t_news WHERE id=%s"
cursor.execute(sql,[id])
con.commit()
except Exception as e:
if "con" in dir():
con.rollback()
print(e)
finally:
if "con" in dir():
con.close()

2.1.3.role_dao.py

from db.mysql_db import pool

class RoleDao:
#查询角色列表
def search_list(self):
try:
con=pool.get_connection()
cursor=con.cursor()
sql="SELECT id,role FROM t_role"
cursor.execute(sql)
result=cursor.fetchall()
return result
except Exception as e:
print(e)
finally:
if "con" in dir():
con.close()

2.1.4.user_dao.py

from db.mysql_db import pool

class UserDao:
#验证用户登录
def login(self,username,password):
try:
con=pool.get_connection()
cursor=con.cursor()
sql="SELECT COUNT(*) FROM t_user WHERE username=%s AND " \
"AES_DECRYPT(UNHEX(password),'HelloWorld')=%s"
cursor.execute(sql,(username,password))
count=cursor.fetchone()[0]
return True if count==1 else False
except Exception as e:
print(e)
finally:
if "con" in dir():
con.close()

#查询用户角色
def search_user_role(self,username):
try:
pass
con=pool.get_connection()
cursor=con.cursor()
sql="SELECT r.role FROM t_user u JOIN t_role r ON u.role_id=r.id " \
"WHERE u.username=%s"
cursor.execute(sql,[username])
role=cursor.fetchone()[0]
return role
except Exception as e:
print(e)
finally:
if "con" in dir():
con.close()

#添加记录
def insert(self,username,password,email,role_id):
try:
con = pool.get_connection()
con.start_transaction()
cursor=con.cursor()
sql="INSERT INTO t_user(username,password,email,role_id) " \
"VALUES(%s,HEX(AES_ENCRYPT(%s,'HelloWorld')),%s,%s)"
cursor.execute(sql,(username,password,email,role_id))
con.commit()
except Exception as e:
if "con" in dir():
con.rollback()
print(e)
finally:
if "con" in dir():
con.close()

#查询用户总页数
def search_count_page(self):
try:
con=pool.get_connection()
cursor=con.cursor()
sql="SELECT CEIL(COUNT(*)/10) FROM t_user"
cursor.execute(sql)
count_page=cursor.fetchone()[0]
return count_page
except Exception as e:
print(e)
finally:
if "con" in dir():
con.close()

#查询用户分页记录
def search_list(self,page):
try:
con=pool.get_connection()
cursor=con.cursor()
sql="SELECT u.id,u.username,r.role " \
"FROM t_user u JOIN t_role r " \
"ON u.role_id=r.id " \
"ORDER BY u.id " \
"LIMIT %s,%s"
cursor.execute(sql,((page-1)*10,10))
result=cursor.fetchall()
return result
except Exception as e:
print(e)
finally:
if "con" in dir():
con.close()

#修改用户信息
def update(self,id,username,password,email,role_id):
try:
con = pool.get_connection()
con.start_transaction()
cursor=con.cursor()
sql="UPDATE t_user SET username=%s," \
"password=HEX(AES_ENCRYPT(%s,'HelloWorld'))," \
"email=%s,role_id=%s " \
"WHERE id=%s"
cursor.execute(sql,(username,password,email,role_id,id))
con.commit()
except Exception as e:
if "con" in dir():
con.rollback()
print(e)
finally:
if "con" in dir():
con.close()

#删除用户
def delete_by_id(self,id):
try:
con = pool.get_connection()
con.start_transaction()
cursor=con.cursor()
sql="DELETE FROM t_user WHERE id=%s"
cursor.execute(sql,[id])
con.commit()
except Exception as e:
if "con" in dir():
con.rollback()
print(e)
finally:
if "con" in dir():
con.close()

2.2.service目录

2.2.1.news_service.py

from db.news_dao import NewsDao

class NewsService:
__news_dao=NewsDao()

# 查询待审批新闻列表
def search_unreview_list(self,page):
result=self.__news_dao.search_unreview_list(page)
return result

# 查询待审批新闻的总页数
def search_unreview_count_page(self):
count_page=self.__news_dao.search_unreview_count_page()
return count_page

# 审批新闻
def update_unreview_news(self, id):
self.__news_dao.update_unreview_news(id)

#查询新闻列表
def search_list(self, page):
result=self.__news_dao.search_list(page)
return result

# 查询新闻总页数
def search_count_page(self):
count_page=self.__news_dao.search_count_page()
return count_page

# 删除新闻
def delete_by_id(self, id):
self.__news_dao.delete_by_id(id)

2.2.2.role_service.py

from db.role_dao import RoleDao

class RoleService:
__role_dao=RoleDao()

# 查询角色列表
def search_list(self):
result=self.__role_dao.search_list()
return result

2.2.3.user_service.py

from db.user_dao import UserDao

class UserService:
__user_dao=UserDao()

# 验证用户登录
def login(self,username,password):
result=self.__user_dao.login(username,password)
return result

# 查询用户角色
def search_user_role(self,username):
role=self.__user_dao.search_user_role(username)
return role

# 添加记录
def insert(self, username, password, email, role_id):
self.__user_dao.insert(username, password, email, role_id)

# 查询用户总页数
def search_count_page(self):
count_page = self.__user_dao.search_count_page()
return count_page

# 查询用户分页记录
def search_list(self, page):
result = self.__user_dao.search_list(page)
return result

# 修改用户信息
def update(self, id, username, password, email, role_id):
self.__user_dao.update(id, username, password, email, role_id)

# 删除用户
def delete_by_id(self, id):
self.__user_dao.delete_by_id(id)

2.3.app.py启动文件

from colorama import Fore,Style,init
init()
from getpass import getpass
from service.user_service import UserService
from service.news_service import NewsService
from service.role_service import RoleService
import os
import sys
import time

__user_service=UserService()
__news_service=NewsService()
__role_service=RoleService()

while True:
os.system("cls")
print(Fore.LIGHTBLUE_EX,"\n\t==================")
print(Fore.LIGHTBLUE_EX,"\n\t欢迎使用新闻管理系统")
print(Fore.LIGHTBLUE_EX, "\n\t==================")
print(Fore.LIGHTGREEN_EX,"\n\t1.登陆系统")
print(Fore.LIGHTGREEN_EX,"\n\t2.退出系统")
print(Style.RESET_ALL)
opt=input("\n\t输入操作编号:")
if opt=="1":
username=input("\n\t用户名:")
password=getpass("\n\t密码:")
result=__user_service.login(username,password)
#登陆成功
if result==True:
#查询角色
role=__user_service.search_user_role(username)
while True:
os.system("cls")
if role=="新闻编辑":
print('test')
elif role=="管理员":
print(Fore.LIGHTGREEN_EX,"\n\t1.新闻管理")
print(Fore.LIGHTGREEN_EX, "\n\t2.用户管理")
print(Fore.LIGHTRED_EX, "\n\tback.退出登陆")
print(Fore.LIGHTRED_EX, "\n\texit.退出系统")
print(Style.RESET_ALL)
opt = input("\n\t输入操作编号:")
if opt=="1":
while True:
os.system("cls")
print(Fore.LIGHTGREEN_EX, "\n\t1.审批新闻")
print(Fore.LIGHTGREEN_EX, "\n\t2.删除新闻")
print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
print(Style.RESET_ALL)
opt = input("\n\t输入操作编号:")
if opt=="1":
page=1
while True:
os.system("cls")
count_page=__news_service.search_unreview_count_page()
result=__news_service.search_unreview_list(page)
for index in range(len(result)):
one=result[index]
print(Fore.LIGHTBLUE_EX, "\n\t%d\t%s\t%s\t%s"%(index+1,one[1],one[2],one[3]))
print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
print(Fore.LIGHTBLUE_EX,"\n\t%d/%d"%(page,count_page))
print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
print(Fore.LIGHTRED_EX, "\n\tprev.上一页")
print(Fore.LIGHTRED_EX, "\n\tnext.下一页")
print(Style.RESET_ALL)
opt = input("\n\t输入操作编号:")
if opt=="back":
break
elif opt=="prev" and page>1:
page-=1
elif opt=="next" and page<count_page:
page+=1
elif int(opt)>=1 and int(opt)<=10:
news_id=result[int(opt)-1][0]
__news_service.update_unreview_news(news_id)
elif opt=="2":
page=1
while True:
os.system("cls")
count_page=__news_service.search_count_page()
result=__news_service.search_list(page)
for index in range(len(result)):
one=result[index]
print(Fore.LIGHTBLUE_EX, "\n\t%d\t%s\t%s\t%s"%(index+1,one[1],one[2],one[3]))
print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
print(Fore.LIGHTBLUE_EX,"\n\t%d/%d"%(page,count_page))
print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
print(Fore.LIGHTRED_EX, "\n\tprev.上一页")
print(Fore.LIGHTRED_EX, "\n\tnext.下一页")
print(Style.RESET_ALL)
opt = input("\n\t输入操作编号:")
if opt=="back":
break
elif opt=="prev" and page>1:
page-=1
elif opt=="next" and page<count_page:
page+=1
elif int(opt)>=1 and int(opt)<=10:
news_id=result[int(opt)-1][0]
__news_service.delete_by_id(news_id)
elif opt=="back":
break
elif opt=="2":
while True:
os.system("cls")
print(Fore.LIGHTGREEN_EX, "\n\t1.添加用户")
print(Fore.LIGHTGREEN_EX, "\n\t2.修改用户")
print(Fore.LIGHTGREEN_EX, "\n\t3.删除用户")
print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
print(Style.RESET_ALL)
opt = input("\n\t输入操作编号:")
if opt=="back":
break
elif opt=="1":
os.system("cls")
username=input("\n\t用户名:")
password = getpass("\n\t密码:")
repassword=getpass("\n\t重复密码:")
if password!=repassword:
print("\n\t两次密码不一致(3秒自动返回)")
time.sleep(3)
continue
email=input("\n\t邮箱:")
result=__role_service.search_list()
for index in range(len(result)):
one=result[index]
print(Fore.LIGHTBLUE_EX,"\n\t%d.%s"%(index+1,one[1]))
print(Style.RESET_ALL)
opt=input("\n\t角色编号:")
role_id=result[int(opt)-1][0]
__user_service.insert(username,password,email,role_id)
print("\n\t保存成功(3秒自动返回)")
time.sleep(3)
elif opt=="2":
page = 1
while True:
os.system("cls")
count_page = __user_service.search_count_page()
result = __user_service.search_list(page)
for index in range(len(result)):
one = result[index]
print(Fore.LIGHTBLUE_EX,
"\n\t%d\t%s\t%s" % (index + 1, one[1], one[2]))
print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
print(Fore.LIGHTBLUE_EX, "\n\t%d/%d" % (page, count_page))
print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
print(Fore.LIGHTRED_EX, "\n\tprev.上一页")
print(Fore.LIGHTRED_EX, "\n\tnext.下一页")
print(Style.RESET_ALL)
opt = input("\n\t输入操作编号:")
if opt == "back":
break
elif opt == "prev" and page > 1:
page -= 1
elif opt == "next" and page < count_page:
page += 1
elif int(opt) >= 1 and int(opt) <= 10:
os.system("cls")
user_id=result[int(opt)-1][0]
username = input("\n\t新用户名:")
password = getpass("\n\t新密码:")
repassword = getpass("\n\t再次输入密码:")
if password!=repassword:
print(Fore.LIGHTRED_EX,"\n\t两次密码不一致(3秒自动返回)")
print(Style.RESET_ALL)
time.sleep(3)
break
email = input("\n\t新邮箱:")
result = __role_service.search_list()
for index in range(len(result)):
one = result[index]
print(Fore.LIGHTBLUE_EX, "\n\t%d.%s" % (index + 1, one[1]))
print(Style.RESET_ALL)
opt = input("\n\t角色编号:")
role_id = result[int(opt) - 1][0]
opt=input("\n\t是否保存(Y/N)")
if opt=="Y" or opt=="y":
__user_service.update(user_id,username,password,email,role_id)
print("\n\t保存成功(3秒自动返回)")
time.sleep(3)
elif opt=="3":
page = 1
while True:
os.system("cls")
count_page = __user_service.search_count_page()
result = __user_service.search_list(page)
for index in range(len(result)):
one = result[index]
print(Fore.LIGHTBLUE_EX,
"\n\t%d\t%s\t%s" % (index + 1, one[1], one[2]))
print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
print(Fore.LIGHTBLUE_EX, "\n\t%d/%d" % (page, count_page))
print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
print(Fore.LIGHTRED_EX, "\n\tprev.上一页")
print(Fore.LIGHTRED_EX, "\n\tnext.下一页")
print(Style.RESET_ALL)
opt = input("\n\t输入操作编号:")
if opt == "back":
break
elif opt == "prev" and page > 1:
page -= 1
elif opt == "next" and page < count_page:
page += 1
elif int(opt) >= 1 and int(opt) <= 10:
os.system("cls")
user_id=result[int(opt)-1][0]
__user_service.delete_by_id(user_id)
print("\n\t删除成功(3秒自动返回)")
time.sleep(3)

if opt=='back':
break;
elif opt=='exit':
sys.exit(0)

else:
print("\n\t登录失败(3秒自动返回)")
time.sleep(3)

elif opt=="2":
sys.exit(0)