ㅇ 일 자 : 2026년 6월 24일(수)
ㅇ 주 제 : 교육자료 27강~33강 예시문제 실습 및 이해하기
▶ PySide6 GUI 채팅 사용자 목록과 귓속말 파일 송수신
1. 파일 전송과 데이터 스트림, 전송 메타데이터에 대한 이해
1.1. 스트림(Stream)의 정의
가. 컴퓨터 과학 (I/O): 파일, 키보드, 네트워크 등에서 데이터가 바이트(Byte) 단위로 흐르는 가상의 연속적인 통로
나. 데이터/이벤트 처리: 데이터를 한 번에 모아서 처리하지 않고, 실시간으로 연속해서 들어오는 데이터를 즉시 수집하고 분석
(예시) Java 8 스트림 API(함수형 데이터 가공), Steam(게임 플랫폼), 미디어 스트리밍(실시간 영상/음악 재생)
1.2. 바이너리 데이터(Binary Data)
0, 1로 이루어진 이진 형식의 데이터로, 텍스트 형식을 제외한 이미지, 동영상, 실행 파일 등 컴퓨터가 직접 이해하는 형태의 데이터
1.3 파일 전송 프로세스 및 메타데이터
패킷의 경계 : 네트워크로 데이터를 보낼 때 데이터가 쪼개지므로, 어디서부터 어디까지가 하나의 파일인지 경계를 구분해야 하는데 이때 패킷의 경계라고 한다.
1.4. 전송 설계
가. 메타데이터(JSON/TXT) 우선 전송 : 파일 이름, 파일 크기(Bytes) 등의 정보를 먼저 전달한다.
나. 바이너리 데이터 전송 : 실제 파일 내용을 나중에 전달하며, 수신 측에서 정상 수신 완료를 확인
1.5. 메타데이터(Metadata)
데이터에 대한 데이터. 즉, 실제 파일 내용이 아니라 파일의 이름, 크기, 유형 등 데이터의 속성을 기술한 정보
2. [27강~30강] 사용자 목록 동기화 및 귓속말 기능 구현
2.1. 사용자 목록 및 상대 선택 UI (27~28강)
가. 사용자 목록 동기화: user_list를 추가하여 서버에 접속한 클라이언트 주소(client_address)를 실시간으로 표시
나. UI 상호작용: 목록에서 특정 사용자를 클릭하면 selected_user에 저장되며 선택 라벨이 변경되고,
선택된 사용자가 퇴장(종료)하면 선택 상태가 자동으로 해제된다
2.2. 귓속말 메시지 전송 및 오류 처리 (29~30강)
가. 귓속말 메커니즘: 전체에 전송하는 broadcast()와 달리, 선택된 대상(selected_user) 한 명의 소켓에만 메시지를 전송
나. 결과 반환 및 UI 표시: 전송 성공 여부를 True/False로 반환하여 발신자에게 실패 결과를 알린다
성공 시 내 화면에는 [귓속말 보냄], 상대 화면에는 [귓속말 받음]이 표시
다. 오류 검증 (사전 검사 & 서버 검증)
라.GUI 자체 사전 검사(4단계): 메시지 공백 여부, 상대 선택 여부, GUI show_error() 처리 등.
마.서버 검증: 자기 자신에게 귓속말을 보내는 행위 차단 및 전송 실패 방어 로직 구축.
바. 시그널과 슬롯(Signal & Slot): PySide6(Qt)의 핵심 이벤트 처리 메커니즘.
버튼 클릭 등의 이벤트(Signal)가 발생하면, 그에 연결된 특정 함수나 메서드(Slot)가 자동으로 실행되는 구조
3. [31강~33강] 텍스트/바이너리 차이 및 파일 분할 전송
3.1. 텍스트와 바이너리 데이터의 변환 (31강)
가. 컴퓨터는 문자열을 그대로 보낼 수 없으므로 인코딩(Encoding)을 통해 바이트로 변환한 뒤 전송하고, 수신 측은 이를 다시 문자열로 디코딩(Decoding)
(예시) Python
text = "늑구 안녕"
data = text.encode("utf-8") # 바이트 데이터로 변환 (b'\xec\x95...')
3.2. 파일 분할 전송(Chunking), 메타데이터 설계 개념 (32~34강)
가. 소켓 점유 방지
대용량 파일을 한 번에 보내면 파일 전송 동안 소켓의 send() 동작이 끝나지 않아 다른 채팅 메시지를 보낼 수 없다. 따라서 파일을 작은 조각(Chunk)으로 나누어 보내야 한다.
나. 데이터 인코딩 (Base64) : 조각낸 바이너리 파일을 안전하게 전송하기 위해 문자열 형태로 변환(Base64)하여 JSON 구조에 담아 보낸다.
다. 구현 및 확인 : * protocol.py에 send_file() 함수 추가, client.py의 ChatClient 클래스에 send_file() 메서드를 구현
라. 파일 전송 완료 후 클라이언트에 "[파일 전송 완료]"가 출력되고 서버 연결이 끊김 없이 유지되는지 확인한다.
마. Base64 : 바이너리(이진) 데이터를 화면에 표시할 수 있는 64개의 ASCII 문자(영문, 숫자 등)로 바꾸는 인코딩 방식
네트워크 전송 중 데이터가 손실되거나 왜곡되는 것을 막기 위해 사용한다.
바. 대용량 파일 전송 시 소켓의 동시성 한계를 극복하기 위해 파일을 조각내어 전송하며, 파일 조각은 Base64로 인코딩되어 JSON 메타데이터와 함께 전송된다. protocol.py와 client.py에 전송 기능을 구현하고, handle_file_end()에서 파일 자원을 정리하며 os 모듈 등을 활용해 저장 경로를 상수(대문자 사용)로 지정하여 파일 전송 시스템을 완성한다.
사. 핵심 구현 내용 및 관련 기술:
- 파일 분할 전송: 대용량 파일은 소켓 send 작업 중 차단 방지를 위해 조각(Chunk) 단위로 분할 전송
- 데이터 인코딩: 파일 조각은 바이너리 데이터를 텍스트로 변환하는 Base64로 인코딩
- 메타데이터/JSON: 파일 전송 시 파일ID, 크기 등을 포함한 정보(Metadata)를 JSON 형태로 함께 교환
- 구현 핵심: protocol.py의 send_file() 함수와 client.py의 메서드 구현, 수신 시 handle_file_end()의 file.close() 호출, 상수(대문자)를 이용한 RECEIVE_DIR 지정.
<< 예시문제 실습하기 >> ** 가상의 환경에서 실습한 화면임 **
1. 27강. 사용자 목록 동기화
27강 소스코드
chat_client / main.py
# 27강. GUI 에서 메세지 보내기 (25강) ### 이전 강의에서 추가한 부분임
# main.py
# main.py
import sys # 시스템 관련 파라미터 및 프로그램 종료를 위한 라이브러리
import socket # 네트워크 통신을 위한 소켓 라이브러리
import threading # 백그라운드에서 메시지를 수신하기 위한 멀티쓰레딩 라이브러리
# PySide6에서 GUI 이벤트 처리 및 시그널-슬롯 구조를 사용하기 위한 모듈 임포트
from PySide6.QtCore import Signal, Slot, Qt
from PySide6.QtWidgets import (
QApplication, # QT 어플리케이션을 제어하는 메인 클래스
QWidget, # 기본 윈도우 창 레이아웃의 베이스 클래스
QTextEdit, # 멀티라인 텍스트 표시 및 편집 창 (채팅 기록 창)
QLineEdit, # 단일라인 텍스트 입력 창 (메시지 입력 창)
QPushButton, # 클릭 가능한 버튼 (접속, 전송 버튼)
QLabel, # 텍스트를 화면에 표시하는 라벨 (상태 표시용)
QVBoxLayout, # 위젯을 수직(세로)으로 배열하는 레이아웃
QHBoxLayout, # 위젯을 수평(가로)으로 배열하는 레이아웃
QMessageBox, # 경고 및 알림 메시지 창을 띄우기 위한 클래스 (추가됨)
)
# 사용자 정의 프로토콜 모듈에서 메시지 송수신 함수를 임포트
from protocol import receive_messages, send_message
# 접속할 서버의 IP 주소와 포트 번호 설정 (로컬 호스트)
HOST = "127.0.0.1"
PORT = 5000
# 1. 메모리 기반 사용자 데이터 (서버 인증용 딕셔너리)
USER_DB = {"user1": "1234", "user2": "5678"}
# 2. 회원가입 페이지 (추가됨)
class RegisterPage(QWidget):
register_success = Signal() # 회원가입 성공 시 발생할 시그널
def __init__(self):
super().__init__()
self.setWindowTitle("회원가입")
self.resize(300, 200)
layout = QVBoxLayout()
self.id_input = QLineEdit()
self.id_input.setPlaceholderText("새로운 아이디")
self.pw_input = QLineEdit()
self.pw_input.setPlaceholderText("새로운 비밀번호")
self.pw_input.setEchoMode(QLineEdit.Password)
self.register_btn = QPushButton("가입하기")
self.cancel_btn = QPushButton("취소")
layout.addWidget(QLabel("회원가입 정보를 입력하세요"))
layout.addWidget(self.id_input)
layout.addWidget(self.pw_input)
layout.addWidget(self.register_btn)
layout.addWidget(self.cancel_btn)
self.setLayout(layout)
self.register_btn.clicked.connect(self.check_register)
self.cancel_btn.clicked.connect(self.close)
def check_register(self):
user_id = self.id_input.text().strip()
password = self.pw_input.text().strip()
if not user_id or not password:
QMessageBox.warning(
self, "오류", "아이디와 비밀번호를 모두 입력해주세요."
)
return
if user_id in USER_DB:
QMessageBox.warning(self, "오류", "이미 존재하는 아이디입니다.")
else:
USER_DB[user_id] = password
QMessageBox.information(
self, "성공", "회원가입이 완료되었습니다!"
)
self.register_success.emit()
self.close()
# 3. 로그인 페이지
class LoginPage(QWidget):
# 로그인 성공 시 실행할 시그널 (아이디 전달)
login_success = Signal(str)
def __init__(self):
super().__init__()
self.setWindowTitle("로그인")
self.resize(300, 200)
layout = QVBoxLayout()
self.id_input = QLineEdit()
self.id_input.setPlaceholderText("아이디")
self.pw_input = QLineEdit()
self.pw_input.setPlaceholderText("비밀번호")
self.pw_input.setEchoMode(QLineEdit.Password)
self.login_btn = QPushButton("로그인")
self.register_btn = QPushButton("회원가입") # 회원가입 이동 버튼 추가
layout.addWidget(QLabel("채팅 프로그램 로그인"))
layout.addWidget(self.id_input)
layout.addWidget(self.pw_input)
layout.addWidget(self.login_btn)
layout.addWidget(self.register_btn)
self.setLayout(layout)
self.login_btn.clicked.connect(self.check_login)
self.register_btn.clicked.connect(self.open_register_page)
self.reg_page = None # 회원가입 창 객체 보유용 변수
def check_login(self):
user_id = self.id_input.text().strip()
password = self.pw_input.text().strip()
# 서버 인증 시뮬레이션
if user_id in USER_DB and USER_DB[user_id] == password:
self.login_success.emit(user_id)
self.close() # 로그인 창 닫기
else:
QMessageBox.warning(
self, "인증 실패", "아이디 또는 비밀번호가 틀렸습니다."
)
def open_register_page(self):
self.reg_page = RegisterPage()
# 회원가입 성공 시 로그인 입력창에 가입한 아이디를 세팅해주면 편리함
self.reg_page.register_success.connect(
lambda: self.id_input.setText(self.reg_page.id_input.text())
)
self.reg_page.show()
# 4. 채팅창 페이지
class ChatWindow(QWidget):
# 백그라운드 쓰레드에서 수신한 메시지 데이터를 메인 GUI 쓰레드로 안전하게 전달하기 위한 커스텀 시그널 정의
message_received = Signal(dict)
def __init__(self):
super().__init__() # 부모 클래스(QWidget)의 생성자 호출
# 초기 변수 및 상태 설정
self.user_id = "" # 현재 로그인한 사용자의 아이디 저장 변수
self.client_socket = None # 서버와 연결할 소켓 객체 저장 변수
self.receive_thread = None # 메시지 수신을 담당할 백그라운드 쓰레드 변수
self.running = False # 수신 루프 실행 여부를 제어하는 플래그
# GUI 창의 기본 속성 설정
self.setWindowTitle("채팅방") # 창 제목 설정
self.resize(560, 480) # 창 크기 설정 (가로 560, 세로 480)
# 채팅 기록창 설정
self.chat_display = QTextEdit()
self.chat_display.setReadOnly(
True
) # 사용자가 직접 수정하지 못하도록 읽기 전용 설정
self.chat_display.setPlaceholderText(
"채팅 메시지가 여기에 표시됩니다."
)
# 메시지 입력창 설정
self.message_input = QLineEdit()
self.message_input.setPlaceholderText("메시지를 입력하세요")
self.message_input.setEnabled(
False
) # 서버 연결 전까지는 입력 불가능하게 설정
# 메시지 전송 버튼 설정
self.send_button = QPushButton("보내기")
self.send_button.setEnabled(
False
) # 서버 연결 전까지는 비활성화 설정
# 상단 서버 접속 영역 설정
self.status_label = QLabel("상태: 서버에 연결되지 않음")
self.connect_button = QPushButton("서버 접속")
# 상단 레이아웃 구성 (상태 라벨 | 접속 버튼)
top_layout = QHBoxLayout()
top_layout.addWidget(self.status_label)
top_layout.addWidget(self.connect_button)
# 하단 레이아웃 구성 (메시지 입력창 | 전송 버튼)
input_layout = QHBoxLayout()
input_layout.addWidget(self.message_input)
input_layout.addWidget(self.send_button)
# 메인 레이아웃 구성 (상단 -> 채팅창 -> 하단 순으로 세로 배치)
main_layout = QVBoxLayout()
main_layout.addLayout(top_layout)
main_layout.addWidget(self.chat_display)
main_layout.addLayout(input_layout)
self.setLayout(
main_layout
) # 구성된 메인 레이아웃을 현재 위젯 창에 적용
# 이벤트 시그널과 핸들러(슬롯) 함수들을 연결 (Connect)
self.connect_button.clicked.connect(
self.connect_to_server
) # 접속 버튼 클릭 시
self.send_button.clicked.connect(self.on_send_clicked) # 전송 버튼 클릭 시
self.message_input.returnPressed.connect(
self.on_send_clicked
) # 입력창에서 엔터키 입력 시
self.message_received.connect(
self.display_message
) # 새 메시지 시그널 수신 시 화면 표시 함수 호출
def set_user(self, user_id):
"""로그인 성공 후 유저 아이디를 세팅하고 타이틀을 변경하는 함수"""
self.user_id = user_id
self.setWindowTitle(f"채팅방 - [{self.user_id}]님")
def connect_to_server(self):
"""서버에 TCP 소켓 연결을 시도하고 수신 쓰레드를 시작하는 함수"""
try:
# IPv4, TCP 프로토콜을 사용하는 소켓 생성 및 서버 연결 시도
self.client_socket = socket.socket(
socket.AF_INET, socket.SOCK_STREAM
)
self.client_socket.connect((HOST, PORT))
# 연결 성공 시 UI 상태 업데이트
self.status_label.setText("상태: 서버에 연결됨")
self.chat_display.append("[시스템] 서버에 연결되었습니다.")
# 버튼 및 입력창 활성화/비활성화 제어
self.connect_button.setEnabled(
False
) # 재접속 방지를 위해 비활성화
self.message_input.setEnabled(True) # 입력 허용
self.send_button.setEnabled(True) # 전송 허용
# 백그라운드 수신 쓰레드 생성 및 시작
self.running = True
self.receive_thread = threading.Thread(target=self.receive_loop)
self.receive_thread.daemon = (
True # 메인 프로그램 종료 시 쓰레드도 자동 종료 설정
)
self.receive_thread.start()
except ConnectionRefusedError:
self.status_label.setText("상태: 서버 연결 실패")
self.chat_display.append(
"[오류] 서버가 실행 중인지 확인하세요."
)
except Exception as error:
self.status_label.setText("상태: 서버 연결 실패")
self.chat_display.append(
f"[오류] 연결 중 문제가 발생했습니다: {error}"
)
def receive_loop(self):
"""백그라운드에서 실행되며 서버로부터 데이터를 지속적으로 수신하는 루프 함수"""
buffer = "" # 데이터가 쪼개져 올 때를 대비한 임시 버퍼 공간
while self.running:
try:
buffer, messages, connected = receive_messages(
self.client_socket, buffer
)
if not connected:
self.message_received.emit(
{
"type": "system",
"content": "서버와의 연결이 끊어졌습니다.",
}
)
break
for message in messages:
self.message_received.emit(message)
except Exception as error:
self.message_received.emit(
{
"type": "error",
"content": f"메시지 수신 중 오류가 발생했습니다: {error}",
}
)
break
self.running = False
def on_send_clicked(self):
"""사용자가 입력한 메시지를 서버로 전송하는 함수"""
text = self.message_input.text().strip()
if not text:
return
if self.client_socket is None:
self.chat_display.append("[오류] 서버에 먼저 접속하세요.")
return
try:
# 보낼 때 발신자 정보(sender)도 프로토콜에 실어 보냅니다.
send_message(
self.client_socket,
{"type": "chat", "sender": self.user_id, "content": text},
)
self.chat_display.append(f"나: {text}")
self.message_input.clear()
except Exception as error:
self.chat_display.append(f"[오류] 메시지 전송 실패: {error}")
@Slot(dict)
def display_message(self, message):
"""수신 쓰레드가 보낸 시그널을 받아 화면에 분류 처리하는 슬롯 함수"""
message_type = message.get("type")
if message_type == "chat":
sender = message.get("sender", "unknown")
content = message.get("content", "")
# 내가 보낸 메시지가 상대방에게서 다시 브로드캐스트되어 오는 중복 출력을 방지
if sender != self.user_id:
self.chat_display.append(f"{sender}: {content}")
elif message_type == "system":
content = message.get("content", "")
self.chat_display.append(f"[시스템] {content}")
elif message_type == "error":
content = message.get("content", "")
self.chat_display.append(f"[오류] {content}")
else:
self.chat_display.append(f"[알 수 없는 메시지] {message}")
# 5. 흐름 제어 및 프로그램 실행 진입점
if __name__ == "__main__":
app = QApplication(sys.argv)
login_page = LoginPage()
chat_window = ChatWindow()
# 로그인 성공 시그널 연결 로직
def on_login_success(user_id):
chat_window.set_user(user_id) # 채팅창에 로그인된 유저 세팅
chat_window.show() # 채팅창 켜기
chat_window.connect_to_server() # 로그인 완료와 동시에 서버 접속 자동 실행
login_page.login_success.connect(on_login_success)
login_page.show() # 처음에 로그인 화면 띄우기
sys.exit(app.exec())
27강 소스코드
chat_server / server.py
# 27강. 사용자 목록 동기화 ### 수정 구현한 부분
# server.py
import socket
import threading
from protocol import receive_messages, send_message
HOST = "127.0.0.1"
PORT = 5000
clients = []
def print_clients():
print(f"현재 접속자 수: {len(clients)}")
def get_user_list(): ### 2단계에서 구현
return [client["address"] for client in clients]
def broadcast_user_list(): ### 2단계에서 구현
user_list_message = {
"type": "user_list",
"users": get_user_list()
}
for client in clients[:]:
try:
send_message(client["socket"], user_list_message)
except Exception:
pass
def broadcast(message, sender_socket=None):
for client in clients[:]:
client_socket = client["socket"] ### 딕셔너리 구조에 맞게 접근
if client_socket != sender_socket:
try:
send_message(client_socket, message)
except Exception:
pass
def send_error(client_socket, content):
send_message(client_socket, {
"type": "error",
"content": content
})
def remove_client(client_socket): ###️ 딕셔너리에서 찾아 제거
for client in clients[:]:
if client["socket"] == client_socket:
clients.remove(client)
return client["address"]
return None
def handle_client(client_socket, client_address):
print(f"클라이언트 처리 시작: {client_address}")
buffer = ""
address_text = str(client_address)
try:
broadcast({
"type": "system",
"content": f"{address_text} 님이 입장했습니다."
}, client_socket)
broadcast_user_list() ### 입장 후 목록 동기화
while True:
try:
buffer, messages, connected = receive_messages(client_socket, buffer)
if not connected:
print(f"클라이언트 연결 끊김: {client_address}")
break
exit_requested = False
for message in messages:
print(f"{client_address} 메시지: {message}")
message_type = message.get("type")
if message_type == "exit":
print(f"클라이언트 종료 요청: {client_address}")
exit_requested = True
break
if message_type == "chat":
content = message.get("content", "").strip()
if not content:
send_error(client_socket, "빈 메시지는 보낼 수 없습니다.")
continue
broadcast({
"type": "chat",
"sender": address_text,
"content": content
}, client_socket)
continue
send_error(client_socket, f"지원하지 않는 메시지 타입입니다: {message_type}")
if exit_requested:
break
except Exception as error:
print(f"클라이언트 처리 오류: {client_address}, {error}")
break
finally: ### with 대신 try/finally로 소켓 관리
try:
client_socket.close()
except Exception:
pass
removed_address = remove_client(client_socket)
broadcast({
"type": "system",
"content": f"{address_text} 님이 퇴장했습니다."
}, client_socket)
broadcast_user_list() ###퇴장 후 목록 동기화
print(f"접속자 목록에서 제거: {removed_address}")
print_clients()
print(f"클라이언트 처리 종료: {client_address}")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind((HOST, PORT))
server_socket.listen()
print("서버를 시작합니다.")
print("클라이언트 접속 대기 중...")
while True:
client_socket, client_address = server_socket.accept()
address_text = str(client_address)
print(f"클라이언트 접속: {client_address}")
clients.append({ ### 소켓+주소 딕셔너리로 추가
"socket": client_socket,
"address": address_text
})
print(f"접속자 목록에 추가: {client_address}")
print_clients()
client_thread = threading.Thread(
target=handle_client,
args=(client_socket, client_address)
)
client_thread.daemon = True
client_thread.start()
2. 28강. 상대 선택 UI 만들기
28강 소스코드
chat_client / main.py
# main.py
import socket # 서버와 TCP 통신을 하기 위한 소켓 모듈
import sys # 프로그램 종료 및 인자 전달을 위한 모듈
import threading # 백그라운드에서 메시지를 계속 받기 위한 스레드 모듈
# PySide6에서 UI 이벤트 처리(시그널, 슬롯)를 위해 가져온 모듈
from PySide6.QtCore import Signal, Slot
# PySide6에서 창을 만들고 위젯들을 배치하기 위해 가져온 모듈들
from PySide6.QtWidgets import (
QApplication,
QWidget,
QTextEdit,
QLineEdit,
QPushButton,
QLabel,
QListWidget,
QVBoxLayout,
QHBoxLayout,
)
# 직접 만든 protocol 모듈에서 메시지 송수신용 함수를 가져옴
from protocol import receive_messages, send_message
# 접속할 서버 정보 (로컬 호스트, 포트 번호 5000)
HOST = "127.0.0.1"
PORT = 5000
class ChatWindow(QWidget):
# 커스텀 시그널 정의: 서버로부터 딕셔너리 형태의 메시지를 받으면 UI 스레드로 전달함
message_received = Signal(dict)
def __init__(self):
super().__init__()
# 초기 상태 변수들 설정
self.client_socket = None # 서버와 연결될 소켓 객체 저장용
self.receive_thread = None # 메시지 수신을 담당할 스레드 객체 저장용
self.running = False # 수신 루프를 돌릴지 제어하는 플래그
self.selected_user = None # 현재 리스트에서 선택한 상대방 이름
# 창 타이틀과 크기 지정
self.setWindowTitle("채팅")
self.resize(700, 480)
# --- UI 위젯들 생성 ---
self.status_label = QLabel("상태: 서버에 연결되지 않음") # 연결 상태 표시 라벨
self.connect_button = QPushButton("서버 접속") # 접속 버튼
self.chat_display = QTextEdit() # 채팅 창 (메시지 보여주는 곳)
self.chat_display.setReadOnly(True) # 사용자가 직접 수정 못 하게 읽기 전용으로 설정
self.chat_display.setPlaceholderText("채팅 메시지가 여기에 표시됩니다.")
self.user_list = QListWidget() # 접속자 명단 리스트 위젯
self.user_list.setMaximumWidth(150) # 가로 길이를 최대 150으로 제한
self.selected_user_label = QLabel("선택한 상대: 없음") # 선택된 귓속말 상대 표시용
self.message_input = QLineEdit() # 메시지 입력창
self.message_input.setPlaceholderText("메시지를 입력하세요")
self.send_button = QPushButton("보내기") # 전송 버튼
# --- 레이아웃 배치 (UI 뼈대 잡기) ---
# 1. 상단 레이아웃: 상태 라벨과 접속 버튼을 가로로 배치
top_layout = QHBoxLayout()
top_layout.addWidget(self.status_label)
top_layout.addWidget(self.connect_button)
# 2. 유저 목록 레이아웃: 선택된 상대 라벨과 유저 리스트를 세로로 배치
user_layout = QVBoxLayout()
user_layout.addWidget(self.selected_user_label)
user_layout.addWidget(self.user_list)
# 3. 중간 레이아웃: 채팅창과 유저 목록 레이아웃을 가로로 배치
content_layout = QHBoxLayout()
content_layout.addWidget(self.chat_display)
content_layout.addLayout(user_layout)
# 4. 하단 레이아웃: 입력창과 전송 버튼을 가로로 배치
input_layout = QHBoxLayout()
input_layout.addWidget(self.message_input)
input_layout.addWidget(self.send_button)
# 5. 메인 레이아웃: 상단, 중간, 하단 레이아웃을 위에서 아래로 순서대로 쌓음
main_layout = QVBoxLayout()
main_layout.addLayout(top_layout)
main_layout.addLayout(content_layout)
main_layout.addLayout(input_layout)
# 위에서 만든 메인 레이아웃을 이 윈도우 창의 레이아웃으로 등록
self.setLayout(main_layout)
# --- 이벤트 연결 (시그널과 슬롯 바인딩) ---
self.connect_button.clicked.connect(self.connect_to_server) # 접속 버튼 클릭 시 서버 연결
self.send_button.clicked.connect(self.on_send_clicked) # 보내기 버튼 클릭 시 메시지 전송
self.message_input.returnPressed.connect(self.on_send_clicked) # 입력창에서 엔터 쳐도 전송
self.message_received.connect(self.display_message) # 메시지 수신 시그널이 오면 화면에 출력
self.user_list.itemClicked.connect(self.on_user_selected) # 유저 리스트 클릭 시 상대방 선택
# 맨 처음에는 연결되지 않은 상태의 UI 세팅을 적용
self.set_connected_state(False)
def set_connected_state(self, connected):
"""연결 상태에 따라 버튼이나 입력창의 활성화/비활성화 상태를 바꿔주는 함수"""
self.connect_button.setEnabled(not connected) # 연결되면 접속 버튼 비활성화
self.message_input.setEnabled(connected) # 연결되어야 입력창 활성화
self.send_button.setEnabled(connected) # 연결되어야 보내기 버튼 활성화
if connected:
self.status_label.setText("상태: 서버에 연결됨")
else:
self.status_label.setText("상태: 서버에 연결되지 않음")
self.user_list.clear() # 연결 끊기면 유저 리스트 초기화
self.selected_user = None
self.selected_user_label.setText("선택한 상대: 없음")
def connect_to_server(self):
"""서버 접속 버튼을 눌렀을 때 실행되는 함수"""
try:
# TCP 소켓을 생성하고 설정된 HOST와 PORT로 접속 시도
self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.client_socket.connect((HOST, PORT))
# 백그라운드에서 실시간으로 메시지를 받을 수신 스레드 가동
self.running = True
self.receive_thread = threading.Thread(target=self.receive_loop)
self.receive_thread.daemon = True # 메인 프로그램 종료 시 스레드도 같이 자동 종료되도록 설정
self.receive_thread.start()
# UI 상태를 연결됨 상태로 전환
self.set_connected_state(True)
self.chat_display.append("[시스템] 서버에 연결되었습니다.")
except ConnectionRefusedError:
# 서버가 켜져있지 않는 등 연결 거부되었을 때 처리
self.client_socket = None
self.connect_button.setEnabled(True)
self.status_label.setText("상태: 서버 연결 실패")
self.chat_display.append("[오류] 서버가 실행 중인지 확인하세요.")
except Exception as error:
# 그 외 예상치 못한 에러 발생 시 처리
self.client_socket = None
self.connect_button.setEnabled(True)
self.status_label.setText("상태: 서버 연결 실패")
self.chat_display.append(f"[오류] 연결 중 문제가 발생했습니다: {error}")
def receive_loop(self):
"""별도의 스레드에서 돌면서 서버로부터 오는 데이터를 무한 대기하는 루프 함수"""
buffer = "" # 잘려서 들어오는 데이터를 임시 저장할 버퍼 공간
while self.running:
try:
# protocol 모듈의 함수를 통해 서버 메시지 수신 시도
buffer, messages, connected = receive_messages(self.client_socket, buffer)
if not connected:
# 서버가 먼저 연결을 끊은 경우
self.message_received.emit({
"type": "system",
"content": "서버와의 연결이 끊어졌습니다."
})
break
# 정상적으로 받아온 메시지가 있다면 하나씩 UI 스레드로 던져줌
for message in messages:
self.message_received.emit(message)
except OSError:
# 소켓이 강제로 닫히거나 에러가 났을 때 루프 탈출
break
except Exception as error:
# 기타 에러 발생 시 UI에 에러 전송 후 루프 탈출
self.message_received.emit({
"type": "error",
"content": f"메시지 수신 중 오류가 발생했습니다: {error}"
})
break
# 루프를 빠져나왔다는 건 연결이 끝났다는 의미이므로 상태 정리
self.running = False
self.message_received.emit({
"type": "state",
"connected": False
})
def on_send_clicked(self):
"""보내기 버튼을 누르거나 엔터를 쳤을 때 실행되는 함수"""
text = self.message_input.text().strip() # 앞뒤 공백 제거한 텍스트 가져오기
if not text:
return # 입력된 내용이 없으면 아무것도 안 함
if self.client_socket is None:
self.chat_display.append("[오류] 서버에 먼저 접속하세요.")
return
# 만약 입력창에 'exit'라고 치면 서버와의 연결을 끊음
if text == "exit":
self.disconnect_from_server()
self.message_input.clear()
return
try:
# protocol 모듈을 사용해 채팅 타입의 데이터를 JSON(or 프로토콜 형태)으로 전송
send_message(self.client_socket, {
"type": "chat",
"content": text
})
# 내가 보낸 메시지를 내 채팅창에도 띄워주고 입력창 비우기
self.chat_display.append(f"나: {text}")
self.message_input.clear()
except Exception as error:
# 전송 실패 시 에러 띄우고 연결 종료 처리
self.chat_display.append(f"[오류] 메시지 전송 실패: {error}")
self.disconnect_from_server()
def disconnect_from_server(self):
"""서버와의 연결을 안전하게 종료하는 함수"""
if self.client_socket is None:
self.set_connected_state(False)
return
try:
# 서버에게 '나 나간다'고 알려주는 exit 메시지 전송 시도
send_message(self.client_socket, {"type": "exit"})
except Exception:
pass # 보내다 실패해도 무시하고 그냥 닫음
self.running = False # 수신 스레드 루프 정지용 플래그 OFF
try:
self.client_socket.close() # 소켓 닫기
except Exception:
pass
self.client_socket = None
self.set_connected_state(False) # UI 상태를 끊김 상태로 변경
self.chat_display.append("[시스템] 서버와의 연결을 종료했습니다.")
@Slot(dict)
def display_message(self, message):
"""수신 스레드가 시그널로 던져준 메시지를 분석해서 실제 화면에 그려주는 슬롯 함수"""
message_type = message.get("type")
# 1. 일반 채팅 메시지일 때
if message_type == "chat":
sender = message.get("sender", "unknown")
content = message.get("content", "")
self.chat_display.append(f"{sender}: {content}")
# 2. 시스템 공지 메시지일 때
elif message_type == "system":
content = message.get("content", "")
self.chat_display.append(f"[시스템] {content}")
# 3. 에러 메시지일 때
elif message_type == "error":
content = message.get("content", "")
self.chat_display.append(f"[오류] {content}")
# 4. 유저 목록 업데이트 지시일 때 (서버가 주기적으로 전체 유저를 쏴줌)
elif message_type == "user_list":
users = message.get("users", [])
self.update_user_list(users)
# 5. 연결 상태가 변했을 때
elif message_type == "state":
connected = message.get("connected", False)
if not connected:
self.client_socket = None
self.set_connected_state(False)
# 6. 그 외 정의되지 않은 메시지 타입 처리
else:
self.chat_display.append(f"[알 수 없는 메시지] {message}")
def on_user_selected(self, item):
"""우측 리스트에서 특정 유저 아이템을 클릭했을 때 실행되는 함수"""
self.selected_user = item.text() # 클릭한 아이템의 텍스트(이름) 저장
self.selected_user_label.setText(f"선택한 상대: {self.selected_user}")
def update_user_list(self, users):
"""서버가 보내준 최신 접속자 명단으로 QListWidget을 갱신하는 함수"""
self.user_list.clear() # 기존 목록 싹 지우기
for user in users:
self.user_list.addItem(user) # 유저 한 명씩 추가
# 만약 내가 선택해뒀던 유저가 방을 나가서 최신 명단에 없다면 선택 해제 처리
if self.selected_user not in users:
self.selected_user = None
self.selected_user_label.setText("선택한 상대: 없음")
def closeEvent(self, event):
"""사용자가 우측 상단 'X' 버튼을 눌러서 프로그램을 끌 때 자동으로 호출되는 이벤트 함수"""
self.disconnect_from_server() # 그냥 꺼지지 말고 서버에 로그아웃 알린 후 소켓 닫기
event.accept() # 창 닫기 요청 승인
# --- 프로그램 시작점 (실행 루프) ---
app = QApplication(sys.argv) # PySide6 애플리케이션 생성
window = ChatWindow() # 내가 만든 채팅창 인스턴스 생성
window.show() # 화면에 띄우기
sys.exit(app.exec()) # GUI 이벤트 루프 시작 (창이 닫힐 때까지 대기 및 프로그램 정상 종료 유도)
28강 소스코드
chat_server / server.py
# 28강. 사용자 목록 동기화 ### 수정 구현한 부분
# server.py
import socket
import threading
# custom 프로토콜 모듈에서 메시지 수신/송신 함수를 가져옴
from protocol import receive_messages, send_message
# 서버가 실행될 IP 주소와 포트 번호 설정
HOST = "127.0.0.1"
PORT = 5000
# 현재 접속한 클라이언트들의 정보를 저장할 리스트 (딕셔너리 형태로 저장할 예정)
clients = []
def print_clients():
"""현재 서버에 접속 중인 클라이언트 수를 콘솔에 출력하는 함수"""
print(f"현재 접속자 수: {len(clients)}")
def get_user_list(): ### 2단계에서 구현
"""현재 접속 중인 모든 클라이언트의 IP/포트 주소 텍스트만 뽑아서 리스트로 반환함"""
return [client["address"] for client in clients]
def broadcast_user_list(): ### 2단계에서 구현
"""접속 중인 모든 클라이언트에게 현재 유저 리스트를 전송해서 동기화하는 함수"""
user_list_message = {
"type": "user_list",
"users": get_user_list() # 위에서 만든 주소 리스트를 담음
}
# clients[:]를 쓰는 이유는 루프 도중 리스트가 변경되어도 안전하게 복사본으로 돌리기 위함이야
for client in clients[:]:
try:
# 각각의 클라이언트 소켓으로 유저 리스트 메시지를 보냄
send_message(client["socket"], user_list_message)
except Exception:
# 에러가 나면 그냥 넘어가 (나중에 종료 처리될 테니까)
pass
def broadcast(message, sender_socket=None):
"""
메시지를 현재 접속한 모든 클라이언트에게 뿌리는 함수
단, sender_socket이 지정되면 메시지를 보낸 본인에게는 다시 보내지 않아
"""
for client in clients[:]:
client_socket = client["socket"] ### 딕셔너리 구조에 맞게 키값으로 소켓을 꺼냄
# 메시지를 보낸 사람이 아닌 다른 클라이언트들에게만 전송
if client_socket != sender_socket:
try:
send_message(client_socket, message)
except Exception:
pass
def send_error(client_socket, content):
"""클라이언트에게 에러 메시지를 정해진 프로토콜 양식에 맞춰 전송하는 함수"""
send_message(client_socket, {
"type": "error",
"content": content
})
def remove_client(client_socket): ###️ 딕셔너리에서 찾아 제거
"""
인자로 받은 소켓을 가진 클라이언트를 찾아서 리스트에서 지우는 함수
지운 클라이언트의 주소를 반환함
"""
for client in clients[:]:
if client["socket"] == client_socket:
clients.remove(client) # 리스트에서 삭제
return client["address"] # 삭제된 클라이언트 주소 반환
return None
def handle_client(client_socket, client_address):
"""클라이언트 한 명당 하나의 스레드로 실행되어 메시지 수신을 전담하는 함수"""
print(f"클라이언트 처리 시작: {client_address}")
buffer = "" # TCP 스트림 데이터 파싱을 위한 개별 버퍼 생성
address_text = str(client_address)
try:
# 새로운 유저가 입장했음을 본인을 제외한 다른 모든 사람에게 알림
broadcast({
"type": "system",
"content": f"{address_text} 님이 입장했습니다."
}, client_socket)
broadcast_user_list() ### 입장했으니 유저들에게 새로운 목록 동기화
# 메시지 수신 무한 루프 시작
while True:
try:
# 데이터를 수신하고 메시지가 잘라져 왔을 때를 대비해 버퍼를 업데이트함
buffer, messages, connected = receive_messages(client_socket, buffer)
# 연결이 끊겼다고 판단되면 루프 탈출
if not connected:
print(f"클라이언트 연결 끊김: {client_address}")
break
exit_requested = False
# 수신된 메시지들을 하나씩 처리
for message in messages:
print(f"{client_address} 메시지: {message}")
message_type = message.get("type")
# 1. 클라이언트가 종료 요청('exit')을 보냈을 때
if message_type == "exit":
print(f"클라이언트 종료 요청: {client_address}")
exit_requested = True
break # 메시지 처리 루프 탈출
# 2. 클라이언트가 일반 채팅('chat')을 보냈을 때
if message_type == "chat":
content = message.get("content", "").strip()
# 빈 메시지 방지 로직
if not content:
send_error(client_socket, "빈 메시지는 보낼 수 없습니다.")
continue # 다음 메시지로 넘어감
# 내용이 잘 있으면 다른 모든 클라이언트에게 채팅 내용 방송
broadcast({
"type": "chat",
"sender": address_text,
"content": content
}, client_socket)
continue
# 정해진 타입(exit, chat)이 아닐 경우 에러 전송
send_error(client_socket, f"지원하지 않는 메시지 타입입니다: {message_type}")
# exit 요청을 받았다면 수신 무한 루프 자체를 빠져나감
if exit_requested:
break
except Exception as error:
print(f"클라이언트 처리 오류: {client_address}, {error}")
break
finally: ###예외가 발생하든 루프를 빠져나왔든 무조건 실행해서 소켓 안전하게 정리
try:
client_socket.close() # 소켓 닫기
except Exception:
pass
# 접속자 리스트에서 해당 클라이언트 지우기
removed_address = remove_client(client_socket)
# 유저가 나갔음을 다른 유저들에게 전송
broadcast({
"type": "system",
"content": f"{address_text} 님이 퇴장했습니다."
}, client_socket)
broadcast_user_list() ###퇴장했으니 유저들에게 바뀐 목록 다시 동기화
print(f"접속자 목록에서 제거: {removed_address}")
print_clients()
print(f"클라이언트 처리 종료: {client_address}")
# 서버 소켓 생성 및 바인딩 (with 구문 사용으로 프로그램 종료 시 소켓 자동 반환)
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
# 소켓 재사용 옵션 설정 (서버가 강제 종료된 후 재실행할 때 포트가 묶여있는 에러 방지)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind((HOST, PORT)) # 설정한 IP와 포트에 서버 소켓 연결
server_socket.listen() # 클라이언트 접속 대기 상태로 진입
print("서버를 시작합니다.")
print("클라이언트 접속 대기 중...")
# 새로운 클라이언트를 계속해서 받아들이는 메인 루프
while True:
# 클라이언트가 접속하면 소켓 객체와 주소를 받아옴 (Blocking 방식)
client_socket, client_address = server_socket.accept()
address_text = str(client_address)
print(f"클라이언트 접속: {client_address}")
### 관리하기 편하도록 소켓과 주소를 하나의 딕셔너리로 묶어서 리스트에 추가
clients.append({
"socket": client_socket,
"address": address_text
})
print(f"접속자 목록에 추가: {client_address}")
print_clients()
# 각 클라이언트마다 독립된 스레드를 생성해서 handle_client 함수를 실행함
client_thread = threading.Thread(
target=handle_client,
args=(client_socket, client_address)
)
# 메인 스레드가 종료되면 이 스레드도 같이 강제 종료되도록 데몬 스레드로 설정
client_thread.daemon = True
client_thread.start() # 스레드 시작
3. 29강. 귓속말 메세지 전송 구현
29강 소스코드
chat_client / main.py
# 29강. 귓속말 메세지 전송 구현
import socket
import sys
import threading
# PySide6에서 UI 이벤트(Signal/Slot) 처리를 위한 모듈 임포트
from PySide6.QtCore import Signal, Slot
# PySide6에서 GUI 구성에 필요한 위젯 및 레이아웃 임포트
from PySide6.QtWidgets import (
QApplication,
QWidget,
QTextEdit,
QLineEdit,
QPushButton,
QLabel,
QListWidget,
QVBoxLayout,
QHBoxLayout,
)
# 별도로 정의된 프로토콜 모듈에서 메시지 송수신 함수 임포트
from protocol import receive_messages, send_message
# 서버 접속 정보 설정 (로컬 호스트, 포트 5000)
HOST = "127.0.0.1"
PORT = 5000
class ChatWindow(QWidget):
# 커스텀 시그널 정의: 서버로부터 받은 딕셔너리 형태의 메시지를 GUI 스레드로 전달 목적
message_received = Signal(dict)
def __init__(self):
super().__init__()
# 네트워크 및 스레드 관련 상태 변수 초기화
self.client_socket = None # 서버와 연결할 소켓
self.receive_thread = None # 메시지 수신을 담당할 백그라운드 스레드
self.running = False # 스레드 루프 제어 플래그
self.selected_user = None # 현재 귓속말 대상으로 선택된 유저 이름
# 메인 윈도우 창 설정
self.setWindowTitle("채팅")
self.resize(700, 480)
# --- UI 위젯 생성 ---
self.status_label = QLabel("상태: 서버에 연결되지 않음")
self.connect_button = QPushButton("서버 접속")
self.chat_display = QTextEdit()
self.chat_display.setReadOnly(True) # 채팅창은 읽기 전용으로 설정
self.chat_display.setPlaceholderText("채팅 메시지가 여기에 표시됩니다.")
self.user_list = QListWidget() # 접속자 목록을 표시할 리스트 위젯
self.user_list.setMaximumWidth(150) # 가로 최대 크기 제한
self.selected_user_label = QLabel("선택한 상대: 없음")
self.message_input = QLineEdit() # 메시지 입력창
self.message_input.setPlaceholderText("메시지를 입력하세요")
self.send_button = QPushButton("보내기")
self.whisper_button = QPushButton("귓속말")
# --- UI 레이아웃 배치 ---
# 1. 상단 레이아웃 (상태 표시, 접속 버튼)
top_layout = QHBoxLayout()
top_layout.addWidget(self.status_label)
top_layout.addWidget(self.connect_button)
# 2. 우측 접속자 목록 레이아웃 (선택된 상대 표시, 리스트)
user_layout = QVBoxLayout()
user_layout.addWidget(self.selected_user_label)
user_layout.addWidget(self.user_list)
# 3. 중앙 레이아웃 (채팅창 + 우측 접속자 목록)
content_layout = QHBoxLayout()
content_layout.addWidget(self.chat_display)
content_layout.addLayout(user_layout)
# 4. 하단 레이아웃 (메시지 입력창, 보내기 버튼, 귓속말 버튼)
input_layout = QHBoxLayout()
input_layout.addWidget(self.message_input)
input_layout.addWidget(self.send_button)
input_layout.addWidget(self.whisper_button)
# 5. 메인 레이아웃 (상단 + 중앙 + 하단 전체 세로 배치)
main_layout = QVBoxLayout()
main_layout.addLayout(top_layout)
main_layout.addLayout(content_layout)
main_layout.addLayout(input_layout)
self.setLayout(main_layout)
# --- 시그널과 슬롯(이벤트 핸들러) 연결 ---
self.connect_button.clicked.connect(self.connect_to_server) # 접속 버튼 클릭 시
self.send_button.clicked.connect(self.on_send_clicked) # 보내기 버튼 클릭 시
self.message_input.returnPressed.connect(self.on_send_clicked) # 입력창에서 엔터 키 입력 시
self.whisper_button.clicked.connect(self.on_whisper_clicked) # 귓속말 버튼 클릭 시
self.message_received.connect(self.display_message) # 메시지 수신 시그널 발생 시 GUI 업데이트
self.user_list.itemClicked.connect(self.on_user_selected) # 유저 목록 클릭 시
# 초기 상태 설정 (연결되지 않음 상태로 위젯 비활성화)
self.set_connected_state(False)
def set_connected_state(self, connected):
"""연결 상태에 따라 UI 위젯들의 활성화/비활성화 상태를 제어하는 함수"""
self.connect_button.setEnabled(not connected) # 연결되면 접속 버튼 비활성화
self.message_input.setEnabled(connected) # 연결되어야 입력창 활성화
self.send_button.setEnabled(connected) # 연결되어야 보내기 버튼 활성화
self.whisper_button.setEnabled(connected) # 연결되어야 귓속말 버튼 활성화
if connected:
self.status_label.setText("상태: 서버에 연결됨")
else:
self.status_label.setText("상태: 서버에 연결되지 않음")
self.user_list.clear()
self.selected_user = None
self.selected_user_label.setText("선택한 상대: 없음")
def connect_to_server(self):
"""서버 소켓 연결을 시도하고 수신용 백그라운드 스레드를 실행하는 함수"""
try:
# TCP 소켓 생성 및 서버 연결
self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.client_socket.connect((HOST, PORT))
# 백그라운드 메시지 수신 스레드 시작
self.running = True
self.receive_thread = threading.Thread(target=self.receive_loop)
self.receive_thread.daemon = True # 메인 프로세스 종료 시 함께 종료되도록 데몬 설정
self.receive_thread.start()
# UI 상태 업데이트
self.set_connected_state(True)
self.chat_display.append("[시스템] 서버에 연결되었습니다.")
except ConnectionRefusedError:
self.client_socket = None
self.connect_button.setEnabled(True)
self.status_label.setText("상태: 서버 연결 실패")
self.chat_display.append("[오류] 서버가 실행 중인지 확인하세요.")
except Exception as error:
self.client_socket = None
self.connect_button.setEnabled(True)
self.status_label.setText("상태: 서버 연결 실패")
self.chat_display.append(f"[오류] 연결 중 문제가 발생했습니다: {error}")
def receive_loop(self):
"""백그라운드 스레드에서 무한 루프를 돌며 서버의 메시지를 수신하는 함수"""
buffer = ""
while self.running:
try:
# protocol 모듈의 함수를 통해 메시지 파싱 및 수신
buffer, messages, connected = receive_messages(self.client_socket, buffer)
if not connected:
# 연결이 끊어졌다면 GUI 스레드로 시스템 알림 발생 후 루프 탈출
self.message_received.emit({
"type": "system",
"content": "서버와의 연결이 끊어졌습니다."
})
break
# 수신된 메시지가 있다면 순차적으로 GUI 스레드에 전송(Emit)
for message in messages:
self.message_received.emit(message)
except OSError:
# 소켓이 강제로 닫히는 등의 OS 에러 발생 시 루프 탈출
break
except Exception as error:
self.message_received.emit({
"type": "error",
"content": f"메시지 수신 중 오류가 발생했습니다: {error}"
})
break
# 루프가 끝나면 스레드를 정지하고 연결 종료 상태를 GUI에 알림
self.running = False
self.message_received.emit({
"type": "state",
"connected": False
})
def on_send_clicked(self):
"""일반 채팅 메시지를 전송하는 슬롯 함수"""
text = self.message_input.text().strip() # 공백 제거 후 텍스트 가져오기
if not text:
return
if self.client_socket is None:
self.chat_display.append("[오류] 서버에 먼저 접속하세요.")
return
# 'exit' 입력 시 접속 종료 처리
if text == "exit":
self.disconnect_from_server()
self.message_input.clear()
return
try:
# 프로토콜 규격(JSON 형태의 딕셔너리)에 맞춰 데이터 전송
send_message(self.client_socket, {
"type": "chat",
"content": text
})
# 내가 보낸 메시지를 화면에 출력하고 입력창 비우기
self.chat_display.append(f"나: {text}")
self.message_input.clear()
except Exception as error:
self.chat_display.append(f"[오류] 메시지 전송 실패: {error}")
self.disconnect_from_server()
def on_whisper_clicked(self):
"""선택된 상대에게 귓속말을 전송하는 슬롯 함수"""
text = self.message_input.text().strip()
if not text:
return
if self.client_socket is None:
self.chat_display.append("[오류] 서버에 먼저 접속하세요.")
return
if self.selected_user is None:
self.chat_display.append("[오류] 귓속말 상대를 선택하세요.")
return
try:
# 귓속말 프로토콜 데이터 전송 (대상 target 포함)
send_message(self.client_socket, {
"type": "whisper",
"target": self.selected_user,
"content": text
})
self.chat_display.append(f"[귓속말 보냄] {self.selected_user}: {text}")
self.message_input.clear()
except Exception as error:
self.chat_display.append(f"[오류] 귓속말 전송 실패: {error}")
self.disconnect_from_server()
def disconnect_from_server(self):
"""서버와의 연결을 안전하게 종료하는 함수"""
if self.client_socket is None:
self.set_connected_state(False)
return
try:
# 서버에 종료하겠다는 의사(exit)를 전달
send_message(self.client_socket, {"type": "exit"})
except Exception:
pass
self.running = False
try:
self.client_socket.close() # 소켓 닫기
except Exception:
pass
self.client_socket = None
self.set_connected_state(False)
self.chat_display.append("[시스템] 서버와의 연결을 종료했습니다.")
@Slot(dict)
def display_message(self, message):
"""message_received 시그널을 전달받아 타입별로 UI에 메시지를 출력하는 슬롯 함수 (GUI 스레드에서 실행됨)"""
message_type = message.get("type")
# 1. 일반 채팅 메시지 처리
if message_type == "chat":
sender = message.get("sender", "unknown")
content = message.get("content", "")
self.chat_display.append(f"{sender}: {content}")
# 2. 시스템 공지 메시지 처리
elif message_type == "system":
content = message.get("content", "")
self.chat_display.append(f"[시스템] {content}")
# 3. 에러 메시지 처리
elif message_type == "error":
content = message.get("content", "")
self.chat_display.append(f"[오류] {content}")
# 4. 실시간 유저 목록 업데이트 처리
elif message_type == "user_list":
users = message.get("users", [])
self.update_user_list(users)
# 5. 수신된 귓속말 처리
elif message_type == "whisper":
sender = message.get("sender", "unknown")
content = message.get("content", "")
self.chat_display.append(f"[귓속말 받음] {sender}: {content}")
# 6. 스레드 루프 종료에 따른 상태 비활성화 처리
elif message_type == "state":
connected = message.get("connected", False)
if not connected:
self.client_socket = None
30강. 귓속말 오류 처리와 표시 방식 실습하기 *** 가상의 주소와 서버 화면 실습 전용 ***
30강. 소스코드
chat_client / main.py
# 30강. 귓속말 오류 처리와 표시 방식
# main.py
import socket
import sys
import threading
# PySide6 구성 요소 임포트 (GUI 및 시그널/슬롯)
from PySide6.QtCore import Signal, Slot
from PySide6.QtWidgets import (
QApplication,
QWidget,
QTextEdit,
QLineEdit,
QPushButton,
QLabel,
QListWidget,
QVBoxLayout,
QHBoxLayout,
)
# 사용자 정의 프로토콜 모듈 임포트 (메시지 송수신 함수)
from protocol import receive_messages, send_message
# 서버 접속 정보 설정
HOST = "127.0.0.1"
PORT = 5000
class ChatWindow(QWidget):
# 멀티스레드에서 수신한 메시지를 메인 GUI 스레드로 전달하기 위한 시그널 정의
message_received = Signal(dict)
def __init__(self):
super().__init__()
# 클라이언트 내부 상태 관리 변수 초기화
self.client_socket = None # 서버와 연결할 소켓
self.receive_thread = None # 메시지 수신을 전담할 스레드
self.running = False # 수신 루프 실행 제어 플래그
self.selected_user = None # 리스트에서 선택된 귓속말 상대방 주소
self.my_address = None # 서버가 할당한 내 고유 주소
# GUI 창 기본 설정
self.setWindowTitle("채팅")
self.resize(700, 480)
# 상단 영역 위젯 (상태 표시 및 접속 버튼)
self.status_label = QLabel("상태: 서버에 연결되지 않음")
self.connect_button = QPushButton("서버 접속")
# 중앙 영역 위젯 (채팅창 및 유저 리스트)
self.chat_display = QTextEdit()
self.chat_display.setReadOnly(True) # 채팅창 수정 불가 설정
self.chat_display.setPlaceholderText("채팅 메시지가 여기에 표시됩니다.")
self.user_list = QListWidget() # 접속자 목록 표시 리스트
self.user_list.setMaximumWidth(150)
# 하단 영역 위젯 (귓속말 상대 표시, 입력창, 전송 버튼)
self.selected_user_label = QLabel("선택한 상대: 없음")
self.message_input = QLineEdit()
self.message_input.setPlaceholderText("메시지를 입력하세요")
self.send_button = QPushButton("보내기")
self.whisper_button = QPushButton("귓속말")
# UI 레이아웃 배치 - 상단 조립
top_layout = QHBoxLayout()
top_layout.addWidget(self.status_label)
top_layout.addWidget(self.connect_button)
# UI 레이아웃 배치 - 우측 접속자 목록 조립
user_layout = QVBoxLayout()
user_layout.addWidget(self.selected_user_label)
user_layout.addWidget(self.user_list)
# UI 레이아웃 배치 - 중앙 콘텐츠 조립
content_layout = QHBoxLayout()
content_layout.addWidget(self.chat_display)
content_layout.addLayout(user_layout)
# UI 레이아웃 배치 - 하단 입력부 조립
input_layout = QHBoxLayout()
input_layout.addWidget(self.message_input)
input_layout.addWidget(self.send_button)
input_layout.addWidget(self.whisper_button)
# 전체 메인 레이아웃 병합 및 등록
main_layout = QVBoxLayout()
main_layout.addLayout(top_layout)
main_layout.addLayout(content_layout)
main_layout.addLayout(input_layout)
self.setLayout(main_layout)
# 위젯 이벤트(시그널)와 제어 메서드(슬롯) 연결
self.connect_button.clicked.connect(self.connect_to_server) # 접속 버튼 클릭
self.send_button.clicked.connect(self.on_send_clicked) # 보내기 버튼 클릭
self.message_input.returnPressed.connect(self.on_send_clicked) # 입력창 엔터키 입력
self.whisper_button.clicked.connect(self.on_whisper_clicked) # 귓속말 버튼 클릭
self.message_received.connect(self.display_message) # 메시지 수신 시그널 발생
self.user_list.itemClicked.connect(self.on_user_selected) # 유저 목록 클릭
# 초기 화면 상태 설정 (미연결 상태 위젯 비활성화)
self.set_connected_state(False)
def show_error(self, content):
"""오류 메시지를 채팅창에 붉은 텍스트 등으로 구분하여 표시하기 위한 래퍼 메서드"""
self.chat_display.append(f"[오류] {content}")
def show_system(self, content):
"""시스템 안내 메시지를 채팅창에 표시하기 위한 래퍼 메서드"""
self.chat_display.append(f"[시스템] {content}")
def set_connected_state(self, connected):
"""연결 상태에 따라 UI 위젯들의 활성화/비활성화 및 텍스트를 제어하는 메서드"""
self.connect_button.setEnabled(not connected) # 연결 상태면 접속 버튼 비활성화
self.message_input.setEnabled(connected)
self.send_button.setEnabled(connected)
self.whisper_button.setEnabled(connected)
if connected:
self.status_label.setText("상태: 서버에 연결됨")
else:
self.status_label.setText("상태: 서버에 연결되지 않음")
self.user_list.clear() # 접속 끊김 시 유저 목록 초기화
self.selected_user = None # 선택된 상대 초기화
self.selected_user_label.setText("선택한 상대: 없음")
self.my_address = None # 내 주소 초기화
def connect_to_server(self):
"""서버 접속을 시도하고 성공 시 수신 백그라운드 스레드를 시작하는 메서드"""
try:
# TCP 소켓 생성 및 서버 연결
self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.client_socket.connect((HOST, PORT))
self.my_address = str(self.client_socket.getsockname()) # 내 고유 IP/Port 저장
# 백그라운드 메시지 수신 스레드 생성 및 실행
self.running = True
self.receive_thread = threading.Thread(target=self.receive_loop)
self.receive_thread.daemon = True # 메인 스레드 종료 시 함께 종료되도록 설정
self.receive_thread.start()
self.set_connected_state(True)
self.show_system("서버에 연결되었습니다.")
except ConnectionRefusedError:
self.client_socket = None
self.connect_button.setEnabled(True)
self.status_label.setText("상태: 서버 연결 실패")
self.show_error("서버가 실행 중인지 확인하세요.")
except Exception as error:
self.client_socket = None
self.connect_button.setEnabled(True)
self.status_label.setText("상태: 서버 연결 실패")
self.show_error(f"연결 중 문제가 발생했습니다: {error}")
def receive_loop(self):
"""백그라운드에서 서버가 보낸 데이터를 지속적으로 수신하는 무한 루프 메서드 (스레드 전용)"""
buffer = ""
while self.running:
try:
# protocol 모듈을 통해 메시지 단위 분할 수신
buffer, messages, connected = receive_messages(self.client_socket, buffer)
if not connected: # 서버가 연결을 끊은 경우
self.message_received.emit({
"type": "system",
"content": "서버와의 연결이 끊어졌습니다."
})
break
# 수신된 메시지들을 하나씩 메인 GUI 스레드로 전달
for message in messages:
self.message_received.emit(message)
except OSError: # 소켓이 강제로 닫힌 경우 루프 종료
break
except Exception as error:
self.message_received.emit({
"type": "error",
"content": f"메시지 수신 중 오류가 발생했습니다: {error}"
})
break
# 루프가 끝나면 연결 종료 처리 시그널 발생
self.running = False
self.message_received.emit({
"type": "state",
"connected": False
})
def on_send_clicked(self):
"""일반 전체 메시지를 전송하는 메서드"""
text = self.message_input.text().strip()
if not text: # 빈 메시지 전송 차단
return
if self.client_socket is None:
self.show_error("서버에 먼저 접속하세요.")
return
if text == "exit": # 특정 명령어 입력 시 연결 종료
self.disconnect_from_server()
self.message_input.clear()
return
try:
# 프로토콜 형태(JSON 규격 dict)에 맞춰 일반 채팅 전송
send_message(self.client_socket, {
"type": "chat",
"content": text
})
self.chat_display.append(f"나: {text}") # 내가 보낸 메시지 화면에 출력
self.message_input.clear()
except Exception as error:
self.show_error(f"메시지 전송 실패: {error}")
self.disconnect_from_server()
def on_whisper_clicked(self):
"""선택한 상대방에게 귓속말을 전송하는 메서드 (예외 처리 기능 포함)"""
text = self.message_input.text().strip()
# 예외 처리 1: 미연결 상태 검증
if self.client_socket is None:
self.show_error("서버에 먼저 접속하세요.")
return
# 예외 처리 2: 내용 빈칸 검증
if not text:
self.show_error("귓속말 내용을 입력하세요.")
return
# 예외 처리 3: 귓속말 대상 미선택 검증
if self.selected_user is None:
self.show_error("귓속말 상대를 먼저 선택하세요.")
return
# 예외 처리 4: 본인에게 전송하는지 검증
if self.selected_user == self.my_address:
self.show_error("자기 자신에게는 귓속말을 보낼 수 없습니다.")
return
try:
# 프로토콜 형태에 맞춰 대상(target)을 지정하여 귓속말 전송
send_message(self.client_socket, {
"type": "whisper",
"target": self.selected_user,
"content": text
})
# 내 화면에 귓속말 발신 내역 기록
self.chat_display.append(f"[귓속말 보냄] {self.selected_user}: {text}")
self.message_input.clear()
except Exception as error:
self.show_error(f"귓속말 전송 실패: {error}")
self.disconnect_from_server()
def on_user_selected(self, item):
"""유저 목록 리스트 위젯의 아이템을 클릭했을 때 선택 항목을 업데이트하는 슬롯"""
self.selected_user = item.text()
self.selected_user_label.setText(f"선택한 상대: {self.selected_user}")
def update_user_list(self, users):
"""서버가 동기화해준 전체 유저 리스트를 UI 목록에 갱신하는 메서드"""
self.user_list.clear()
for user in users:
# 내 주소는 목록에 노출하되 별도 표시를 해주거나 구분을 위해 유지
if user == self.my_address:
self.user_list.addItem(f"{user} (나)")
else:
self.user_list.addItem(user)
def disconnect_from_server(self):
"""소켓을 안전하게 닫고 상태를 미연결로 되돌리는 종료 메서드"""
if self.client_socket is None:
self.set_connected_state(False)
return
try:
send_message(self.client_socket, {"type": "exit"}) # 서버에 종료 알림 전송
except Exception:
pass
self.running = False
try:
self.client_socket.close() # 소켓 해제
except Exception:
pass
self.client_socket = None
self.set_connected_state(False)
self.show_system("서버와의 연결을 종료했습니다.")
@Slot(dict)
def display_message(self, message):
30강 소스코드
chat_server / server.py
# 30강. 귓속말 오류 처리와 표시 방식
# server.py
import socket
import threading
from protocol import receive_messages, send_message
# 서버 접속 정보 설정 (로컬 호스트 및 포트 5000)
HOST = "127.0.0.1"
PORT = 5000
# 접속한 클라이언트 정보(딕셔너리)들을 저장할 리스트
clients = []
# 현재 접속 중인 클라이언트 수를 콘솔에 출력하는 함수
def print_clients():
print(f"현재 접속자 수: {len(clients)}")
# 현재 접속자들의 고유 주소(IP, Port 문자열) 목록을 반환하는 함수
def get_user_list():
return [client["address"] for client in clients]
# 모든 클라이언트에게 최신 접속자 명단을 전송하는 함수
def broadcast_user_list():
user_list_message = {
"type": "user_list",
"users": get_user_list()
}
# 리스트 복사본([:])을 사용하여 안전하게 순회하며 명단 전송
for client in clients[:]:
try:
send_message(client["socket"], user_list_message)
except Exception:
pass # 전송 실패 시 무시 (연결 끊김은 다른 곳에서 처리)
# 메시지를 보낸 사람을 제외한 모든 접속자에게 메시지를 브로드캐스트하는 함수
def broadcast(message, sender_socket=None):
for client in clients[:]:
client_socket = client["socket"]
# 본인이 아닌 경우에만 메시지 전송
if client_socket != sender_socket:
try:
send_message(client_socket, message)
except Exception:
pass
# 특정 클라이언트에게 에러 메시지를 형식을 갖추어 전송하는 함수
def send_error(client_socket, content):
send_message(client_socket, {
"type": "error",
"content": content
})
# 접속을 종료한 클라이언트를 목록에서 제거하고 해당 주소를 반환하는 함수
def remove_client(client_socket):
for client in clients[:]:
if client["socket"] == client_socket:
clients.remove(client)
return client["address"]
return None
# 주소(address) 또는 닉네임(nickname)으로 특정 클라이언트를 검색하는 함수
def find_client_by_identifier(identifier):
for client in clients:
if client["address"] == identifier:
return client
if client.get("nickname") == identifier:
return client
return None
# 지정한 대상(타겟)에게 일대일 귓속말 메시지를 전송하는 함수
def send_whisper(sender_identifier, target_identifier, content):
target_client = find_client_by_identifier(target_identifier)
# 대상이 접속 리스트에 없는 경우 실패 반환
if target_client is None:
return False
try:
# 대상 클라이언트 소켓으로 귓속말 패킷 전송
send_message(target_client["socket"], {
"type": "whisper",
"sender": sender_identifier,
"content": content
})
return True
except Exception:
return False
# 개별 클라이언트와의 통신을 독립적으로 처리하는 스레드 함수
def handle_client(client_socket, client_address):
print(f"클라이언트 처리 시작: {client_address}")
buffer = "" # TCP 데이터 스트림을 메시지 단위로 조립하기 위한 버퍼
address_text = str(client_address)
try:
# 클라이언트가 입장했음을 다른 모든 사용자에게 알림
broadcast({
"type": "system",
"content": f"{address_text} 님이 입장했습니다."
}, client_socket)
# 새로운 유저가 들어왔으므로 유저 목록 갱신 및 전송
broadcast_user_list()
# 메시지 수신 대기 루프
while True:
try:
# 프로토콜 모듈을 통해 버퍼와 메시지 패킷 파싱
buffer, messages, connected = receive_messages(client_socket, buffer)
# 클라이언트가 소켓 연결을 정상/비정상 종료한 경우 루프 탈출
if not connected:
print(f"클라이언트 연결 끊김: {client_address}")
break
exit_requested = False
# 파싱된 메시지들을 순차적으로 처리
for message in messages:
print(f"{client_address} 메시지: {message}")
message_type = message.get("type")
# 1. 종료(exit) 메시지 처리
if message_type == "exit":
print(f"클라이언트 종료 요청: {client_address}")
exit_requested = True
break
# 2. 일반 채팅(chat) 메시지 처리
if message_type == "chat":
content = message.get("content", "").strip()
if not content:
send_error(client_socket, "빈 메시지는 보낼 수 없습니다.")
continue
# 보낸 사람 제외 전체 사용자에게 채팅 전달
broadcast({
"type": "chat",
"sender": address_text,
"content": content
}, client_socket)
continue
# 3. 귓속말(whisper) 메시지 처리 및 오류 예외 처리
if message_type == "whisper":
target = message.get("target", "")
content = message.get("content", "").strip()
# 검증 1: 귓속말 대상 지정 여부 확인
if not target:
send_error(client_socket, "귓속말 대상을 선택하세요.")
continue
# 검증 2: 귓속말 내용 입력 여부 확인
if not content:
send_error(client_socket, "귓속말 내용을 입력하세요.")
continue
# 검증 3: 본인에게 보내는지 확인
if target == address_text:
send_error(client_socket, "자기 자신에게는 귓속말을 보낼 수 없습니다.")
continue
# 귓속말 전송 시도
success = send_whisper(address_text, target, content)
# 전송 실패 시 (대상이 방을 나간 경우 등) 처리
if not success:
send_error(client_socket, "선택한 사용자가 더 이상 접속 중이 아닙니다.")
broadcast_user_list() # 동기화를 위해 목록 재갱신
continue
# 4. 알 수 없는 메시지 타입 처리
send_error(client_socket, f"지원하지 않는 메시지 타입입니다: {message_type}")
if exit_requested:
break
except Exception as error:
print(f"클라이언트 처리 오류: {client_address}, {error}")
break
# 연결 끊김, 루프 종료, 에러 발생 시 최종 자원 정리 구간
finally:
try:
client_socket.close() # 클라이언트 소켓 닫기
except Exception:
pass
# 전역 접속자 리스트에서 제거
removed_address = remove_client(client_socket)
# 다른 사용자들에게 퇴장 알림 공지
broadcast({
"type": "system",
"content": f"{address_text} 님이 퇴장했습니다."
}, client_socket)
# 유저 목록 리프레시 공지
broadcast_user_list()
print(f"접속자 목록에서 제거: {removed_address}")
print_clients()
print(f"클라이언트 처리 종료: {client_address}")
# 메인 서버 소켓 생성 및 설정
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
# 소켓 재사용 옵션 설정 (서버 재시작 시 포트 바인딩 에러 방지)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind((HOST, PORT)) # 지정한 주소와 포트에 소켓 결합
server_socket.listen() # 클라이언트 연결 요청 대기 상태 진입
print("서버를 시작합니다.")
print("클라이언트 접속 대기 중...")
# 무한 루프를 돌며 다중 클라이언트 접속을 수락
while True:
# 새로운 클라이언트 접속 승인
client_socket, client_address = server_socket.accept()
address_text = str(client_address)
print(f"클라이언트 접속: {client_address}")
# 접속자 관리 리스트에 소켓 및 주소 정보 등록
clients.append({
"socket": client_socket,
"address": address_text
})
print(f"접속자 목록에 추가: {client_address}")
print_clients()
# 멀티스레딩을 통해 각 클라이언트 전용 handle_client 함수 실행
client_thread = threading.Thread(
target=handle_client,
args=(client_socket, client_address)
)
client_thread.daemon = True # 메인 스레드가 종료되면 서브 스레드도 함께 종료되도록 설정
client_thread.start() # 스레드 시작
31강. 텍스트와 바이너리 데이터의 차이
가. 텍스트 데이터와 바이너리 데이터의 핵심 차이는 사람이 직접 읽을 수 있는가와 컴퓨터가 효율적으로 처리하는가의 여부
나. 모든 데이터는 컴퓨터 내부에서 0과 1의 이진수(바이너리) 형태로 저장되지만, 이를 해석하고 표현하는 방식에서 차이가 발생
텍스트 데이터 (Text Data) "사람이 이해할 수 있는 문자 코드로 이루어진 데이터"
- 가독성: 메모장 같은 편집기로 열었을 때 사람이 바로 읽고 이해 가능
- 인코딩: 문자를 표현하기 위해 ASCII, UTF-8, CP949 같은 문자 인코딩 규격을 사용
- 데이터 구조: 줄바꿈 문자(\n, \r\n)와 공백을 기준으로 데이터의 경계를 구분
- 대표 예시: TXT, HTML, JSON, XML, 소스 코드 파일(.py, .java 등)
다. 바이너리 데이터 (Binary Data) : 컴퓨터가 직접 읽고 실행하거나 특정 프로그램이 해석해야 하는 데이터
- 가독성: 메모장으로 열면 깨진 문자나 알 수 없는 기호()
- 효율성: 문자에 국한되지 않고 이미지, 소리, 실행 명령 등을 0과 1로 압축하여 저장하므로 공간 효율성이 높다.
- 데이터 구조: 고정된 크기(바이트)의 헤더 정보나 파일 포맷 구조에 따라 데이터를 해석
- 대표 예시: 이미지(PNG, JPG), 오디오/비디오(MP3, MP4), 실행 파일(EXE), 압축 파일(ZIP) 및 일반적인 BIN 파일
32강. 파일 전송 개념과 메타데이터 설계
가. 파일 전송(File Transfer)과 메타데이터 설계(Metadata Design)는 시스템 간 데이터를 안전하고 정확하게 주고받기 위한 개념. 데이터 유실 없이 전송되려면 파일 본문뿐만 아니라, 그 파일을 설명하는 구조화된 데이터(메타데이터)의 명확한 정의가 필수적
나. 파일 전송 (File Transfer) 핵심 개념 및 용어
파일을 한 시스템(송신측)에서 다른 시스템(수신측)으로 이동하는 과정에서 사용되는 기술적 용어
- 페이로드 (Payload) : 전송되는 데이터의 실제 본문 내용을 의미. 헤더나 메타데이터를 제외한 순수 파일 데이터
- 프로토콜 (Protocol) : 컴퓨터 간에 파일을 안전하게 주고받기 위해 약속한 통신 규약 (예: FTP, SFTP, HTTP/HTTPS)
- 체크섬 (Checksum) / 해시 (Hash): 전송된 파일이 원래 파일과 완전히 일치하는지(무결성) 검증하기 위해 산출하는 고유한 문자열 값 (예: MD5, SHA-256)
- 청크 (Chunk) / 블록 (Block): 대용량 파일을 한 번에 보내지 않고 네트워크 과부하를 막기 위해 쪼갠 작은 데이터 단위
- 이어받기 (Resume / Partial Content): 전송 중 네트워크가 끊겼을 때 처음부터 다시 받지 않고, 중단된 지점부터 이어서 전송하는 기술
다. 메타데이터 설계 (Metadata Design) 용어 정의
메타데이터란 "데이터를 설명하는 데이터"이며, 이를 시스템에 맞게 표준화하고 구조를 짜는 것을 메타데이터 설계
※ 메타데이터의 3대 분류
- 기술(설명) 메타데이터 (Descriptive Metadata): 콘텐츠를 검색하고 식별하기 위한 정보 (예: 파일 이름, 작성자, 키워드)
- 구조 메타데이터 (Structural Metadata): 복합적인 파일들이 어떻게 결합해 있는지 관계를 나타냄 (예: 압축파일 내부 폴더 구조, 비디오와 자막 파일의 연결 관계)
- 관리 메타데이터 (Administrative Metadata): 파일의 관리, 권한, 보존을 위한 정보 (예: 생성 일시, 파일 크기, 접근 권한, DRM 보안 정보)
※ 설계 단계별 핵심 용어
- 데이터 사전 (Data Dictionary): 메타데이터 설계 시 사용되는 모든 데이터 요소(필드)의 이름, 의미, 타입, 제약조건을 기록해 둔 명세서
- 스키마 (Schema): 메타데이터가 어떤 구조와 관계를 가지고 저장되는지 정의한 전체적인 데이터 구조의 틀
- 데이터 표준화 (Data Standardization): 시스템 간 데이터 오해를 없애기 위해 용어의 단어, 형식, 코드 구성을 일관되게 규칙화하는 작업
- 네임스페이스 (Namespace): 메타데이터 요소의 이름이 서로 충돌하지 않도록 도메인별로 범위를 구분해 놓은 이름 공간
라. 파일 전송용 메타데이터 설계 예시 (JSON 포맷)
실제 개발 환경에서 파일 전송 기능을 구현할 때, 파일 본문(바이너리)과 함께 전송하는 메타데이터 스키마 설계의 직관적인 예시
예시 (json)
{
"file_id": "FILE_20260625_001", // 파일 고유 식별자 (ID)
"file_name": "report_v2.pdf", // 파일 이름
"file_size_bytes": 1048576, // 파일 크기 (바이트 단위)
"mime_type": "application/pdf", // 파일 형식 (바이너리/텍스트 종류 구분)
"checksum_sha256": "a1b2c3d4...", // 무결성 검증용 해시값
"created_at": "2026-06-25T10:11:00Z"// 생성 일시 (표준 시각 포맷)
}
33강. 파일을 조각으로 나누어 보내기
1. 테스트 스크립트로 전송 확인 하기
main.py에 파일 버튼 미연결 상태하 임시 테스트 파일로 확인해 본다.
33강 테스트 소스코드
# test_send_file.py (임시 테스트 파일 — 확인 후 삭제 가능)
import socket
import sys
sys.path.insert(0, "chat_client")
from protocol import send_file
HOST = "127.0.0.1"
PORT = 5000
test_path = sys.argv[1] if len(sys.argv) > 1 else "sample.png"
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.connect((HOST, PORT))
success = send_file(sock, test_path, "tester")
if success:
print(f"[파일 전송 완료] {test_path}")
else:
print(f"[파일 전송 실패] {test_path}")~ $ 터미널 1 >> python chat_server/server.py
~ $ 터미널 2 >> python test_send_file.py sample.png
※ 서버에 file_info, file_chunk, file_end 처리 코드가 없어서 오류가 발생할 수 있고, 34강에서 서버와 클라이언트의 수신 처리 코드를 추가해서 실습해 본다.
33강 소스코드
chat_client / protocol.py
import json
import base64
import os
import time
CHUNK_SIZE = 65536
def send_message(sock, message):
data = json.dumps(message).encode("utf-8")
length = len(data)
sock.sendall(length.to_bytes(4, "big") + data)
def recv_message(sock):
raw_length = sock.recv(4)
if not raw_length:
return None
length = int.from_bytes(raw_length, "big")
data = b""
while len(data) < length:
packet = sock.recv(length - len(data))
if not packet:
return None
data += packet
return json.loads(data.decode("utf-8"))
def send_file(sock, file_path, sender):
filename = os.path.basename(file_path)
size = os.path.getsize(file_path)
file_id = f"{filename}_{int(time.time() * 1000)}"
send_message(sock, {
"type": "file_info",
"file_id": file_id,
"filename": filename,
"size": size,
"sender": sender
})
try:
with open(file_path, "rb") as file:
while True:
chunk = file.read(CHUNK_SIZE)
if not chunk:
break
encoded = base64.b64encode(chunk).decode("utf-8")
send_message(sock, {
"type": "file_chunk",
"file_id": file_id,
"data": encoded
})
except Exception as error:
send_message(sock, {
"type": "error",
"content": f"파일 전송 실패: {error}"
})
return False
send_message(sock, {
"type": "file_end",
"file_id": file_id
})
return True끝.
'학습 로그 > Python3' 카테고리의 다른 글
| D+53[Python] PySide6 GUI 연동하기(22~26강) (0) | 2026.06.23 |
|---|---|
| D+52[Python] 다중 접속과 브로드캐스팅메시지 & 프로토콜 (0) | 2026.06.22 |
| D+51 [Python] 채팅서버 용어의 이해 및 예시문제 실습 (0) | 2026.06.19 |
| D+50[Python] PySide6 & Server-Client 예제 실습 (0) | 2026.06.19 |
| [D+50] LLM (Gemini, lovable, gabia) 활용한 개인 웹페이지 제작(실습) (0) | 2026.06.18 |