3. 채팅 만들기
1. 목표
2. 서버
더보기
import sys
from PySide6.QtWidgets import (
QApplication, QMainWindow, QWidget,
QVBoxLayout, QTextEdit
)
from PySide6.QtNetwork import QTcpServer, QTcpSocket, QHostAddress
class ServerWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("채팅 서버")
self.resize(500, 400)
# UI
central = QWidget(self)
self.setCentralWidget(central)
layout = QVBoxLayout(central)
self.log = QTextEdit(self)
self.log.setReadOnly(True)
layout.addWidget(self.log)
# Network
self.server = QTcpServer(self)
self.server.newConnection.connect(self.on_new_connection)
self.clients = set() # 연결된 클라이언트 소켓들
self.buffers = {} # 소켓별 수신 버퍼
self.nicknames = {} # 소켓별 닉네임
host = QHostAddress("127.0.0.1")
port = 8888
if self.server.listen(host, port):
self.log.append("서버가 시작되었습니다.")
self.log.append(f"대기 주소: {host.toString()}:{port}")
else:
self.log.append("서버 시작에 실패했습니다.")
def on_new_connection(self):
socket = self.server.nextPendingConnection()
self.clients.add(socket)
self.buffers[socket] = bytearray()
# 기본 닉네임(중복 방지용으로 간단히 고유값 부여)
default_nick = f"Guest-{int(socket.socketDescriptor())}"
self.nicknames[socket] = default_nick
socket.readyRead.connect(lambda s=socket: self.on_ready_read(s))
socket.disconnected.connect(lambda s=socket: self.on_disconnected(s))
self.log.append(f"클라이언트가 접속했습니다. (임시 닉네임: {default_nick})")
# 접속한 본인에게 안내 메시지(선택)
self.send_to(socket, f"시스템: 닉네임을 설정하세요. 예) NICK 홍길동")
def on_ready_read(self, socket: QTcpSocket):
data = bytes(socket.readAll())
self.buffers[socket].extend(data)
while b"\n" in self.buffers[socket]:
line, _, rest = self.buffers[socket].partition(b"\n")
self.buffers[socket] = rest
text = line.decode("utf-8", errors="replace").strip()
if not text:
continue
# 1) 닉네임 설정 명령 처리: "NICK <nickname>"
if text.upper().startswith("NICK "):
new_nick = text[5:].strip()
if not new_nick:
self.send_to(socket, "시스템: 닉네임이 비어있습니다.")
continue
old = self.nicknames.get(socket, "Unknown")
self.nicknames[socket] = new_nick
self.log.append(f"닉네임 변경: {old} -> {new_nick}")
self.broadcast(f"시스템: {old} 님이 {new_nick} 님으로 변경했습니다.")
continue
# 2) 일반 메시지 처리
nick = self.nicknames.get(socket, "Unknown")
formatted = f"{nick}: {text}"
self.log.append(f"수신 메시지: {formatted}")
self.broadcast(formatted)
def broadcast(self, message: str):
data = (message + "\n").encode("utf-8")
for client in list(self.clients):
client.write(data)
def send_to(self, socket: QTcpSocket, message: str):
socket.write((message + "\n").encode("utf-8"))
def on_disconnected(self, socket: QTcpSocket):
nick = self.nicknames.get(socket, "Unknown")
if socket in self.clients:
self.clients.remove(socket)
self.buffers.pop(socket, None)
self.nicknames.pop(socket, None)
socket.deleteLater()
self.log.append(f"클라이언트 연결이 종료되었습니다. ({nick})")
self.broadcast(f"시스템: {nick} 님이 퇴장했습니다.")
if __name__ == "__main__":
app = QApplication(sys.argv)
win = ServerWindow()
win.show()
sys.exit(app.exec())
3. 클라이언트
더보기
import sys
from PySide6.QtWidgets import (
QApplication, QMainWindow, QWidget,
QVBoxLayout, QTextEdit, QLineEdit, QPushButton
)
from PySide6.QtNetwork import QTcpSocket
class ClientWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("채팅 클라이언트")
self.resize(500, 420)
# UI
central = QWidget(self)
self.setCentralWidget(central)
layout = QVBoxLayout(central)
self.chat = QTextEdit(self)
self.chat.setReadOnly(True)
# 닉네임 입력
self.nick_input = QLineEdit(self)
self.nick_input.setPlaceholderText("닉네임을 입력하세요 (예: 홍길동)")
self.nick_btn = QPushButton("닉네임 설정", self)
self.nick_btn.clicked.connect(self.send_nick)
self.input = QLineEdit(self)
self.input.setPlaceholderText("메시지를 입력하세요")
self.send_btn = QPushButton("전송", self)
self.send_btn.clicked.connect(self.send_message)
layout.addWidget(self.chat)
layout.addWidget(self.nick_input)
layout.addWidget(self.nick_btn)
layout.addWidget(self.input)
layout.addWidget(self.send_btn)
# Network
self.socket = QTcpSocket(self)
self.socket.readyRead.connect(self.on_ready_read)
self.socket.connected.connect(self.on_connected)
self.socket.disconnected.connect(self.on_disconnected)
self.socket.connectToHost("127.0.0.1", 8888)
self.buffer = bytearray()
self.nickname = "" # 로컬 닉네임 저장(표시/편의)
def on_connected(self):
self.chat.append("서버에 연결되었습니다.")
def on_disconnected(self):
self.chat.append("서버와의 연결이 종료되었습니다.")
def send_nick(self):
nick = self.nick_input.text().strip()
if not nick:
return
# 서버 프로토콜: NICK <nickname>\n
self.socket.write((f"NICK {nick}\n").encode("utf-8"))
self.nickname = nick
self.chat.append(f"시스템: 닉네임을 '{nick}' 으로 설정 요청했습니다.")
def send_message(self):
text = self.input.text().strip()
if not text:
return
# (선택) 닉네임이 없으면 먼저 설정 유도
if not self.nickname:
self.chat.append("시스템: 먼저 닉네임을 설정하세요.")
return
self.socket.write((text + "\n").encode("utf-8"))
self.input.clear()
def on_ready_read(self):
data = bytes(self.socket.readAll())
self.buffer.extend(data)
while b"\n" in self.buffer:
line, _, rest = self.buffer.partition(b"\n")
self.buffer = rest
message = line.decode("utf-8", errors="replace")
self.chat.append(message)
if __name__ == "__main__":
app = QApplication(sys.argv)
win = ClientWindow()
win.show()
sys.exit(app.exec())
4. 과제
더보기
- 닉네임 추가하기
- 서버에서 접속자 수 표시
- 특정 단어 금지 필터
- 서버에서만 공지 메시지 보내기
- 시간 표시 추가