目录

Python实例题

题目

要求：

解题思路：

代码实现：

Python实例题

题目

基于 Flask 的在线聊天系统

要求：

  • 使用 Flask 框架构建一个实时在线聊天系统，支持以下功能：
    • 用户注册、登录和个人资料管理
    • 一对一实时聊天功能
    • 群聊功能
    • 消息通知和未读消息提示
    • 在线用户状态显示
  • 使用 Flask-SocketIO 实现实时通信。
  • 使用 SQLite 数据库存储用户、聊天记录等信息。
  • 添加美观的前端界面，支持响应式设计。

解题思路：

  • 使用 Flask 蓝图组织代码结构。
  • 通过 Flask-Login 处理用户认证。
  • 使用 Flask-SocketIO 实现实时消息推送。
  • 使用 Flask-SQLAlchemy 管理数据库操作。
  • 结合 Bootstrap 实现美观的前端界面。

代码实现：

from flask import Flask, render_template, request, redirect, url_for, flash, session, Blueprint
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
from werkzeug.security import generate_password_hash, check_password_hash
from flask_socketio import SocketIO, emit, join_room, leave_room, send
import os
import time
from datetime import datetime# 初始化 Flask 应用
app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key-here'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(app.root_path, 'chat.db')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False# 初始化数据库
db = SQLAlchemy(app)# 初始化 Flask-Login
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'auth.login'# 初始化 SocketIO
socketio = SocketIO(app, async_mode='eventlet')# 数据模型
class User(UserMixin, db.Model):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(80), unique=True, nullable=False)password_hash = db.Column(db.String(120), nullable=False)avatar = db.Column(db.String(100), default='default.png')online = db.Column(db.Boolean, default=False)last_seen = db.Column(db.DateTime)messages_sent = db.relationship('Message', foreign_keys='Message.sender_id',backref='sender', lazy='dynamic')messages_received = db.relationship('Message', foreign_keys='Message.recipient_id',backref='recipient', lazy='dynamic')def set_password(self, password):self.password_hash = generate_password_hash(password)def check_password(self, password):return check_password_hash(self.password_hash, password)def get_unread_messages_count(self):return Message.query.filter_by(recipient_id=self.id, read=False).count()class Message(db.Model):id = db.Column(db.Integer, primary_key=True)sender_id = db.Column(db.Integer, db.ForeignKey('user.id'))recipient_id = db.Column(db.Integer, db.ForeignKey('user.id'))content = db.Column(db.Text, nullable=False)timestamp = db.Column(db.DateTime, default=datetime.utcnow)read = db.Column(db.Boolean, default=False)class ChatRoom(db.Model):id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(100), nullable=False)is_group = db.Column(db.Boolean, default=False)creator_id = db.Column(db.Integer, db.ForeignKey('user.id'))members = db.relationship('User', secondary='chat_room_member', backref='chat_rooms')class ChatRoomMember(db.Model):id = db.Column(db.Integer, primary_key=True)chat_room_id = db.Column(db.Integer, db.ForeignKey('chat_room.id'))user_id = db.Column(db.Integer, db.ForeignKey('user.id'))joined_at = db.Column(db.DateTime, default=datetime.utcnow)class GroupMessage(db.Model):id = db.Column(db.Integer, primary_key=True)chat_room_id = db.Column(db.Integer, db.ForeignKey('chat_room.id'))sender_id = db.Column(db.Integer, db.ForeignKey('user.id'))content = db.Column(db.Text, nullable=False)timestamp = db.Column(db.DateTime, default=datetime.utcnow)# 用户加载回调
@login_manager.user_loader
def load_user(user_id):return User.query.get(int(user_id))# 认证蓝图
auth = Blueprint('auth', __name__)@auth.route('/register', methods=['GET', 'POST'])
def register():if request.method == 'POST':username = request.form['username']password = request.form['password']if User.query.filter_by(username=username).first():flash('用户名已存在', 'error')return redirect(url_for('auth.register'))user = User(username=username)user.set_password(password)db.session.add(user)db.session.commit()flash('注册成功，请登录', 'success')return redirect(url_for('auth.login'))return render_template('register.html')@auth.route('/login', methods=['GET', 'POST'])
def login():if request.method == 'POST':username = request.form['username']password = request.form['password']user = User.query.filter_by(username=username).first()if user and user.check_password(password):login_user(user)user.online = Trueuser.last_seen = datetime.utcnow()db.session.commit()# 存储用户会话信息session['user_id'] = user.idflash('登录成功', 'success')return redirect(url_for('main.index'))else:flash('用户名或密码错误', 'error')return render_template('login.html')@auth.route('/logout')
@login_required
def logout():# 更新用户在线状态user = User.query.get(current_user.id)user.online = Falseuser.last_seen = datetime.utcnow()db.session.commit()logout_user()flash('已注销', 'success')return redirect(url_for('auth.login'))# 主蓝图
main = Blueprint('main', __name__)@main.route('/')
@login_required
def index():# 获取所有用户（除当前用户外）users = User.query.filter(User.id != current_user.id).order_by(User.username).all()# 获取当前用户的所有聊天组chat_rooms = ChatRoom.query.join(ChatRoomMember).filter(ChatRoomMember.user_id == current_user.id).all()# 获取未读消息数量unread_counts = {}for user in users:unread_counts[user.id] = Message.query.filter_by(recipient_id=current_user.id,sender_id=user.id,read=False).count()return render_template('index.html', users=users, chat_rooms=chat_rooms, unread_counts=unread_counts)@main.route('/user/<int:user_id>/chat')
@login_required
def user_chat(user_id):recipient = User.query.get_or_404(user_id)# 将当前对话的消息标记为已读Message.query.filter_by(sender_id=recipient.id,recipient_id=current_user.id,read=False).update({Message.read: True})db.session.commit()# 获取所有用户（除当前用户外）users = User.query.filter(User.id != current_user.id).order_by(User.username).all()# 获取当前用户的所有聊天组chat_rooms = ChatRoom.query.join(ChatRoomMember).filter(ChatRoomMember.user_id == current_user.id).all()# 获取未读消息数量unread_counts = {}for user in users:unread_counts[user.id] = Message.query.filter_by(recipient_id=current_user.id,sender_id=user.id,read=False).count()return render_template('user_chat.html', recipient=recipient, users=users, chat_rooms=chat_rooms, unread_counts=unread_counts)@main.route('/group/<int:group_id>/chat')
@login_required
def group_chat(group_id):chat_room = ChatRoom.query.get_or_404(group_id)# 检查当前用户是否是该组的成员is_member = ChatRoomMember.query.filter_by(chat_room_id=group_id,user_id=current_user.id).first() is not Noneif not is_member:flash('你不是该组的成员', 'error')return redirect(url_for('main.index'))# 获取所有用户（除当前用户外）users = User.query.filter(User.id != current_user.id).order_by(User.username).all()# 获取当前用户的所有聊天组chat_rooms = ChatRoom.query.join(ChatRoomMember).filter(ChatRoomMember.user_id == current_user.id).all()# 获取未读消息数量unread_counts = {}for user in users:unread_counts[user.id] = Message.query.filter_by(recipient_id=current_user.id,sender_id=user.id,read=False).count()return render_template('group_chat.html', chat_room=chat_room, users=users, chat_rooms=chat_rooms, unread_counts=unread_counts)@main.route('/create-group', methods=['GET', 'POST'])
@login_required
def create_group():if request.method == 'POST':group_name = request.form['group_name']member_ids = request.form.getlist('members')if not group_name:flash('组名不能为空', 'error')return redirect(url_for('main.create_group'))# 创建新组group = ChatRoom(name=group_name, is_group=True, creator_id=current_user.id)db.session.add(group)db.session.commit()# 添加创建者为成员member = ChatRoomMember(chat_room_id=group.id, user_id=current_user.id)db.session.add(member)# 添加其他成员for member_id in member_ids:if int(member_id) != current_user.id:member = ChatRoomMember(chat_room_id=group.id, user_id=member_id)db.session.add(member)db.session.commit()flash('组创建成功', 'success')return redirect(url_for('main.group_chat', group_id=group.id))# 获取所有用户（除当前用户外）users = User.query.filter(User.id != current_user.id).order_by(User.username).all()return render_template('create_group.html', users=users)@main.route('/profile', methods=['GET', 'POST'])
@login_required
def profile():if request.method == 'POST':username = request.form['username']avatar = request.form['avatar']if username != current_user.username and User.query.filter_by(username=username).first():flash('用户名已存在', 'error')return redirect(url_for('main.profile'))current_user.username = usernamecurrent_user.avatar = avatardb.session.commit()flash('个人资料已更新', 'success')return redirect(url_for('main.profile'))return render_template('profile.html')# SocketIO 事件处理
@socketio.on('connect')
def handle_connect():user_id = session.get('user_id')if user_id:user = User.query.get(user_id)if user:user.online = Trueuser.last_seen = datetime.utcnow()db.session.commit()# 通知所有在线用户更新用户状态update_user_status(user)@socketio.on('disconnect')
def handle_disconnect():user_id = session.get('user_id')if user_id:user = User.query.get(user_id)if user:user.online = Falseuser.last_seen = datetime.utcnow()db.session.commit()# 通知所有在线用户更新用户状态update_user_status(user)@socketio.on('private_message')
def handle_private_message(data):sender_id = session.get('user_id')recipient_id = data['recipient_id']content = data['content']if sender_id and recipient_id and content:# 保存消息到数据库message = Message(sender_id=sender_id,recipient_id=recipient_id,content=content)db.session.add(message)db.session.commit()# 发送消息给接收者recipient = User.query.get(recipient_id)if recipient.online:emit('new_message', {'sender_id': sender_id,'recipient_id': recipient_id,'content': content,'timestamp': message.timestamp.strftime('%Y-%m-%d %H:%M:%S')}, room=str(recipient_id))# 发送消息给发送者自己emit('new_message', {'sender_id': sender_id,'recipient_id': recipient_id,'content': content,'timestamp': message.timestamp.strftime('%Y-%m-%d %H:%M:%S')}, room=str(sender_id))@socketio.on('join_room')
def handle_join_room(data):room_id = data['room_id']join_room(room_id)@socketio.on('leave_room')
def handle_leave_room(data):room_id = data['room_id']leave_room(room_id)@socketio.on('group_message')
def handle_group_message(data):sender_id = session.get('user_id')chat_room_id = data['chat_room_id']content = data['content']if sender_id and chat_room_id and content:# 保存消息到数据库group_message = GroupMessage(chat_room_id=chat_room_id,sender_id=sender_id,content=content)db.session.add(group_message)db.session.commit()# 发送消息到群组emit('new_group_message', {'sender_id': sender_id,'chat_room_id': chat_room_id,'content': content,'timestamp': group_message.timestamp.strftime('%Y-%m-%d %H:%M:%S')}, room=str(chat_room_id))def update_user_status(user):"""通知所有在线用户更新用户状态"""emit('user_status_update', {'user_id': user.id,'online': user.online,'last_seen': user.last_seen.strftime('%Y-%m-%d %H:%M:%S') if user.last_seen else ''}, broadcast=True)# 注册蓝图
app.register_blueprint(main)
app.register_blueprint(auth)# 创建数据库表
with app.app_context():db.create_all()if __name__ == '__main__':socketio.run(app, debug=True)

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如若转载，请注明出处：http://www.pswp.cn/web/87215.shtml
繁体地址，请注明出处：http://hk.pswp.cn/web/87215.shtml
英文地址，请注明出处：http://en.pswp.cn/web/87215.shtml

如若内容造成侵权/违法违规/事实不符，请联系英文站点网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

v-bind指令

v-bind指令

好的&#xff0c;我们来学习 v-bind 指令。这个指令是理解 Vue 数据驱动思想的基石。 核心功能&#xff1a;v-bind 的作用是将一个或多个 HTML 元素的 attribute (属性) 或一个组件的 prop (属性) 动态地绑定到 Vue 实例的数据上。 简单来说&#xff0c;它在你的数据和 HTML …
阅读更多...
【设计模式04】单例模式

【设计模式04】单例模式

前言 整个系统中只会出现要给实例&#xff0c;比如Spring中的Bean基本都是单例的 UML类图 无 代码示例 package com.sw.learn.pattern.B_create.c_singleton;public class Main {public static void main(String[] args) {// double check locking 线程安全懒加载 ⭐️ //…
阅读更多...
飞算科技依托 JavaAI 核心技术，打造企业级智能开发全场景方案

飞算科技依托 JavaAI 核心技术，打造企业级智能开发全场景方案

在数字经济蓬勃发展的当下&#xff0c;企业对智能化开发的需求愈发迫切。飞算数智科技&#xff08;深圳&#xff09;有限公司&#xff08;简称 “飞算科技”&#xff09;作为自主创新型数字科技公司与国家级高新技术企业&#xff0c;凭借深厚的技术积累与创新能力&#xff0c;以…
阅读更多...
20250701【二叉树公共祖先】|Leetcodehot100之236【pass】今天计划

20250701【二叉树公共祖先】|Leetcodehot100之236【pass】今天计划

20250701 思路与错误记录1.二叉树的数据结构与初始化1.1数据结构1.2 初始化 2.解题 完整代码今天做了什么 题目 思路与错误记录 1.二叉树的数据结构与初始化 1.1数据结构 1.2 初始化 根据列表&#xff0c;顺序存储构建二叉树 def build_tree(nodes, index0):# idx是root开始…
阅读更多...
Web应用开发 --- Tips

Web应用开发 --- Tips

Web应用开发 --- Tips General后端需要做参数校验代码风格和Api设计风格的一致性大于正确性数据入库时间应由后端记录在对Api修改的时候&#xff0c;要注意兼容情况&#xff0c;避免breaking change 索引对于查询字段&#xff0c;注意加索引对于唯一的字段&#xff0c;考虑加唯…
阅读更多...
CSS 安装使用教程

CSS 安装使用教程

一、CSS 简介 CSS&#xff08;Cascading Style Sheets&#xff0c;层叠样式表&#xff09;是用于为 HTML 页面添加样式的语言。通过 CSS 可以控制网页元素的颜色、布局、字体、动画等&#xff0c;是前端开发的三大核心技术之一&#xff08;HTML、CSS、JavaScript&#xff09;。…
阅读更多...
机器学习中为什么要用混合精度训练

机器学习中为什么要用混合精度训练

目录 FP16与显存占用关系机器学习中一般使用混合精度训练&#xff1a;FP16计算 FP32存储关键变量。 FP16与显存占用关系 显存&#xff08;Video RAM&#xff0c;简称 VRAM&#xff09;是显卡&#xff08;GPU&#xff09;专用的内存。 FP32&#xff08;单精度浮点&#xff09;&…
阅读更多...
[附源码+数据库+毕业论文+答辩PPT]基于Spring+MyBatis+MySQL+Maven+vue实现的中小型企业财务管理系统，推荐！

[附源码+数据库+毕业论文+答辩PPT]基于Spring+MyBatis+MySQL+Maven+vue实现的中小型企业财务管理系统，推荐！

摘 要 现代经济快节奏发展以及不断完善升级的信息化技术&#xff0c;让传统数据信息的管理升级为软件存储&#xff0c;归纳&#xff0c;集中处理数据信息的管理方式。本中小型企业财务管理就是在这样的大环境下诞生&#xff0c;其可以帮助管理者在短时间内处理完毕庞大的数据信…
阅读更多...
华为云Flexus+DeepSeek征文 | 对接华为云ModelArts Studio大模型：AI赋能投资理财分析与决策

华为云Flexus+DeepSeek征文 | 对接华为云ModelArts Studio大模型：AI赋能投资理财分析与决策

引言&#xff1a;AI金融&#xff0c;开启智能投资新时代​​ 随着人工智能技术的飞速发展&#xff0c;金融投资行业正迎来前所未有的变革。​​华为云ModelArts Studio​​结合​​Flexus高性能计算​​与​​DeepSeek大模型​​&#xff0c;为投资者提供更精准、更高效的投资…
阅读更多...
从模型部署到AI平台：云原生环境下的大模型平台化演进路径

从模型部署到AI平台：云原生环境下的大模型平台化演进路径

&#x1f4dd;个人主页&#x1f339;&#xff1a;慌ZHANG-CSDN博客 &#x1f339;&#x1f339;期待您的关注 &#x1f339;&#x1f339; 一、引言&#xff1a;部署只是起点&#xff0c;平台才是终局 在过去一年&#xff0c;大语言模型的飞速发展推动了AI生产力浪潮。越来越多…
阅读更多...
UI前端大数据可视化创新：利用AR/VR技术提升用户沉浸感

UI前端大数据可视化创新：利用AR/VR技术提升用户沉浸感

hello宝子们...我们是艾斯视觉擅长ui设计、前端开发、数字孪生、大数据、三维建模、三维动画10年经验!希望我的分享能帮助到您!如需帮助可以评论关注私信我们一起探讨!致敬感谢感恩! 在大数据与沉浸式技术高速发展的今天&#xff0c;传统二维数据可视化已难以满足复杂数据场景的…
阅读更多...
MacOS 安装brew 国内源【超简洁步骤】

MacOS 安装brew 国内源【超简洁步骤】

​/bin/zsh -c "$(curl -fsSL https://gitee.com/cunkai/HomebrewCN/raw/master/Homebrew.sh)"请输入序号&#xff1a;1
阅读更多...
GENESIS64：全球知名的工业设备监控与可视化平台

GENESIS64：全球知名的工业设备监控与可视化平台

一、概述 GENESIS64是一款由ICONICS开发的先进工业自动化软件平台&#xff0c;专为实现实时数据可视化、智能化监控及管理而设计。该平台采用模块化架构&#xff0c;具有高效的数据处理能力和灵活的扩展性&#xff0c;适用于各类工业环境&#xff0c;帮助企业实现自动化运营&a…
阅读更多...
RNN（Recurrent Neural Network，循环神经网络）家族详解（RNN，LSTM，GRU）

RNN（Recurrent Neural Network，循环神经网络）家族详解（RNN，LSTM，GRU）

文章目录 一、RNN基础&#xff1a;序列建模的核心思想1.1 RNN的本质与核心机制1.2 应用场景与结构分类 二、传统RNN&#xff1a;序列模型的起点2.1 内部结构与数学表达2.2 计算示例2.3 RNN在Pytorch中的API2.4 代码示例2.5 优缺点与梯度问题 三、LSTM&#xff1a;门控机制破解长…
阅读更多...
多云密钥统一管理实战：CKMS对接阿里云/华为云密钥服务

多云密钥统一管理实战：CKMS对接阿里云/华为云密钥服务

某保险公司因阿里云KMS密钥与华为云密钥割裂管理&#xff0c;导致勒索事件中解密失败&#xff01;据统计&#xff0c;73%企业因多云密钥分散管理引发数据恢复延迟&#xff08;IDC 2024&#xff09;。本文将详解安当CKMS统一纳管方案&#xff0c;实现跨云密钥全生命周期管控&…
阅读更多...
光伏接入承载力计算仿真：基于图计算技术的自动建模技术研究

光伏接入承载力计算仿真：基于图计算技术的自动建模技术研究

光伏接入承载力计算仿真：基于图计算技术的自动建模技术研究 一、 引言：挑战与机遇 光伏发电的大规模接入对中低压配电网的安全稳定运行带来了巨大挑战。精确评估电网对光伏的承载力（Hosting Capacity, HC）是保障消纳与安全的关键。传统承载力评估严重依赖电网仿真，而仿真…
阅读更多...
如何在Excel中每隔几行取一行

如何在Excel中每隔几行取一行

如何在Excel中每隔几行取一行 摘要&#xff1a; Excel中快速实现每隔n行取一行的技巧&#xff1a;使用OFFSET函数配合ROW函数即可实现。公式为OFFSET(起始单元格,(ROW(A1)-1)*n,)&#xff0c;其中n为间隔行数。例如从A2开始每2行取一行&#xff0c;公式为OFFSET(A2,(ROW(A1)-1)…
阅读更多...
【MariaDB】MariaDB Server 11.3.0 Alpha下载、安装、配置

【MariaDB】MariaDB Server 11.3.0 Alpha下载、安装、配置

MariaDB是一个开源关系型数据库管理系统&#xff08;RDBMS&#xff09;&#xff0c;由MySQL的原始开发者Michael Widenius主导开发。作为MySQL的分支&#xff0c;MariaDB旨在保持与MySQL的高度兼容性&#xff0c;同时提供性能优化、新功能和更好的开源承诺。 目录 MariaDB下载 …
阅读更多...
如何保证缓存和数据库的双写一致性

如何保证缓存和数据库的双写一致性

程序员面试资料大全&#xff5c;各种技术书籍等资料-1000G IDEA开发工具- FREE 一、双写一致性问题本质 在分布式系统中&#xff0c;缓存与数据库双写一致性指当数据被修改时&#xff0c;如何确保缓存&#xff08;如Redis&#xff09;和数据库&#xff08;如MySQL&#xff09…
阅读更多...
Qt 5.9 XML文件写入指南

Qt 5.9 XML文件写入指南

Qt 5.9 XML文件写入指南 在Qt 5.9中&#xff0c;有多种方法可以编写XML文件。下面我将介绍三种主要方法&#xff0c;并提供完整的代码示例和最佳实践。 三种XML写入方法对比 方法优点缺点适用场景QXmlStreamWriter高效、内存占用低无树形结构大型XML文件QDomDocument树形结构…
阅读更多...
最新文章

Copyright @ 2022~2023