当前位置：首页 > news >正文

python源码：基于fastapi+websocket双向信息通道的简易网页聊天室

news 2025/8/27 21:01:43

前言

由于我的另一个想法，我需要使用双向通信，并最终选择了fastapi模块中的WebSocket方法来实现这个目的。

为了能够尽快掌握它，我设计了这个《基于fastapi+websocket双向信息通道的简易网页聊天室》，并且具备以下功能：

用户进入退出提示、发送及广播文字消息和图片，这两个主体功能

代码及图片

效果图

代码

0、安装模块

pip install fastapi uvicorn

1、websocket_chatRoom.py

import jsonfrom fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from tools import ws_demoapp = FastAPI()
# 增加防跨域组件
origins = ['*']
app.add_middleware(CORSMiddleware,allow_origins=origins,allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)manager = ws_demo.ConnectionManager()@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket):await manager.connect(websocket)user_id = manager.active_connections[websocket]try:await manager.send_one_message(websocket, f'您的userid：{user_id}', user_id, False)while True:data = await websocket.receive_text()data = json.loads(data)message = data.get("message")is_image = data.get("is_image", False)await manager.send_message(message, user_id, is_image)except WebSocketDisconnect:await manager.disconnect(websocket, user_id)if __name__ == '__main__':import uvicornuvicorn.run(app, host="0.0.0.0", port=int(8848))

2、 ws_demo.py

from fastapi import WebSocket
from typing import Dict
import uuid
import json
import base64# 维护连接的 WebSocket 客户端列表
class ConnectionManager:def __init__(self):self.active_connections: Dict[WebSocket, str] = {}async def connect(self, websocket: WebSocket):user_id = str(uuid.uuid4())  # 生成一个唯一的匿名IDawait websocket.accept()self.active_connections[websocket] = user_id# 广播新用户进入聊天室await self.broadcast_user_status("joined", websocket, user_id)async def disconnect(self, websocket: WebSocket, user_id):self.active_connections.pop(websocket, None)# 广播用户离开聊天室await self.broadcast_user_status("left", websocket, user_id)# 发送消息：文字或者图片async def send_message(self, message: str, user_id: str, is_image: bool = False):formatted_message = {'user': user_id,'message': message,'is_image': is_image}for connection, connection_id in self.active_connections.items():await connection.send_json(formatted_message)# 发送消息：文字或者图片async def send_one_message(self, websocket: WebSocket, message: str, user_id: str, is_image: bool = False):formatted_message = {'user': user_id,'message': message,'is_image': is_image}for connection, connection_id in self.active_connections.items():if connection == websocket:await connection.send_json(formatted_message)breakasync def broadcast_user_status(self, status: str, websocket, user_id):message = json.dumps({"user": user_id,"userStatus": status,"message": f"{user_id} {'已进入聊天室' if status == 'joined' else '已离线'}","is_image": False})for connection in self.active_connections:if connection != websocket:await connection.send_text(message)

3、show.html

<!DOCTYPE html>
<html>
<head><title>WebSocket Chat</title><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>WebSocket Chat</title><style>body {font-family: Arial, sans-serif;background-color: #f4f4f4;margin: 0;padding: 20px;}h1 {text-align: center;color: #333;margin-bottom: 20px;}#messages {border: 1px solid #ccc;height: 500px;overflow-y: auto;padding: 10px;background-color: #fff;border-radius: 5px;box-shadow: 0 2px 5px rgba(0, 0, 0, 0.1);margin-bottom: 20px;  /* 添加底部间距 */}.message {margin: 10px 0;}.message img {max-width: 100%;border-radius: 5px;}#inputContainer {display: flex;flex-direction: column; /* 改为垂直排列 */gap: 10px; /* 添加间距 */}.inputRow {display: flex;align-items: center; /* 垂直居中 */gap: 10px; /* 添加间距 */}#messageInput {flex: 1;padding: 10px;border: 1px solid #ccc;border-radius: 5px;transition: border-color 0.3s;}#messageInput:focus {border-color: #5cb85c; /* 聚焦时边框颜色 */outline: none; /* 去掉默认聚焦轮廓 */}#sendButton, #sendImageButton {padding: 10px 15px;background-color: #5cb85c;color: white;border: none;border-radius: 5px;cursor: pointer;transition: background-color 0.3s;}#sendButton:hover, #sendImageButton:hover {background-color: #4cae4c;}#imageInput {padding: 10px;}#imageInput:hover {cursor: pointer; /* 鼠标悬停提示 */}</style><script>document.addEventListener("DOMContentLoaded", function() {const socket = new WebSocket("ws://localhost:8848/ws/chat");const messages = document.getElementById("messages");const input = document.getElementById("messageInput");socket.onmessage = function(event) {const message = document.createElement("div");if (event.data.startsWith("{")) {// 如果数据是 JSON 格式，则可能是特殊消息（例如系统通知）const data = JSON.parse(event.data);if (data.is_image) {// 显示图片message.textContent = data.user + ": ";const imgElement = document.createElement("img");imgElement.src = data.message;message.appendChild(imgElement);}else {message.textContent = data.user + ": " + data.message;}messages.appendChild(message);}messages.scrollTop = messages.scrollHeight;  // 自动滚动到最新消息};document.getElementById("sendButton").addEventListener("click", function() {const message = input.value;socket.send(JSON.stringify({message: message }));input.value = "";});document.getElementById("sendImageButton").addEventListener("click", function() {const fileInput = document.getElementById("imageInput");const file = fileInput.files[0];if (file) {const reader = new FileReader();reader.onload = function(e) {const base64Image = e.target.result;// 发送 Base64 编码的图片socket.send(JSON.stringify({message: base64Image, is_image: true }));};reader.readAsDataURL(file);} else {alert("Please select an image to send.");}});});</script>
</head>
<body><h1>Chat Room with Image Support</h1><div id="messages"></div><div id="inputContainer"><div class="inputRow"><input id="messageInput" type="text" placeholder="Type a message..." /><button id="sendButton">Send Text</button></div><div class="inputRow"><input id="imageInput" type="file" accept="image/*" /><button id="sendImageButton">Send Image</button></div></div>
</body>
</html>