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,timedelta
def 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':
pass
elif choice =='4':
create_work_orders()
elif choice =='5':
pass
elif choice =='6':
pass
elif choice =='7':
system.login_out()
print("已经退出系统，")
break
else:
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("暂无变电站数据")
return
print("请选择变电站:")
for i,sub in enumerate(substations,1):
print(f"{i}. {sub['substation_name']:<15} {sub['location']:<20}")
try:
choice=int(input("请选择编号:"))-1
if 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("没有可用的变电站")
return
print("选择关联的设备")
for i,sub in enumerate(substations,1):
print(f"{i}. {sub['substation_name']:<15} ")
try:
sub_choice=int(input("选择变电站"))-1
if 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,Grid
from pyecharts.globals import ThemeType
class 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 users
where 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 True
except Exception as e:
logging.error(f"登录失败{e}")
return False
def 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 None
db.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 = None
def get_substations(self):
query="""
select s.*,u.full_name as operator_name
from substations s
left join users u on s.operator_id = u.user_id
order 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_id
where e.substation_id=%s
order 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()
