D+55 [Project] 팀프로젝트 채팅앱 프로그램(개발 계획)

ㅇ 일 자 : 2026년 6월 25일(목)~7. 2.(목) 

ㅇ 주 제 : 팀 프로젝트(2), 사내 개발자 채팅앱 프로그램 개발

ㅇ 목 표 : Pyside6 & GUI를 활용한 두번째 팀프로젝트  
              - 팀채널 : 사내 개발팀 소통, 배포, 오픈, 클로즈,

              - 커뮤니티 : 프로젝트별 방, 실시간 커뮤니티, 소스코드 전송(파일 보내기) 

 

1. 개발 계획 산출물

1.1. 산출물 목차

[개발 계획 단계] 1. 프로그램 개발 계획서 2. 유스케이스 3. 요구사항 분석서 4. 순서도 5. 화면구성도 6. 목업 7. 일정표 8. API 명세서 9. 테이블 명세서 10. ERD(엔티티 관계도)

[개발 완료 단계] 프로젝트 완료 보고서     1. 프로젝트 소개     2. 프로그램 구현 영상 및 스크린샷     3. 요구사항 명세서    4. 유스케이스    5. 순 서 도    6. 메뉴 구조도    7. 개발 일정표    8. 프로그램 구현 파일      8.1. 패키지 파일 목록      8.2. 소스 코드    9. 프로그램 개발 후기   10. 프로그램 개발 일지

 

1.2. 개발 문서 핵심 용어 정의

• 개발 계획서: 프로젝트의 목표, 범위, 예천 예산, 인력 구성, 추진 방향을 총망라하여 기록한 전체 로드맵 문서
• 유스케이스: 사용자의 관점에서 시스템이 제공해야 하는 기능과 상호작용 과정을 시나리오 형태로 표준화하여 표현한 것
• 요구사항 명세서: 구축할 시스템이 반드시 갖추어야 할 기능적·비기능적 조건과 제약사항을 명확하고 구체적으로 기록한 문서
• 테이블 명세서: 데이터베이스에 저장될 테이블의 구조, 데이터 타입, 주키(PK) 여부, 컬럼 설명 등을 상세히 기록한 설계도
• 순서도: 프로그램이나 프로세스의 처리 절차와 논리적 흐름을 표준화된 기호(도형)와 화살표로 시각화한 다이어그램
• 개발 일정표: 프로젝트의 전체 기간 동안 수행할 작업들을 일정에 따라 배치하고, 마일스톤과 담당자를 지정해 둔 일정 관리표(예: 간트차트, 색상그래프 형식 등)

1.3. 폭포수 문서의 유기적 관계 (동작 방식)

• 강한 종속성(Dependency): 요구사항 명세서가 승인되지 않으면 유스케이스를 작성할 수 없고, 설계가 끝나지 않으면 개발 일정표와 코딩을 시작할 수 없다.
• 계약서의 역할: 각 문서는 고객과의 합의점(Baseline)이 됩니다. 문서에 없는 기능은 개발하지 않으며, 변경하려면 복잡한 '변경 관리 절차'를 거쳐야 함
• 이정표(Milestone) 기반 관리: 개발 일정표는 프로젝트 초기에 정한 종료일을 맞추기 위해 엄격하게 통제되며, 각 문서의 완료 시점이 중요한 마일스톤이다.

1.4. 권장 작성 순서 : 프로젝트는 큰 그림에서 시작하여 점차 구체적인 설계로 진행된다.

1. 개발 계획서 (전체 방향 설정)
2. 요구사항 명세서 (필요한 기능 정의)
3. 유스케이스 (기능의 구체적 시나리오화)
4. 순서도 & 테이블 명세서 (동시 진행: 로직 및 데이터 설계)
5. 개발 일정표 (구체적 업무 기반 일정 산정)

1.5. 문서 간의 유기적 관계

• 개발 계획서 → 요구사항 명세서
  • 계획서에 담긴 프로젝트의 '범위(Scope)'를 바탕으로, 구체적으로 무엇(What)을 개발할지 기능별로 쪼개어 요구사항 명세서를 작성합니다.
• 요구사항 명세서 → 유스케이스
  • 요구사항 명세서의 기능들을 사용자가 실제로 어떻게 사용하는지(How) 사용자 관점의 시나리오로 구체화한 것이 유스케이스입니다.
• 유스케이스 → 순서도 & 테이블 명세서
  • 유스케이스의 상호작용 흐름을 시스템의 논리적 절차로 바꾸면 순서도가 됩니다.
  • 유스케이스 시나리오에서 오고 가는 데이터(회원 정보, 게시글 등)를 추출하여 저장할 공간을 만들면 테이블 명세서가 됩니다.
• 구체적 설계(순서도/테이블) → 개발 일정표
  • 구현해야 할 로직(순서도)과 데이터 구조(테이블)의 복잡도를 파악해야만, 개발자별로 정확한 작업 소요 시간(M/M)을 계산하여 현실적인 개발 일정표를 짤 수 있습니다.

 

2. 프로젝트 계획서

2.1. 프로젝트 개발 계획서 : 별도 첨부 

2.2. Protocol 함수 작성
- 파일 선택 : [Client / Protocol]사용자가 전송할 파일 선택,

- Base64 전송 : [Client / Protocol] 선택한 파일을 Base64로 변환하여 서버 전송
- 파일 수신 : [Client / Protocol] 채팅방 참여자가 전송된 파일 정보를 수신
- 파일 정보 표시 : [Client / Protocol] 파일명, 파일 크기, 전송 상태 표시
- 파일 오류 처리 : [Client / Protocol] 파일 미선택, 빈 파일, 전송 실패 안내

2.3. Server 와  Client  연동할 때 구성 정리 및 "예시코드" 참고

※ 서버에서 클라이언트 응답 메세지는 password, id 같은 요청 필드가 없는 구조이므로 응답 전용메세지는  validate() 대신 빌더 함수 만으로 생성함

 

[protocol.py]

클래스(함수)	내     용
MsgType	login, message  타입 상수 + all 집합
field	id, roon_id, 필드키 상수
validate( )	protocol Error 발생
build_*( )	클라이언트--> 서버   build_login, build_mesage, build_file 8개
build_*_response()	서버--> 클라이언트  build_login_response, build_system, build_error 7개
encode() / decode_lie()	직력, 역직렬화 유틸
strcamParser	버퍼  TCP 스트림 파서 (client, server) 대처
decode_file_data()	base64 파일 데이터 디스크 저장
__main__	빌더, 검증, 파서 동작  test

 

Protocol.py

[client.py]

from protocol import build_login, encode, Stream Parser, MsgType

# 빌더로 페이로더 생성
Pyload = build_login(uid, password))
self._sook.sendall(encode (payload))

#수신루프를 streamparser로 교체
parser = streamparser()
for mag in parser.fead(chunk):
    self._handle_received(msg)

 

[server.py]

from Protocol improt validate, build_login_response, build_error, ProtocolError.enc
try:
    validate(data)
except ProtocolError ase:
    send_json(conn,build_error(Str(e)))
        Continue
# 응답 빌더 사용
Send_Json(conn, build_login_response (Success = True, uid = uid))

 

[학습 프로그램 예시] 

1. server.py

# server.py

import socket
import threading

from protocol import receive_messages, send_message

HOST = "127.0.0.1"
PORT = 5000

clients = []

users ={
    "alice": "1234",
    "bob": "abcd"
}

def print_clients():
    print(f"현재 접속자 수: {len(clients)}")

def get_user_list():  # 
    return [client["address"] for client in clients]

def broadcast_user_list():  # 
    user_list_message = {
        "type": "user_list",
        "user_list": get_user_list()
    }

    for client in clients[:]:
        try:
            send_message(client["socket"], user_list_message)
        except Exception:
            if client in clients:
                clients.remove(client)

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:
                if client in clients:
                    clients.remove(client)

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 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 = ""
    address_text = str(client_address)

    broadcast({
        "type": "system",
        "content": f"{address_text} 님이 입장했습니다."
    }, client_socket)
    broadcast_user_list()  #  입장 후 목록 동기화

    with client_socket:
        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 == "register":
                        username = message.get("username",)
                        password = message.get("password",)

                        if username in users:
                            send_message(client_socket, {
                                "type": "auth_result",
                                "success": False,
                                "message" : "이미 사용중인 아이디 입니다."
                            })
                        else:
                            users[username] = password
                            send_message(client_socket, {
                                "type": "auth_result",
                                "success": True,
                                "message" : "회원가입 완료, 로그인 하세요"
                            })
                        continue

                    # 로그인 처리
                    if message_type == "login":
                        username = message.get("username",)
                        password = message.get("password",)

                        if users.get(username) == password:
                            send_message(client_socket, {
                                "type": "auth_result",
                                "success": True,
                                "message" : f"{username}님 환영합니다."
                            })
                        else:
                            send_message(client_socket, {
                                "type": "auth_result",
                                "success": False,
                                "message" : "아이디 또는 비밀번호가 틀렸습니다."
                            })
                        continue

                    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

                    if message_type == "whisper":
                        target = message.get("target", "")
                        content = message.get("content", "")
                        if not target:
                            send_error(client_socket, "귓속말 대상을 선택하세요")
                            continue
                        if not content:
                            send_error(client_socket, "귓속말 내용을 입력하세요")
                            continue
                        success = send_whisper(address_text, target, content)
                        if target == address_text:
                            send_error(client_socket, "자기 자신에게는 귓속말을 보낼 수 없습니다.")
                        if not success:
                            send_error(client_socket, "귓속말 대상을 찾을 수 없습니다.")
                            broadcast_user_list()
                            continue
                        continue

                    send_error(client_socket, f"지원하지 않는 메시지 타입입니다: {message_type}")

                if exit_requested:
                    break

            except Exception as error:
                print(f"클라이언트 처리 오류: {client_address}, {error}")
                break

    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.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.start()

       

2. protocol.py

# 1팀 사내 채팅앱 프로젝트  protocol.py

import json
import base64
import os
import time

# 파일 전송 시 사용할 청크(조각)의 크기 설정 (64KB)
CHUNK_SIZE = 65536

def send_message(sock, message):
    # 소켓을 통해 JSON 메시지를 전송하는 함수
    # 메시지 길이를 먼저 4바이트(Big-endian)로 보내고, 그 뒤에 실제 데이터를 전송한다.

    # 1. 딕셔너리 형태의 메시지를 JSON 문자열로 변환 후 UTF-8 바이트로 인코딩
    data = json.dumps(message).encode("utf-8")
    # 2. 인코딩된 데이터의 전체 길이 계산
    length = len(data)
    # 3. 데이터 길이(4바이트 정수)와 실제 데이터를 결합하여 소켓으로 일괄 전송
    sock.sendall(length.to_bytes(4, "big") + data)

def recv_message(sock):
    # 소켓으로부터 전송된 JSON 메시지를 수신하는 함수
    # 앞선 4바이트에서 메시지 길이를 파악한 후, 그 길이만큼 데이터를 정확히 읽어온다

    # 1. 메시지의 길이를 나타내는 헤더(4바이트) 수신
    raw_length = sock.recv(4)\
    # 수신된 데이터가 없으면 소켓 연결이 종료된 것으로 간주하고 None 반환
    if not raw_length:
        return None
    # 2. 바이트 배열을 정수형(정해진 데이터 길이)으로 변환
    length = int.from_bytes(raw_length, "big")
    data = b""
    # 3. 지정된 길이만큼 데이터를 루프를 돌며 완전하게 수신
    while len(data) < length:
        # 아직 받지 못한 남은 크기만큼 데이터 요청
        packet = sock.recv(length - len(data))
        # 데이터 수신 중 연결이 끊긴 경우 None 반환
        if not packet:
            return None
        # 수신한 패킷을 기존 데이터에 누적
        data += packet
    # 4. 수신 완료된 바이트 데이터를 문자열로 디코딩 후 다시 JSON(딕셔너리)으로 파싱하여 반환
    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)
    # 현재 시간을 밀리초 단위까지 활용하여 고유한 파일 식별자(ID) 생성
    file_id = f"{filename}_{int(time.time() * 1000)}"
    # 1단계: 수신 측에 파일 정보(메타데이터)를 먼저 전송
    send_message(sock, {
        "type": "file_info",
        "file_id": file_id,
        "filename": filename,
        "size": size,
        "sender": sender
    })

    try:
        # 2단계: 파일을 바이너리 읽기(rb) 모드로 열어 청크 단위로 전송 진행
        with open(file_path, "rb") as file:
            while True:
                # 지정한 크기(64KB)만큼 파일 읽기
                chunk = file.read(CHUNK_SIZE)
                # 더 이상 읽을 데이터가 없으면 루프 탈출
                if not chunk:
                    break
                # 이진 데이터를 텍스트 기반의 JSON에 담기 위해 base64로 인코딩 후 문자열 변환
                encoded = base64.b64encode(chunk).decode("utf-8")
                # 파일 조각 데이터 전송
                send_message(sock, {
                    "type": "file_chunk",
                    "file_id": file_id,
                    "data": encoded
                })
    except Exception as error:
        # 파일 읽기 및 전송 중 예외 발생 시 에러 메시지 전송 후 실패(False) 반환
        send_message(sock, {
            "type": "error",
            "content": f"파일 전송 실패: {error}"
        })
        return False

    # 3단계: 파일의 모든 조각이 성공적으로 전송되었음을 알리는 종료 메시지 전송
    send_message(sock, {
        "type": "file_end",
        "file_id": file_id
    })
    return True

끝.