1. 实现下工单
------程序效果图：----
✅连接到MySQL数据库
⚡ 欢迎使用XXX电网管理系统====================================*****登录系统******
请输入用户名:admin
请输入用密码:admin
✅连接到MySQL数据库
恭喜您admin登录成功了====================================国家电网管理系统
====================================
当前用户: 系统管理员 (admin)
1. 查看变电站列表
2. 查看设备列表
3. 查看最新电力数据
4. 创建工单
5. 查看我的工单
6. 系统日志
7. 退出
------------------------------------
请选择功能[请输入1--7之间的数字]：4
工单标题：test
工单的描述test
✅连接到MySQL数据库
选择关联的设备
1. 东区变电站
2. 中心变电站
选择变电站1
✅连接到MySQL数据库
您选择了东区变电站的设备
1. 电容器组C1 capacitor
选择设备:1
✅连接到MySQL数据库
指派给工程师:
1. engineer1 engineer
选择工程师1
优先级:low/medium/high/critical 默认的是[medium]: high
✅连接到MySQL数据库
工单创建成功！工单号是:1====================================国家电网管理系统
====================================
当前用户: 系统管理员 (admin)
1. 查看变电站列表
2. 查看设备列表
3. 查看最新电力数据
4. 创建工单
5. 查看我的工单
6. 系统日志
7. 退出
------------------------------------
请选择功能[请输入1--7之间的数字]：7
✅连接到MySQL数据库
已经退出系统，
实现代码：
"""
@Project : 01-python-learn
@File : main.py
@IDE : PyCharm
@Author : 刘庆东
@Date : 2025/9/25 9:10主函数类
"""
from power_grid_system import system
from database import db
from datetime import datetime,timedeltadef main():print("⚡ 欢迎使用XXX电网管理系统")while not system.current_user:if not login_menu():retry=input("登录失败，重试吗？(y/n)")if retry.lower() != "y":print("再见！")return while True:main_menu()try:choice=input("请选择功能[请输入1--7之间的数字]：").strip()if choice =='1':view_substations()elif choice =='2':view_equipment()elif choice =='3':passelif choice =='4':create_work_orders()elif choice =='5':passelif choice =='6':passelif choice =='7':system.login_out()print("已经退出系统，")breakelse:print("无效选择，请重新输入:")except Exception as e:print(f"程序被中断，再见！原因是:{e}")break
def view_substations():substations = system.get_substations()print("变电站列表:")print(f"{'ID':<4} {'名称':<15} {'位置':<20}"f" {'电压(KV)':<10} "f"{'容量(MA)':<10} {'状态':<10} {'负责人'}")print("_" * 36)for sub in substations:print(f"{sub['substation_id']:<4}"f"{sub['substation_name']:<15}"f"{sub['location']:<20}"f"{sub['voltage_level_kV']:<10}"f"{sub['capacity_MVA']:<10}"f"{sub['status']:<15}"f"{sub['operator_name']}")
def view_equipment():substations=system.get_substations()if not substations:print("暂无变电站数据")returnprint("请选择变电站:")for i,sub in enumerate(substations,1):print(f"{i}. {sub['substation_name']:<15} {sub['location']:<20}")try:choice=int(input("请选择编号:"))-1if 0<=choice<len(substations):selected_sub = substations[choice]equipments=system.get_equipment_by_substation_id(selected_sub['substation_id'])print(f"变电站{sub['substation_name']}的设备列表:")print("-"*36)print(f"{'ID':<4} {'名称':<20} {'类型':<12} {'型号':<15} {'序列号':<15} {'状态'}")print("-" * 36)for eq in equipments:print(f"{eq['equipment_id']:<4} {eq['equipment_name']:<20} {eq['equipment_type']:<12} "f"{eq['model']:<15} {eq['serial_number']:<15} {eq['status']}")print("-"*36)except Exception as e:print(f"请输入有效的数字:{e}")
def create_work_orders():title=input("工单标题：").strip()description=input("工单的描述").strip() substations=system.get_substations()if not substations:print("没有可用的变电站")returnprint("选择关联的设备")for i,sub in enumerate(substations,1):print(f"{i}. {sub['substation_name']:<15} ")try:sub_choice=int(input("选择变电站"))-1if sub_choice<0 or sub_choice>=len(substations):print("无效的选择")returnselected_sub = substations[sub_choice]equipments=system.get_equipment_by_substation_id(selected_sub['substation_id'])if not equipments:print("该变电站木有设备")returnprint(f"您选择了{selected_sub['substation_name']}的设备")for i,eq in enumerate(equipments,1):print(f"{i}. {eq['equipment_name']} {eq['equipment_type']}")eq_choice=int(input("选择设备:"))-1if eq_choice<0 or eq_choice>=len(equipments):print("无效选择")returnselected_eq= equipments[eq_choice]engineers=[u for u in system.get_users() if u['role']=='engineer']if not engineers:print("无空闲的工程师!")returnprint("指派给工程师:")for i,eng in enumerate(engineers,1):print(f"{i}. {eng['username']} {eng['role']}")eng_choice=int(input("选择工程师"))-1if eng_choice<0 or eng_choice>=len(engineers):print("无效的选择")returnassigned_to=engineers[eng_choice]['user_id']priority=input("优先级:low/medium/high/critical 默认的是[medium]: ").strip() or "medium"if priority not in ["low", "medium", "high", "critical"]:priority='medium'start_time=(datetime.now() + timedelta(hours=1)).strftime('%Y-%m-%d %H:%M:%S')end_time=(datetime.now() + timedelta(hours=3)).strftime('%Y-%m-%d %H:%M:%S')order_id=system.create_work_order(title=title,description=description,equipment_id=selected_eq['equipment_id'],substation_id=selected_sub['substation_id'],assigned_to=assigned_to,priority=priority,scheduled_start = start_time,scheduled_end = end_time)print(f"工单创建成功！工单号是:{order_id}")except Exception as e:print(f"输入有错误:{e}")
def main_menu():print("\n" + "=" * 36)print(" 国家电网管理系统")print("=" * 36)print(f"当前用户: {system.current_user['full_name']} ({system.current_user['role']})")print("1. 查看变电站列表")print("2. 查看设备列表")print("3. 查看最新电力数据")print("4. 创建工单")print("5. 查看我的工单")print("6. 系统日志")print("7. 退出")print("-" * 36)
def login_menu():print("\n" + "=" * 36)print(" *****登录系统******")username=input("请输入用户名:")password=input("请输入用密码:")success_login=system.login(username, password)if success_login:print(f"恭喜您{username}登录成功了")return Trueelse:print(f"系统登录失败")return Falseprint("\n" + "=" * 36)if __name__ == '__main__':main()
实现代码：
"""
@Project : 01-python-learn
@File : power_grid_system.py
@IDE : PyCharm
@Author : 刘庆东
@Date : 2025/9/24 14:01核心业务类:刚开始认为技术最难其实 业务最难"""
from database import db
import bcrypt
"""
datetime 表示当前具体的时刻 2025年9月24日 下午02:06
timedelta 代表的是两个时间点之间的差值 比如说 1天 4小时23分钟后进行休息
"""
from datetime import datetime,timedelta
import logging from pyecharts import options as opts"""
Line 上课学过使用场景:股票走势、气温变化、销售额度Bar 上课学过使用场景：月业绩排名、各个地区的人口数量Pie 大饼图使用场景:市场份额占比、预算分配Gauge 仪表盘使用场景: KPI、进度、完成率Grid 布局网格使用场景: 将一个折线图 和一个柱状图并排显示 ! """
from pyecharts.charts import Line,Bar,Pie,Gauge,Gridfrom pyecharts.globals import ThemeTypeclass PowerGridSystem:def __init__(self):self.current_user=None def login(self,username,password):try:query="""select user_id,username,password_hash,role,full_name from userswhere username=%s and STATUS='active'"""result= db.execute_query(query,(username,))if result:user =result[0]if bcrypt.checkpw(password.encode('utf-8'),user['password_hash'].encode('utf-8')):self.current_user=user self.log_action("login","user",user['user_id'],f"用户{username}登录")return Trueexcept Exception as e:logging.error(f"登录失败{e}")return Falsedef log_action(self,action, target_type, target_id, details):query="""insert into system_logs (user_id, action, target_type, target_id, details)values (%s,%s,%s,%s,%s)"""user_id= self.current_user['user_id'] if self.current_user else Nonedb.execute_update(query,(user_id,action,target_type,target_id,details))def login_out(self):if self.current_user:self.log_action("logout",'user',self.current_user['user_id'],f"用户{self.current_user['username']}登出")self.current_user = Nonedef get_substations(self):query="""select s.*,u.full_name as operator_name from substations sleft join users u on s.operator_id = u.user_idorder by s.substation_name; """return db.execute_query(query)def get_equipment_by_substation_id(self,substation_id):sql="""select e.*,s.substation_name from equipment e join substations s on e.substation_id = s.substation_idwhere e.substation_id=%sorder by e.equipment_name"""return db.execute_query(sql,(substation_id,))def get_users(self):query="""select user_id,username,password_hash,role,full_name from users"""return db.execute_query(query)"""order_id=system.create_work_order(titile=titile,description=description,equipment_id=selected_eq['equipment_id'],substation_id=selected_sub['substation_id'],assigned_to=assigned_to,priority=priority,scheduled_start = datetime.now() + timedelta(hours=1),scheduled_end = datetime.now() + timedelta(hours=3))"""def create_work_order(self,title,description,equipment_id,substation_id,assigned_to,priority,scheduled_start,scheduled_end):query="""insert into work_orders (title,description,equipment_id,substation_id,assigned_to,priority,scheduled_start,scheduled_end)values (%s,%s,%s,%s,%s,%s,%s,%s)"""return db.execute_update(query,(title,description,equipment_id,substation_id,assigned_to,priority,scheduled_start,scheduled_end))
system= PowerGridSystem()
