大家好！我是程序猿老A，感谢您阅读本文，欢迎一键三连哦。

💞当前专栏：Python毕业设计

精彩专栏推荐👇🏻👇🏻👇🏻

🎀 Java毕业设计
🌎微信小程序毕业设计

开发环境

• 开发语言：Python
• 框架：Flask
• Python版本：python3.7.7
• 数据库：mysql 5.7（一定要5.7版本）
• 数据库工具：Navicat11
• 开发软件：PyCharm
• 浏览器：谷歌浏览器

演示视频

论文目录

【如需全文或源码请按文末获取联系】

一、项目简介

系统采用 Python 语言开发，结合 Django 框架实现高性能的后端服务，并利用 MySQL 数据库存储用户数据、用电记录及预测模型参数等信息。功能模块涵盖个人中心、用户管理、家庭成员信息维护、多种用电类型分类、详细的用电记录查询、用电数据分析、个性化的用电建议以及核心的用电预测功能。通过深度学习算法对历史用电数据进行学习和建模，系统能够准确预测未来用电趋势，帮助用户优化用电行为，降低能源消耗。系统管理模块提供权限控制、数据备份与恢复等功能，确保系统的稳定运行。

二、系统设计

2.1软件功能模块设计

系统主要包括个人中心、用户、家庭成员、用电类型、用电记录、用电数据、用电建议、用电预测、系统管理等功能。系统总体功能结构图如图4-1所示。

2.2数据库设计

本系统的整体E-R实体属性如图4-7所示。

三、系统项目部分截图

3.1管理员功能实现

管理员主页面作为系统控制中心，提供全面的管理功能。页面通常详细列出所有管理模块，包括个人中心、用户、家庭成员、用电类型、用电记录、用电数据、用电建议、用电预测、系统管理等，确保管理员能够高效地进行日常管理工作。整个页面布局清晰，功能模块化，便于管理员快速定位和操作。管理员主页界面如图5-1所示：

在系统中，管理员点击“用电类型”模块后，可输入用电类型名称进行查询操作。查询结果以列表形式展示，管理员可在此列表中添加新的用电类型，或删除现有类型。管理员还能点击列表中的用电类型，查看其详细信息，包括类型描述、关联设备等。支持对用电类型详细信息进行修改或删除操作，以满足系统管理需求；用电类型界面如图5-2所示：

管理员点击“用电记录”模块，输入天气状况后，可查询与之相关的用电记录。查询结果显示为列表，管理员可在此基础上添加新的用电记录，或删除现有记录。列表中包含类型耗电、电器耗电和天气分析等信息。管理员还可点击具体记录，查看其详细信息，系统会提供用电建议。管理员能够对用电记录的详细信息进行修改或删除操作，以满足管理需求；用电记录界面如图5-3所示：

该看板集成了家庭用电管理的核心数据展示功能。它包括用户统计、类型耗电、电器耗电、用户总数、用电记录总数、用电数据总数等关键指标，提供天气分析、用电预测和月电表数等可视化信息。通过直观的图表和数据呈现，管理员可以快速了解系统的整体运行情况，便于高效管理和决策。看板界面如图5-6所示：

3.2用户功能实现

用户主页面提供个人中心、家庭成员管理、用电记录查询、用电数据分析、用电建议、用电预测及公告信息等功能，满足用户日常用电管理与节能需求。用户主页界面如图5-7所示：

四、部分核心代码

# coding:utf-8
__author__ = "ila"

import logging, os, json, configparser
import time
import numbers
import requests
from werkzeug.utils import redirect

from flask import request, jsonify,session
from sqlalchemy.sql import func,and_,or_,case
from sqlalchemy import cast, Integer,Float
from api.models.brush_model import *
from . import main_bp
from utils.codes import *
from utils.jwt_auth import Auth
from configs import configs
from utils.helper import *
import random
import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr
from email.header import Header
from utils.baidubce_api import BaiDuBce
from api.models.config_model import config
# 注册接口
@main_bp.route("/pythonu0iaagil/householdmembers/register", methods=['POST'])
def pythonu0iaagil_householdmembers_register():
    if request.method == 'POST':#post请求
        msg = {'code': normal_code, 'message': 'success', 'data': [{}]}
        req_dict = session.get("req_dict")


        #创建新用户数据
        error = householdmembers.createbyreq(householdmembers, householdmembers, req_dict)
        if error!=None and error is Exception:
            msg['code'] = crud_error_code
            msg['msg'] = "注册用户已存在"
        else:
            msg['data'] = error
        #返回结果
        return jsonify(msg)

# 登录接口
@main_bp.route("/pythonu0iaagil/householdmembers/login", methods=['GET','POST'])
def pythonu0iaagil_householdmembers_login():
    if request.method == 'GET' or request.method == 'POST':#get、post请求
        msg = {"code": normal_code, "msg": "success", "data": {}}
        #获取用户名和密码参数
        req_dict = session.get("req_dict")
        req_model = session.get("req_dict")
        try:
            del req_model['role']
        except:
            pass

        #根据用户名获取用户数据
        datas = householdmembers.getbyparams(householdmembers, householdmembers, req_model)
        if not datas:#如果为空则代表账号密码错误或用户不存在
            msg['code'] = password_error_code
            msg['msg']='密码错误或用户不存在'
            return jsonify(msg)


        req_dict['id'] = datas[0].get('id')
        try:
            del req_dict['mima']
        except:
            pass

        #新建用户缓存数据并返回结果
        return Auth.authenticate(Auth, householdmembers, req_dict)


# 登出接口
@main_bp.route("/pythonu0iaagil/householdmembers/logout", methods=['POST'])
def pythonu0iaagil_householdmembers_logout():
    if request.method == 'POST':#post请求
        msg = {
            "msg": "退出成功",
            "code": 0
        }
        req_dict = session.get("req_dict")

        return jsonify(msg)

# 重置密码接口
@main_bp.route("/pythonu0iaagil/householdmembers/resetPass", methods=['POST'])
def pythonu0iaagil_householdmembers_resetpass():
    if request.method == 'POST':#post请求
        msg = {"code": normal_code, "msg": "success"}
        #获取传递的参数
        req_dict = session.get("req_dict")

        if req_dict.get('mima') != None:
            req_dict['mima'] = '123456'
        #更新重置后的密码
        error = householdmembers.updatebyparams(householdmembers, householdmembers, req_dict)

        if error != None:
            msg['code'] = crud_error_code
            msg['msg'] = error
        else:
            msg['msg'] = '密码已重置为：123456'
        return jsonify(msg)

# 获取会话信息接口
@main_bp.route("/pythonu0iaagil/householdmembers/session", methods=['GET'])
def pythonu0iaagil_householdmembers_session():
    if request.method == 'GET':#get请求
        msg = {"code": normal_code, "data": {}}
        #获取token里的id，查找对应的用户数据返回
        req_dict={"id":session.get('params').get("id")}
        msg['data']  = householdmembers.getbyparams(householdmembers, householdmembers, req_dict)[0]

        return jsonify(msg)

# 分类接口（后端）
@main_bp.route("/pythonu0iaagil/householdmembers/page", methods=['GET'])
def pythonu0iaagil_householdmembers_page():
    if request.method == 'GET':#get请求
        msg = {"code": normal_code, "msg": "success",  "data":{"currPage":1,"totalPage":1,"total":1,"pageSize":10,"list":[]}}
        #获取传递的参数
        req_dict = session.get("req_dict")
        userinfo = session.get("params")
        try:#判断是否有消息
            __hasMessage__=householdmembers.__hasMessage__
        except:
            __hasMessage__=None
        if __hasMessage__ and __hasMessage__!="否":
            tablename=session.get("tablename")
            if tablename!="users" and session.get("params")!=None and householdmembers!='chat':
                req_dict["userid"]=session.get("params").get("id")

        tablename=session.get("tablename")
        #非管理员账号则需要判断用户的相应权限
        if tablename!="users" :
            mapping_str_to_object = {}
            for model in Base_model._decl_class_registry.values():
                if hasattr(model, '__tablename__'):
                    mapping_str_to_object[model.__tablename__] = model

            try:#是否有管理员权限
                __isAdmin__=mapping_str_to_object[tablename].__isAdmin__
            except:
                __isAdmin__=None
            try:#是否有用户权限
                __authSeparate__ =mapping_str_to_object[tablename].__authSeparate__
            except:
                __authSeparate__ = None

            if __isAdmin__!="是" and __authSeparate__ == "是" and session.get("params")!=None:
                req_dict["userid"]=session.get("params").get("id")
            else:
                try:
                    del req_dict["userid"]
                except:
                    pass

            # 当前表也是有管理员权限的表
            if  __isAdmin__ == "是" and 'householdmembers' != 'forum':
                if req_dict.get("userid") and 'householdmembers' != 'chat':
                    del req_dict["userid"]
            else:
                #非管理员权限的表,判断当前表字段名是否有userid
                if tablename!="users" and 'householdmembers'[:7]!='discuss'and "userid" in householdmembers.getallcolumn(householdmembers,householdmembers):
                    req_dict["userid"] = session.get("params").get("id")

        # 判断用户权限对应字段
        tablename = session.get("tablename")
        if tablename == 'user':
            req_dict['useraccount'] = userinfo['useraccount']
            if 'userid' in req_dict.keys():
                del req_dict["userid"]
        clause_args = []
        or_clauses = or_(*clause_args)
        #查询列表数据
        msg['data']['list'], msg['data']['currPage'], msg['data']['totalPage'], msg['data']['total'], \
        msg['data']['pageSize']  = householdmembers.page(householdmembers, householdmembers, req_dict, or_clauses)
        return jsonify(msg)

# 排序接口
@main_bp.route("/pythonu0iaagil/householdmembers/autoSort", methods=['GET'])
def pythonu0iaagil_householdmembers_autosort():
    if request.method == 'GET':#get请求
        msg = {"code": normal_code, "msg": "success",  "data":{"currPage":1,"totalPage":1,"total":1,"pageSize":10,"list":[]}}
        #获取传递的参数
        req_dict = session.get("req_dict")
        req_dict['sort']='clicktime'
        req_dict['order']='desc'

        try:#获取需要排序的内容
            __browseClick__= householdmembers.__browseClick__
        except:
            __browseClick__=None
        #根据排序字段进行排序
        if __browseClick__ =='是':
            req_dict['sort']='clicknum'
        elif __browseClick__ =='时长':
            req_dict['sort']='browseduration'
        else:
            req_dict['sort']='clicktime'
        #获取排序内容
        msg['data']['list'], msg['data']['currPage'], msg['data']['totalPage'], msg['data']['total'], \
        msg['data']['pageSize']  = householdmembers.page(householdmembers, householdmembers, req_dict)

        return jsonify(msg)

#查询单条数据
@main_bp.route("/pythonu0iaagil/householdmembers/query", methods=['GET'])
def pythonu0iaagil_householdmembers_query():
    if request.method == 'GET':#get请求
        msg = {"code": normal_code, "msg": "success",  "data":{}}
        #获取传递的参数，根据参数获取单条结果
        req_dict = session.get("req_dict")
        query = db.session.query(householdmembers)
        for key, value in req_dict.items():
            query = query.filter(getattr(householdmembers, key) == value)
        query_result = query.first()
        query_result.__dict__.pop('_sa_instance_state', None)
        msg['data'] = query_result.__dict__
        return jsonify(msg)

# 分页接口（前端）
@main_bp.route("/pythonu0iaagil/householdmembers/list", methods=['GET'])
def pythonu0iaagil_householdmembers_list():
    if request.method == 'GET':#get请求
        msg = {"code": normal_code, "msg": "success",  "data":{"currPage":1,"totalPage":1,"total":1,"pageSize":10,"list":[]}}
        #获取传递的参数
        req_dict = session.get("req_dict")
        if req_dict.__contains__('vipread'):
            del req_dict['vipread']
            
        userinfo = session.get("params")

        try:#判断是否有列表权限
            __foreEndListAuth__=householdmembers.__foreEndListAuth__
        except:
            __foreEndListAuth__=None
        #不需要权限判断就去掉userid
        if __foreEndListAuth__ and __foreEndListAuth__!="否":
            tablename=session.get("tablename")
            if tablename!="users" and session.get("params")!=None:
                req_dict['userid']=session.get("params").get("id")

        tablename=session.get("tablename")

        if 'luntan' in 'householdmembers':
            if 'userid' in req_dict.keys():
                del req_dict["userid"]

        if 'discuss' in 'householdmembers':
            if 'userid' in req_dict.keys():
                del req_dict["userid"]
        #根据封装的req_dict字典去筛选获取列表数据
        msg['data']['list'], msg['data']['currPage'], msg['data']['totalPage'], msg['data']['total'], \
        msg['data']['pageSize']  = householdmembers.page(householdmembers, householdmembers, req_dict)
        return jsonify(msg)

# 保存接口（后端）
@main_bp.route("/pythonu0iaagil/householdmembers/save", methods=['POST'])
def pythonu0iaagil_householdmembers_save():
    if request.method == 'POST':#post请求
        msg = {"code": normal_code, "msg": "success", "data": {}}
        #获取传递的参数
        req_dict = session.get("req_dict")
        for key in req_dict:#将空值转为None
            if req_dict[key] == '':
                req_dict[key] = None

        #保存数据
        error= householdmembers.createbyreq(householdmembers, householdmembers, req_dict)
        if error!=None and error is Exception:
            msg['code'] = crud_error_code
            msg['msg'] = error
        else:
            msg['data'] = error
        return jsonify(msg)

# 添加接口（前端）
@main_bp.route("/pythonu0iaagil/householdmembers/add", methods=['POST'])
def pythonu0iaagil_householdmembers_add():
    if request.method == 'POST':#post请求
        msg = {"code": normal_code, "msg": "success", "data": {}}
        #获取参数
        req_dict = session.get("req_dict")
        #判断用户权限
        try:
            __foreEndListAuth__=householdmembers.__foreEndListAuth__
        except:
            __foreEndListAuth__=None
        #不需要权限则去掉userid
        if __foreEndListAuth__ and __foreEndListAuth__!="否":
            tablename=session.get("tablename")
            if tablename!="users":
                req_dict['userid']=session.get("params").get("id")

        #保存数据
        error= householdmembers.createbyreq(householdmembers, householdmembers, req_dict)
        if error!=None and error is Exception:
            msg['code'] = crud_error_code
            msg['msg'] = error
            return jsonify(msg)
        else:
            msg['data'] = error
        return jsonify(msg)

# 踩、赞接口
@main_bp.route("/pythonu0iaagil/householdmembers/thumbsup/<id_>", methods=['GET'])
def pythonu0iaagil_householdmembers_thumbsup(id_):
    if request.method == 'GET':#get请求
        msg = {"code": normal_code, "msg": "success", "data": {}}
        req_dict = session.get("req_dict")
        id_=int(id_)
        type_=int(req_dict.get("type",0))
        #获取要踩赞的记录
        rets=householdmembers.getbyid(householdmembers, householdmembers,id_)
        update_dict={
            "id":id_,
        }
        #加减数据
        if type_==1:#赞
            update_dict["thumbsupnum"]=int(rets[0].get('thumbsupnum'))+1
        elif type_==2:#踩
            update_dict["crazilynum"]=int(rets[0].get('crazilynum'))+1
        #更新记录
        error = householdmembers.updatebyparams(householdmembers, householdmembers, update_dict)
        if error!=None:
            msg['code'] = crud_error_code
            msg['msg'] = error
        return jsonify(msg)

# 获取详情信息（后端）
@main_bp.route("/pythonu0iaagil/householdmembers/info/<id_>", methods=['GET'])
def pythonu0iaagil_householdmembers_info(id_):
    if request.method == 'GET':#get请求
        msg = {"code": normal_code, "msg": "success", "data": {}}
        #根据id获取对应记录
        data = householdmembers.getbyid(householdmembers, householdmembers, int(id_))
        if len(data)>0:
            msg['data']=data[0]
        #浏览点击次数
        try:
            __browseClick__= householdmembers.__browseClick__
        except:
            __browseClick__=None

        if __browseClick__  and  "clicknum"  in householdmembers.__table__.columns:
            click_dict={"id":int(id_),"clicknum":str(int(data[0].get("clicknum") or 0)+1)}#增加点击次数
            ret=householdmembers.updatebyparams(householdmembers,householdmembers,click_dict)#更新记录
            if ret!=None:
                msg['code'] = crud_error_code
                msg['msg'] = ret
        return jsonify(msg)

# 获取详情信息（前端）
@main_bp.route("/pythonu0iaagil/householdmembers/detail/<id_>", methods=['GET'])
def pythonu0iaagil_householdmembers_detail(id_):
    if request.method == 'GET':#get请求
        msg = {"code": normal_code, "msg": "success", "data": {}}
        #根据id获取对应记录
        data = householdmembers.getbyid(householdmembers, householdmembers, int(id_))
        if len(data)>0:
            msg['data']=data[0]

        #浏览点击次数
        try:
            __browseClick__= householdmembers.__browseClick__
        except:
            __browseClick__=None

        if __browseClick__ and "clicknum" in householdmembers.__table__.columns:
            click_dict={"id":int(id_),"clicknum":str(int(data[0].get("clicknum") or 0)+1)}#增加点击次数
            ret=householdmembers.updatebyparams(householdmembers,householdmembers,click_dict)#更新记录
            if ret!=None:
                msg['code'] = crud_error_code
                msg['msg'] = ret
        return jsonify(msg)

# 更新接口
@main_bp.route("/pythonu0iaagil/householdmembers/update", methods=['POST'])
def pythonu0iaagil_householdmembers_update():
    if request.method == 'POST':#post请求
        msg = {"code": normal_code, "msg": "success", "data": {}}
        req_dict = session.get("req_dict")
        #如果存在密码或点击次数则不更新
        if req_dict.get("mima") and "mima" not in householdmembers.__table__.columns :
            del req_dict["mima"]
        if req_dict.get("password") and "password" not in householdmembers.__table__.columns :
            del req_dict["password"]
        try:
            del req_dict["clicknum"]
        except:
            pass

        #更新记录
        error = householdmembers.updatebyparams(householdmembers, householdmembers, req_dict)
        if error!=None:
            msg['code'] = crud_error_code
            msg['msg'] = error

        return jsonify(msg)

# 删除接口
@main_bp.route("/pythonu0iaagil/householdmembers/delete", methods=['POST'])
def pythonu0iaagil_householdmembers_delete():
    if request.method == 'POST':#post请求
        msg = {"code": normal_code, "msg": "success", "data": {}}
        req_dict = session.get("req_dict")
        #删除记录
        error=householdmembers.delete(
            householdmembers,
            req_dict
        )
        if error!=None:
            msg['code'] = crud_error_code
            msg['msg'] = error
        return jsonify(msg)

# 投票接口
@main_bp.route("/pythonu0iaagil/householdmembers/vote/<int:id_>", methods=['POST'])
def pythonu0iaagil_householdmembers_vote(id_):
    if request.method == 'POST':#post请求
        msg = {"code": normal_code, "msg": "success"}
        #根据id获取记录
        data= householdmembers.getbyid(householdmembers, householdmembers, int(id_))
        for i in data:
            #增加投票数并更新记录
            votenum=i.get('votenum')
            if votenum!=None:
                params={"id":int(id_),"votenum":votenum+1}
                error=householdmembers.updatebyparams(householdmembers,householdmembers,params)
                if error!=None:
                    msg['code'] = crud_error_code
                    msg['msg'] = error
        return jsonify(msg)




#获取所有记录列表
@main_bp.route("/pythonu0iaagil/householdmembers/lists", methods=['GET'])
def pythonu0iaagil_householdmembers_lists():
    if request.method == 'GET':#get请求
        msg = {"code": normal_code, "msg": "success", "data": []}
        list,_,_,_,_ = householdmembers.page(householdmembers,householdmembers,{})
        msg['data'] = list
        return jsonify(msg)

获取源码或论文

如需对应的论文或源码，以及其他定制需求，也可以下方微❤联系。