Python实现私信功能全指南：从设计到实战

私信功能是社交类 Web 应用的核心交互模块，它让用户之间能够进行私密交流。本文将详细介绍如何使用 Python Web 技术栈构建一个完整的私信系统，涵盖数据库设计、后端 API 开发、前端交互及功能扩展等关键环节。

一、功能需求分析

一个基础且实用的私信系统应包含以下核心功能：

发送文本私信
查看私信列表（显示最近联系人）
查看与特定用户的聊天记录
未读消息提醒与计数
标记消息为已读
删除消息（支持单条删除）

二、技术选型

本次实现采用当前流行的 Python Web 技术组合：

后端框架：Flask 2.0+（轻量灵活，适合快速开发）
数据库：SQLite（开发环境）/MySQL（生产环境）
ORM：SQLAlchemy（简化数据库操作）
前端：HTML5 + JavaScript + Bootstrap 5
异步通信：AJAX（可扩展为 WebSocket 实现实时通信）
部署：Gunicorn + Nginx（生产环境）

三、数据库设计

我们需要设计两张核心表：用户表和私信表。使用 SQLAlchemy 的 ORM 方式定义数据模型。

1. 数据模型定义（models.py）

python

运行"www.stats.gov.cn.nxbest.cn", "www.cq.gov.cn.nxbest.cn", "gov.zb.nxbest.cn",

from datetime import datetime
from flask_sqlalchemy import SQLAlchemy
from werkzeug.security import generate_password_hash, check_password_hash

db = SQLAlchemy()

class User(db.Model):
    """用户模型"""
    __tablename__ = 'users'
    
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=True, nullable=False)
    password_hash = db.Column(db.String(128), nullable=False)
    avatar = db.Column(db.String(255), default='/static/avatars/default.png')
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    
    # 定义密码设置方法
    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)

class Message(db.Model):
    """私信模型"""
    __tablename__ = 'messages'
    
    id = db.Column(db.Integer, primary_key=True)
    sender_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    receiver_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    content = db.Column(db.Text, nullable=False)
    is_read = db.Column(db.Boolean, default=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    deleted = db.Column(db.Boolean, default=False)
    
    # 关联关系
    sender = db.relationship('User', foreign_keys=[sender_id], backref='sent_messages')
    receiver = db.relationship('User', foreign_keys=[receiver_id], backref='received_messages')
    
    # 索引
    __table_args__ = (
        db.Index('idx_sender_receiver', 'sender_id', 'receiver_id'),
        db.Index('idx_receiver_unread', 'receiver_id', 'is_read'),
    )

2. 数据库迁移

使用 Flask-Migrate 管理数据库迁移：

"gov.wap.nxbest.cn", "LzLjhb.cn", "www.LzLjhb.cn", "m.LzLjhb.cn", "wap.LzLjhb.cn",

python

运行"gov.LzLjhb.cn", "web.LzLjhb.cn", "www.stats.gov.cn.LzLjhb.cn", "www.cq.gov.cn.LzLjhb.cn",

# app.py 中初始化迁移
from flask_migrate import Migrate

migrate = Migrate(app, db)

执行迁移命令：

bash

运行

flask db init    # 初始化迁移环境（仅首次）
flask db migrate -m "initial tables"  # 创建迁移脚本
flask db upgrade  # 应用迁移

四、核心功能实现

1. 应用初始化（app.py）

python

运行"gov.zb.LzLjhb.cn", "gov.wap.LzLjhb.cn", "gzjxhL.cn", "www.gzjxhL.cn", "m.gzjxhL.cn",

from flask import Flask, session, redirect, url_for, request, render_template, jsonify
from models import db, User, Message
import os
from functools import wraps

app = Flask(__name__)
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'dev-key-for-testing')
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///message_system.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# 初始化数据库
db.init_app(app)

# 登录装饰器
def login_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if 'user_id' not in session:
            return redirect(url_for('login', next=request.url))
        return f(*args, **kwargs)
    return decorated_function

2. 私信功能核心逻辑（message_service.py）

python

运行"wap.gzjxhL.cn", "gov.gzjxhL.cn", "web.gzjxhL.cn", "www.stats.gov.cn.gzjxhL.cn",

from models import db, Message, User
from sqlalchemy import or_, and_

class MessageService:
    @staticmethod
    def send_message(sender_id, receiver_id, content):
        """发送私信"""
        if not content.strip() or sender_id == receiver_id:
            return False
            
        message = Message(
            sender_id=sender_id,
            receiver_id=receiver_id,
            content=content.strip()
        )
        db.session.add(message)
        db.session.commit()
        return True
    
    @staticmethod
    def get_chat_history(user_id, target_user_id):
        """获取与指定用户的聊天记录"""
        # 标记未读消息为已读
        unread_messages = Message.query.filter(
            Message.sender_id == target_user_id,
            Message.receiver_id == user_id,
            Message.is_read == False
        ).all()
        
        for msg in unread_messages:
            msg.is_read = True
        db.session.commit()
        
        # 获取聊天记录
        messages = Message.query.filter(
            or_(
                and_(
                    Message.sender_id == user_id,
                    Message.receiver_id == target_user_id,
                    Message.deleted == False
                ),
                and_(
                    Message.sender_id == target_user_id,
                    Message.receiver_id == user_id,
                    Message.deleted == False
                )
            )
        ).order_by(Message.created_at).all()
        
        return messages
    
    @staticmethod
    def get_message_contacts(user_id):
        """获取用户的私信联系人列表"""
        # 子查询：获取每个联系人的最新消息
        subquery = db.session.query(
            Message.id,
            or_(
                Message.sender_id == user_id,
                Message.receiver_id == user_id
            ).label('is_involved'),
            # 确定联系人ID
            db.case(
                [(Message.sender_id == user_id, Message.receiver_id)],
                else_=Message.sender_id
            ).label('contact_id'),
            Message.created_at
        ).filter(
            or_(
                Message.sender_id == user_id,
                Message.receiver_id == user_id
            ),
            Message.deleted == False
        ).subquery()
        
        # 获取每个联系人的最新消息ID
        latest_msg_ids = db.session.query(
            subquery.c.contact_id,
            db.func.max(subquery.c.created_at).label('max_time')
        ).group_by(subquery.c.contact_id).subquery()
        
        # 关联查询获取完整信息
        contacts = db.session.query(
            User.id,
            User.username,
            User.avatar,
            Message.content,
            Message.created_at,
            # 计算未读消息数量
            db.session.query(db.func.count(Message.id)).filter(
                Message.receiver_id == user_id,
                Message.sender_id == User.id,
                Message.is_read == False,
                Message.deleted == False
            ).label('unread_count')
        ).join(
            Message, Message.id == db.session.query(
                subquery.c.id
            ).join(
                latest_msg_ids,
                and_(
                    subquery.c.contact_id == latest_msg_ids.c.contact_id,
                    subquery.c.created_at == latest_msg_ids.c.max_time
                )
            ).scalar_subquery()
        ).join(
            User, User.id == latest_msg_ids.c.contact_id
        ).order_by(Message.created_at.desc()).all()
        
        return contacts
    
    @staticmethod
    def get_unread_total(user_id):
        """获取未读消息总数"""
        return Message.query.filter(
            Message.receiver_id == user_id,
            Message.is_read == False,
            Message.deleted == False
        ).count()
    
    @staticmethod
    def delete_message(message_id, user_id):
        """删除消息（逻辑删除）"""
        message = Message.query.filter(
            Message.id == message_id,
            or_(
                Message.sender_id == user_id,
                Message.receiver_id == user_id
            )
        ).first()
        
        if not message:
            return False
            
        message.deleted = True
        db.session.commit()
        return True

3. 路由与视图（app.py 续）

python

运行

from message_service import MessageService

# 首页/登录页
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        user = User.query.filter_by(username=username).first()
        
        if user and user.check_password(password):
            session['user_id'] = user.id
            session['username'] = user.username
            return redirect(url_for('message_list'))
            
    return render_template('login.html')

# 私信列表页
@app.route('/messages')
@login_required
def message_list():
    user_id = session['user_id']
    contacts = MessageService.get_message_contacts(user_id)
    unread_total = MessageService.get_unread_total(user_id)
    return render_template('message_list.html', 
                         contacts=contacts, 
                         unread_total=unread_total)

# 聊天页面
@app.route('/chat/<int:target_id>')
@login_required
def chat(target_id):
    user_id = session['user_id']
    target_user = User.query.get_or_404(target_id)
    messages = MessageService.get_chat_history(user_id, target_id)
    return render_template('chat.html', 
                         target_user=target_user, 
                         messages=messages)

# 发送消息API
@app.route('/api/send-message', methods=['POST'])
@login_required
def send_message_api():
    data = request.json
    receiver_id = data.get('receiver_id')
    content = data.get('content')
    
    if not receiver_id or not content:
        return jsonify({'success': False, 'message': '参数不完整'})
        
    success = MessageService.send_message(
        session['user_id'], 
        receiver_id, 
        content
    )
    
    return jsonify({
        'success': success,
        'message': '发送成功' if success else '发送失败'
    })

# 删除消息API
@app.route('/api/delete-message/<int:message_id>', methods=['POST'])
@login_required
def delete_message_api(message_id):
    success = MessageService.delete_message(message_id, session['user_id'])
    return jsonify({
        'success': success,
        'message': '删除成功' if success else '删除失败'
    })

# 检查新消息API
@app.route('/api/check-new-messages/<int:target_id>')
@login_required
def check_new_messages(target_id):
    user_id = session['user_id']
    new_count = Message.query.filter(
        Message.sender_id == target_id,
        Message.receiver_id == user_id,
        Message.is_read == False,
        Message.deleted == False
    ).count()
    
    return jsonify({
        'has_new': new_count > 0,
        'count': new_count
    })

if __name__ == '__main__':
    app.run(debug=True)

4. 前端页面实现

私信列表页面（templates/message_list.html）

html

预览

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>我的私信</title>
    <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
    <style>
        .contact-item {
            padding: 15px;
            border-bottom: 1px solid #eee;
            transition: background-color 0.3s;
        }
        .contact-item:hover {
            background-color: #f8f9fa;
        }
        .unread-badge {
            background-color: #dc3545;
            color: white;
        }
    </style>
</head>
<body>
    <div class="container mt-4">
        <h3>我的私信 
            {% if unread_total > 0 %}
                <span class="badge unread-badge">{{ unread_total }}</span>
            {% endif %}
        </h3>
        
        <div class="list-group mt-3">
            {% if not contacts %}
                <div class="alert alert-info">暂无消息记录</div>
            {% else %}
                {% for contact in contacts %}
                    <a href="{{ url_for('chat', target_id=contact.id) }}" 
                       class="text-decoration-none text-dark">
                        <div class="contact-item d-flex justify-content-between align-items-center">
                            <div class="d-flex align-items-center">
                                <img src="{{ contact.avatar }}" alt="头像" 
                                     width="50" height="50" class="rounded-circle me-3">
                                <div>
                                    <div class="fw-bold">{{ contact.username }}</div>
                                    <div class="text-muted small">{{ contact.content|truncate(20) }}</div>
                                </div>
                            </div>
                            <div class="text-end">
                                <small class="text-muted">
                                    {{ contact.created_at.strftime('%m-%d %H:%M') }}
                                </small>
                                {% if contact.unread_count > 0 %}
                                    <div class="badge unread-badge mt-1">
                                        {{ contact.unread_count }}
                                    </div>
                                {% endif %}
                            </div>
                        </div>
                    </a>
                {% endfor %}
            {% endif %}
        </div>
    </div>

    <script src="https://cdn.jsdelivr.net/npm/jquery@3.6.0/dist/jquery.min.js"></script>
</body>
</html>

聊天页面（templates/chat.html）

html

预览

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>与 {{ target_user.username }} 聊天</title>
    <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
    <style>
        .chat-container {
            height: 500px;
            overflow-y: auto;
            padding: 15px;
            border: 1px solid #eee;
            border-radius: 5px;
            margin-bottom: 15px;
        }
        .message {
            margin-bottom: 15px;
            max-width: 70%;
        }
        .sent {
            margin-left: auto;
        }
        .sent .message-content {
            background-color: #0d6efd;
            color: white;
        }
        .received .message-content {
            background-color: #e9ecef;
        }
        .message-content {
            padding: 10px 15px;
            border-radius: 18px;
        }
        .message-time {
            font-size: 0.7rem;
            color: #6c757d;
            margin-top: 5px;
        }
        .sent .message-time {
            text-align: right;
        }
    </style>
</head>
<body>
    <div class="container mt-4">
        <a href="{{ url_for('message_list') }}" class="btn btn-secondary mb-3">返回私信列表</a>
        
        <div class="card">
            <div class="card-header d-flex align-items-center">
                <img src="{{ target_user.avatar }}" alt="头像" 
                     width="40" height="40" class="rounded-circle me-2">
                <h5 class="mb-0">{{ target_user.username }}</h5>
            </div>
            
            <div class="card-body">
                <div class="chat-container" id="chatContainer">
                    {% for msg in messages %}
                        <div class="message {{ 'sent' if msg.sender_id == session.user_id else 'received' }}">
                            <div class="message-content">{{ msg.content }}</div>
                            <div class="message-time">
                                {{ msg.created_at.strftime('%m-%d %H:%M') }}
                            </div>
                        </div>
                    {% endfor %}
                </div>
                
                <div class="input-group">
                    <textarea id="messageInput" class="form-control" 
                              rows="3" placeholder="输入消息内容..."></textarea>
                    <button id="sendBtn" class="btn btn-primary">发送</button>
                </div>
            </div>
        </div>
    </div>

    <script src="https://cdn.jsdelivr.net/npm/jquery@3.6.0/dist/jquery.min.js"></script>
    <script>
        const targetUserId = {{ target_user.id }};
        
        // 页面加载完成后滚动到底部
        $(document).ready(function() {
            scrollToBottom();
            
            // 发送消息事件绑定
            $('#sendBtn').click(sendMessage);
            
            // 按Enter发送消息，Shift+Enter换行
            $('#messageInput').keydown(function(e) {
                if (e.key === 'Enter' && !e.shiftKey) {
                    e.preventDefault();
                    sendMessage();
                }
            });
            
            // 定时检查新消息
            setInterval(checkNewMessages, 3000);
        });
        
        // 发送消息
        function sendMessage() {
            const content = $('#messageInput').val().trim();
            if (!content) return;
            
            $.ajax({
                url: '{{ url_for('send_message_api') }}',
                type: 'POST',
                contentType: 'application/json',
                data: JSON.stringify({
                    receiver_id: targetUserId,
                    content: content
                }),
                success: function(response) {
                    if (response.success) {
                        $('#messageInput').val('');
                        // 重新加载页面显示新消息
                        location.reload();
                    } else {
                        alert(response.message);
                    }
                }
            });
        }
        
        // 滚动到底部
        function scrollToBottom() {
            const container = $('#chatContainer');
            container.scrollTop(container[0].scrollHeight);
        }
        
        // 检查新消息
        function checkNewMessages() {
            $.get('{{ url_for('check_new_messages', target_id=target_user.id) }}',
                function(response) {
                    if (response.has_new) {
                        // 有新消息，刷新页面
                        location.reload();
                    }
                }
            );
        }
    </script>
</body>
</html>

五、功能扩展与优化建议

1. 实时通信升级

当前实现使用 AJAX 轮询检查新消息，可升级为 WebSocket 实现真正的实时通信：

使用 Flask-SocketIO 扩展
实现消息实时推送，避免频繁轮询
添加 "正在输入" 状态提示

python

运行

# WebSocket示例代码（需安装Flask-SocketIO）
from flask_socketio import SocketIO, emit

socketio = SocketIO(app)

@socketio.on('send_message')
def handle_send_message(data):
    # 处理消息发送逻辑
    # ...
    # 向接收者推送消息
    emit('new_message', message_data, room=receiver_room)