Python简易IM系统：Socket+CSV实现即时通信教学骨架

简介：本资源是一套基于Python开发的简易微信系统课程设计源码，面向计算机专业本科生及Python初学者，用于理解客户端-服务器架构、内存数据库管理与基础网络通信原理。压缩包共20个文件，含7个核心Python源码（如server.py、client.py、database.py等）、4个CSV数据表（user.csv、friend.csv等）、3个编译后pyc文件、1个Word课程报告、1个README说明文档及1个可执行客户端exe，整体仅598KB，轻量易部署。已有306人学习下载，适合课程设计实践、毕设参考或分布式系统入门教学。读者可完整获得从服务端启动/关闭机制（通过system.py调用）、三层数据库操作封装（内存层→功能层→指令层），到GUI界面、消息存储与好友关系管理的全链路实现逻辑，并附带图文并茂的课程报告与清晰目录结构，便于快速理解模块职责与调试路径。

1. 这不是微信客户端，而是一套可调试、可拆解的即时通信教学骨架

你双击wechat.exe看到登录界面，输入账号密码能发消息、加好友、查历史记录——但它既不连腾讯服务器，也不走微信协议。这是一个用纯 Python 实现的本地局域网级简易微信系统，核心逻辑全部摊开在server.py、client.py、database.py和GUI.py里。它不追求高并发或加密传输，而是把「用户注册→好友关系建立→消息收发→离线存储」这整条链路，用最直白的内存操作+CSV 文件持久化+TCP socket 通信串起来。适合课程设计学生快速理解 IM 系统分层结构，也适合 Python 初学者拿来做 socket 编程、GUI 事件驱动、文件 I/O 和简单状态管理的综合练手项目。所有模块耦合度低，database.py只管增删查，server.py只管指令路由，GUI.py只响应按钮点击——改一行代码就能看到效果，而不是面对几百行黑盒框架无从下手。

2. 三层数据库架构：从 CSV 文件读取到内存缓存再到业务封装

2.1 底层数据表操作：CSV 文件即数据库，database.py是唯一入口

整个系统没有使用 SQLite 或 MySQL，而是将user.csv、friend.csv、message.csv三张表作为持久化载体。database.py封装了所有底层操作，关键函数包括：

# database.py def load_csv(filename): """从CSV文件加载数据到内存列表，返回二维列表""" if not os.path.exists(filename): return [] with open(filename, 'r', encoding='utf-8') as f: reader = csv.reader(f) return list(reader) def save_csv(filename, data): """将内存中的二维列表写入CSV文件""" with open(filename, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerows(data)

提示：load_csv()在服务器启动时被server.py调用一次，将全部 CSV 内容载入内存；后续所有查询、插入、删除均在内存中进行，save_csv()仅在关键操作（如注册成功、发送消息）后触发落盘。这种“内存优先+异步落盘”策略避免了每次操作都读写磁盘，但需注意程序异常退出会导致最新变更丢失——课程设计中可接受，生产环境必须加事务或改用 SQLite。

2.1.1 表结构与字段含义必须对齐

三张 CSV 文件采用固定列顺序，无表头，靠位置索引访问字段：

注意：message.csv中is_read=0表示未读，客户端收到消息后需主动更新该字段为1；friend.csv中(a,b)和(b,a)视为同一关系，代码中需双向判断，否则会出现“A 加了 B，但 B 好友列表里没 A”的逻辑漏洞。

2.2 中间层封装：规避空指针与越界访问的健壮性包装

database.py提供的get_user_by_name(username)、add_friend(user_a, user_b)等函数，并非直接调用load_csv()，而是先做存在性校验再执行操作。例如：

# database.py def get_user_by_name(username): users = load_csv('user.csv') for row in users: if len(row) >= 2 and row[0] == username: # 防止空行或字段不足导致 index error return row return None # 显式返回 None，而非抛异常，上层可安全判断 def add_friend(user_a, user_b): friends = load_csv('friend.csv') # 检查是否已存在该关系（双向） exists = any((row[0] == user_a and row[1] == user_b) or (row[0] == user_b and row[1] == user_a) for row in friends if len(row) >= 2) if not exists: friends.append([user_a, user_b, '1']) save_csv('friend.csv', friends) return True return False

2.2.1 为什么必须做len(row) >= 2判断？

原始 CSV 文件可能因手动编辑产生空行或字段缺失（如zhangsan,,张三），若直接row[1]访问会触发IndexError。课程设计中常见错误是忽略 CSV 解析的鲁棒性，导致服务器一收到异常格式数据就崩溃。此处用len(row) >= N显式兜底，比try-except更利于定位问题源头。

2.3 顶层指令映射：server.py如何将 socket 指令转为数据库操作

服务器监听 TCP 端口（默认8080），接收客户端发来的纯文本指令，格式为CMD|arg1|arg2|...。server.py的核心循环如下：

# server.py while True: conn, addr = server_socket.accept() data = conn.recv(1024).decode('utf-8').strip() if not data: continue parts = data.split('|') cmd = parts[0] if cmd == 'REGISTER': username, password, nickname = parts[1], parts[2], parts[3] result = db.register_user(username, password, nickname) # 调用中间层封装 conn.send(f"RESULT|{result}".encode('utf-8')) elif cmd == 'LOGIN': username, password = parts[1], parts[2] result = db.login_user(username, password) # 封装了密码比对和状态更新 conn.send(f"RESULT|{result}".encode('utf-8')) elif cmd == 'SEND_MSG': sender, receiver, content = parts[1], parts[2], parts[3] db.save_message(sender, receiver, content) # 自动添加时间戳并落盘 conn.send(b"RESULT|OK")

参数说明：parts[0]是命令名，parts[1:]是参数。所有指令均以|分隔，避免空格歧义；SEND_MSG不返回消息内容，只确认接收成功，防止敏感信息泄露；RESULT|OK是约定响应格式，客户端client.py严格按此解析。

3. 客户端 GUI 与网络通信：PyQt5 实现的事件驱动交互流

3.1 GUI 层结构：GUI.py与client.py的职责分离

GUI.py仅负责界面渲染与事件绑定，所有业务逻辑（登录、发消息、刷新好友列表）由client.py处理。这种分离让界面可替换（比如换成 Tkinter）而不影响通信逻辑。主窗口类WeChatWindow继承自QMainWindow，关键控件包括：

• QLineEdit：用户名、密码、消息输入框
• QPushButton：登录、发送、添加好友
• QListWidget：在线好友列表（实时更新）
• QTextEdit：消息显示区域（只读）

事件绑定示例：

# GUI.py self.login_btn.clicked.connect(self.on_login_click) # 点击登录按钮 → 触发 on_login_click 方法 self.send_btn.clicked.connect(self.on_send_click) def on_login_click(self): username = self.username_input.text().strip() password = self.password_input.text().strip() if not username or not password: self.show_error("用户名和密码不能为空") return # 调用 client.py 的 login 方法，传入回调函数处理响应 client.login(username, password, self.on_login_success, self.on_login_fail)

3.1.1 回调函数机制：避免 GUI 线程阻塞

client.py的login()函数内部启动独立线程执行 socket 连接与通信，防止conn.recv()阻塞主线程导致界面冻结：

# client.py def login(username, password, success_cb, fail_cb): def _do_login(): try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(('127.0.0.1', 8080)) # 默认连接本地服务器 sock.send(f"LOGIN|{username}|{password}".encode('utf-8')) response = sock.recv(1024).decode('utf-8') if response.startswith("RESULT|OK"): success_cb(username) # 在主线程执行回调 else: fail_cb("登录失败：" + response.split('|', 1)[1]) except Exception as e: fail_cb(f"连接失败：{str(e)}") finally: sock.close() threading.Thread(target=_do_login, daemon=True).start()

注意：daemon=True确保线程随主程序退出而终止；success_cb和fail_cb是 GUI 层传入的函数对象，用于在登录成功后跳转主界面或弹出错误提示，这是 PyQt5 中处理异步操作的标准做法。

3.2 消息收发流程：从输入框到 CSV 文件的完整链路

用户在输入框键入文字，点击“发送”后，经历以下步骤：

1. GUI.py获取receiver（当前选中好友）和content（输入框文本）
2. client.py构造SEND_MSG|zhangsan|lisi|你好啊并发送至服务器
3. server.py解析指令，调用db.save_message('zhangsan', 'lisi', '你好啊')
4. database.py在内存中追加新消息行，并调用save_csv('message.csv', ...)持久化
5. 服务器返回RESULT|OK，客户端收到后清空输入框
6. 关键点：消息仅存于message.csv，不推送给接收方！接收方需主动轮询（每 2 秒发一次GET_MSGS|lisi指令）获取新消息

# client.py - 轮询逻辑 def start_polling(receiver): def _poll(): while True: try: sock = socket.socket() sock.connect(('127.0.0.1', 8080)) sock.send(f"GET_MSGS|{receiver}".encode('utf-8')) resp = sock.recv(4096).decode('utf-8') if resp.startswith("MESSAGES|"): msgs = resp[9:].split('||') # 消息用 || 分隔 for msg in msgs: if msg.strip(): # 在 GUI 主线程中更新消息显示 QApplication.postEvent( gui_window, UpdateMessageEvent(msg) ) time.sleep(2) # 每2秒轮询一次 except: break threading.Thread(target=_poll, daemon=True).start()

参数说明：GET_MSGS|lisi返回格式为MESSAGES|zhangsan|lisi|你好啊|2024-06-15 10:22:31|0||zhangsan|lisi|收到|2024-06-15 10:23:05|0，客户端按||拆分后，再按|解析每条消息的发送者、接收者、内容、时间、已读状态。

4. 服务端启停与调试：绕过 Windows 任务管理器的精准控制方案

4.1 启动服务器：system.py的隐藏开关与进程隔离

system.py是一个轻量级控制脚本，其核心是调用 Windowsstart /b python server.py后台启动服务端进程。但默认状态下，关键行被注释：

# system.py # print(system('start /b python server.py')) # ← 此行需取消注释才能启动 # print(system('taskkill /im python.exe /f')) # ← 此行需取消注释才能强制关闭

提示：start /b启动的进程不创建新终端窗口，避免干扰；/b参数确保后台运行，否则server.py会独占命令行窗口。若需查看日志，可临时改为start python server.py，此时会弹出独立 CMD 窗口显示print()输出。

4.1.1 如何验证服务器是否真正运行？

不要依赖窗口是否存在，而应检查端口占用：

netstat -ano | findstr :8080

若返回类似TCP 0.0.0.0:8080 0.0.0.0:0 LISTENING 12345，则PID 12345对应的进程即为server.py。可通过tasklist | findstr 12345确认进程名。

4.2 关闭服务器：taskkill的精确匹配与风险规避

直接taskkill /im python.exe /f会杀死所有 Python 进程，包括你正在调试的client.py或database.py。更安全的做法是限定进程名：

# system.py - 改进版关闭逻辑 import os os.system('taskkill /f /fi "imagename eq python.exe" /fi "services eq "')

但本项目更推荐在server.py中加入优雅退出机制：监听特定指令（如SHUTDOWN），收到后执行save_csv()并break循环。修改server.py主循环：

# server.py while True: conn, addr = server_socket.accept() data = conn.recv(1024).decode('utf-8').strip() if data == 'SHUTDOWN': # 新增指令 print("收到关机指令，正在保存数据...") db.save_all() # 假设 database.py 提供此方法 break # ...原有指令处理逻辑 server_socket.close() print("服务器已停止")

然后在system.py中发送关机指令：

# system.py - 安全关闭方式 def shutdown_server(): try: sock = socket.socket() sock.connect(('127.0.0.1', 8080)) sock.send(b'SHUTDOWN') sock.close() print("服务器已收到关机指令") except: print("无法连接服务器，尝试强制关闭...") os.system('taskkill /f /im python.exe /fi "pid gt 0"') # 调用 shutdown_server() 替代原 taskkill 命令

4.3 调试技巧：用print()定位数据流向断点

当消息发不出或好友列表为空时，不要盲目重启，按以下顺序加print()：

1. 在client.py的send_message()开头打印f"准备发送：{cmd}"
2. 在server.py的recv()后打印f"收到指令：{data}"
3. 在database.py的save_message()中打印f"写入消息到CSV：{row}"
4. 在GUI.py的on_login_success()中打印f"登录成功，当前用户：{username}"

关键观察点：若第 1 步有输出但第 2 步无输出，说明客户端未连上服务器（检查 IP/端口）；若第 2 步有输出但第 3 步无输出，说明指令解析失败（检查|分隔符是否被误删）；若第 3 步有输出但message.csv无新增，说明save_csv()路径错误（确认工作目录是项目根目录）。

5. 课程设计落地技巧：三处可立即加分的改造点

5.1 添加消息时间戳格式化：从2024-06-15 10:22:31到10:22

message.csv存储的是datetime.now().strftime('%Y-%m-%d %H:%M:%S')，但 GUI 显示时应精简。在GUI.py解析消息后，对时间字段做二次处理：

# GUI.py - 消息显示前处理 def format_time(timestamp_str): try: dt = datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S') return dt.strftime('%H:%M') # 只显示时:分 except: return timestamp_str # 解析失败则原样返回 # 在更新消息文本时调用 msg_text = f"[{format_time(time_field)}] {sender}: {content}"

效果：避免消息区域被长日期挤占，符合主流 IM 界面习惯；try-except兜底防止 CSV 中时间格式异常导致整个消息渲染失败。

5.2 实现“已读回执”：客户端主动标记is_read=1

当前message.csv中is_read字段始终为0，无法体现消息是否被对方查看。可在client.py的消息接收逻辑中，对自己作为接收者的消息批量更新状态：

# client.py - 接收消息后立即标记已读 def mark_messages_as_read(receiver): messages = db.load_csv('message.csv') updated = False for row in messages: if len(row) >= 5 and row[1] == receiver and row[4] == '0': # receiver 匹配且未读 row[4] = '1' # 标记为已读 updated = True if updated: db.save_csv('message.csv', messages)

调用时机：在UpdateMessageEvent处理完新消息后立即执行mark_messages_as_read(current_user)。这样下次启动客户端时，历史消息将正确显示“已读”。

5.3 修复好友列表重复：friend.csv双向去重查询

get_friends(username)函数当前只查user_a == username，漏掉user_b == username的情况。修正如下：

# database.py def get_friends(username): friends = load_csv('friend.csv') result = set() # 用 set 去重 for row in friends: if len(row) >= 2: if row[0] == username and row[2] == '1': # user_a 是自己 result.add(row[1]) elif row[1] == username and row[2] == '1': # user_b 是自己 result.add(row[0]) return list(result)

为什么用set？避免同一好友因多次添加操作（如 A 加 B 后 B 又加 A）在列表中出现两次；list(result)转回列表供QListWidget.addItems()使用。此修复能让课程报告中“好友关系管理”章节的截图更严谨。

实际部署时，只需在server.py启动前手动清空user.csv、friend.csv、message.csv三文件，即可获得干净测试环境。所有功能均不依赖外部服务，一台装有 Python 3.6+ 和 PyQt5 的 Windows 机器即可完整运行。

本文还有配套的精品资源，点击获取