크크오톡 팀프로젝트 - 1일차 - 개발일지 [2023-07-11 학습일지]

(개발 계획서와 팀 일지는 프로젝트 게시판에 업로드하여 개인 일지는 간소화하여 작성하겠습니다)

---

 

금일 소스코드 작성한 것

 

1. JSON을.. 활용해야할 것 같아서 우선 이렇게 작성해보았다. ( toJSON, __eq__ 함수)

 

import json


class Message:
    def __init__(self, message_id, sender_user_id, talk_room_id, contents, send_time_stamp, long_contents_id):
        self.message_id = message_id
        self.sender_user_id = sender_user_id
        self.talk_room_id = talk_room_id
        self.contents = contents
        self.send_time_stamp = send_time_stamp
        self.long_contents_id = long_contents_id

    def toJSON(self):
        return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)

    def __repr__(self):
        return f'{self.__dict__}'

    def __eq__(self, other):
        if isinstance(other, Message) and \
                self.message_id == other.message_id and \
                self.sender_user_id == other.sender_user_id and \
                self.talk_room_id == other.talk_room_id and \
                self.contents == other.contents and \
                self.send_time_stamp == other.send_time_stamp and \
                self.long_contents_id == other.long_contents_id:
            return True
        return False

 

2. 클라이언트 프로토타입 위젯 코드 (완성 못함)

-> 현재는 쓰레드를 아직 어떻게 사용할지 구체적으로 정하지 못해, 함수 위치의 대격변이 예상됨

 

class ClientPrototypeWidget(QtWidgets.QWidget, Ui_prototype):
    ENCODED_DOT = bytes('.', 'utf-8')
    ENCODED_PASS = bytes('pass', 'utf-8')

    def __init__(self, client_app):
        super().__init__()
        self.setupUi(self)
        self.client_app = client_app
        self.valid_duplication_id = False
        self.qthread = WorkerServerThread(self)
        self.set_btn_trigger()
        self.set_init_label()
        self.encoder = KKOEncoder()
        self.decoder = KKODecoder()

    def set_init_label(self):
        self.initialize_app()
        self.setWindowTitle("성혁이를 위한 프로토타입 위젯")

    def set_btn_trigger(self):
        self.btn_init.clicked.connect(lambda state: self.initialize_app())
        self.btn_check_same_id.clicked.connect(lambda state: self.send_join_id_for_assert_same_username())
        self.btn_join.clicked.connect(lambda state: self.send_join_id_and_pw_for_join_access())
        self.btn_login.clicked.connect(lambda state: self.send_login_id_and_pw_for_login_access())
        self.btn_send_message.clicked.connect(lambda state: self.send_message_to_chat_room())
        self.btn_transfer_file.clicked.connect(lambda state: self.send_file_to_chat_room())

    def initialize_app(self):
        self.btn_init.clicked.connect(lambda state: self.initialize_app())
        self.text_edit_chat_room.clear()
        self.text_edit_for_send_chat.clear()
        self.line_edit_for_join_id.clear()
        self.line_edit_for_join_pw.clear()
        self.line_edit_for_join_nick.clear()
        self.text_edit_chat_room.clear()
        self.valid_duplication_id = False


    # client function =================================
    def send_join_id_for_assert_same_username(self):
        input_username = self.line_edit_for_join_id.text()
        self.client_app.send(b"/assert_username")  # 헤더를 붙이고 보내는 동작(?)
        self.client_app.recv(1024)  # "." 받음
        self.client_app.send(input_username.encode())  # 실제 내용을 붙여서 보내는 동작
        return_result = self.client_app.recv(1024).decode('utf-8')  # 응답 받기
        if return_result == 'pass':
            self.valid_duplication_id = True
            return QtWidgets.QMessageBox.about(self, "가능", "중복 없는 아이디, 써도됌")
        elif return_result == '.':
            return QtWidgets.QMessageBox.about(self, "불가능", "중복 아이디, 새로 쓰기")

    def send_join_id_and_pw_for_join_access(self):
        if self.valid_duplication_id is False:
            QtWidgets.QMessageBox.about(self, "어허", "아이디 중복확인 먼저 시행해주세요")
            return
        join_username = self.line_edit_for_join_id.text()
        join_pw = self.line_edit_for_join_pw.text()
        join_nickname = self.line_edit_for_join_nick.text()
        join_user = User(None, join_username, join_pw, join_nickname)
        user_json_str = join_user.toJSON()
        self.client_app.send("/join_user")
        self.client_app.recv(1024)  # "." 받음
        self.client_app.send(user_json_str.encode())
        return_result = self.client_app.recv(1024).decode('utf-8')  # 응답 받기
        if return_result == 'pass':
            return QtWidgets.QMessageBox.about(self, "성공", "회원가입 성공")
        elif return_result == '.':
            return QtWidgets.QMessageBox.about(self, "실패", "회원가입 실패")

    def send_login_id_and_pw_for_login_access(self):
        login_username = self.line_edit_for_login_id.text()
        login_pw = self.line_edit_for_login_pw.text()
        sending_message = login_username + '%' + login_pw
        self.client_app.send("/login")
        self.client_app.recv(1024)  # "." 받음
        self.client_app.send(sending_message.encode())
        return_result = self.client_app.recv(1024).decode('utf-8')  # 응답 받기

        if return_result == 'pass':
            return QtWidgets.QMessageBox.about(self, "성공", "login 성공")
        elif return_result == '.':
            return QtWidgets.QMessageBox.about(self, "실패", "login 실패")


    def send_message_to_chat_room(self):
        txt_message = self.text_edit_for_send_chat.toPlainText()
        #todo: send 메시지

    def send_file_to_chat_room(self):
        save_excel_dialog = QtWidgets.QMessageBox.question(self, "파일 업로드", "파일을 업로드합니까?")
        if save_excel_dialog == QtWidgets.QMessageBox.Yes:
            save_path_file_name, _, = QtWidgets.QFileDialog.getSaveFileName(self, '파일 저장', './')
            print(f"{save_path_file_name} send 로직 실행")
        # todo: send 메시지

 

3. 서버쪽 프로토타입 위젯

이하동문

 

class ServerControllerWidget(QtWidgets.QWidget, Ui_server_controller):
    ENCODED_DOT = bytes('.', 'utf-8')
    ENCODED_PASS = bytes('pass', 'utf-8')

    def __init__(self, server_obj, db_connector):
        global global_total_user_list
        super().__init__()
        self.setupUi(self)
        self.server = server_obj
        self.db_conn = db_connector
        # self.qthread = WorkerServerThread(self, global_total_user_list)
        self.check_timer = None
        self.set_initial_label()
        self.set_btn_trigger()
        self.set_timer_to_check_server_status()

    def set_timer_to_check_server_status(self):
        self.check_timer = QtCore.QTimer(self)
        self.check_timer.setInterval(1000)
        self.check_timer.timeout.connect(lambda: self.assert_server_status())
        self.check_timer.start()

    def assert_server_status(self):
        # if self.server.status == ""
        # self.label_server_status.setText("🟢 가동중")
        # self.label_server_status.setText("🟠 시작중")
        # self.label_server_status.setText("🔴 종료됨")
        pass

    def set_initial_label(self):
        # todo check serve status logic
        self.label_server_status.setText("🔴 종료됨")

    def set_btn_trigger(self):
        self.btn_run.clicked.connect(lambda state: self.server_run())
        self.btn_stop.clicked.connect(lambda state: self.server_stop())

    def server_run(self):
        self.qthread.set_work(self.server.run)
        self.qthread.start()  # 쓰레드 동작시킴


    def server_stop(self):
        self.server.stop()

    def listening(self):
        while True:
            msg = self.server.recv(1024).decode("utf-8")
            if msg == '/assert_username':
                self.server.send(self.ENCODED_DOT)
                response = self.server.recv(1024).decode("utf-8")
                if self.db_conn.assert_same_join_id(response) is True:
                    self.server.send(self.ENCODED_PASS)
                else:
                    self.server.send('.'.encode())
            elif msg == '/join_user':
                self.server.send(self.ENCODED_DOT)
                response = self.server.recv(1024).decode("utf-8")
                if self.join_access(response) is True:
                    self.server.send('pass'.encode(encoding='utf-8'))
                else:
                    self.server.send(self.ENCODED_PASS)
            elif msg == '/login':
                self.server.send(self.ENCODED_DOT)
                response = self.server.recv(1024).decode("utf-8")
                if self.join_access(response) is True:
                    self.server.send(self.ENCODED_PASS)
                else:
                    self.server.send(self.ENCODED_DOT)

    def assert_same_join_id(self, input_username):
        return self.db_conn.assert_same_login_id(input_username)

    def join_access(self, join_user_json_str: str):
        join_user_obj = self.decoder.decode(join_user_json_str)
        join_user_obj: User
        if self.db_conn.insert_user(join_user_obj) is not False:
            print(f"유저 {join_user_obj.nickname} 가입 성공")
            return True
        else:
            return False

    def login_access(self, login_message: str):
        username, pw = login_message.split('%')
        if self.db_conn.user_log_in(username, pw):
            print(f"유저 {username} 로그인 성공")
            return True
        else:
            return False

 

4. OO님 유닛테스트 만들어준 것

네트워크 파트땜에 2개만 추가하고 더 못해주었음.

 

 

class UnitTest(TestCase):
    def setUp(self):  # 각 테스트마다 실행되는 함수임, Create tables 하기땜에 싹 밀고 새로 생긴다고 보면됌
        # 운영상 DB를 건들이지 않기위해 테스트용 db를 이용함
        self.conn = DBConnector(test_option=True)
        self.conn.create_tables()

    def test_유저_객체_만들어서_DB_insert_하면_ID가_1부터_시작_AUTO_INCREMENT_USER_OBJ_반환(self):
        user_a = User(None, 'abc1234', '1234', '뿡뿡이')
        user_b = User(None, 'def5678', '5678', '짱구')
        user_a = self.conn.insert_user(user_a)
        user_b = self.conn.insert_user(user_b)

        self.assertEqual(1, user_a.user_id)
        self.assertEqual(2, user_b.user_id)

    def test_유저_객체_만들면_find_all_했을때_다_불러와져야함(self):
        user_a = User(None, 'abc1234', '1234', '뿡뿡이')
        user_b = User(None, 'def5678', '5678', '짱구')
        user_c = User(None, 'charming_park', '1234', '맹구')
        user_a = self.conn.insert_user(user_a)
        user_b = self.conn.insert_user(user_b)
        user_c = self.conn.insert_user(user_c)

        insert_user_list = [user_a, user_b, user_c]

        found_user_list = self.conn.find_all_user()
        self.assertEqual(3, len(found_user_list))  # 3개 넣었으니까 3이 되야함

        # db에 넣은거랑 직접 넣은거랑 갖고 있는 인스턴스 변수값이 같은지 확인

        for (db_instance, insert_instance) in zip(found_user_list, insert_user_list):
            self.assertIsInstance(db_instance, User)
            self.assertEqual(db_instance.user_id, insert_instance.user_id)
            self.assertEqual(db_instance.username, insert_instance.username)
            self.assertEqual(db_instance.password, insert_instance.password)
            self.assertEqual(db_instance.nickname, insert_instance.nickname)

 

5. 클라이언트쪽 테스트 런처

완성되면 클라이언트쪽 소스코드로 복사 될 예정임

 

if __name__ == '__main__':
    app = QApplication(sys.argv)
    client = ClientApp()
    proto_widget = ClientPrototypeWidget(client)
    proto_widget.show()

    sys.excepthook = lambda exctype, value, traceback: show_error_message(str(value), traceback)

    sys.exit(app.exec_())

 

6. 서버쪽 테스트 런처

완성되면 서버쪽 소스코드로 복사 될 예정임

 

import sys

from PyQt5.QtWidgets import QApplication

from Code.domain.class_db_connector import DBConnector
from Code.front.class_client_controller import WindowController
from Code.network.class_client_prototype import ClientPrototypeWidget
from Code.network.class_server import Server
from Code.network.class_server_controller_widget import ServerControllerWidget
from Common.common_module import *





if __name__ == '__main__':
    app = QApplication(sys.argv)
    server = Server()
    db_conn = DBConnector(test_option=True)
    db_conn.create_tables()
    proto_widget = ServerControllerWidget(server, db_conn)
    proto_widget.show()

    sys.excepthook = lambda exctype, value, traceback: show_error_message(str(value), traceback)

    sys.exit(app.exec_())

 

7. JSON 인코더 & 디코더

우선적으로 User, Message 도메인에 대해서만 구현해놓은 상태

 

from Code.domain.class_message import Message
from Code.domain.class_user import User
from json import JSONEncoder, JSONDecoder


class KKOEncoder(JSONEncoder):

    def __init__(self):
        super().__init__()

    def default(self, o):
        return o.__dict__


class KKODecoder(JSONDecoder):
    def __init__(self):
        super().__init__()

    def decode(self, o, **kwargs):
        dict_obj = super().decode(o, **kwargs)
        if 'user_id' in dict_obj.keys():
            return User(dict_obj['user_id'], dict_obj['username'], dict_obj['password'], dict_obj['nickname'])
        elif 'message_id' in dict_obj.keys():
            return Message(dict_obj['message_id'], dict_obj['sender_user_id'], dict_obj['talk_room_id'],
                           dict_obj['contents'], dict_obj['send_time_stamp'], dict_obj['long_contents_id'])

 

 

내일은 정말정말로 프로토타입이 제작되어야한다..

 

어떻게든 잘 완성시켜보고 싶다.

 

감사합니다.