시연 영상
기능 영상 ) 회원 가입을 지원함
기능 영상 ) 침여중인 채팅방 인원을 보여줌
이하는 소스 코드입니다.
감사합니다.
# Code/domain/class_db_connect
import sqlite3
from Code.domain.class_user import User
from Code.domain.class_user_talk_room import UserTalkRoom
from Code.domain.class_talk_room import TalkRoom
from Code.domain.class_message import Message
from Code.domain.class_long_contents import LongContents
class DBConnector:
_instance = None
def __new__(cls, test_option=None):
if not isinstance(cls._instance, cls):
cls._instance = object.__new__(cls)
return cls._instance
def __init__(self, test_option=None):
self.conn = None
self.test_option = test_option
def start_conn(self):
if self.test_option is True:
self.conn = sqlite3.connect('db_test.db')
else:
self.conn = sqlite3.connect('main_db.db')
return self.conn.cursor()
def end_conn(self):
if self.conn is not None:
self.conn.close()
self.conn = None
def commit_db(self):
if self.conn is not None:
self.conn.commit()
else:
raise f"cannot commit database! {self.__name__}"
# CREATE TABLES =======================================================================
def create_tables(self):
c = self.start_conn()
c.executescript("""
DROP TABLE IF EXISTS user;
CREATE TABLE "user" (
"user_id" INTEGER,
"username" TEXT NOT NULL UNIQUE,
"password" TEXT NOT NULL,
"nickname" TEXT NOT NULL,
PRIMARY KEY("user_id" AUTOINCREMENT)
);
DROP TABLE IF EXISTS user_talk_room;
CREATE TABLE "user_talk_room" (
"user_talk_room_id" INTEGER,
"user_id" INTEGER,
"talk_room_id" INTEGER,
PRIMARY KEY("user_talk_room_id" AUTOINCREMENT)
);
DROP TABLE IF EXISTS talk_room;
CREATE TABLE "talk_room" (
"talk_room_id" INTEGER,
"talk_room_name" TEXT NOT NULL,
"open_time_stamp" TEXT NOT NULL,
PRIMARY KEY("talk_room_id" AUTOINCREMENT)
);
DROP TABLE IF EXISTS message;
CREATE TABLE "message" (
"message_id" INTEGER,
"sender_user_id" INTEGER,
"talk_room_id" INTEGER,
"send_time_stamp" TEXT,
"contents" TEXT,
"long_contents_id" INTEGER,
PRIMARY KEY("message_id" AUTOINCREMENT)
);
DROP TABLE IF EXISTS long_contents;
CREATE TABLE "long_contents" (
"long_contents_id" INTEGER,
"contents_type" INTEGER,
"long_text" TEXT,
"image" BLOB,
PRIMARY KEY("long_contents_id" AUTOINCREMENT)
);
""")
self.commit_db()
self.end_conn()
# user 테이블========================================================================================
# 인자로 들어온 User객체의 user_id 이미 존재하는지 확인, 존재하는 경우 update 함수 실행하는 로직 구현
def insert_user(self, user_object: User):
c = self.start_conn()
user_id = user_object.user_id
user_name = user_object.username
password = user_object.password
nickname = user_object.nickname
users_id = c.execute('select * from user where user_id = ?', (user_id,)).fetchone()
if users_id is None:
c.execute('insert into user(username, password, nickname) values (?, ?, ?)',
(user_name, password, nickname))
self.commit_db()
inserted_user_row = c.execute('select * from user order by user_id desc limit 1').fetchone()
inserted_user_obj = User(*inserted_user_row)
self.end_conn()
return inserted_user_obj
else:
updated_user_obj = self.update_user(user_object)
return updated_user_obj
def update_user(self, user_object: User):
c = self.start_conn()
user_id = user_object.user_id
user_name = user_object.username
password = user_object.password
nickname = user_object.nickname
c.execute('update user set username=?, password = ?, nickname=? where user_id = ?',
(user_name, password, nickname, user_id))
self.commit_db()
updated_user_row = c.execute('select * from user where user_id = ?', (user_id,)).fetchone()
updated_user_obj = User(*updated_user_row)
self.end_conn()
return updated_user_obj
# User 찾기
def find_all_user(self):
c = self.start_conn()
user_data = c.execute('select * from user').fetchall()
if user_data is None:
return None
all_user_obj_list = list()
for row_user in user_data:
all_user_obj_list.append(User(*row_user))
self.end_conn()
return all_user_obj_list
def find_user_by_username(self, username):
c = self.start_conn()
user_data = c.execute('select * from user where username = ?', (username,)).fetchone()
if user_data is None:
return None
user_object = User(*user_data)
self.end_conn()
return user_object
def find_user_by_user_id(self, user_id):
c = self.start_conn()
user_data = c.execute('select * from user where user_id = ?', (user_id,)).fetchone()
if user_data is None:
return None
user_object = User(*user_data)
self.end_conn()
return user_object
# user 삭제
def delete_user_by_username(self, username: str):
c = self.start_conn()
deleted_user = c.execute('select * from user where username = ?', (username,)).fetchone()
deleted_user_obj = User(*deleted_user)
c.execute('delete from user where username = ?', (username,))
self.commit_db()
self.end_conn()
return deleted_user_obj
def delete_user_by_user_id(self, user_id):
c = self.start_conn()
deleted_user = c.execute('select * from user where user_id = ?', (user_id,)).fetchone()
deleted_user_obj = User(*deleted_user)
c.execute('delete from user where user_id = ?', (user_id,))
self.commit_db()
self.end_conn()
return deleted_user_obj
# id 기준으로 함수로 하나 더 만들기
# user_talk_room================================================
# 새로운 채팅방 생성 함수 호출 및 user_talk_room 테이블 정보 추가
# def insert_user_talk_room(self, user_id, talk_room_name, open_time_stamp):
# talk_room_obj = self.create_talk_room(talk_room_name, open_time_stamp)
# c = self.start_conn()
# talk_room_id = talk_room_obj.talk_room_id
# c.execute('insert into user_talk_room (user_id, talk_room_id) values (?, ?)', (user_id, talk_room_id))
# self.commit_db()
# self.end_conn()
def insert_user_talk_room(self, user_talk_room_obj: UserTalkRoom):
c = self.start_conn()
# user_talk_room_id = user_talk_room_obj.user_talk_room_id
user_id = user_talk_room_obj.user_id
talk_room_id = user_talk_room_obj.talk_room_id
c.execute('insert into user_talk_room (user_id, talk_room_id) values (?, ?)', (user_id, talk_room_id))
self.commit_db()
inserted_user_talk_room_row = c.execute('select * from user_talk_room order by user_talk_room_id desc limit 1').fetchone()
inserted_user_talk_room_obj = UserTalkRoom(*inserted_user_talk_room_row)
self.end_conn()
return inserted_user_talk_room_obj
def find_all_user_talk_room(self):
c = self.start_conn()
all_user_talk_room_obj_list = list()
user_talk_room_rows = c.execute('select * from user_talk_room').fetchall()
for row in user_talk_room_rows:
user_talk_room_obj = UserTalkRoom(*row)
all_user_talk_room_obj_list.append(user_talk_room_obj)
self.end_conn()
return all_user_talk_room_obj_list
# 유저가 속한 모든 채팅방(TalkRoom) 객체 리스트로 반환
def find_user_talk_room_by_user_id(self, user_id: int) -> list[UserTalkRoom]:
c = self.start_conn()
users_talk_room_obj_list = list()
user_talk_rooms = c.execute('select * from user_talk_room where user_id = ?', (user_id,)).fetchall()
# talk_rooms = c.execute('select * from talk_room').fetchall()
for user_talk_row in user_talk_rooms:
user_talk_room_obj = UserTalkRoom(*user_talk_row)
users_talk_room_obj_list.append(user_talk_room_obj)
self.end_conn()
return users_talk_room_obj_list
def find_user_talk_room_by_username(self, username: str) -> list[UserTalkRoom]:
c = self.start_conn()
username_row = c.execute('select * from user where username = ?', (username,)).fetchone()
user_id = username_row[0]
self.end_conn()
result = self.find_user_talk_room_by_user_id(user_id)
return result
# talk_room_id에 해당하는 톡방에 있는 유저 객체 반환
def find_user_by_talk_room_id(self, talk_room_id: int) -> list[User]:
c = self.start_conn()
all_user = list()
involved_user = c.execute('select * from user_talk_room where talk_room_id = ?', (talk_room_id,)).fetchall()
if len(involved_user) == 0:
return None
for user in involved_user:
user_id = user[1]
user = c.execute('select * from user where user_id = ?', (user_id,)).fetchone()
all_user.append(User(*user))
self.end_conn()
return all_user
# 회원이 자신이 속한 모든 톡방을 나간 경우(탈퇴 등-이미 탈퇴 함수 있음), user_id / username에 따라 삭제(유저가 속한 모든 톡방 나감)
def delete_user_talk_room_by_user_id(self, user_id):
c = self.start_conn()
c.execute('delete from user_talk_room where user_id = ?', (user_id,))
self.commit_db()
self.end_conn()
def delete_user_talk_room_by_username_and_talk_room_id(self, user_name, talk_room_id):
out_user_id = self.find_user_by_username(user_name).user_id
self.delete_user_talk_room_by_user_id_and_talk_room_id(out_user_id, talk_room_id)
# 회원 탈퇴한 경우 user테이블 삭제 함수 및 user_talk_room 삭제 함수 호출(user 및 user_talk_room 모두 삭제)
def delete_withdrawal_user_talk_room_by_username(self, user_name: str):
withdrawal_user_id = self.delete_user_by_username(user_name).user_id
self.delete_user_talk_room_by_user_id(withdrawal_user_id)
def delete_withdrawal_user_talk_room_by_user_id(self, user_id):
self.delete_user_by_user_id(user_id)
self.delete_user_talk_room_by_user_id(user_id)
# talk_room 테이블=======================================================
# 새로운 채팅방 생성
def insert_talk_room(self, talk_room_obj: TalkRoom):
talk_room_name = talk_room_obj.talk_room_name
open_time_stamp = talk_room_obj.open_time_stamp
c = self.start_conn()
c.execute('insert into talk_room (talk_room_name, open_time_stamp) values (?, ?)',
(talk_room_name, open_time_stamp))
self.commit_db()
created_talk_room = c.execute('select * from talk_room order by talk_room_id desc limit 1').fetchone()
created_talk_room_obj = TalkRoom(*created_talk_room)
self.end_conn()
return created_talk_room_obj
# def create_talk_room(self, talk_room_name, open_time_stamp):
# c = self.start_conn()
# c.execute('insert into talk_room (talk_room_name, open_time_stamp) values (?, ?)',
# (talk_room_name, open_time_stamp))
# self.commit_db()
# created_talk_room = c.execute('select * from talk_room order by talk_room_id desc limit 1').fetchone()
# created_talk_room_obj = TalkRoom(*created_talk_room)
# self.end_conn()
# return created_talk_room_obj
def find_all_talk_room(self):
c = self.start_conn()
all_talk_room_obj_list = list()
all_talk_room_rows = c.execute('select * from talk_room').fetchall()
for talk_room_row in all_talk_room_rows:
talk_room_obj = TalkRoom(*talk_room_row)
all_talk_room_obj_list.append(talk_room_obj)
self.end_conn()
return all_talk_room_obj_list
# talk_room_id로 talk_room_obj 반환 : user_talk_room_obj 의 talk_room_id로 talk_room_obj 반환가능
def find_talk_room_by_talk_room_id(self, talk_room_id):
c = self.start_conn()
row_data = c.execute('select * from talk_room where talk_room_id = ?', (talk_room_id,)).fetchone()
talk_room_obj = TalkRoom(*row_data)
self.end_conn()
return talk_room_obj
# def find_talk_room_by_talk_room_id(self, user_talk_room_obj: UserTalkRoom):
# c = self.start_conn()
# talk_room_id = user_talk_room_obj.talk_room_id
# row_data = c.execute('select * from talk_room where talk_room_id = ?', (talk_room_id,)).fetchone()
# talk_room_obj = TalkRoom(*row_data)
# self.end_conn()
# return talk_room_obj
# message테이블===================================================================================
def insert_message(self, sender_user_id, talk_room_id, send_time_stamp, contents=None, long_contents_id=None):
c = self.start_conn()
c.execute('''insert into message (sender_user_id, talk_room_id, send_time_stamp, contents, long_contents_id)
values (?, ?, ?, ?, ?)''', (sender_user_id, talk_room_id, send_time_stamp, contents, long_contents_id))
self.commit_db()
inserted_message_row = c.execute('select * from message order by message_id desc limit 1').fetchone()
sender_user_obj = self.find_user_by_user_id(sender_user_id)
inserted_message_obj = Message(*inserted_message_row, sender_user_obj)
self.end_conn()
return inserted_message_obj
# 메세지 객체 받아 db에 저장
def create_message(self, message_obj: Message):
sender_user_id = message_obj.sender_user_id
talk_room_id = message_obj.talk_room_id
send_time_stamp = message_obj.send_time_stamp
contents = message_obj.contents
long_contents_id = message_obj.long_contents_id
inserted_message_obj = self.insert_message(sender_user_id, talk_room_id, send_time_stamp, contents, long_contents_id)
return inserted_message_obj
def find_message_by_message_id(self, message_id):
c = self.start_conn()
message_row = c.execute('select * from message where message_id = ?', (message_id,)).fetchone()
sender_user_id = message_row[1]
sender_user_obj = self.find_user_by_user_id(sender_user_id)
message_obj = Message(*message_row, sender_user_obj)
self.end_conn()
return message_obj
def find_message_by_sender_user_id(self, sender_user_id) -> list[Message]:
c = self.start_conn()
message_rows = c.execute('select * from message where sender_user_id = ?', (sender_user_id,)).fetchall()
message_obj_list = list()
for message_row in message_rows:
sender_user_obj = self.find_user_by_user_id(sender_user_id)
message_obj = Message(*message_row, sender_user_obj)
message_obj_list.append(message_obj)
self.end_conn()
return message_obj_list
def find_message_by_talk_room_id(self, talk_room_id) -> list[Message]:
c = self.start_conn()
message_rows = c.execute('select * from message where talk_room_id = ?', (talk_room_id,)).fetchall()
message_obj_list = list()
for message_row in message_rows:
sender_user_id = message_row[1]
sender_user_obj = self.find_user_by_user_id(sender_user_id)
message_obj = Message(*message_row, sender_user_obj)
message_obj_list.append(message_obj)
self.end_conn()
return message_obj_list
# long_contents 테이블==========================================================================================
# contents_type - 0 : long_text, 1: image
def insert_long_contents(self, contents_type, long_text=None, image=None):
c = self.start_conn()
if contents_type == 0 and long_text is not None:
c.execute('insert into long_contents (contents_type, long_text) values (?, ?)', (contents_type, long_text))
self.commit_db()
inserted_long_contents = c.execute(
'select * from long_contents order by long_contents_id desc limit 1').fetchone()
inserted_long_contents_obj = LongContents(*inserted_long_contents)
self.end_conn()
return inserted_long_contents_obj
elif contents_type == 0 and long_text is None:
return f"콘텐츠타입{contents_type} 과 롱텍스트{long_text} 불일치, 이미지{image}"
elif contents_type == 1 and image is not None:
c.execute('insert into long_contents (contents_type, image) values (?, ?)', (contents_type, image))
self.commit_db()
inserted_long_contents = c.execute(
'select * from long_contents order by long_contents_id desc limit 1').fetchone()
inserted_long_contents_obj = LongContents(*inserted_long_contents)
self.end_conn()
return inserted_long_contents_obj
elif contents_type == 1 and image is None:
return f"콘텐츠타입{contents_type} 과 이미지{image} 불일치, 롱텍스트{long_text}"
def create_long_contents(self, long_contents_obj:LongContents):
contents_type = long_contents_obj.contents_type
long_text = long_contents_obj.long_text
image = long_contents_obj.image
inserted_long_contents_obj = self.insert_long_contents(contents_type, long_text=long_text, image=image)
return inserted_long_contents_obj
def find_long_contents_by_long_contents_id(self, long_contents_id):
c = self.start_conn()
long_contents_row = c.execute('select * from long_contents where long_contents_id = ?', (long_contents_id,)).fetchone()
long_contents_obj = LongContents(*long_contents_row)
self.end_conn()
return long_contents_obj
# ui 우선 사용 함수=============================================================================================
# 사용자 아이디 중복 확인
def assert_same_login_id(self, inserted_id):
c = self.start_conn()
username_id = c.execute('select * from user where username = ?', (inserted_id,)).fetchone()
if username_id is None:
print('사용 가능한 아이디 입니다.') # 사용 가능 아이디
return True
else:
print('사용 불가능한 아이디 입니다.') # 사용불가
return False
# 회원가입용 함수(insert_user함수 호출)
def user_sign_up(self, insert_id, insert_pw, nickname):
useable_id = self.assert_same_login_id(insert_id)
if useable_id is False:
return False
c = self.start_conn()
last_user_row = c.execute('select * from user order by user_id desc limit 1').fetchone()
if last_user_row is None:
user_id = 1
else:
user_id = last_user_row[0] + 1
sign_up_user_obj = User(user_id, insert_id, insert_pw, nickname)
self.end_conn()
sing_up_obj = self.insert_user(sign_up_user_obj)
return sing_up_obj
# 사용자 로그인 함수
def user_log_in(self, login_id, login_pw):
c = self.start_conn()
exist_user = c.execute('select * from user where username = ? and password = ?',
(login_id, login_pw)).fetchone()
self.end_conn()
if exist_user is not None:
print('로그인 성공')
login_user_obj = User(*exist_user)
return login_user_obj
else:
print('아이디 혹은 비밀번호를 잘못 입력했습니다.')
return False
# 해당 talkroom에 존재하지 않는 user 반환 객체 리스트 반환(초대 가능한 사람 리스트)
def uninvited_users_from_talk_room(self, talk_room_id):
c = self.start_conn()
invited_user = c.execute('select * from user_talk_room where talk_room_id = ?', (talk_room_id,)).fetchall()
invited_user_id = list()
for user in invited_user:
invited_user_id.append(user[1])
print(invited_user_id)
uninvited_user_obj_list = list()
uninvited_user_rows = c.execute("select * from user where user_id not in (" + ",".join("?" * len(invited_user_id)) + ")", invited_user_id).fetchall()
for un_user_row in uninvited_user_rows:
un_user_obj = User(*un_user_row)
uninvited_user_obj_list.append(un_user_obj)
self.end_conn()
return uninvited_user_obj_list
#Code/domain/class_long_contents.py
import json
class LongContents:
def __init__(self, contents_id, contents_type, long_text, image):
self.contents_id = contents_id
self.contents_type = contents_type
self.long_text = long_text
self.image = image
def toJSON(self):
return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True)
def __repr__(self):
return f'{self.__dict__}'
def __eq__(self, other):
if isinstance(other, LongContents) and \
self.contents_id == other.contents_id and \
self.contents_type == other.contents_type and \
self.long_text == other.long_text and \
self.image == other.image:
return True
return False# Code/domain/class_message.py
import json
class Message:
def __init__(self, message_id, sender_user_id, talk_room_id, send_time_stamp, contents, long_contents_id, user_obj=None):
self.message_id = message_id
self.sender_user_id = sender_user_id
self.talk_room_id = talk_room_id
self.send_time_stamp = send_time_stamp
self.contents = contents
self.long_contents_id = long_contents_id
self.user_obj = user_obj
def toJSON(self):
obj_str = json.dumps(self, default=lambda o: o.__dict__)
return obj_str
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.send_time_stamp == other.send_time_stamp and \
self.contents == other.contents and \
self.long_contents_id == other.long_contents_id:
return True
return False#Code/domain/class_talk_room.py
import json
import datetime
class TalkRoom:
def __init__(self, talk_room_id, talk_room_name, open_time_stamp):
self.talk_room_id = talk_room_id
self.talk_room_name = talk_room_name
self.open_time_stamp = open_time_stamp
self.talk_room_user_list = list()
def get_datetime(self):
datetime_obj = datetime.datetime.strptime(self.open_time_stamp)
return datetime_obj
def __str__(self):
return f"{self.__repr__()}"
def __repr__(self):
return f"{self.__dict__}"
def toJSON(self):
origin_data = dict()
origin_data.update({"talk_room_id": self.talk_room_id})
origin_data.update({"talk_room_name": self.talk_room_name})
origin_data.update({"open_time_stamp": self.open_time_stamp})
result = json.dumps(origin_data, default=lambda o: o.__dict__)
return result
def to_dict(self):
origin_data = dict()
origin_data.update({"talk_room_id": self.talk_room_id})
origin_data.update({"talk_room_name": self.talk_room_name})
origin_data.update({"open_time_stamp": self.open_time_stamp})
result =f'{origin_data}'
return result
def append_user(self, user):
if isinstance(user, list):
self.talk_room_user_list.extend(user)
else:
self.talk_room_user_list.append(user)
def __eq__(self, other):
if isinstance(other, TalkRoom) and \
self.talk_room_id == other.talk_room_id and \
self.talk_room_name == other.talk_room_name and \
self.open_time_stamp == other.open_time_stamp:
return True#Code/domain/class_user.py
import json
import random
class User:
def __init__(self, user_id, username, password, nickname):
self.user_id = user_id
self.username = username
self.password = password
self.nickname = nickname
def __str__(self):
return f"{self.__repr__()}"
def __repr__(self):
return f"{self.__dict__}"
def toJSON(self) -> str:
return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True)
def __eq__(self, other):
if isinstance(other, User) and \
self.user_id == other.user_id and \
self.username == other.username and \
self.password == other.password and \
self.nickname == other.nickname:
return True
return False
def __lt__(self, other):
return self.nickname < other.nickname#Code/domain/class_user_talk_room.py
import json
class UserTalkRoom:
def __init__(self, user_talk_room_id, user_id, talk_room_id):
self.user_talk_room_id = user_talk_room_id
self.user_id = user_id
self.talk_room_id = talk_room_id
def __str__(self):
return f"{self.__repr__()}"
def __repr__(self):
return f"{self.__dict__}"
def toJSON(self):
return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True)
def __eq__(self, other):
if isinstance(other, UserTalkRoom) and \
self.user_talk_room_id == other.user_talk_room_id and \
self.user_id == other.user_id and \
self.talk_room_id == other.talk_room_id:
return True
return False#Code/front/class_client_controller.py
import datetime
import time
from PyQt5 import QtWidgets, QtCore
from PyQt5.QtCore import QPoint, Qt, pyqtSignal
from Code.domain.class_db_connector import DBConnector
from Code.front.class_custom_message_box import NoFrameMessageBox
from Code.front.widget_friend_list_page import FriendListWidget
from Code.front.widget_join_page import JoinWidget
from Code.front.widget_login_page import LoginWidget
from Code.front.widget_talk_room_list_page import TalkRoomListWidget
from Code.front.widget_make_talk_room import InviteFriendListWidget
from Code.front.widget_talk_room_page import TalkRoomWidget
from Code.front.widget_search_talk_room_member_list_page import UserListWidgetInTalkRoom
from Code.front.widget_invite_user_in_chat_room import InviteFriendListWidgetInTalkRoom
from Code.front.widget_profile_page import ProfilePage
from Code.domain.class_user import User
from Code.network.class_client import ClientApp
from Common.class_json import KKOEncoder, KKODecoder
from Common.common_module import get_now_time_str
class WindowController(QtWidgets.QWidget):
# signal 클래스 변수
assert_same_id_signal = pyqtSignal(bool)
sign_up_signal = pyqtSignal(bool)
log_in_signal = pyqtSignal(bool)
enter_square_signal = pyqtSignal(bool)
all_user_list_signal = pyqtSignal(str)
user_talk_room_signal = pyqtSignal(str)
talk_room_user_list_se_signal = pyqtSignal(int, str)
out_talk_room_signal = pyqtSignal(bool)
send_msg_se_signal = pyqtSignal(str)
invite_user_talk_room_signal = pyqtSignal(bool)
make_talk_room_signal = pyqtSignal(int)
talk_room_msg_signal = pyqtSignal(str)
def __init__(self, client_app=ClientApp):
assert isinstance(client_app, ClientApp)
super().__init__()
self.client_app = client_app # db연결 인스턴스
self.client_app.set_widget(self)
self.db_connector = None
# Domain 인스턴스
self.widget_login_page = LoginWidget(self) # 로그인 화면 ui 클래스 함수화
self.widget_friend_list_page = FriendListWidget(self) # 친구창 ui 클래스 함수화
self.widget_join = JoinWidget(self) # 회원 가입창 ui 클래스 함수화
self.widget_talk_room_list = TalkRoomListWidget(self)
self.widget_ask_to_make_talk_room = InviteFriendListWidget(self)
self.widget_add_member_in_chat_room = InviteFriendListWidgetInTalkRoom(self)
self.widget_profile_window = ProfilePage(self)
self.widget_talk_room = TalkRoomWidget(self, 1)
self.widget_room_member_list = UserListWidgetInTalkRoom(self, self.widget_talk_room)
self.stored_user_list = list() # 가입되어있는 모든 유저의 정보
self.stored_talk_room_id_list = list() # 로그인한 유저의 친구 리스트
self.talk_room_related_user_id_list = list()
self.stored_user_obj = None
self.valid_duplication_id = None # 중복 체크 여부 확인 변수 -> 기초 false set / 중복확인 후 결과 따라 True 됨
self.join_id = None # 회원가입에서 적힌 아이디
self.join_pw = None # 회원가입에서 적힌 비밀번호
self.join_nickname = None # 회원가입에서 적힌 닉네임
self.encoder = KKOEncoder()
self.decoder = KKODecoder()
self.initial_trigger_setting()
# ui 동작 관련 변수
self.list_widget_geometry_x = None
self.list_widget_geometry_y = None
self.drag_start_position = QPoint(0, 0)
def mousePressEvent(self, widget, event):
self.drag_start_position = QPoint(widget.x(), widget.y())
if event.button() == Qt.LeftButton:
self.drag_start_position = event.globalPos() - widget.frameGeometry().topLeft()
event.accept()
def mouseMoveEvent(self, widget, event):
if event.buttons() == Qt.LeftButton:
widget.move(event.globalPos() - self.drag_start_position)
event.accept() def get_user_self(self):
return self.client_app.get_user_self()
def get_message_list(self, talk_room_id):
stored_dictionary_message = self.client_app.stored_talk_message
if talk_room_id not in stored_dictionary_message.keys():
stored_dictionary_message.update({talk_room_id: list()})
result_list = stored_dictionary_message[talk_room_id]
return result_list
def get_all_user_list(self):
user_list = self.client_app.get_all_user_list()
if len(user_list) == 0:
self.client_app.send_all_user_list()
time.sleep(0.5)
return self.client_app.all_user_list_in_memory
def get_user_by_id(self, user_id):
return self.client_app.get_user_by_id(user_id)
def get_talk_room_by_room_id(self, talk_room_id):
return self.client_app.get_talk_room_by_room_id(talk_room_id)
def get_already_invited_user_list(self):
return self.client_app.get_already_invited_user_list()
def get_not_yet_invited_user_list(self):
return self.client_app.get_not_yet_invited_user_list()
def send_make_talk_room_user_id(self, talk_room_name, selected_user_id_list):
# todo: 채팅방을 만들때 초대할 유저의 정보, 토크룸의 이름을 보낸다
now_time = get_now_time_str()
self.client_app.send_make_talk_room(talk_room_name, selected_user_id_list, now_time)
def uninvited_user_list(self, talk_room_id):
"""
todo: 만일 채팅방에서 초대되지 않은 인원을 파악하고자 할 때 유저 리스트를 반환함
:param talk_room_id:
:return:
"""
def show_profile_page(self, user_id):
self.client_app: ClientApp
selected_user = [x for x in self.client_app.all_user_list_in_memory if x.user_id == user_id]
selected_user = selected_user[0]
self.widget_profile_window.set_profile_user_data(selected_user)
self.widget_profile_window.show()
def show_talk_room(self, talk_room_id):
"""
채팅방 보여주는 메서드
채팅방에 초대 되어있지 않은 유저 목록 저장
:return:
"""
print('show try')
self.client_app.selected_talk_room_id = talk_room_id
self.widget_talk_room.set_talk_room_id(talk_room_id)
self.widget_talk_room.show()
# def get_friend_profile(self):
def show_member_plus(self):
"""
채팅방에 멤버추가 위젯 보여주는 메서드
:return:
"""
self.widget_add_member_in_chat_room.show()
def show_room_member_list(self):
"""
채팅방에 있는 멤버 목록을 보여주는 메서드
:return:
"""
self.widget_room_member_list.show()
def show_make_talk_room(self):
"""
채팅방 만드는 위젯 보여주는 메서드
:return:
"""
self.widget_ask_to_make_talk_room.show() def show_talk_room_list_page(self):
"""
채팅방 보여주는 메서드
:return:
"""
cp = self.widget_friend_list_page.frameGeometry().topLeft()
cp: QPoint
self.list_widget_geometry_x = cp.x()
self.list_widget_geometry_y = cp.y()
self.widget_talk_room_list.setGeometry(self.list_widget_geometry_x, self.list_widget_geometry_y,
self.widget_talk_room_list.width(),
self.widget_talk_room_list.width())
self.widget_talk_room_list.show()
def show_friend_list(self):
"""
친구창 보여주는 메서드
:return:
"""
cp = self.widget_talk_room_list.frameGeometry().topLeft()
cp: QPoint
self.list_widget_geometry_x = cp.x()
self.list_widget_geometry_y = cp.y()
self.widget_friend_list_page.setGeometry(self.list_widget_geometry_x, self.list_widget_geometry_y,
self.widget_friend_list_page.width(),
self.widget_friend_list_page.height())
self.widget_friend_list_page.show()
# 채팅방 리스트 화면 띄우는 메서드
def show_join_page(self):
"""
회원 가입창 띄우는 메서드
"""
self.widget_join.show()
def run(self):
"""
로그인 화면 띄우는 메서드
"""
self.widget_login_page.show()
def show_login_success(self):
# todo: 친구창과 채팅방을 띄울때 본인의 정보를 따로 처리해야한다
self.widget_friend_list_page.show() # 친구창 띄우는 함수
@staticmethod
def clear_widget(widget):
if widget.layout() is not None:
while widget.layout().count() > 0:
item = widget.layout().takeAt(0)
if item.widget():
item.widget().deleteLater()
# 1대1대화 시 해당 멤버가 속한 방 talk_room_id 찾기
def find_already_created_room(self, target_user_id):
talk_room_list = self.get_all_talk_room()
user_self_id = self.get_user_self().user_id
for talk_room in talk_room_list:
talk_room_user_list = talk_room.talk_room_user_list
if len(talk_room_user_list) == 2 and [user_self_id, target_user_id] in talk_room_user_list:
return talk_room.talk_room_id
else:
return False
# 레이아웃 안에있는 위젯들 삭제
def open_one_to_one_chat_room(self, target_user_id):
found_talk_room_id = self.find_already_created_room(target_user_id)
print('found id', found_talk_room_id)
if found_talk_room_id is False:
target_user_obj = self.get_user_by_id(target_user_id)
invite_list = list()
invite_list.append(self.get_user_self().user_id) # 본인 포함이므로
invite_list.append(target_user_id)
now_time_stamp = get_now_time_str()
self.client_app.send_make_talk_room(target_user_obj.nickname, invite_list, now_time_stamp)
else:
self.show_talk_room(found_talk_room_id)
def open_talk_room_widget_after_recv_talk_room_id(self, talk_room_id: int = None):
self.show_talk_room(talk_room_id) # ==== 클라이언트 response 함수 ================================================================
def initial_trigger_setting(self):
self.valid_duplication_id = False
self.assert_same_id_signal.connect(self.assert_same_name_res)
self.sign_up_signal.connect(self.sign_up_res)
self.log_in_signal.connect(self.log_in_res)
self.enter_square_signal.connect(self.enter_square_res)
self.all_user_list_signal.connect(self.all_user_list_res)
self.user_talk_room_signal.connect(self.user_talk_room_list_res)
self.talk_room_user_list_se_signal.connect(self.talk_room_user_list_se_res)
self.out_talk_room_signal.connect(self.out_talk_room_res)
self.send_msg_se_signal.connect(self.send_msg_se_res)
self.invite_user_talk_room_signal.connect(self.invite_user_talk_room_res)
self.make_talk_room_signal.connect(self.make_talk_room_res)
self.talk_room_msg_signal.connect(self.load_message_history)
# client function =================================
# 클라 -> 서버 아이디 중복 체크 요청
def assert_same_username(self, join_username): # 아이디 중복
self.client_app.send_join_id_for_assert_same_username(join_username)
def assert_same_name_res(self, return_result: bool):
if return_result is True:
self.valid_duplication_id = True
return NoFrameMessageBox(self, "가능", "중복 없는 아이디, 써도됌", "about")
elif return_result is False:
return NoFrameMessageBox(self, "불가능", "중복 아이디, 새로 쓰기", "about")
# 클라 -> 서버 회원가입 요청
def join_access(self):
join_username = self.widget_join.lineEdit_join_username.text()
join_pw = self.widget_join.lineedit_join_pw.text()
join_nickname = self.widget_join.lineedit_join_user_nickname.text()
self.client_app.send_join_id_and_pw_for_join_access(join_username, join_pw, join_nickname)
# 서버 -> 클라 회원가입 결과 체크 결과 대응
def sign_up_res(self, return_result: bool):
if return_result is True:
result = NoFrameMessageBox(self, "성공", "회원가입 성공", "about")
self.widget_join.close()
return
elif return_result is False:
return NoFrameMessageBox(self, "실패", "회원가입 실패", "about")
# 클라 -> 서버 로그인 요청
def assert_login_data(self, login_id, login_pw):
"""
기능1:
로그인 화면에서 로그인 승인 버튼에 시그널을 주면 서버에 ID,PW가 저장 되있고 일치여부 요청
"""
self.client_app.user_id = None
self.client_app.username = login_id
self.client_app.user_pw = login_pw
self.client_app.user_nickname = None
self.client_app.send_login_id_and_pw_for_login_access(login_id, login_pw)
def log_in_res(self, return_result: bool):
if return_result is True:
# 모든건 로그인 버튼을 누르면 시작한다. 나중에 수정
# 받아와야할 정보 : 전체 회원 리스트, 본인이 포함된 톡방 리스트
login_user = self.get_user_self()
self.widget_friend_list_page.login_user_obj = login_user
self.all_user_list_req()
self.user_talk_room_list_req()
time.sleep(0.05)
self.show_login_success()
# self.out_talk_room()
return NoFrameMessageBox(self, "성공", "login 성공", "about")
elif return_result is False:
return NoFrameMessageBox(self, "실패", "login 실패", "about")
# 클라 -> 서버 초기 체팅방 입장, 로그인시 실행
def enter_square(self):
self.client_app.send_enter_square()
# 전체 회원방 메시지 요청
# 서버 -> 클라 초기 체팅방 입장 결과 체크 def enter_square_res(self):
# 화면 띄우기? 화면전환?
# 전체 회원방 메시지 저장
print("초기방 입장 완료")
# 클라 -> 서버 유저 리스트 요청, 로그인시 할 수도있음
def all_user_list_req(self):
self.client_app.send_all_user_list()
# 서버 -> 클라 유저 리스트 정보 받음
def all_user_list_res(self, return_result: str):
user_list = self.decoder.decode_any(return_result)
self.widget_login_page.close()
self.client_app.all_user_list_in_memory = user_list
self.widget_friend_list_page.show()
# 클라 -> 서버 채팅방 리스트 요청
def get_all_talk_room(self):
return self.client_app.talk_room_list_in_memory
def get_talk_by_room_id(self, talk_room_id):
return self.client_app.get_talk_by_room_id(talk_room_id)
def user_talk_room_list_req(self):
self.client_app.send_user_talk_room_list()
# 서버 -> 클라 채팅방 리스트 정보 받음
def user_talk_room_list_res(self, return_result: str):
result_list = self.decoder.decode_any(return_result)
for talk_room in result_list:
self.talk_room_user_list_se(talk_room.talk_room_id)
self.client_app.send_talk_room_msg(talk_room.talk_room_id)
self.widget_talk_room_list.talk_room_list = self.client_app.talk_room_list_in_memory.copy()
self.widget_talk_room_list.refresh_chat_room_list()
# 클라 -> 서버 채팅방 관련 유저 정보 요청
# 방 아이디를 넘겨줘야 할듯 하다.
def talk_room_user_list_se(self, talk_room_id):
self.client_app.send_talk_room_user_list_se(talk_room_id)
# 서버 -> 클라 톡방 유저 객체 정보 획득
def talk_room_user_list_se_res(self, talk_room_id:int, return_result: str):
related_talk_room_user_list = self.decoder.decode_any(return_result)
found_talk_room = None
if len(self.get_all_talk_room()) == 0:
time.sleep(0.5)
for talk_room in self.get_all_talk_room():
temp_num = talk_room.talk_room_id
if temp_num == talk_room_id:
found_talk_room = talk_room
break
if found_talk_room is None:
raise "채팅방 못찾음"
found_talk_room.talk_room_user_list.clear()
found_talk_room.append_user(related_talk_room_user_list)
print(found_talk_room, end="!!!")
print(found_talk_room.talk_room_user_list)
def out_talk_room(self):
self.client_app.send_out_talk_room(talk_room_id)
# 채팅방 나가기 결과 반환
# 메세지 박스를 화면 전환 해주세요 def out_talk_room_res(self, return_result: bool):
if return_result is True:
return NoFrameMessageBox(self, "성공", "방탈출 성공", "about")
elif return_result is False:
return NoFrameMessageBox(self.widget_join, "실패", "방탈출 실패", "about")
# 화면 전환후 채팅방 목록 불러오기
# 클라 -> 서버 메시지 전달
def send_msg_se(self, talk_room_id, txt_message):
self.client_app.send_send_msg_se(talk_room_id, txt_message)
# 서버 -> 클라 메시지 받기
def send_msg_se_res(self, return_result: str):
message = self.decoder.decode_any(return_result)
self.client_app.store_message(message)
self.widget_talk_room.show()
# todo: send 메시지
# 클라 -> 서버 단톡방 초대 요청
def invite_user_talk_room_res(self, return_result: bool):
self.show_talk_room(self.client_app.selected_talk_room_id)
# 채팅방 개설하기
def make_talk_room_res(self, return_result: int):
self.show_talk_room(return_result)
def load_message_history(self, return_result: str):
# self.widget_talk_room.hide()
# self.widget_talk_room.show()
print('메시지 저장 완료')
def send_file_to_chat_room(self):
save_excel_dialog = NoFrameMessageBox(self, "파일 업로드", "파일을 업로드합니까?", "question").result
if save_excel_dialog is True:
save_path_file_name, _, = QtWidgets.QFileDialog.getSaveFileName(self, '파일 저장', './')
print(f"{save_path_file_name} send 로직 실행")
# todo: send 메시지# Code/network/class_client.py
import datetime
import socket
import time
from threading import *
from Code.domain.class_message import Message
from Code.domain.class_talk_room import TalkRoom
from Code.domain.class_user import User
from Code.domain.class_user_talk_room import UserTalkRoom
from Common.class_json import KKODecoder
class ClientApp:
HOST = '127.0.0.1'
PORT = 9999
BUFFER = 50000
FORMAT = "utf-8"
HEADER_LENGTH = 30
assert_username = "assert_username"
join_user = "join_user"
login = "login"
enter_square = "enter_square"
all_user_list = "all_user_list"
user_talk_room_list = "user_talk_room_list"
talk_room_user_list_se = 'talk_room_user_list_se'
out_talk_room = 'out_talk_room'
send_msg_se = 'send_msg_se'
invite_user_talk_room = 'invite_user_talk_room'
make_talk_room = 'make_talk_room'
talk_room_msg = 'talk_room_msg'
send_msg_c_room = "send_msg_c_room"
send_alarm_c_room = "send_alarm_c_room"
send_ = "send_alarm_c_room"
HEADER_LIST = {
assert_username: assert_username.encode(FORMAT),
join_user: join_user.encode(FORMAT),
login: login.encode(FORMAT),
send_msg_c_room: send_msg_c_room.encode(FORMAT),
send_alarm_c_room: send_alarm_c_room.encode(FORMAT),
}
def __init__(self):
# 서버 소켓 설정
self.client_socket = None
self.config = None
self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.client_socket.connect((self.HOST, self.PORT))
# self.client_socket.setblocking(False)
self.user_id = None
self.user_pw = None
self.username = None
self.user_nickname = None
self.stored_talk_message = dict()
self.receive_thread = Thread(target=self.receive_message)
self.receive_thread.daemon = True
self.receive_thread.start()
self.talk_room_list_in_memory = list()
self.client_widget = None
self.decoder = KKODecoder()
self.all_user_list_in_memory = list()
self.selected_talk_room_id = 1
self.buffer_guest_list = None
#
# client function =================================
def set_widget(self, widget_):
self.client_widget = widget_
def get_user_self(self):
return User(self.user_id, self.username, self.user_pw, self.user_nickname)
def get_user_by_id(self, user_id):
find_list = [x for x in self.all_user_list_in_memory if x.user_id == user_id]
found_user = find_list[0]
return found_user
def get_all_user_list(self):
return self.all_user_list_in_memory.copy()
def get_not_yet_invited_user_list(self):
total_user_list = self.get_all_user_list().copy()
now_talk_room_id = self.selected_talk_room_id
talk_room_obj = self.get_talk_room_by_room_id(now_talk_room_id)
user_list = talk_room_obj.talk_room_user_list.copy()
result_list = [x for x in total_user_list if x not in user_list]
for user in result_list:
print(user.user_id)
return result_list
def get_already_invited_user_list(self):
now_talk_room_id = self.selected_talk_room_id
talk_room_obj = self.get_talk_room_by_room_id(now_talk_room_id)
user_list = talk_room_obj.talk_room_user_list.copy()
return user_list
def get_talk_room_by_room_id(self, talk_room_id: int):
assert isinstance(talk_room_id, int)
result = None
for talk_room in self.talk_room_list_in_memory:
temp_num = talk_room.talk_room_id
if temp_num == talk_room_id:
result = talk_room
break
if result is None:
raise '결과 찾기 실패'
found_talk_room = result
return found_talk_room
def send_join_id_for_assert_same_username(self, input_username: str):
data_msg = f"{input_username:<{self.BUFFER - self.HEADER_LENGTH}}".encode(self.FORMAT)
data_msg_length = len(data_msg)
request_msg = self.assert_username
header_msg = f"{request_msg:<{self.HEADER_LENGTH}}".encode(self.FORMAT)
self.client_socket.send(header_msg + data_msg) # 헤더를 붙이고 보내는 동작(?)
def send_join_id_and_pw_for_join_access(self, join_username, join_pw, join_nickname):
join_user = User(None, join_username, join_pw, join_nickname)
user_json_str = join_user.toJSON()
request_msg = self.join_user
result = self.fixed_volume(request_msg, user_json_str)
self.client_socket.send(result)
def send_login_id_and_pw_for_login_access(self, login_username, login_pw):
# 로그인 파일이 있다고 가정하고 제작
login_user = User(None, login_username, login_pw, None)
login_user_str = login_user.toJSON()
request_msg = self.login
result = self.fixed_volume(request_msg, login_user_str)
self.client_socket.send(result) # 응답 받기
# 로그인시 가져와야하는 정보들을 한번에 묶을수 있을까? 헤더 그리고 반환값이 달라서 문제임
def send_enter_square(self):
user_object = User(self.user_id, self.username, self.user_pw, self.user_nickname)
user_object_str = user_object.toJSON()
request_msg = self.enter_square
result = self.fixed_volume(request_msg, user_object_str)
self.client_socket.send(result)
def send_all_user_list(self):
user_object = User(self.user_id, self.username, self.user_pw, self.user_nickname)
user_object_str = user_object.toJSON()
request_msg = self.all_user_list
result = self.fixed_volume(request_msg, user_object_str)
self.client_socket.send(result)
def send_user_talk_room_list(self):
user_object = User(self.user_id, self.username, self.user_pw, self.user_nickname)
user_object_str = user_object.toJSON()
request_msg = self.user_talk_room_list
result = self.fixed_volume(request_msg, user_object_str)
self.client_socket.send(result)
def send_talk_room_user_list_se(self, talk_room_id):
"""
해당 방 정보 유저 갱신
:param talk_room_id:
:return:
"""
self.selected_talk_room_id = talk_room_id # 해당 선택된 톡방 저장
user_talk_room_obj = UserTalkRoom(None, self.user_id, talk_room_id)
user_talk_room_obj_str = user_talk_room_obj.toJSON()
request_msg = self.talk_room_user_list_se
result = self.fixed_volume(request_msg, user_talk_room_obj_str)
self.client_socket.send(result)
def send_out_talk_room(self, talk_room_id):
user_talk_room_obj = UserTalkRoom(None, self.user_id, talk_room_id)
user_talk_room_obj_str = user_talk_room_obj.toJSON()
request_msg = self.out_talk_room
result = self.fixed_volume(request_msg, user_talk_room_obj_str)
self.client_socket.send(result)
def send_send_msg_se(self, talk_room_id, msg):
msg_obj = Message(None, self.user_id, talk_room_id, str(datetime.datetime.now()), msg, None,
User(self.user_id, self.username, self.user_pw, self.user_nickname))
# self.store_message(msg_obj)
msg_obj_str = msg_obj.toJSON()
request_msg = self.send_msg_se
result = self.fixed_volume(request_msg, msg_obj_str)
self.client_socket.send(result)
def send_invite_user_talk_room(self, talk_room_id, invite_user):
user_talk_room_obj = UserTalkRoom(None, invite_user, talk_room_id)
user_talk_room_obj_str = user_talk_room_obj.toJSON()
request_msg = self.invite_user_talk_room
result = self.fixed_volume(request_msg, user_talk_room_obj_str)
self.client_socket.send(result)
def send_make_talk_room(self, room_name, guest_user_id_list: list[int], open_time_stamp):
"""2번에 나눠서 로직 구성됨, 1) 방 개설 / 2) 방 입장시키기"""
self.buffer_guest_list = guest_user_id_list
create_room = TalkRoom(None, room_name, open_time_stamp)
create_room_str = create_room.toJSON()
reqeust_msg = self.make_talk_room
result = self.fixed_volume(reqeust_msg, create_room_str)
self.client_socket.send(result)
def invite_guest_user(self, talk_room_id):
# 방 아이디를 부여받아야함. talk_room_id 부여 받고 스레드상에서 발동됨
guest_user_id_list = self.buffer_guest_list
for guest_id in guest_user_id_list:
self.send_invite_user_talk_room(talk_room_id, guest_id)
self.send_talk_room_user_list_se(talk_room_id)
self.buffer_guest_list.clear()
def send_talk_room_msg(self, talk_room_id):
user_talk_room_obj = UserTalkRoom(None, self.user_id, talk_room_id)
user_talk_room_obj_str = user_talk_room_obj.toJSON()
reqeust_msg = self.talk_room_msg
result = self.fixed_volume(reqeust_msg, user_talk_room_obj_str)
self.client_socket.send(result)
# 크기 고정으로 만들어 주는 함수
def fixed_volume(self, header, data):
header_msg = f"{header:<{self.HEADER_LENGTH}}".encode(self.FORMAT)
data_msg = f"{data:<{self.BUFFER - self.HEADER_LENGTH}}".encode(self.FORMAT)
return header_msg + data_msg
def store_message(self, message_obj):
"""
클라이언트 stored_talk_message 변수에 메시지를 저장하는 함수
:param message_obj:
:return:
"""
talk_room_id = message_obj.talk_room_id
if talk_room_id not in self.stored_talk_message.keys():
self.stored_talk_message.update({talk_room_id: list()})
target_message_list = self.stored_talk_message[talk_room_id]
target_message_list.append(message_obj)
def send_file_to_chat_room(self):
# todo: send 메시지
pass
def receive_message(self):
while True:
# self.return_result = self.client_socket.recv(self.BUFFER).decode(self.FORMAT)
return_result = self.client_socket.recv(self.BUFFER).decode(self.FORMAT)
response_header = return_result[:self.HEADER_LENGTH].strip()
response_data = return_result[self.HEADER_LENGTH:].strip()
print(f"CLIENT RECEIVED: ({response_header},{response_data})")
# 아이디 중복 확인 결과
if response_header == self.assert_username:
if response_data == 'pass':
self.client_widget.assert_same_id_signal.emit(True)
elif response_data == '.':
self.client_widget.assert_same_id_signal.emit(False)
# 회원 가입 중복 확인 결과
elif response_header == self.join_user:
if response_data == 'pass':
self.client_widget.sign_up_signal.emit(True)
elif response_data == '.':
self.client_widget.sign_up_signal.emit(False)
# 로그인 신청 확인 결과, 로그인한 유저정보 저장?
elif response_header == self.login:
if response_data == '.':
self.client_widget.log_in_signal.emit(False)
else:
object_data = self.decoder.decode_any(response_data)
self.username = object_data.username
self.user_id = object_data.user_id
self.user_pw = object_data.password
self.user_nickname = object_data.nickname
self.client_widget.log_in_signal.emit(True)
# 초기 단톡방 입장 반환
elif response_header == self.enter_square:
if response_data == 'pass':
self.client_widget.enter_square_signal.emit(True)
# 본인 제외 모든 유저 정보
elif response_header == self.all_user_list:
if response_data == '.':
print('오류?')
else:
self.all_user_list_in_memory = self.decoder.decode_any(response_data)
self.client_widget.all_user_list_signal.emit(response_data)
# 채팅방 리스트 정보
elif response_header == self.user_talk_room_list:
self.talk_room_list_in_memory = self.decoder.decode_any(response_data)
for talk_room_obj in self.talk_room_list_in_memory:
self.send_talk_room_user_list_se(talk_room_obj.talk_room_id)
self.client_widget.user_talk_room_signal.emit(response_data)
# 채팅방 참여 유저 정보
elif self.talk_room_user_list_se in response_header:
if response_data == '.':
print('아무도 없는 방')
else:
if len(self.talk_room_list_in_memory) == 0:
time.sleep(0.05) # 정보가 받아질 때까지 대기
try:
header_str, talk_room_id_str = response_header.split("%")
print(header_str, talk_room_id_str)
talk_room_id = int(talk_room_id_str.strip())
self.client_widget.talk_room_user_list_se_signal.emit(talk_room_id, response_data)
except:
return
# 방나가기
elif response_header == self.out_talk_room:
if response_data == 'pass':
self.client_widget.out_talk_room_signal.emit(True)
elif response_data == '.':
self.client_widget.out_talk_room_signal.emit(False)
# 메시지 받기
elif response_header == self.send_msg_se:
msg_obj = self.decoder.decode_any(response_data)
# if msg_obj.sender_user_id == self.user_id:
# pass
# else:
# self.client_widget.send_msg_se_signal.emit(response_data)
self.client_widget.send_msg_se_signal.emit(response_data)
# 상대방 초대
elif response_header == self.invite_user_talk_room:
if response_data == 'pass':
self.client_widget.invite_user_talk_room_signal.emit(True)
elif response_data == '.':
self.client_widget.invite_user_talk_room_signal.emit(False)
# 방 만들기
elif response_header == self.make_talk_room:
talk_room_obj = self.decoder.decode_any(response_data)
self.invite_guest_user(talk_room_obj.talk_room_id)
self.talk_room_list_in_memory.append(talk_room_obj)
self.client_widget.make_talk_room_signal.emit(talk_room_obj.talk_room_id)
# 메시지 받아보기
elif response_header == self.talk_room_msg:
message_list = self.decoder.decode_any(response_data)
for m in message_list:#Code/network/class_client_prototype.py
from threading import Thread
from PyQt5 import QtCore, QtGui, QtWidgets
from Code.domain.class_db_connector import DBConnector
from Code.domain.class_user import User
from Code.network.class_worker_thread import WorkerServerThread
from Code.network.server_ui.ui_chat_room import Ui_prototype
from Code.network.server_ui.ui_server_controller_widget import Ui_server_controller
from Common.class_json import KKODecoder, KKOEncoder
from Common.common_module import *
from PyQt5.QtCore import pyqtSignal
class ClientPrototypeWidget(QtWidgets.QWidget, Ui_prototype):
ENCODED_DOT = bytes('.', 'utf-8')
ENCODED_PASS = bytes('pass', 'utf-8')
# signal 클래스 변수
assert_same_id_signal = pyqtSignal(bool)
sign_up_signal = pyqtSignal(bool)
log_in_signal = pyqtSignal(bool)
enter_square_signal = pyqtSignal(bool)
all_user_list_signal = pyqtSignal(str)
user_talk_room_signal = pyqtSignal(str)
talk_room_user_list_se_signal = pyqtSignal(int, str)
out_talk_room_signal = pyqtSignal(bool)
send_msg_se_signal = pyqtSignal(str)
invite_user_talk_room_signal = pyqtSignal(bool)
make_talk_room_signal = pyqtSignal(int)
talk_room_msg_signal = pyqtSignal(str)
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()
self.set_client_know_each_other()
self.assert_same_id_signal.connect(self.assert_same_name_res)
self.sign_up_signal.connect(self.sign_up_res)
self.log_in_signal.connect(self.log_in_res)
self.enter_square_signal.connect(self.enter_square_res)
self.all_user_list_signal.connect(self.all_user_list_res)
self.user_talk_room_signal.connect(self.user_talk_room_list_res)
self.talk_room_user_list_se_signal.connect(self.talk_room_user_list_se_res)
self.out_talk_room_signal.connect(self.out_talk_room_res)
self.send_msg_se_signal.connect(self.send_msg_se_res)
self.invite_user_talk_room_signal.connect(self.invite_user_talk_room_res)
self.make_talk_room_signal.connect(self.make_talk_room_res)
self.talk_room_msg_signal.connect(self.talk_room_msg_res)
def set_client_know_each_other(self):
self.client_app.set_widget(self)
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.assert_same_username())
self.btn_join.clicked.connect(lambda state: self.join_access())
self.btn_login.clicked.connect(lambda state: self.login_access())
self.btn_send_message.clicked.connect(lambda state: self.send_msg_se())
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 assert_same_username(self):
input_username = self.line_edit_for_join_id.text()
self.client_app.send_join_id_for_assert_same_username(input_username) # 헤더를 붙이고 보내는 동작(?)
# 서버 -> 클라 아이디 중복 체크 결과 대응
def assert_same_name_res(self, return_result: bool):
if return_result is True:
self.valid_duplication_id = True
return QtWidgets.QMessageBox.about(self, "가능", "중복 없는 아이디, 써도됌")
elif return_result is False:
return QtWidgets.QMessageBox.about(self, "불가능", "중복 아이디, 새로 쓰기")
# 클라 -> 서버 회원가입 요청
def 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()
self.client_app.send_join_id_and_pw_for_join_access(join_username, join_pw, join_nickname)
# 서버 -> 클라 회원가입 결과 체크 결과 대응
def sign_up_res(self, return_result: bool):
if return_result is True:
return QtWidgets.QMessageBox.about(self, "성공", "회원가입 성공")
elif return_result is False:
return QtWidgets.QMessageBox.about(self, "실패", "회원가입 실패")
# 클라 -> 서버 로그인 요청
def login_access(self):
login_username = self.line_edit_for_login_id.text()
login_pw = self.line_edit_for_login_pw.text()
self.client_app.send_login_id_and_pw_for_login_access(login_username, login_pw)
# 서버 -> 클라 로그인 결과 체크 결과 대응
def log_in_res(self, return_result: bool):
if return_result is True:
# 모든건 로그인 버튼을 누르면 시작한다. 나중에 수정
self.enter_square()
self.all_user_list()
self.user_talk_room_list()
self.talk_room_user_list_se()
# self.out_talk_room()
self.talk_room_msg()
return QtWidgets.QMessageBox.about(self, "성공", "login 성공")
elif return_result is False:
return QtWidgets.QMessageBox.about(self, "실패", "login 실패")
# 클라 -> 서버 초기 체팅방 입장, 로그인시 실행
def enter_square(self):
self.client_app.send_enter_square()
# 서버 -> 클라 초기 체팅방 입장 결과 체크
def enter_square_res(self):
# 화면 띄우기? 화면전환?
print("초기방 입장 완료")
# 클라 -> 서버 유저 리스트 요청, 로그인시 할 수도있음
def all_user_list(self):
self.client_app.send_all_user_list()
# 서버 -> 클라 유저 리스트 정보 받음
def all_user_list_res(self, return_result: str):
all_user_list = self.decoder.decode(return_result)
print('가입한 유저 정보', all_user_list)
# 클라 -> 서버 채팅방 리스트 요청
def user_talk_room_list(self):
self.client_app.send_user_talk_room_list()
# 서버 -> 클라 채팅방 리스트 정보 받음
def user_talk_room_list_res(self, return_result: str):
talk_room_list = self.decoder.decode(return_result)
print('존재하는 방 리스트', talk_room_list)
# 클라 -> 서버 채팅방 관련 유저 정보 요청
# 방 아이디를 넘겨줘야 할듯 하다.
def talk_room_user_list_se(self):
# self.client_app.send_talk_room_user_list_se(talk_room_id)
pass
# 서버 -> 클라 톡방 유저 객체 정보 획득
def talk_room_user_list_se_res(self, talk_room_id:int, return_result: str):
user_list = self.decoder.decode(return_result)
print('방에 존재하는 유저 정보', user_list)
# 클라 -> 서버 채팅방 나가기 요청
# 방 아이디를 넘겨줘야 할듯 하다
def out_talk_room(self):
# self.client_app.send_out_talk_room(talk_room_id)
pass
# 채팅방 나가기 결과 반환
# 메세지 박스를 화면 전환 해주세요
def out_talk_room_res(self, return_result: bool):
if return_result is True:
return QtWidgets.QMessageBox.about(self, "성공", "방탈출 성공")
elif return_result is False:
return QtWidgets.QMessageBox.about(self, "실패", "방탈출 실패")
# 화면 전환후 채팅방 목록 불러오기
# 클라 -> 서버 메시지 전달
def send_msg_se(self):
txt_message = self.text_edit_for_send_chat.toPlainText()
self.text_edit_for_send_chat.clear()
self.text_edit_chat_room.appendPlainText(txt_message)
self.client_app.send_send_msg_se(1, txt_message)
# 서버 -> 클라 메시지 받기
def send_msg_se_res(self, return_result: str):
message = self.decoder.decode_any(return_result)
self.text_edit_chat_room.appendPlainText(
f"{message.user_obj.nickname} : {message.contents} > {message.send_time_stamp}")
# todo: send 메시지
# 클라 -> 서버 단톡방 초대 요청
def invite_user_talk_room(self):
# self.client_app.send_invite_user_talk_room(talk_room_id, invite_user)
pass
# 서버 -> 클라 단톡방 초대 완료
def invite_user_talk_room_res(self, return_result: bool):
print('초대완료')
# 채팅방 개설하기
def make_talk_room(self):
# 시간은 어떻게 받을 지몰라서 그대로 둠. user_id도 같인 이유
# self.client_app.send_make_talk_room(room_name, guest_list, open_time_stmp)
pass
def make_talk_room_res(self, return_result: bool):
print('개설완료')
# 단톡방 리스트 갱신하는 파일 만들기
# 채팅방 입장시 클라 -> 서버 이전 message 내용 전송
def talk_room_msg(self):
self.client_app.send_talk_room_msg(talk_room_id=1)
# 서버 -> 클라 message obj 내용 받기
def talk_room_msg_res(self, return_result: str):
room_msg = self.decoder.decode_any(return_result)
# 오브젝트들 잘 나오는지 확인
for i in room_msg:
print(i)
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 메시지'광주인력개발원' 카테고리의 다른 글
| | 미니 의료 정보 시스템 | - 1일차 - | 개발일지 [2023-07-17 학습일지] (0) | 2023.07.30 |
|---|---|
| | 미니 의료 정보 시스템 | - 0일차 - | 개발계획서 및 개발일지 [2023-07-17 학습일지] (0) | 2023.07.30 |
| 크크오톡 팀프로젝트 - 5일차 - 개발일지 [2023-07-15 학습일지] (0) | 2023.07.30 |
| [팀 개발일지] 5일차 | 크크오톡 | 1조 Going To Pro Level 팀 | 파이썬 단체 채팅 프로그램 | (0) | 2023.07.30 |
| 크크오톡 팀프로젝트 - 4일차 - 개발일지 [2023-07-14 학습일지] (0) | 2023.07.30 |