You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

594 lines
20 KiB

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

# P2P Network Communication - Chat Widget
"""
聊天窗口组件
实现消息输入、发送、显示和历史加载
需求: 3.1, 3.2, 3.3, 3.4, 3.5, 9.2
"""
import logging
from datetime import datetime
from typing import Optional, List
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QTextEdit, QLineEdit,
QPushButton, QLabel, QScrollArea, QFrame, QSizePolicy
)
from PyQt6.QtCore import Qt, pyqtSignal, QTimer
from PyQt6.QtGui import QTextCursor, QKeyEvent
from shared.models import ChatMessage, MessageType, UserInfo
logger = logging.getLogger(__name__)
class MessageBubble(QFrame):
"""消息气泡组件"""
def __init__(self, message: ChatMessage, is_self: bool = False, parent=None):
super().__init__(parent)
self.message = message
self.is_self = is_self
self._setup_ui()
def _setup_ui(self) -> None:
"""设置UI"""
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 5, 10, 5)
layout.setSpacing(2)
# 根据消息类型显示不同内容
if self.message.content_type == MessageType.IMAGE:
# 图片消息 - 显示缩略图
content_widget = self._create_image_widget()
elif self.message.content_type == MessageType.FILE_REQUEST:
# 文件消息 - 显示文件信息
content_widget = self._create_file_widget()
else:
# 文本消息
content_widget = QLabel(self.message.content)
content_widget.setWordWrap(True)
content_widget.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
# 时间戳
time_str = self.message.timestamp.strftime("%H:%M")
time_label = QLabel(time_str)
time_label.setStyleSheet("color: #999; font-size: 10px;")
# 状态指示
status_text = "✓✓" if self.message.is_read else ("" if self.message.is_sent else "")
status_label = QLabel(status_text)
status_label.setStyleSheet("color: #999; font-size: 10px;")
if self.is_self:
# 自己发送的消息,右对齐
self.setStyleSheet("""
QFrame {
background-color: #dcf8c6;
border-radius: 10px;
margin-left: 50px;
}
""")
layout.setAlignment(Qt.AlignmentFlag.AlignRight)
bottom_layout = QHBoxLayout()
bottom_layout.addStretch()
bottom_layout.addWidget(time_label)
bottom_layout.addWidget(status_label)
layout.addWidget(content_widget)
layout.addLayout(bottom_layout)
else:
# 对方发送的消息,左对齐
self.setStyleSheet("""
QFrame {
background-color: white;
border-radius: 10px;
margin-right: 50px;
border: 1px solid #eee;
}
""")
layout.setAlignment(Qt.AlignmentFlag.AlignLeft)
layout.addWidget(content_widget)
layout.addWidget(time_label)
def _create_image_widget(self) -> QWidget:
"""创建图片显示组件"""
import os
from PyQt6.QtGui import QPixmap
container = QWidget()
layout = QVBoxLayout(container)
layout.setContentsMargins(0, 0, 0, 0)
file_path = self.message.content
file_name = os.path.basename(file_path)
# 尝试加载图片
if os.path.exists(file_path):
pixmap = QPixmap(file_path)
if not pixmap.isNull():
# 缩放图片
scaled = pixmap.scaled(200, 200, Qt.AspectRatioMode.KeepAspectRatio,
Qt.TransformationMode.SmoothTransformation)
image_label = QLabel()
image_label.setPixmap(scaled)
image_label.setCursor(Qt.CursorShape.PointingHandCursor)
image_label.setToolTip(f"点击打开: {file_path}")
image_label.mousePressEvent = lambda e: self._open_file(file_path)
layout.addWidget(image_label)
else:
layout.addWidget(QLabel(f"[图片] {file_name}"))
else:
layout.addWidget(QLabel(f"[图片] {file_name}\n(文件不存在)"))
return container
def _create_file_widget(self) -> QWidget:
"""创建文件显示组件"""
import os
container = QWidget()
layout = QHBoxLayout(container)
layout.setContentsMargins(0, 0, 0, 0)
file_path = self.message.content
file_name = os.path.basename(file_path)
# 文件图标
icon_label = QLabel("📄")
icon_label.setStyleSheet("font-size: 24px;")
layout.addWidget(icon_label)
# 文件信息
info_layout = QVBoxLayout()
name_label = QLabel(file_name)
name_label.setStyleSheet("font-weight: bold;")
info_layout.addWidget(name_label)
if os.path.exists(file_path):
size = os.path.getsize(file_path)
size_str = self._format_size(size)
size_label = QLabel(size_str)
size_label.setStyleSheet("color: #666; font-size: 11px;")
info_layout.addWidget(size_label)
else:
info_layout.addWidget(QLabel("(文件不存在)"))
layout.addLayout(info_layout)
# 点击打开
container.setCursor(Qt.CursorShape.PointingHandCursor)
container.setToolTip(f"点击打开: {file_path}")
container.mousePressEvent = lambda e: self._open_file(file_path)
return container
def _format_size(self, size: int) -> str:
"""格式化文件大小"""
if size < 1024:
return f"{size} B"
elif size < 1024 * 1024:
return f"{size / 1024:.1f} KB"
elif size < 1024 * 1024 * 1024:
return f"{size / (1024 * 1024):.1f} MB"
else:
return f"{size / (1024 * 1024 * 1024):.1f} GB"
def _open_file(self, file_path: str):
"""打开文件"""
import os
import subprocess
import sys
if not os.path.exists(file_path):
return
try:
if sys.platform == 'win32':
os.startfile(file_path)
elif sys.platform == 'darwin':
subprocess.run(['open', file_path])
else:
subprocess.run(['xdg-open', file_path])
except Exception as e:
logger.error(f"Failed to open file: {e}")
class ChatWidget(QWidget):
"""
聊天窗口组件
实现消息输入和发送 (需求 3.1)
实现消息显示和历史加载 (需求 3.2, 3.5, 9.2)
实现消息状态显示 (需求 3.3, 3.4)
"""
# 信号定义
message_sent = pyqtSignal(str, str) # peer_id, content
file_send_requested = pyqtSignal(str) # peer_id
image_send_requested = pyqtSignal(str) # peer_id
voice_call_requested = pyqtSignal(str) # peer_id
voice_call_end_requested = pyqtSignal() # 挂断通话
def __init__(self, parent=None):
super().__init__(parent)
self._current_peer_id: Optional[str] = None
self._current_peer_info: Optional[UserInfo] = None
self._messages: List[ChatMessage] = []
self._in_call: bool = False # 是否在通话中
self._setup_ui()
self._connect_signals()
logger.info("ChatWidget initialized")
def _setup_ui(self) -> None:
"""设置UI"""
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
# 聊天对象信息栏
self._header = self._create_header()
layout.addWidget(self._header)
# 消息显示区域
self._message_area = self._create_message_area()
layout.addWidget(self._message_area, 1)
# 输入区域
self._input_area = self._create_input_area()
layout.addWidget(self._input_area)
def _create_header(self) -> QWidget:
"""创建头部信息栏"""
header = QFrame()
header.setFixedHeight(50)
header.setStyleSheet("""
QFrame {
background-color: #f5f5f5;
border-bottom: 1px solid #ddd;
}
""")
layout = QHBoxLayout(header)
layout.setContentsMargins(15, 0, 15, 0)
self._peer_name_label = QLabel("选择联系人开始聊天")
self._peer_name_label.setStyleSheet("font-size: 16px; font-weight: bold;")
layout.addWidget(self._peer_name_label)
layout.addStretch()
self._peer_status_label = QLabel("")
self._peer_status_label.setStyleSheet("color: #666;")
layout.addWidget(self._peer_status_label)
return header
def _create_message_area(self) -> QWidget:
"""创建消息显示区域"""
scroll_area = QScrollArea()
scroll_area.setWidgetResizable(True)
scroll_area.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
scroll_area.setStyleSheet("""
QScrollArea {
border: none;
background-color: #e5ddd5;
}
""")
self._message_container = QWidget()
self._message_layout = QVBoxLayout(self._message_container)
self._message_layout.setContentsMargins(10, 10, 10, 10)
self._message_layout.setSpacing(10)
self._message_layout.addStretch()
scroll_area.setWidget(self._message_container)
self._scroll_area = scroll_area
return scroll_area
def _create_input_area(self) -> QWidget:
"""创建输入区域"""
input_frame = QFrame()
input_frame.setStyleSheet("""
QFrame {
background-color: #f0f0f0;
border-top: 1px solid #ddd;
}
""")
layout = QVBoxLayout(input_frame)
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(10)
# 功能按钮行
button_layout = QHBoxLayout()
button_layout.setSpacing(5)
self._file_btn = QPushButton("📎")
self._file_btn.setFixedSize(30, 30)
self._file_btn.setToolTip("发送文件")
self._file_btn.clicked.connect(self._on_file_btn_clicked)
button_layout.addWidget(self._file_btn)
self._image_btn = QPushButton("🖼️")
self._image_btn.setFixedSize(30, 30)
self._image_btn.setToolTip("发送图片")
self._image_btn.clicked.connect(self._on_image_btn_clicked)
button_layout.addWidget(self._image_btn)
self._voice_btn = QPushButton("📞")
self._voice_btn.setFixedSize(30, 30)
self._voice_btn.setToolTip("语音通话")
self._voice_btn.clicked.connect(self._on_voice_btn_clicked)
button_layout.addWidget(self._voice_btn)
# 挂断按钮(默认隐藏)
self._hangup_btn = QPushButton("📵")
self._hangup_btn.setFixedSize(30, 30)
self._hangup_btn.setToolTip("挂断通话")
self._hangup_btn.setStyleSheet("""
QPushButton {
background-color: #ff4444;
border-radius: 15px;
color: white;
}
QPushButton:hover {
background-color: #ff0000;
}
""")
self._hangup_btn.clicked.connect(self._on_hangup_btn_clicked)
self._hangup_btn.hide() # 默认隐藏
button_layout.addWidget(self._hangup_btn)
# 通话状态标签(默认隐藏)
self._call_status_label = QLabel("")
self._call_status_label.setStyleSheet("color: #4a90d9; font-weight: bold;")
self._call_status_label.hide()
button_layout.addWidget(self._call_status_label)
button_layout.addStretch()
layout.addLayout(button_layout)
# 输入行
input_layout = QHBoxLayout()
input_layout.setSpacing(10)
self._input_edit = QLineEdit()
self._input_edit.setPlaceholderText("输入消息...")
self._input_edit.setMinimumHeight(40)
self._input_edit.setStyleSheet("""
QLineEdit {
border: 1px solid #ddd;
border-radius: 20px;
padding: 0 15px;
background-color: white;
}
""")
self._input_edit.returnPressed.connect(self._on_send_clicked)
input_layout.addWidget(self._input_edit, 1)
self._send_btn = QPushButton("发送")
self._send_btn.setMinimumHeight(40)
self._send_btn.setMinimumWidth(80)
self._send_btn.setStyleSheet("""
QPushButton {
background-color: #4a90d9;
color: white;
border: none;
border-radius: 20px;
font-size: 14px;
}
QPushButton:hover {
background-color: #3a80c9;
}
QPushButton:pressed {
background-color: #2a70b9;
}
QPushButton:disabled {
background-color: #ccc;
}
""")
self._send_btn.clicked.connect(self._on_send_clicked)
input_layout.addWidget(self._send_btn)
layout.addLayout(input_layout)
return input_frame
def _connect_signals(self) -> None:
"""连接信号"""
pass
def _on_send_clicked(self) -> None:
"""处理发送按钮点击"""
if not self._current_peer_id:
return
content = self._input_edit.text().strip()
if not content:
return
self.message_sent.emit(self._current_peer_id, content)
self._input_edit.clear()
def _on_file_btn_clicked(self) -> None:
"""处理文件按钮点击"""
if self._current_peer_id:
self.file_send_requested.emit(self._current_peer_id)
def _on_image_btn_clicked(self) -> None:
"""处理图片按钮点击"""
if self._current_peer_id:
self.image_send_requested.emit(self._current_peer_id)
def _on_voice_btn_clicked(self) -> None:
"""处理语音按钮点击"""
if self._current_peer_id:
self.voice_call_requested.emit(self._current_peer_id)
def _on_hangup_btn_clicked(self) -> None:
"""处理挂断按钮点击"""
self.voice_call_end_requested.emit()
self.set_call_state(False)
def _scroll_to_bottom(self) -> None:
"""滚动到底部"""
QTimer.singleShot(100, lambda: self._scroll_area.verticalScrollBar().setValue(
self._scroll_area.verticalScrollBar().maximum()
))
def _clear_messages(self) -> None:
"""清空消息显示"""
while self._message_layout.count() > 1:
item = self._message_layout.takeAt(0)
if item.widget():
item.widget().deleteLater()
# ==================== 公共方法 ====================
def set_peer(self, peer_id: str, peer_info: Optional[UserInfo] = None) -> None:
"""
设置当前聊天对象
实现用户切换聊天对象 (需求 9.2)
WHEN 用户切换聊天对象 THEN P2P_Client SHALL 加载并显示与该用户的聊天历史
Args:
peer_id: 对等端ID
peer_info: 对等端用户信息
"""
self._current_peer_id = peer_id
self._current_peer_info = peer_info
if peer_info:
self._peer_name_label.setText(peer_info.display_name or peer_info.username)
self._peer_status_label.setText(peer_info.status.value)
else:
self._peer_name_label.setText(peer_id)
self._peer_status_label.setText("")
# 清空并加载历史消息
self._clear_messages()
self._messages.clear()
# 启用输入
self._input_edit.setEnabled(True)
self._send_btn.setEnabled(True)
self._input_edit.setFocus()
logger.info(f"Chat peer set: {peer_id}")
def add_message(self, message: ChatMessage, is_self: bool = False) -> None:
"""
添加消息到显示区域
实现消息显示 (需求 3.2)
WHEN P2P_Client 收到文本消息 THEN P2P_Client SHALL 立即显示消息内容和发送者信息
Args:
message: 聊天消息
is_self: 是否是自己发送的消息
"""
self._messages.append(message)
bubble = MessageBubble(message, is_self)
# 在stretch之前插入
self._message_layout.insertWidget(self._message_layout.count() - 1, bubble)
self._scroll_to_bottom()
def load_history(self, messages: List[ChatMessage], current_user_id: str) -> None:
"""
加载聊天历史
实现消息历史加载 (需求 3.5)
WHEN 显示消息历史 THEN P2P_Client SHALL 按时间顺序展示所有消息记录
Args:
messages: 消息列表(应按时间排序)
current_user_id: 当前用户ID
"""
self._clear_messages()
self._messages = messages.copy()
for msg in messages:
is_self = msg.sender_id == current_user_id
bubble = MessageBubble(msg, is_self)
self._message_layout.insertWidget(self._message_layout.count() - 1, bubble)
self._scroll_to_bottom()
logger.info(f"Loaded {len(messages)} messages")
def update_message_status(self, message_id: str, is_sent: bool = False, is_read: bool = False) -> None:
"""
更新消息状态
实现消息状态显示 (需求 3.3, 3.4)
WHEN 消息发送成功 THEN P2P_Client SHALL 显示发送成功的状态标识
IF 消息发送失败 THEN P2P_Client SHALL 显示错误提示并提供重试选项
Args:
message_id: 消息ID
is_sent: 是否已发送
is_read: 是否已读
"""
for msg in self._messages:
if msg.message_id == message_id:
msg.is_sent = is_sent
msg.is_read = is_read
break
# 刷新显示简化实现实际应只更新对应的bubble
if self._current_peer_id and self._messages:
current_user_id = self._messages[0].sender_id if self._messages else ""
self.load_history(self._messages, current_user_id)
def clear(self) -> None:
"""清空聊天窗口"""
self._current_peer_id = None
self._current_peer_info = None
self._clear_messages()
self._messages.clear()
self._peer_name_label.setText("选择联系人开始聊天")
self._peer_status_label.setText("")
self._input_edit.setEnabled(False)
self._send_btn.setEnabled(False)
@property
def current_peer_id(self) -> Optional[str]:
"""获取当前聊天对象ID"""
return self._current_peer_id
def set_call_state(self, in_call: bool, status_text: str = "") -> None:
"""
设置通话状态
Args:
in_call: 是否在通话中
status_text: 状态文本(如"通话中...""正在呼叫..."
"""
self._in_call = in_call
if in_call:
self._voice_btn.hide()
self._hangup_btn.show()
self._call_status_label.setText(status_text or "通话中...")
self._call_status_label.show()
else:
self._voice_btn.show()
self._hangup_btn.hide()
self._call_status_label.hide()
self._call_status_label.setText("")
@property
def is_in_call(self) -> bool:
"""是否在通话中"""
return self._in_call