要实现一个简单的IM（即时通讯）系统，支持用户注册、登录和聊天记录存储，你可以使用Python和mysql数据库。以下是一个基本的实现示例： 要使用MySQL创建表并通过Python提供一个API服务，你可以使用Flask框架来实现API服务，并使用PyMySQL库来连接MySQL数据库。以下是一个基本的实现步骤：
1. 安装所需库
首先，确保你安装了Flask和PyMySQL库：
|
1 |
pip install flask pymysql |
2. MySQL数据库设置
|
1 |
docker run --hostname=a5ddc3708f2e --env=MYSQL_ROOT_PASSWORD=123456 --env=MYSQL_DATABASE=jwordpress --env=TZ=Asia/Shanghai --env=LANG=en_US.UTF-8 --env=PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin --env=GOSU_VERSION=1.7 --env=MYSQL_MAJOR=5.7 --env=MYSQL_VERSION=5.7.26-1debian9 --volume=D:\IdeaProjects\Jwordpress-parent-s02-81f3dea303558f4388ef39435a52ea1cfab22904\docker\mysql\my.cnf:/etc/mysql/my.cnf:rw --volume=D:\IdeaProjects\Jwordpress-parent-s02-81f3dea303558f4388ef39435a52ea1cfab22904\docker\mysql\init-file.sql:/etc/mysql/init-file.sql:rw --volume=D:\IdeaProjects\Jwordpress-parent-s02-81f3dea303558f4388ef39435a52ea1cfab22904\docker\mysql\data:/var/lib/mysql:rw --volume=D:\IdeaProjects\Jwordpress-parent-s02-81f3dea303558f4388ef39435a52ea1cfab22904\docker\mysql\docker-entrypoint-initdb.d:/docker-entrypoint-initdb.d:rw --volume=/var/lib/mysql --network=docker_default -p 3306:3306 --restart=unless-stopped --label='com.docker.compose.config-hash=f33622a4d32e092d39a39c3dc0bd2259df09b24ad897567bcaa7f7fa0630b019' --label='com.docker.compose.container-number=1' --label='com.docker.compose.depends_on=' --label='com.docker.compose.image=sha256:a1aa4f76fab910095dfcf4011f32fbe7acdb84c46bb685a8cf0a75e7d0da8f6b' --label='com.docker.compose.oneoff=False' --label='com.docker.compose.project=docker' --label='com.docker.compose.project.config_files=D:\IdeaProjects\Jwordpress-parent-s02-81f3dea303558f4388ef39435a52ea1cfab22904\docker\docker-compose.yml' --label='com.docker.compose.project.working_dir=D:\IdeaProjects\Jwordpress-parent-s02-81f3dea303558f4388ef39435a52ea1cfab22904\docker' --label='com.docker.compose.service=mysql' --label='com.docker.compose.version=2.21.0' --runtime=runc -d registry.cn-hangzhou.aliyuncs.com/zhengqing/mysql:5.7 |
假设你已经在MySQL中创建了一个数据库，接下来创建用户和消息表。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
CREATE DATABASE chat_db;
USE chat_db;
CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(255) UNIQUE NOT NULL, password VARCHAR(255) NOT NULL );
CREATE TABLE messages ( id INT AUTO_INCREMENT PRIMARY KEY, sender VARCHAR(255) NOT NULL, receiver VARCHAR(255) NOT NULL, message TEXT NOT NULL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ); |
3. 创建Flask API服务
一个简单的Flask应用，提供注册、登录和发送消息的API。 详见main.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
from flask import Flask, request, jsonify import pymysql import bcrypt
app = Flask(__name__)
# Database connection configuration db_config = { 'host': 'localhost', 'user': 'root', 'password': '123456', 'database': 'chat_db' }
def get_db_connection(): return pymysql.connect(**db_config)
@app.route('/register', methods=['POST']) def register(): data = request.json username = data['username'] password = data['password']
hashed = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())
conn = get_db_connection() cursor = conn.cursor()
try: cursor.execute("INSERT INTO users (username, password) VALUES (%s, %s)", (username, hashed)) conn.commit() return jsonify({'message': 'Registration successful!'}), 201 except pymysql.IntegrityError: return jsonify({'message': 'Username already exists!'}), 400 finally: cursor.close() conn.close()
@app.route('/login', methods=['POST']) def login(): data = request.json username = data['username'] password = data['password']
conn = get_db_connection() cursor = conn.cursor()
try: cursor.execute("SELECT password FROM users WHERE username = %s", (username,)) result = cursor.fetchone()
if result and bcrypt.checkpw(password.encode('utf-8'), result[0].encode('utf-8')): return jsonify({'message': 'Login successful!'}), 200 else: return jsonify({'message': 'Incorrect username or password!'}), 401 finally: cursor.close() conn.close()
@app.route('/send_message', methods=['POST']) def send_message(): data = request.json sender = data['sender'] receiver = data['receiver'] message = data['message']
conn = get_db_connection() cursor = conn.cursor()
cursor.execute("INSERT INTO messages (sender, receiver, message) VALUES (%s, %s, %s)", (sender, receiver, message)) conn.commit() cursor.close() conn.close()
return jsonify({'message': 'Message sent successfully!'}), 201
@app.route('/get_messages', methods=['GET']) def get_messages(): user1 = request.args.get('user1') user2 = request.args.get('user2')
conn = get_db_connection() cursor = conn.cursor()
cursor.execute('''SELECT sender, receiver, message, timestamp FROM messages WHERE (sender = %s AND receiver = %s) OR (sender = %s AND receiver = %s) ORDER BY timestamp''', (user1, user2, user2, user1))
messages = cursor.fetchall() cursor.close() conn.close()
return jsonify(messages), 200
if __name__ == '__main__': app.run(debug=True) |
4. 运行API服务
将上述代码保存为一个Python文件（例如main.py），然后运行：
|
1 |
python main.py |
这将启动一个Flask开发服务器，你可以通过POST请求来注册和登录用户，通过GET请求来获取聊天记录。
5. 测试
post http://127.0.0.1:5000/register
|
1 2 3 4 |
{ "username": "alice", "password": "password123" } |
post http://127.0.0.1:5000/register
|
1 2 3 4 |
{ "username": "bob", "password": "password123" } |
post http://127.0.0.1:5000/send_message
|
1 2 3 4 5 |
{ "sender": "alice", "receiver": "bob", "message": "Hello Bob!" } |
get http://127.0.0.1:5000/get_messages?user1=alice&user2=bob
|
1 2 |
[ [ "alice", "bob", "Hello Bob!", "Fri, 15 Nov 2024 16:06:33 GMT" ] ] |
请注意，这个示例是一个基本实现，适用于学习和测试。在生产环境中，你需要考虑更多的安全性和性能优化，例如使用HTTPS、添加身份验证令牌等。
