# 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