1、前言
首先，关于源码的获取，本人提供了三种方式：
- 直接从文章里面Ctrl+C，Ctrl+V，然后按照我已给的文件结构搞一下即可；
- 通过积分下载上传到CSDN的资源；
- 点开本人的主页，点击“查看详细资料”，添加好友获取源码文件（如果有问题同样可以通过这里问），本人承诺无特殊情况，三小时内将无条件提供源码（所谓特殊情况仅指时间上的，毕竟挂上去的是我的副号，获取信息可能不及时，见谅）。
注：关于第二个获取方式……实际上用第三种方式就行，和第二种方式获得的文件没任何区别，再不济可以用第一种，真的真的没必要用第二种（但是做慈善的话我也欢迎，毕竟我也需要一点积分用来下载资源）
万万没想到啊，我大学以来虽然上了Python的课程，反而我写的体量稍大的程序都不属于Python课程的，而是分别归属于数据库原理和计算机网络（即本文所述程序）。这确实是我未曾设想的道路。不过虽然是意料之外，却也是情理之中。单单一个Python如果不配合应用场景，使用需求，也写不出什么能提高综合代码水平的稍复杂的程序。
故，本程序融合了Socket编程（TCP与UDP）与Python-MySQL连接，包含登录检查，注册检查，游戏自动匹配等功能，分为两大板块四个Servlet程序（两个客户端程序，两个服务端程序）。
2、系统实机演示
实机演示视频
3、系统分析与设计
（1）主要软件与工具
工具：HUAWEI MATEBOOK D14（Windows 10）
软件：PyCharm Community Edition 2021.3.2，MySQL 8.0.28
（2）图片素材
直接插入图片有水印……我这边就直接直接放到CSDN资源里面去了（是免积分的，放心下载，我心没那么黑）。里面包含黑子白子和棋盘块三个素材。
（3）系统需求分析
对于一个简易的基于Socket的联机五子棋，其基本功能包括登录注册部分、匹配系统和游戏部分。其中：
- 登录注册需求包括：登录验证，注册信息验证，注册信息写入，向服务器传递本机IP等；
- 匹配系统需求包括：自动对申请游戏的玩家进行匹配；
- 游戏部分的需求包括：回合制下棋，判断胜负，相关提示信息。
（4）系统设计与响应流程
- 数据库部分：需要在MySQL中建立用于存储用户登录注册信息的表（table user），和用于游戏匹配队列的表（table ip）。此处由于之前存在一个dbms_analyze库及其中的user表，因此未重新建库建表；
- 主客户端（ClientSignIn.py）：主客户端主要用于构建前端界面、进行登录和注册的基本验证（非空验证和格式校验），以及根据服务器的反馈完成其他校验并给予反馈。同时在完成登录后向服务器提交IP地址，并根据服务器给出的反馈充当游戏的服务端或客户端角色；
- 主服务器（ServerSign.py）：主服务器主要用于连接数据库进行操作，负责进行登录资质校验并给予反馈、注册并给予反馈、检测游戏登录队列并进行反馈；
- 戏客户端（ClientGobang.py）与游戏服务端（ServerGobang.py）：此处的服务端和客户端与之前的服务端和客户端在运行时是互相独立的，服务端和客户端之间是直接连接的，数据并不通过主服务器中转，主服务器仅提供匹配服务；
- C2.py：与主客户端完全一致，无实际意义，仅作为测试用。
如下为主程序响应流程：
| ClientSignIn.py | ServerSign.py |
| 启动并启动监听 | |
| 启动并弹出页面 | 等待 |
| （可选）启动注册 | 等待 |
| 输入信息并点击注册 | 等待 |
| 初次校验格式和非空检查、密码一致性检查 | 等待 |
| 建立套接字，建立连接 | 等待 |
| 拼接为字符串，加入标签后传递至主服务器 | 等待 |
| 等待 | 主服务器接收信息并分割字符串 |
| 等待 | 根据标签启动注册写入函数 |
| 等待 | 根据写入是否成功判定逻辑值 |
| 等待 | 将逻辑值转化为0和1标签并发送回主客户端 |
| 接收标签 | 等待 |
| 关闭套接字 | 等待 |
| 根据标签给出注册成功或用户已存在提示 | 等待 |
| （可选）主界面登录 | 等待 |
| 输入信息并点击登录 | 等待 |
| 进行非空检查 | 等待 |
| 建立套接字，建立连接 | 等待 |
| 拼接为字符串，加入标签后传递至主服务器 | 等待 |
| 等待 | 主服务器接收信息并分割字符串 |
| 等待 | 根据标签启动登录校验函数 |
| 等待 | 进行查询，检查是否能返回相应结果，并判定逻辑值 |
| 等待 | 将逻辑值转化为0和1标签并发送回主客户端 |
| 接收标签 | 等待 |
| 关闭套接字 | 等待 |
| 根据标签给出欢迎或密码账号错误提示 | 等待 |
| 若登陆成功： | 等待 |
| 建立套接字，建立连接 | 等待 |
| 获取本机局域网IP | 等待 |
| 将用户名和IP拼接为字符串，加入标签后传递至服务器 | 等待 |
| 等待 | 主服务器接收信息并分割字符串 |
| 等待 | 根据标签启动队列判断函数 |
| 等待 | 队列判断函数：若之前ip表中不存在有匹配记录，则向其中写入接收到的用户名和IP，若之前的ip表中存在有匹配记录，则不进行任何数据库操作，根据上述结果判定逻辑值 |
| 等待 | 根据逻辑值，若有匹配记录，则获取匹配记录的IP并发回，然后从库中删除已有匹配记录；若没有匹配记录，发回字符串0，等待3分钟后也删除匹配记录（都匹配了三分钟了还没匹配上，不得不说该重新匹配了，肯定不是程序的问题，一定是网络的问题） |
| 接收信息 | 等待 |
| 关闭套接字 | 等待 |
| 若信息为0则直接启动游戏服务端函数，IP为本机IP，端口440；若信息为已有游戏服务端IP，则启动游戏客户端函数，直连已存在的游戏服务端 | 等待 |
如下为游戏响应流程：
| ClientGobang.py | ServerGobang.py |
| 启动并建立棋盘，开始监听 | |
| 启动并建立棋盘，建立连接 | |
| 落子，向游戏客户端发送已落子的坐标 | |
| 接收坐标 | 判定是否胜利（主要是判断是否胜利，此时服务端已落子，因此更新的是己方棋子，只有可能自己胜利或未判定） |
| 判定是否失败（主要是判断是否失败，因为胜利或失败的信息并不发送，而是单独判断，而此时客户端未落子，因此判断的是己方是否失败） | 未判定显示等待对方下棋，胜利则显示“胜利”并终止游戏 |
| 为判定显示己方下棋，失败则显示 | |
| 落子，向游戏服务端发送已落子坐标 | |
| 判定是否胜利（主要是判断是否胜利，此时客户端已落子，因此更新的是己方棋子，只有可能自己胜利或未判定） | …… |
| 未判定显示等待对方下棋，胜利则显示“胜利”并终止游戏 | …… |
| …… | …… |
| 若断开连接 | |
| 消息发送出现问题，显示对方掉线，终止游戏 |
4、代码与关键注释、文件简析
数据库部分由于使用了之前创建的一个用户和用户表，因此此处不再详述，可参此篇；
CilentSignIn.py：
"""
-*- coding: utf-8 -*-
@File : ClientSignIn.py
@author: 刘子忻
@CSDN : 山河之书Liu_Zixin
@Time : 2022/12/31 10:19
"""
import re
import socket as sk
import tkinter as tk
from tkinter import messagebox
import Gobang.ServerGobang
import Gobang.ClientGobang
server_ip = "192.168.3.18" # 服务器ip地址
server_port = 8888 # 服务器端口号
# 配置主窗口
window = tk.Tk() # 新建主窗口
window.title("Gobang -- Sign In") # 设置主窗口名称
window.geometry("450x300") # 设置主窗口大小
tk.Label(window, text="WELCOME TO GOBANG GAME").place(x=135, y=60) # 位于窗口的一些文字
# 设置User输入系
username = tk.Label(window, text="User:") # 输入提示
username.place(x=100, y=150) # 设置位置
username_str = tk.StringVar() # 设置读取文字
username_input = tk.Entry(window, width=20, textvariable=username_str) # 设置输入框，宽20
username_input.place(x=180, y=150) # 设置输入框位置
# 设置Password输入系
password = tk.Label(window, text="Password:")
password.place(x=100, y=190)
password_str = tk.StringVar()
password_input = tk.Entry(window, width=20, textvariable=password_str, show="*") # 此输入框输入的内容会被*替代
password_input.place(x=180, y=190)
# 登录函数
def SignIn():
username_info = username_input.get() # 获取前端传来的用户名和密码
password_info = password_input.get()
if username_info == "" or password_info == "": # 非空校验
tk.messagebox.showerror("ACCESS DENIED", "User name and password should not be empty.") # 账号密码不能为空提示
else:
client_socket = sk.socket(sk.AF_INET, sk.SOCK_STREAM) # 新建套接字
client_socket.connect((server_ip, server_port)) # 新建连接
info = "0 " + username_info + " " + password_info # 拼接字符串，利用空格作为分割，0是标记此信息来源的标签
client_socket.sendall(bytes(info, encoding="utf-8")) # 发送信息
# waiting......
# 等待回复
info_back = client_socket.recv(1024).decode() # 获取回复信息
client_socket.close() # 关闭套接字
if info_back == "0": # 返回信息如果为0（也就是不匹配，没有对应的用户名密码对）
tk.messagebox.showerror("ACCESS DENIED", "Wrong user name or password.") # 账号密码错误提示
else:
tk.messagebox.showinfo("ACCESS PERMITTED", "Welcome!") # 欢迎（必须关掉这个欢迎后才能启动下面的）
ipv4 = sk.gethostbyname(sk.gethostname()) # 获取本机局域网IP地址
ip_socket = sk.socket(sk.AF_INET, sk.SOCK_STREAM)
ip_socket.connect((server_ip, server_port))
info_ip = "2 " + username_info + " " + ipv4 # 此处使用的信息来源标签是2
ip_socket.sendall(bytes(info_ip, encoding="utf8"))
info_ip_back = ip_socket.recv(1024).decode()
ip_socket.close()
Quit() # 关闭主窗口，顺便退出登录
if info_ip_back == "0": # 返回的信息如果是0，就表明本机将作为游戏服务端，因此启动服务端代码，使用本机IP
Gobang.ServerGobang.ServerGobang(ipv4, 400)
else: # 如果不是（即返回数据库中存储的已有游戏服务端的IP地址），说明本机将作为客户端依据给出的IP地址直连作为游戏服务端的另一台主机
Gobang.ClientGobang.ClientGobang(info_ip_back, 400)
def SignUp():
def SaveUser():
username_info = username_new_input.get() # 获取前端传来的用户名、密码和确认密码
password_info = password_new_input.get()
password_confirm_info = password_confirm_input.get()
pattern_password = r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)[0-9a-zA-Z]{8,}$" # 匹配8位以上包含大小写字母和数字的密码
result_password = re.match(pattern_password, password_info) # 进行匹配，匹配失败返回None
if username_info == "" or password_info == "" or password_confirm_info == "": # 非空检验
tk.messagebox.showerror("ERROR", "All information should not be empty.")
elif result_password is None: # 如果密码格式匹配失败
tk.messagebox.showerror("ERROR", "Password must contain more than 8 characters, \n"
"including numbers, upper and lower case letters.") # 弹出密码格式匹配失败提示
elif password_info != password_confirm_info: # 密码和确认密码不一致
tk.messagebox.showerror("ERROR", "Inconsistent password.") # 弹出密码不一致错误提示
else: # 通过初步校验后
client_socket = sk.socket(sk.AF_INET, sk.SOCK_STREAM) # 同上之理建立socket通讯
client_socket.connect((server_ip, server_port))
info = "1 " + username_info + " " + password_info # 此处使用的信息来源标签是1，字符串不需要拼接确认密码
client_socket.sendall(bytes(info, encoding="utf-8"))
info_back = client_socket.recv(1024).decode()
client_socket.close()
if info_back == "0": # 如果信息为0（也就是出现违反完整性的操作，此处只可能是存在主键出现重复值，也就是用户存在）
tk.messagebox.showerror("ERROR", "Existed User.")
else:
tk.messagebox.showinfo("SUCCESS", "Sign up successfully")
window_sign_up.destroy() # 注册成功则关闭注册窗口
window_sign_up = tk.Toplevel(window) # 新建注册窗口，如下操作与建立主窗口类似
window_sign_up.geometry("350x200")
window_sign_up.title("Sign Up")
# 设置新用户输入系
username_new = tk.Label(window_sign_up, text="user:")
username_new.place(x=10, y=10)
username_new_str = tk.StringVar()
username_new_input = tk.Entry(window_sign_up, width=20, textvariable=username_new_str)
username_new_input.place(x=150, y=10)
# 设置密码输入系
password_new = tk.Label(window_sign_up, text="password:")
password_new.place(x=10, y=40)
password_new_str = tk.StringVar()
password_new_input = tk.Entry(window_sign_up, width=20, textvariable=password_new_str, show="*")
password_new_input.place(x=150, y=40)
# 设置确认密码输入系
password_confirm = tk.Label(window_sign_up, text="confirm password:")
password_confirm.place(x=10, y=70)
password_confirm_str = tk.StringVar()
password_confirm_input = tk.Entry(window_sign_up, width=20, textvariable=password_confirm_str, show="*")
password_confirm_input.place(x=150, y=70)
tk.Label(window_sign_up, text="Password must contain more than 8 characters, \n"
"including numbers, upper and lower case letters").place(x=10, y=100)
# 确认注册按钮
confirm_sign_up = tk.Button(window_sign_up, text="Sign Up", command=SaveUser)
confirm_sign_up.place(x=150, y=150)
# 关闭主窗口（同时结束运行，退出登录）
def Quit():
window.destroy()
login = tk.Button(window, text="Sign In", command=SignIn) # 设置登录按键
login.place(x=140, y=230) # 设置登录按键位置
logup = tk.Button(window, text="Sign Up", command=SignUp) # 设置注册按键
logup.place(x=210, y=230) # 设置注册按键位置
logdown = tk.Button(window, text="Quit", command=Quit) # 设置退出按键
logdown.place(x=300, y=230) # 设置退出按键
window.mainloop() # 主循环
ServerSign.py：
"""
-*- coding: utf-8 -*-
@File : ServerSign.py
@author: 刘子忻
@CSDN : 山河之书Liu_Zixin
@Time : 2022/12/31 12:02
"""
import pymysql as py
import socket as sk
# 用于登录资质校验的函数
def SignIn():
conn_db = py.connect(host="127.0.0.1", user="lzx", password="lzx", port=3306, database="dbms_report",
charset="utf8") # 配置数据库连接
cursor = conn_db.cursor() # 游标
sql = "select * from user where user_name='%s' and user_password='%s'" % (list_info[1], list_info[2]) # SQL语句
cursor.execute(sql) # 执行SQL语句
result_set = cursor.fetchall() # 捕获查询结果
flag = False # 用于判定的逻辑值
if result_set == (): # 如果检索结果为空（即没有对应的匹配对象），则更改逻辑值
flag = True
conn_db.commit() # commit事务
conn_db.close() # 关闭连接
return flag # 返回逻辑值
# 用于写入注册信息的函数（顺便校验是否存在重名）
def SignUp():
conn_db = py.connect(host="127.0.0.1", user="lzx", password="lzx", port=3306, database="dbms_report",
charset="utf8")
cursor = conn_db.cursor()
sql = "insert into user(user_name,user_password,user_e_mail,province_name,score) " \
"values('%s','%s','','山东省',0)" % (list_info[1], list_info[2])
try:
cursor.execute(sql)
flag = False
except py.err.IntegrityError: # 如果出现完整性错误，说明存在重名（主键）
flag = True
conn_db.commit() # 此处的事务commit十分重要，不然前面的SQL执行了也白执行
conn_db.close()
return flag
def Confirm():
conn_db = py.connect(host="127.0.0.1", user="lzx", password="lzx", port=3306, database="dbms_report",
charset="utf8")
cursor = conn_db.cursor()
sql_1 = "select * from ip" # 第一步先将ip表中的信息提出来
cursor.execute(sql_1)
result_set = cursor.fetchall()
conn_db.commit()
if result_set == (): # 为空数组则表名这是之前没有未匹配的主机存在
sql_2 = "insert into ip(user_name,ip_address) values('%s','%s')" % (list_info[1], list_info[2])
cursor.execute(sql_2)
conn_db.commit() # 将匹配信息写入数据
conn_db.close()
flag = True
else:
conn_db.close()
flag = False
return flag
while True:
socket_server = sk.socket()
socket_server.setsockopt(sk.SOL_SOCKET, sk.SO_REUSEADDR, 1)
socket_server.bind(("192.168.3.18", 8888))
socket_server.listen(10)
conn, addr = socket_server.accept()
info_received = conn.recv(1024).decode()
list_info = info_received.split(" ") # 把函数写在前面就是不想一直传参，容易出现错误
identifier = list_info[0] # 之前传来的消息中都存在标签，读取标签即可启动相应的函数
if identifier == "0":
if SignIn():
conn.send(bytes("0", encoding="utf8")) # 根据不同的响应状态发送标签
else:
conn.send(bytes("1", encoding="utf8"))
conn.close()
elif identifier == "1":
if SignUp():
conn.send(bytes("0", encoding="utf8"))
else:
conn.send(bytes("1", encoding="utf8"))
conn.close()
elif identifier == "2": # 标签为2，代表发送的信息是用于匹配的
if Confirm(): # 如果之前没有正在匹配的服务端，则返回True
conn.send(bytes("0", encoding="utf8")) # 直接发回标签，主客户端启动游戏服务端
conn.close()
else: # 如果游戏服务端已存在
conn_db2 = py.connect(host="127.0.0.1", user="lzx", password="lzx", port=3306, database="dbms_report",
charset="utf8")
cursor2 = conn_db2.cursor()
sql_3 = "select * from ip" # 读取出游戏服务端的信息
cursor2.execute(sql_3)
result_set2 = cursor2.fetchall()
conn_db2.commit()
ip_info = result_set2[0][1] # 读取出游戏服务端IP，然后发回，用于启动客户端
conn.send(bytes(ip_info, encoding="utf8")) # 完成匹配自然要把原有的服务端信息清除
sql_4 = "truncate table ip"
cursor2.execute(sql_4)
conn_db2.commit()
conn_db2.close()
conn.close()
ClientGobang.py：
"""
-*- coding: utf-8 -*-
@File : ClientGobang.py
@author: 刘子忻
@CSDN : 山河之书Liu_Zixin
@Time : 2023/01/02 1:39
"""
import pygame
import sys
from pygame.locals import *
from collections import Counter
import json
import select
import socket
def ClientGobang(ip, port):
# 界面初始化
screen = pygame.display.set_mode((1200, 720))
pygame.display.set_caption("Gobang--Client")
pygame.init()
# 图片导入
img_board = pygame.image.load('chess_board.png')
img_bchess = pygame.image.load('black_chess.jpg')
img_wchess = pygame.image.load('white_chess.jpg')
# 颜色
white = (255, 255, 255)
black = (0, 0, 0)
# 用于传送的数据
msg = []
# 棋盘定义
chess_board = [[]]
def SetChessBoard():
x, y = 0, 0
while True:
if x == 1200:
x = 0
y += 40
if y < 720:
chess_board.append([])
if y == 720:
break
chess_board[-1].append([x, y])
x += 40
SetChessBoard()
# 棋盘格子是否落子
chess_exist = [[0 for _ in range(30)] for _ in range(18)]
# 黑白棋子初始化
black_chess, white_chess = [], []
wcx, wcy, bcx, bcy = [], [], [], [] # white_chess_x
def DrawBoard():
for i in chess_board:
for j in i:
screen.blit(img_board, (j[0], j[1]))
pygame.display.update()
# 默认棋子类型为0
def SetChess():
if event.type == MOUSEBUTTONDOWN:
pos = pygame.mouse.get_pos()
for i in range(len(chess_board)):
for j in range(len(chess_board[i])):
if chess_board[i][j][0] < pos[0] < chess_board[i][j][0] + 40 and chess_board[i][j][1] < pos[1] < \
chess_board[i][j][1] + 40:
if chess_exist[i][j] == 0:
white_chess.append([i, j])
wcx.append(white_chess[-1][0])
wcy.append(white_chess[-1][1])
msg.extend((i, j))
chess_exist[i][j] = 1
pygame.display.update()
return 1
def DrawChess():
for i in white_chess:
screen.blit(img_wchess, (i[1] * 40, i[0] * 40))
for i in black_chess:
screen.blit(img_bchess, (i[1] * 40, i[0] * 40))
pygame.display.update()
def RowColumnWin(x, m, n, chess):
for i in x:
if x[i] >= 5:
xy = []
for j in chess:
if j[m] == i:
xy.append(j[n])
xy.sort()
count = 0
for j in range(len(xy) - 1):
if xy[j] + 1 == xy[j + 1]:
count += 1
else:
count = 0
if count >= 4:
return 1
def DiagonalWin(chess):
x, y = [], []
chess.sort()
for i in chess:
x.append(i[0])
y.append(i[1])
c, first, last = 0, 0, 0
for i in range(len(x) - 1):
if x[i + 1] != x[i]:
if x[i] + 1 == x[i + 1]:
c += 1
last = i + 1
else:
if c < 4:
first = i + 1
c = 0
else:
last = i
print(last)
break
else:
last = i + 1
if c >= 4:
dis = []
for i in range(first, last + 1):
dis.append(x[i] - y[i])
count = Counter(dis)
for i in count:
if count[i] >= 5:
return 1
dis = []
x2 = [i * (-1) for i in x]
for i in range(first, last + 1):
dis.append(x2[i] - y[i])
count = Counter(dis)
for i in count:
if count[i] >= 5:
return 1
def GameOver():
wcx_count, wcy_count, bcx_count, bcy_count = Counter(wcx), Counter(wcy), Counter(bcx), Counter(bcy)
if RowColumnWin(wcx_count, 0, 1, white_chess) == 1:
return 1
elif RowColumnWin(bcx_count, 0, 1, black_chess) == 1:
return 0
elif RowColumnWin(wcy_count, 1, 0, white_chess) == 1:
return 1
elif RowColumnWin(bcy_count, 1, 0, black_chess) == 1:
return 0
elif DiagonalWin(white_chess) == 1:
return 1
elif DiagonalWin(black_chess) == 1:
return 0
def DrawText(text, x, y, size):
pygame.font.init()
fontObj = pygame.font.SysFont('SimHei', size)
textSurfaceObj = fontObj.render(text, True, white, black)
textRectObj = textSurfaceObj.get_rect()
textRectObj.center = (x, y)
screen.blit(textSurfaceObj, textRectObj)
pygame.display.update()
buf_size = 1024
addr = (ip, port)
# 连接服务器
tcpCliSock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
tcpCliSock.connect(addr)
inputs = [tcpCliSock]
DrawBoard()
settable = 0
while True:
rs, ws, es = select.select(inputs, [], [], 0)
for r in rs:
if r is tcpCliSock:
data, addr_2 = r.recvfrom(buf_size)
DrawText(" YOUR TURN ", 600, 25, 40)
data = json.loads(data)
settable = 1
black_chess.append(data)
bcx.append(data[0])
bcy.append(data[1])
for event in pygame.event.get():
if event.type == QUIT:
tcpCliSock.close()
pygame.quit()
sys.exit()
if settable == 1:
if SetChess() == 1:
DrawText("WAITING FOR OPPONENT", 600, 25, 40)
settable = 0
msg1 = json.dumps(msg)
tcpCliSock.sendto(msg1.encode(), addr)
msg = []
DrawChess()
if GameOver() == 1:
DrawText("YOU WIN", 600, 360, 40)
while True:
for event in pygame.event.get():
if event.type == QUIT:
pygame.quit()
sys.exit()
elif GameOver() == 0:
DrawText("YOU FAILED", 600, 360, 40)
while True:
for event in pygame.event.get():
if event.type == QUIT:
pygame.quit()
sys.exit()
ServerGobang.py：
"""
-*- coding: utf-8 -*-
@File : ServerGobang.py
@author: 刘子忻
@CSDN : 山河之书Liu_Zixin
@Time : 2023/01/02 1:41
"""
import pygame
import sys
from pygame.locals import *
from collections import Counter
import json
import select
import socket
def ServerGobang(ip, port):
# 界面初始化
screen = pygame.display.set_mode((1200, 720))
pygame.display.set_caption('Gobang--Server')
pygame.init()
# 图片导入
img_board = pygame.image.load('../Client/chess_board.png') # 棋盘
img_bchess = pygame.image.load('../Client/black_chess.jpg') # 黑子
img_wchess = pygame.image.load('../Client/white_chess.jpg') # 白子
# 后面的字和字的背景颜色，之所以设置字的背景颜色就是为了遮盖之前的字
white = (255, 255, 255)
black = (0, 0, 0)
# 用于传送的数据
msg = []
# 棋盘定义——二元组
chess_board = [[]]
def SetChessBoard(): # 根据尺寸信息设置棋盘网格
x, y = 0, 0
while True:
if x == 1200: # 长边界
x = 0
y += 40
if y < 720:
chess_board.append([])
if y == 720: # 宽边界
break
chess_board[-1].append([x, y])
x += 40
SetChessBoard()
# 棋盘格子是否落子
chess_exist = [[0 for _ in range(30)] for _ in range(18)] # 棋盘边界
# 黑白棋子初始化
black_chess, white_chess = [], []
wcx, wcy, bcx, bcy = [], [], [], []
# 绘制棋盘的图像
def DrawBoard():
for i in chess_board:
for j in i:
screen.blit(img_board, (j[0], j[1]))
pygame.display.update()
# 默认棋子类型为1（黑棋）
def SetChess():
if event.type == MOUSEBUTTONDOWN:
pos = pygame.mouse.get_pos()
for i in range(len(chess_board)):
for j in range(len(chess_board[i])):
if chess_board[i][j][0] < pos[0] < chess_board[i][j][0] + 40 and chess_board[i][j][1] < pos[1] < \
chess_board[i][j][1] + 40:
if chess_exist[i][j] == 0:
black_chess.append([i, j])
bcx.append(black_chess[-1][0])
bcy.append(black_chess[-1][1])
msg.extend((i, j))
chess_exist[i][j] = 1
pygame.display.update()
return 1
# 绘制棋子
def DrawChess():
for i in white_chess:
screen.blit(img_wchess, (i[1] * 40, i[0] * 40))
for i in black_chess:
screen.blit(img_bchess, (i[1] * 40, i[0] * 40))
pygame.display.update()
# 判定行或者列的胜利
def RowColumnWin(x, m, n, chess):
for i in x:
if x[i] >= 5:
xy = []
for j in chess:
if j[m] == i:
xy.append(j[n])
xy.sort()
count = 0
for j in range(len(xy) - 1):
if xy[j] + 1 == xy[j + 1]:
count += 1
else:
count = 0
if count >= 4:
return 1
# 判定对角线的胜利
def DiagonalWin(chess):
x, y = [], []
chess.sort()
for i in chess:
x.append(i[0])
y.append(i[1])
c, first, last = 0, 0, 0
for i in range(len(x) - 1):
if x[i + 1] != x[i]:
if x[i] + 1 == x[i + 1]:
c += 1
last = i + 1
else:
if c < 4:
first = i + 1
c = 0
else:
last = i
print(last)
break
else:
last = i + 1
if c >= 4:
dis = []
for i in range(first, last + 1):
dis.append(x[i] - y[i])
count = Counter(dis)
for i in count:
if count[i] >= 5:
return 1
dis = []
x2 = [i * (-1) for i in x]
for i in range(first, last + 1):
dis.append(x2[i] - y[i])
count = Counter(dis)
for i in count:
if count[i] >= 5:
return 1
# 利用上述的；两个判别函数，对不同情况下的胜利进行判定并返回胜利或失败
def GameOver():
wcx_count, wcy_count, bcx_count, bcy_count = Counter(wcx), Counter(wcy), Counter(bcx), Counter(bcy)
if RowColumnWin(wcx_count, 0, 1, white_chess) == 1:
return 0
elif RowColumnWin(bcx_count, 0, 1, black_chess) == 1:
return 1
elif RowColumnWin(wcy_count, 1, 0, white_chess) == 1:
return 0
elif RowColumnWin(bcy_count, 1, 0, black_chess) == 1:
return 1
elif DiagonalWin(white_chess) == 1:
return 0
elif DiagonalWin(black_chess) == 1:
return 1
# 为后续显示提示做准备
def DrawText(text, x, y, size):
pygame.font.init()
fontObj = pygame.font.SysFont('SimHei', size)
textSurfaceObj = fontObj.render(text, True, white, black)
textRectObj = textSurfaceObj.get_rect()
textRectObj.center = (x, y)
screen.blit(textSurfaceObj, textRectObj)
pygame.display.update()
buf_size = 1024
addr = (ip, port)
# 定义服务器属性
conn_gobang = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
conn_gobang.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # 对socket的配置重用ip和端口号
conn_gobang.bind(addr)
conn_gobang.listen(1)
inputs = [conn_gobang]
DrawBoard()
settable = 1
link = False
while True:
rs, ws, es = select.select(inputs, [], [], 0) # 获取连接信息
for r in rs:
if r is conn_gobang:
link = True
udp, addr = conn_gobang.accept()
inputs.append(udp)
else:
data, addr = r.recvfrom(buf_size)
disconnected = not data
DrawText(" YOUR TURN ", 600, 20, 40)
if disconnected:
inputs.remove(r)
DrawText("LOSE CONNECTION", 600, 360, 40)
while True:
for event in pygame.event.get():
if event.type == QUIT:
pygame.quit()
sys.exit()
else:
data = json.loads(data)
settable = 1
white_chess.append(data)
wcx.append(data[0])
wcy.append(data[1])
for event in pygame.event.get():
if event.type == QUIT:
conn_gobang.close()
pygame.quit()
sys.exit()
if link:
if settable == 1:
if SetChess() == 1:
DrawText("WAITING FOR OPPONENT", 600, 20, 40)
settable = 0
msg1 = json.dumps(msg)
udp.send(msg1.encode())
msg = []
DrawChess()
if GameOver() == 1:
DrawText("YOU WIN", 600, 360, 40)
while True:
for event in pygame.event.get():
if event.type == QUIT:
pygame.quit()
sys.exit()
elif GameOver() == 0:
DrawText("YOU FAILED", 600, 360, 40)
while True:
for event in pygame.event.get():
if event.type == QUIT:
pygame.quit()
sys.exit()
5、系统各部分完成形式及参考代码
| 文件名 | 完成形式 | 参考代码 |
| ClientSignIn.py | 自主完成 | [1][2] |
| ServerSign.py | 自主完成 | [1][2] |
| ClientGobang.py | 参照改写 | [3] |
| ServerGobang.py | 参照改写 | [3] |
参考代码：
[1] 基于RSA加密和Tkinter可视化的密码存储程序（可用于期末作业设计、Python练习、实用应用；抗错误输入、抗密码盗取）二：登录、注册界面_山河之书Liu_Zixin的博客-CSDN博客_tkinter保存多个账号密码
[2]python-mysql期末实验：三、程序登录、注册界面的制作_山河之书Liu_Zixin的博客-CSDN博客_python+mysql 实现用户登录,注册界面
[3]python网络编程案例—五子棋游戏_鹏鹏写代码的博客-CSDN博客_python五子棋网络
