MySQL(11) - Python+MySQL开发新闻管理系统
2022-05-25 10:50
2621 查看
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 result2.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)
相关文章推荐
- Python开发【项目】:学员管理系统(mysql)
- PHP+MYSQL新闻系统开发之PHP100视频教程62 修正版
- Python的系统管理_11_ConfigParser读取配置文件
- PHP+MYSQL新闻系统开发之admin_news_add.php和admin_news_list.php
- 在Cloud Studio用Python3+Django开发一个简单的人员管理系统
- 利用python实战开发一个web管理系统框架
- 使用 python 管理 mysql 开发工具箱 - 2
- python开发信息管理系统
- 视频教程-jsp servlet mysql实现图书管理系统实战开发教程-Java
- springboot开发:新闻管理系统之添加和修改新闻
- PHP基础示例:用PHP+Mysql编写简易新闻管理系统[转]
- 安卓SQLite数据库操作,半小时开发新闻管理系统,纯干货
- Pycharm+Django+Python+MySQL开发 后台管理数据库
- 体育馆管理系统开发日记 11
- Python基于mysql实现学生管理系统
- JAVA EE学习-02 开发基于Mysql、Servlet、JSP的作业管理系统
- python实现学生管理系统(mysql+面向对象)
- java/php/net/python新闻发布及管理系统设计
- 【Aspx应用开发平台教程】架构篇:微系统构件实例-新闻内容管理构件
- 用python开发微信图文消息回复系统——新闻检索系统