ppy4sjqvf
/
chatProgram
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

确保UI模块测试通过

Browse Source
main
杨博文 2 weeks ago
parent e4c02e8585
commit f119e3eabc
12 changed files with 4731 additions and 0 deletions
Show Stats Download Patch File Download Diff File

29
client/ui/__init__.py
Unescape View File

@ -0,0 +1,29 @@
# P2P Network Communication - UI Module
"""
用户界面模块
提供PyQt6图形界面组件
需求: 9.1
"""
__version__ = "0.1.0"
from client.ui.main_window import MainWindow
from client.ui.login_dialog import LoginDialog
from client.ui.chat_widget import ChatWidget
from client.ui.contact_list import ContactListWidget
from client.ui.file_transfer_widget import FileTransferWidget
from client.ui.media_player_widget import MediaPlayerWidget
from client.ui.voice_call_widget import VoiceCallWidget
from client.ui.system_tray import SystemTrayManager
__all__ = [
'MainWindow',
'LoginDialog',
'ChatWidget',
'ContactListWidget',
'FileTransferWidget',
'MediaPlayerWidget',
'VoiceCallWidget',
'SystemTrayManager',
]

426
client/ui/chat_widget.py
Unescape View File

@ -0,0 +1,426 @@
# 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)
# 消息内容
content_label = QLabel(self.message.content)
content_label.setWordWrap(True)
content_label.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_label)
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_label)
layout.addWidget(time_label)
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
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._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)
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 _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

331
client/ui/contact_list.py
Unescape View File

@ -0,0 +1,331 @@
# P2P Network Communication - Contact List Widget
"""
联系人列表组件
显示在线用户列表和联系人管理
需求: 9.1, 2.3, 2.4
"""
import logging
from typing import Optional, Dict, List
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QListWidget, QListWidgetItem,
QLabel, QLineEdit, QPushButton, QMenu, QFrame
)
from PyQt6.QtCore import Qt, pyqtSignal, QSize
from PyQt6.QtGui import QIcon, QColor, QAction
from shared.models import UserInfo, UserStatus
logger = logging.getLogger(__name__)
class ContactItem(QListWidgetItem):
"""联系人列表项"""
def __init__(self, user_info: UserInfo, parent: Optional[QListWidget] = None):
super().__init__(parent)
self.user_info = user_info
self._update_display()
def _update_display(self) -> None:
"""更新显示"""
status_icons = {
UserStatus.ONLINE: "🟢",
UserStatus.OFFLINE: "⚫",
UserStatus.BUSY: "🔴",
UserStatus.AWAY: "🟡",
}
icon = status_icons.get(self.user_info.status, "⚫")
display_name = self.user_info.display_name or self.user_info.username
self.setText(f"{icon} {display_name}")
# 设置提示信息
self.setToolTip(
f"用户名: {self.user_info.username}\n"
f"状态: {self.user_info.status.value}\n"
f"ID: {self.user_info.user_id}"
)
def update_status(self, status: UserStatus) -> None:
"""更新用户状态"""
self.user_info.status = status
self._update_display()
class ContactListWidget(QWidget):
"""
联系人列表组件
实现联系人列表面板 (需求 9.1)
WHEN 用户请求查看在线用户列表 THEN P2P_Client SHALL 显示当前所有在线用户 (需求 2.3)
WHEN 用户选择一个在线用户 THEN P2P_Client SHALL 显示该用户的基本信息和连接状态 (需求 2.4)
"""
# 信号定义
contact_selected = pyqtSignal(str) # 选中联系人时发出，参数为user_id
contact_double_clicked = pyqtSignal(str) # 双击联系人时发出
refresh_requested = pyqtSignal() # 请求刷新联系人列表
def __init__(self, parent: Optional[QWidget] = None):
super().__init__(parent)
self._contacts: Dict[str, ContactItem] = {}
self._setup_ui()
self._connect_signals()
logger.info("ContactListWidget initialized")
def _setup_ui(self) -> None:
"""设置UI"""
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
# 搜索栏
search_frame = QFrame()
search_frame.setStyleSheet("""
QFrame {
background-color: #f5f5f5;
border-bottom: 1px solid #ddd;
}
""")
search_layout = QHBoxLayout(search_frame)
search_layout.setContentsMargins(8, 8, 8, 8)
self._search_input = QLineEdit()
self._search_input.setPlaceholderText("搜索联系人...")
self._search_input.setClearButtonEnabled(True)
self._search_input.textChanged.connect(self._filter_contacts)
search_layout.addWidget(self._search_input)
self._refresh_btn = QPushButton("🔄")
self._refresh_btn.setFixedSize(30, 30)
self._refresh_btn.setToolTip("刷新联系人列表")
self._refresh_btn.clicked.connect(self._on_refresh_clicked)
search_layout.addWidget(self._refresh_btn)
layout.addWidget(search_frame)
# 联系人列表
self._list_widget = QListWidget()
self._list_widget.setStyleSheet("""
QListWidget {
border: none;
background-color: white;
}
QListWidget::item {
padding: 10px;
border-bottom: 1px solid #eee;
}
QListWidget::item:selected {
background-color: #e3f2fd;
color: black;
}
QListWidget::item:hover {
background-color: #f5f5f5;
}
""")
self._list_widget.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self._list_widget.customContextMenuRequested.connect(self._show_context_menu)
layout.addWidget(self._list_widget, 1)
# 底部状态栏
status_frame = QFrame()
status_frame.setStyleSheet("""
QFrame {
background-color: #f5f5f5;
border-top: 1px solid #ddd;
}
""")
status_layout = QHBoxLayout(status_frame)
status_layout.setContentsMargins(8, 4, 8, 4)
self._status_label = QLabel("0 位联系人在线")
self._status_label.setStyleSheet("color: #666; font-size: 12px;")
status_layout.addWidget(self._status_label)
layout.addWidget(status_frame)
def _connect_signals(self) -> None:
"""连接信号"""
self._list_widget.itemClicked.connect(self._on_item_clicked)
self._list_widget.itemDoubleClicked.connect(self._on_item_double_clicked)
def _on_item_clicked(self, item: QListWidgetItem) -> None:
"""处理项目点击"""
if isinstance(item, ContactItem):
self.contact_selected.emit(item.user_info.user_id)
def _on_item_double_clicked(self, item: QListWidgetItem) -> None:
"""处理项目双击"""
if isinstance(item, ContactItem):
self.contact_double_clicked.emit(item.user_info.user_id)
def _on_refresh_clicked(self) -> None:
"""处理刷新按钮点击"""
self.refresh_requested.emit()
def _filter_contacts(self, text: str) -> None:
"""过滤联系人列表"""
text = text.lower()
for i in range(self._list_widget.count()):
item = self._list_widget.item(i)
if isinstance(item, ContactItem):
visible = (
text in item.user_info.username.lower() or
text in (item.user_info.display_name or "").lower()
)
item.setHidden(not visible)
def _show_context_menu(self, pos) -> None:
"""显示右键菜单"""
item = self._list_widget.itemAt(pos)
if not isinstance(item, ContactItem):
return
menu = QMenu(self)
chat_action = QAction("发送消息", self)
chat_action.triggered.connect(lambda: self.contact_double_clicked.emit(item.user_info.user_id))
menu.addAction(chat_action)
menu.addSeparator()
info_action = QAction("查看资料", self)
info_action.triggered.connect(lambda: self._show_user_info(item.user_info))
menu.addAction(info_action)
menu.exec(self._list_widget.mapToGlobal(pos))
def _show_user_info(self, user_info: UserInfo) -> None:
"""显示用户信息"""
from PyQt6.QtWidgets import QMessageBox
QMessageBox.information(
self,
"用户信息",
f"用户名: {user_info.username}\n"
f"显示名: {user_info.display_name or '未设置'}\n"
f"状态: {user_info.status.value}\n"
f"ID: {user_info.user_id}"
)
def _update_status_label(self) -> None:
"""更新状态标签"""
online_count = sum(
1 for item in self._contacts.values()
if item.user_info.status == UserStatus.ONLINE
)
total_count = len(self._contacts)
self._status_label.setText(f"{online_count}/{total_count} 位联系人在线")
# ==================== 公共方法 ====================
def add_contact(self, user_info: UserInfo) -> None:
"""
添加联系人
Args:
user_info: 用户信息
"""
if user_info.user_id in self._contacts:
# 更新现有联系人
self.update_contact_status(user_info.user_id, user_info.status)
return
item = ContactItem(user_info)
self._list_widget.addItem(item)
self._contacts[user_info.user_id] = item
self._update_status_label()
logger.debug(f"Contact added: {user_info.username}")
def remove_contact(self, user_id: str) -> None:
"""
移除联系人
Args:
user_id: 用户ID
"""
if user_id not in self._contacts:
return
item = self._contacts.pop(user_id)
row = self._list_widget.row(item)
self._list_widget.takeItem(row)
self._update_status_label()
logger.debug(f"Contact removed: {user_id}")
def update_contact_status(self, user_id: str, status: UserStatus) -> None:
"""
更新联系人状态
Args:
user_id: 用户ID
status: 新状态
"""
if user_id not in self._contacts:
return
self._contacts[user_id].update_status(status)
self._update_status_label()
logger.debug(f"Contact status updated: {user_id} -> {status.value}")
def set_contacts(self, users: List[UserInfo]) -> None:
"""
设置联系人列表
Args:
users: 用户信息列表
"""
self.clear_contacts()
for user in users:
self.add_contact(user)
def clear_contacts(self) -> None:
"""清空联系人列表"""
self._list_widget.clear()
self._contacts.clear()
self._update_status_label()
def get_selected_contact(self) -> Optional[UserInfo]:
"""
获取当前选中的联系人
Returns:
选中的用户信息，如果没有选中则返回None
"""
item = self._list_widget.currentItem()
if isinstance(item, ContactItem):
return item.user_info
return None
def get_contact(self, user_id: str) -> Optional[UserInfo]:
"""
获取指定联系人信息
Args:
user_id: 用户ID
Returns:
用户信息
"""
if user_id in self._contacts:
return self._contacts[user_id].user_info
return None
def select_contact(self, user_id: str) -> None:
"""
选中指定联系人
Args:
user_id: 用户ID
"""
if user_id in self._contacts:
self._list_widget.setCurrentItem(self._contacts[user_id])

673
client/ui/file_transfer_widget.py
Unescape View File

@ -0,0 +1,673 @@
# P2P Network Communication - File Transfer Widget
"""
文件传输界面组件
实现文件选择、传输进度显示和图片预览
需求: 4.1, 4.3, 4.6, 5.2, 5.3, 5.4, 5.6
"""
import logging
import os
from typing import Optional, Dict
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton,
QProgressBar, QFileDialog, QScrollArea, QFrame, QMessageBox,
QDialog, QSlider, QSizePolicy
)
from PyQt6.QtCore import Qt, pyqtSignal, QSize
from PyQt6.QtGui import QPixmap, QTransform
from shared.models import TransferProgress, TransferStatus
logger = logging.getLogger(__name__)
class TransferItem(QFrame):
"""传输项组件"""
cancel_requested = pyqtSignal(str) # file_id
def __init__(self, file_id: str, file_name: str, total_size: int,
is_upload: bool = True, parent=None):
super().__init__(parent)
self.file_id = file_id
self.file_name = file_name
self.total_size = total_size
self.is_upload = is_upload
self._setup_ui()
def _setup_ui(self) -> None:
"""设置UI"""
self.setFrameShape(QFrame.Shape.StyledPanel)
self.setStyleSheet("""
QFrame {
background-color: white;
border: 1px solid #ddd;
border-radius: 5px;
margin: 2px;
}
""")
layout = QVBoxLayout(self)
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(5)
# 文件信息行
info_layout = QHBoxLayout()
icon = "⬆️" if self.is_upload else "⬇️"
self._name_label = QLabel(f"{icon} {self.file_name}")
self._name_label.setStyleSheet("font-weight: bold;")
info_layout.addWidget(self._name_label, 1)
self._size_label = QLabel(self._format_size(self.total_size))
self._size_label.setStyleSheet("color: #666;")
info_layout.addWidget(self._size_label)
self._cancel_btn = QPushButton("✕")
self._cancel_btn.setFixedSize(20, 20)
self._cancel_btn.setStyleSheet("""
QPushButton {
border: none;
color: #999;
}
QPushButton:hover {
color: red;
}
""")
self._cancel_btn.clicked.connect(lambda: self.cancel_requested.emit(self.file_id))
info_layout.addWidget(self._cancel_btn)
layout.addLayout(info_layout)
# 进度条
self._progress_bar = QProgressBar()
self._progress_bar.setRange(0, 100)
self._progress_bar.setValue(0)
self._progress_bar.setTextVisible(True)
self._progress_bar.setStyleSheet("""
QProgressBar {
border: 1px solid #ddd;
border-radius: 3px;
text-align: center;
height: 20px;
}
QProgressBar::chunk {
background-color: #4a90d9;
border-radius: 2px;
}
""")
layout.addWidget(self._progress_bar)
# 状态行
status_layout = QHBoxLayout()
self._status_label = QLabel("等待中...")
self._status_label.setStyleSheet("color: #666; font-size: 12px;")
status_layout.addWidget(self._status_label)
status_layout.addStretch()
self._speed_label = QLabel("")
self._speed_label.setStyleSheet("color: #666; font-size: 12px;")
status_layout.addWidget(self._speed_label)
layout.addLayout(status_layout)
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):.2f} GB"
def _format_speed(self, speed: float) -> str:
"""格式化传输速度"""
return f"{self._format_size(int(speed))}/s"
def update_progress(self, progress: TransferProgress) -> None:
"""更新进度"""
percent = int(progress.progress_percent)
self._progress_bar.setValue(percent)
self._speed_label.setText(self._format_speed(progress.speed))
if progress.eta > 0:
eta_str = f"剩余 {int(progress.eta)}秒"
else:
eta_str = ""
self._status_label.setText(f"传输中... {eta_str}")
def set_status(self, status: TransferStatus) -> None:
"""设置状态"""
status_texts = {
TransferStatus.PENDING: "等待中...",
TransferStatus.IN_PROGRESS: "传输中...",
TransferStatus.COMPLETED: "✓ 完成",
TransferStatus.FAILED: "✕ 失败",
TransferStatus.CANCELLED: "已取消",
TransferStatus.PAUSED: "已暂停",
}
self._status_label.setText(status_texts.get(status, "未知"))
if status == TransferStatus.COMPLETED:
self._progress_bar.setValue(100)
self._cancel_btn.hide()
self._progress_bar.setStyleSheet("""
QProgressBar {
border: 1px solid #ddd;
border-radius: 3px;
text-align: center;
height: 20px;
}
QProgressBar::chunk {
background-color: #4caf50;
border-radius: 2px;
}
""")
elif status == TransferStatus.FAILED:
self._cancel_btn.hide()
self._progress_bar.setStyleSheet("""
QProgressBar {
border: 1px solid #ddd;
border-radius: 3px;
text-align: center;
height: 20px;
}
QProgressBar::chunk {
background-color: #f44336;
border-radius: 2px;
}
""")
class FileTransferWidget(QWidget):
"""
文件传输界面组件
实现文件选择对话框 (需求 4.1)
实现传输进度显示 (需求 4.3)
实现图片预览和查看 (需求 5.2, 5.3, 5.4, 5.6)
"""
# 信号定义
send_file_requested = pyqtSignal(str, str) # peer_id, file_path
cancel_transfer_requested = pyqtSignal(str) # file_id
def __init__(self, parent=None):
super().__init__(parent)
self._transfers: Dict[str, TransferItem] = {}
self._setup_ui()
logger.info("FileTransferWidget initialized")
def _setup_ui(self) -> None:
"""设置UI"""
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
# 标题栏
header = QFrame()
header.setFixedHeight(40)
header.setStyleSheet("""
QFrame {
background-color: #f5f5f5;
border-bottom: 1px solid #ddd;
}
""")
header_layout = QHBoxLayout(header)
header_layout.setContentsMargins(10, 0, 10, 0)
title_label = QLabel("文件传输")
title_label.setStyleSheet("font-weight: bold;")
header_layout.addWidget(title_label)
header_layout.addStretch()
self._clear_btn = QPushButton("清除已完成")
self._clear_btn.clicked.connect(self._clear_completed)
header_layout.addWidget(self._clear_btn)
layout.addWidget(header)
# 传输列表
scroll_area = QScrollArea()
scroll_area.setWidgetResizable(True)
scroll_area.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
self._transfer_container = QWidget()
self._transfer_layout = QVBoxLayout(self._transfer_container)
self._transfer_layout.setContentsMargins(5, 5, 5, 5)
self._transfer_layout.setSpacing(5)
self._transfer_layout.addStretch()
scroll_area.setWidget(self._transfer_container)
layout.addWidget(scroll_area, 1)
# 空状态提示
self._empty_label = QLabel("暂无传输任务")
self._empty_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._empty_label.setStyleSheet("color: #999;")
self._transfer_layout.insertWidget(0, self._empty_label)
def _clear_completed(self) -> None:
"""清除已完成的传输"""
to_remove = []
for file_id, item in self._transfers.items():
# 检查是否已完成（简化判断）
if item._progress_bar.value() == 100:
to_remove.append(file_id)
for file_id in to_remove:
self.remove_transfer(file_id)
def _update_empty_state(self) -> None:
"""更新空状态显示"""
self._empty_label.setVisible(len(self._transfers) == 0)
# ==================== 公共方法 ====================
def add_transfer(self, file_id: str, file_name: str, total_size: int,
is_upload: bool = True) -> None:
"""
添加传输任务
Args:
file_id: 文件ID
file_name: 文件名
total_size: 文件大小
is_upload: 是否是上传
"""
if file_id in self._transfers:
return
item = TransferItem(file_id, file_name, total_size, is_upload)
item.cancel_requested.connect(self._on_cancel_requested)
self._transfer_layout.insertWidget(self._transfer_layout.count() - 1, item)
self._transfers[file_id] = item
self._update_empty_state()
logger.debug(f"Transfer added: {file_name}")
def update_transfer_progress(self, file_id: str, progress: TransferProgress) -> None:
"""
更新传输进度
实现传输进度显示 (需求 4.3)
WHILE 文件传输进行中 THEN P2P_Client SHALL 显示传输进度百分比和传输速度
Args:
file_id: 文件ID
progress: 进度信息
"""
if file_id in self._transfers:
self._transfers[file_id].update_progress(progress)
def update_transfer_status(self, file_id: str, status: TransferStatus) -> None:
"""
更新传输状态
Args:
file_id: 文件ID
status: 状态
"""
if file_id in self._transfers:
self._transfers[file_id].set_status(status)
def remove_transfer(self, file_id: str) -> None:
"""
移除传输任务
Args:
file_id: 文件ID
"""
if file_id not in self._transfers:
return
item = self._transfers.pop(file_id)
self._transfer_layout.removeWidget(item)
item.deleteLater()
self._update_empty_state()
def _on_cancel_requested(self, file_id: str) -> None:
"""处理取消请求"""
self.cancel_transfer_requested.emit(file_id)
def show_file_dialog(self, peer_id: str) -> Optional[str]:
"""
显示文件选择对话框
实现文件选择对话框 (需求 4.1)
WHEN 用户选择发送文件 THEN File_Transfer_Module SHALL 允许用户从本地选择任意类型的文件
Args:
peer_id: 目标用户ID
Returns:
选择的文件路径，如果取消则返回None
"""
file_path, _ = QFileDialog.getOpenFileName(
self,
"选择文件",
"",
"所有文件 (*.*)"
)
if file_path:
self.send_file_requested.emit(peer_id, file_path)
return file_path
return None
def show_image_dialog(self, peer_id: str) -> Optional[str]:
"""
显示图片选择对话框
Args:
peer_id: 目标用户ID
Returns:
选择的图片路径
"""
file_path, _ = QFileDialog.getOpenFileName(
self,
"选择图片",
"",
"图片文件 (*.jpg *.jpeg *.png *.gif *.bmp);;所有文件 (*.*)"
)
if file_path:
self.send_file_requested.emit(peer_id, file_path)
return file_path
return None
def show_save_dialog(self, file_name: str) -> Optional[str]:
"""
显示保存文件对话框
实现用户选择保存位置 (需求 4.6)
WHEN 接收到文件 THEN P2P_Client SHALL 允许用户选择保存位置
Args:
file_name: 默认文件名
Returns:
保存路径
"""
file_path, _ = QFileDialog.getSaveFileName(
self,
"保存文件",
file_name,
"所有文件 (*.*)"
)
return file_path if file_path else None
def show_image_preview(self, image_path: str) -> None:
"""
显示图片预览
实现图片预览和查看 (需求 5.4)
WHEN 用户点击图片缩略图 THEN P2P_Client SHALL 打开图片查看器显示完整图片
Args:
image_path: 图片路径
"""
dialog = ImagePreviewDialog(image_path, self)
dialog.exec()
def create_image_thumbnail(self, image_path: str) -> Optional[ImageThumbnail]:
"""
创建图片缩略图组件
实现图片缩略图预览 (需求 5.2, 5.3)
Args:
image_path: 图片路径
Returns:
缩略图组件
"""
if not os.path.exists(image_path):
return None
thumbnail = ImageThumbnail(image_path)
thumbnail.clicked.connect(self.show_image_preview)
return thumbnail
class ImagePreviewDialog(QDialog):
"""
图片预览对话框
实现图片预览和查看 (需求 5.3, 5.4)
WHEN 图片传输完成 THEN P2P_Client SHALL 在聊天窗口内直接显示图片缩略图
WHEN 用户点击图片缩略图 THEN P2P_Client SHALL 打开图片查看器显示完整图片
"""
def __init__(self, image_path: str, parent=None):
super().__init__(parent)
self.image_path = image_path
self._rotation = 0
self._zoom_level = 100
self._setup_ui()
self._load_image()
def _setup_ui(self) -> None:
"""设置UI"""
self.setWindowTitle("图片预览")
self.setMinimumSize(600, 500)
self.resize(800, 600)
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
# 图片显示区域
scroll_area = QScrollArea()
scroll_area.setWidgetResizable(True)
scroll_area.setAlignment(Qt.AlignmentFlag.AlignCenter)
scroll_area.setStyleSheet("background-color: #2d2d2d;")
self._image_label = QLabel()
self._image_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
scroll_area.setWidget(self._image_label)
layout.addWidget(scroll_area, 1)
# 工具栏
toolbar = QFrame()
toolbar.setFixedHeight(50)
toolbar.setStyleSheet("""
QFrame {
background-color: #f5f5f5;
border-top: 1px solid #ddd;
}
""")
toolbar_layout = QHBoxLayout(toolbar)
toolbar_layout.setContentsMargins(10, 5, 10, 5)
toolbar_layout.setSpacing(10)
# 缩放控制
zoom_out_btn = QPushButton("➖")
zoom_out_btn.setFixedSize(30, 30)
zoom_out_btn.clicked.connect(self._zoom_out)
toolbar_layout.addWidget(zoom_out_btn)
self._zoom_slider = QSlider(Qt.Orientation.Horizontal)
self._zoom_slider.setRange(25, 400)
self._zoom_slider.setValue(100)
self._zoom_slider.setFixedWidth(150)
self._zoom_slider.valueChanged.connect(self._on_zoom_changed)
toolbar_layout.addWidget(self._zoom_slider)
zoom_in_btn = QPushButton("➕")
zoom_in_btn.setFixedSize(30, 30)
zoom_in_btn.clicked.connect(self._zoom_in)
toolbar_layout.addWidget(zoom_in_btn)
self._zoom_label = QLabel("100%")
self._zoom_label.setFixedWidth(50)
toolbar_layout.addWidget(self._zoom_label)
toolbar_layout.addStretch()
# 旋转按钮
rotate_left_btn = QPushButton("↺")
rotate_left_btn.setFixedSize(30, 30)
rotate_left_btn.setToolTip("向左旋转")
rotate_left_btn.clicked.connect(self._rotate_left)
toolbar_layout.addWidget(rotate_left_btn)
rotate_right_btn = QPushButton("↻")
rotate_right_btn.setFixedSize(30, 30)
rotate_right_btn.setToolTip("向右旋转")
rotate_right_btn.clicked.connect(self._rotate_right)
toolbar_layout.addWidget(rotate_right_btn)
toolbar_layout.addStretch()
# 关闭按钮
close_btn = QPushButton("关闭")
close_btn.clicked.connect(self.close)
toolbar_layout.addWidget(close_btn)
layout.addWidget(toolbar)
def _load_image(self) -> None:
"""加载图片"""
if not os.path.exists(self.image_path):
self._image_label.setText("图片不存在")
return
self._original_pixmap = QPixmap(self.image_path)
if self._original_pixmap.isNull():
self._image_label.setText("无法加载图片")
return
self._update_display()
def _update_display(self) -> None:
"""更新图片显示"""
if not hasattr(self, '_original_pixmap') or self._original_pixmap.isNull():
return
# 应用旋转
transform = QTransform()
transform.rotate(self._rotation)
rotated = self._original_pixmap.transformed(transform, Qt.TransformationMode.SmoothTransformation)
# 应用缩放
scale = self._zoom_level / 100.0
scaled = rotated.scaled(
int(rotated.width() * scale),
int(rotated.height() * scale),
Qt.AspectRatioMode.KeepAspectRatio,
Qt.TransformationMode.SmoothTransformation
)
self._image_label.setPixmap(scaled)
def _zoom_in(self) -> None:
"""放大"""
self._zoom_slider.setValue(min(400, self._zoom_level + 25))
def _zoom_out(self) -> None:
"""缩小"""
self._zoom_slider.setValue(max(25, self._zoom_level - 25))
def _on_zoom_changed(self, value: int) -> None:
"""缩放变化"""
self._zoom_level = value
self._zoom_label.setText(f"{value}%")
self._update_display()
def _rotate_left(self) -> None:
"""向左旋转"""
self._rotation = (self._rotation - 90) % 360
self._update_display()
def _rotate_right(self) -> None:
"""向右旋转"""
self._rotation = (self._rotation + 90) % 360
self._update_display()
class ImageThumbnail(QFrame):
"""
图片缩略图组件
实现图片缩略图预览 (需求 5.2)
WHEN 图片发送前 THEN P2P_Client SHALL 显示图片缩略图预览
"""
clicked = pyqtSignal(str) # image_path
def __init__(self, image_path: str, thumbnail_size: tuple = (150, 150), parent=None):
super().__init__(parent)
self.image_path = image_path
self.thumbnail_size = thumbnail_size
self._setup_ui()
def _setup_ui(self) -> None:
"""设置UI"""
self.setFrameShape(QFrame.Shape.StyledPanel)
self.setCursor(Qt.CursorShape.PointingHandCursor)
self.setStyleSheet("""
QFrame {
background-color: white;
border: 1px solid #ddd;
border-radius: 5px;
}
QFrame:hover {
border-color: #4a90d9;
}
""")
layout = QVBoxLayout(self)
layout.setContentsMargins(5, 5, 5, 5)
layout.setSpacing(5)
# 缩略图
self._image_label = QLabel()
self._image_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._image_label.setFixedSize(*self.thumbnail_size)
if os.path.exists(self.image_path):
pixmap = QPixmap(self.image_path)
if not pixmap.isNull():
scaled = pixmap.scaled(
*self.thumbnail_size,
Qt.AspectRatioMode.KeepAspectRatio,
Qt.TransformationMode.SmoothTransformation
)
self._image_label.setPixmap(scaled)
else:
self._image_label.setText("无法加载")
else:
self._image_label.setText("图片不存在")
layout.addWidget(self._image_label)
# 文件名
file_name = os.path.basename(self.image_path)
if len(file_name) > 20:
file_name = file_name[:17] + "..."
name_label = QLabel(file_name)
name_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
name_label.setStyleSheet("font-size: 11px; color: #666;")
layout.addWidget(name_label)
def mousePressEvent(self, event) -> None:
"""鼠标点击事件"""
if event.button() == Qt.MouseButton.LeftButton:
self.clicked.emit(self.image_path)
super().mousePressEvent(event)

304
client/ui/login_dialog.py
Unescape View File

@ -0,0 +1,304 @@
# P2P Network Communication - Login Dialog
"""
登录对话框
实现用户注册和登录界面
需求: 2.1
"""
import logging
import uuid
import re
from typing import Optional
from PyQt6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit,
QPushButton, QMessageBox, QFormLayout, QFrame, QTabWidget,
QWidget, QCheckBox
)
from PyQt6.QtCore import Qt, pyqtSignal
from shared.models import UserInfo, UserStatus
logger = logging.getLogger(__name__)
class LoginDialog(QDialog):
"""
登录对话框
实现用户名设置界面和登录/注册流程 (需求 2.1)
WHEN 用户首次使用应用程序 THEN P2P_Client SHALL 允许用户设置唯一的用户名和显示名称
"""
# 信号定义
login_successful = pyqtSignal(UserInfo)
def __init__(self, parent=None):
super().__init__(parent)
self._user_info: Optional[UserInfo] = None
self._setup_ui()
logger.info("LoginDialog initialized")
def _setup_ui(self) -> None:
"""设置UI"""
self.setWindowTitle("登录 / 注册")
self.setFixedSize(450, 400)
self.setModal(True)
layout = QVBoxLayout(self)
layout.setSpacing(15)
layout.setContentsMargins(20, 20, 20, 20)
# 标题
title_label = QLabel("P2P 通信应用")
title_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
title_label.setStyleSheet("font-size: 28px; font-weight: bold; color: #4a90d9;")
layout.addWidget(title_label)
subtitle_label = QLabel("安全、快速的点对点通信")
subtitle_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
subtitle_label.setStyleSheet("font-size: 12px; color: #666;")
layout.addWidget(subtitle_label)
layout.addSpacing(10)
# 选项卡
self._tab_widget = QTabWidget()
self._tab_widget.addTab(self._create_login_tab(), "登录")
self._tab_widget.addTab(self._create_register_tab(), "注册")
layout.addWidget(self._tab_widget)
# 设置焦点
self._login_username_input.setFocus()
def _create_login_tab(self) -> QWidget:
"""创建登录选项卡"""
tab = QWidget()
layout = QVBoxLayout(tab)
layout.setSpacing(15)
layout.setContentsMargins(20, 20, 20, 20)
# 表单
form_layout = QFormLayout()
form_layout.setSpacing(12)
self._login_username_input = QLineEdit()
self._login_username_input.setPlaceholderText("请输入用户名")
self._login_username_input.setMinimumHeight(38)
self._login_username_input.setStyleSheet(self._get_input_style())
form_layout.addRow("用户名:", self._login_username_input)
self._login_server_input = QLineEdit()
self._login_server_input.setPlaceholderText("服务器地址:端口")
self._login_server_input.setText("127.0.0.1:8888")
self._login_server_input.setMinimumHeight(38)
self._login_server_input.setStyleSheet(self._get_input_style())
form_layout.addRow("服务器:", self._login_server_input)
layout.addLayout(form_layout)
# 记住设置
self._remember_checkbox = QCheckBox("记住我的设置")
layout.addWidget(self._remember_checkbox)
layout.addStretch()
# 登录按钮
self._login_btn = QPushButton("登录")
self._login_btn.setMinimumHeight(45)
self._login_btn.setStyleSheet(self._get_button_style())
self._login_btn.clicked.connect(self._on_login)
layout.addWidget(self._login_btn)
# 快捷键
self._login_username_input.returnPressed.connect(self._on_login)
return tab
def _create_register_tab(self) -> QWidget:
"""创建注册选项卡"""
tab = QWidget()
layout = QVBoxLayout(tab)
layout.setSpacing(15)
layout.setContentsMargins(20, 20, 20, 20)
# 表单
form_layout = QFormLayout()
form_layout.setSpacing(12)
self._reg_username_input = QLineEdit()
self._reg_username_input.setPlaceholderText("3-20个字符，字母数字下划线")
self._reg_username_input.setMinimumHeight(38)
self._reg_username_input.setStyleSheet(self._get_input_style())
form_layout.addRow("用户名:", self._reg_username_input)
self._reg_display_name_input = QLineEdit()
self._reg_display_name_input.setPlaceholderText("显示给其他用户的名称（可选）")
self._reg_display_name_input.setMinimumHeight(38)
self._reg_display_name_input.setStyleSheet(self._get_input_style())
form_layout.addRow("显示名:", self._reg_display_name_input)
self._reg_server_input = QLineEdit()
self._reg_server_input.setPlaceholderText("服务器地址:端口")
self._reg_server_input.setText("127.0.0.1:8888")
self._reg_server_input.setMinimumHeight(38)
self._reg_server_input.setStyleSheet(self._get_input_style())
form_layout.addRow("服务器:", self._reg_server_input)
layout.addLayout(form_layout)
# 提示信息
hint_label = QLabel("用户名将作为您的唯一标识，注册后不可更改")
hint_label.setStyleSheet("color: #999; font-size: 11px;")
layout.addWidget(hint_label)
layout.addStretch()
# 注册按钮
self._register_btn = QPushButton("注册并登录")
self._register_btn.setMinimumHeight(45)
self._register_btn.setStyleSheet(self._get_button_style())
self._register_btn.clicked.connect(self._on_register)
layout.addWidget(self._register_btn)
return tab
def _get_input_style(self) -> str:
"""获取输入框样式"""
return """
QLineEdit {
border: 1px solid #ddd;
border-radius: 5px;
padding: 0 12px;
background-color: #fafafa;
}
QLineEdit:focus {
border-color: #4a90d9;
background-color: white;
}
"""
def _get_button_style(self) -> str:
"""获取按钮样式"""
return """
QPushButton {
background-color: #4a90d9;
color: white;
border: none;
border-radius: 5px;
font-size: 15px;
font-weight: bold;
}
QPushButton:hover {
background-color: #3a80c9;
}
QPushButton:pressed {
background-color: #2a70b9;
}
QPushButton:disabled {
background-color: #ccc;
}
"""
def _validate_username(self, username: str) -> tuple:
"""
验证用户名
Args:
username: 用户名
Returns:
(is_valid, error_message)
"""
if not username:
return False, "请输入用户名"
if len(username) < 3:
return False, "用户名至少需要3个字符"
if len(username) > 20:
return False, "用户名不能超过20个字符"
if not re.match(r'^[a-zA-Z0-9_]+$', username):
return False, "用户名只能包含字母、数字和下划线"
return True, ""
def _on_login(self) -> None:
"""处理登录"""
username = self._login_username_input.text().strip()
is_valid, error_msg = self._validate_username(username)
if not is_valid:
QMessageBox.warning(self, "错误", error_msg)
self._login_username_input.setFocus()
return
# 创建用户信息
self._user_info = UserInfo(
user_id=str(uuid.uuid4()),
username=username,
display_name=username,
status=UserStatus.ONLINE
)
logger.info(f"User logged in: {username}")
self.login_successful.emit(self._user_info)
self.accept()
def _on_register(self) -> None:
"""处理注册"""
username = self._reg_username_input.text().strip()
display_name = self._reg_display_name_input.text().strip()
is_valid, error_msg = self._validate_username(username)
if not is_valid:
QMessageBox.warning(self, "错误", error_msg)
self._reg_username_input.setFocus()
return
# 创建用户信息
self._user_info = UserInfo(
user_id=str(uuid.uuid4()),
username=username,
display_name=display_name or username,
status=UserStatus.ONLINE
)
logger.info(f"User registered and logged in: {username}")
self.login_successful.emit(self._user_info)
self.accept()
def get_user_info(self) -> Optional[UserInfo]:
"""
获取用户信息
Returns:
用户信息，如果未登录则返回None
"""
return self._user_info
def get_server_address(self) -> tuple:
"""
获取服务器地址
Returns:
(host, port) 元组
"""
# 根据当前选项卡获取服务器地址
if self._tab_widget.currentIndex() == 0:
server = self._login_server_input.text().strip()
else:
server = self._reg_server_input.text().strip()
if ":" in server:
host, port = server.split(":", 1)
try:
return host, int(port)
except ValueError:
return host, 8888
return server, 8888

550
client/ui/main_window.py
Unescape View File

@ -0,0 +1,550 @@
# P2P Network Communication - Main Window
"""
主窗口模块
实现应用程序主界面，包含联系人列表、聊天窗口和功能按钮
需求: 9.1
"""
import logging
from typing import Optional, Dict, Any
from PyQt6.QtWidgets import (
QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QSplitter, QStatusBar, QToolBar, QMenuBar,
QMenu, QMessageBox, QLabel, QFrame
)
from PyQt6.QtCore import Qt, pyqtSignal, QSize
from PyQt6.QtGui import QAction, QIcon, QCloseEvent
from config import ClientConfig
from shared.models import UserInfo, UserStatus, Message, MessageType
logger = logging.getLogger(__name__)
class MainWindow(QMainWindow):
"""
主窗口
实现应用程序主界面 (需求 9.1)
WHEN P2P_Client 启动 THEN P2P_Client SHALL 显示主界面包含联系人列表、聊天窗口和功能按钮
"""
# 信号定义
message_received = pyqtSignal(Message)
user_status_changed = pyqtSignal(str, UserStatus)
connection_state_changed = pyqtSignal(str)
def __init__(self, config: Optional[ClientConfig] = None, parent: Optional[QWidget] = None):
"""
初始化主窗口
Args:
config: 客户端配置
parent: 父窗口
"""
super().__init__(parent)
self.config = config or ClientConfig()
self._current_user: Optional[UserInfo] = None
self._current_chat_peer: Optional[str] = None
# 子组件引用（延迟导入避免循环依赖）
self._contact_list: Optional[QWidget] = None
self._chat_widget: Optional[QWidget] = None
self._file_transfer_widget: Optional[QWidget] = None
self._media_player_widget: Optional[QWidget] = None
self._voice_call_widget: Optional[QWidget] = None
self._system_tray: Optional[Any] = None
self._setup_ui()
self._setup_menu_bar()
self._setup_tool_bar()
self._setup_status_bar()
self._connect_signals()
logger.info("MainWindow initialized")
def _setup_ui(self) -> None:
"""设置UI布局"""
self.setWindowTitle("P2P 通信应用")
self.setMinimumSize(800, 600)
self.resize(self.config.window_width, self.config.window_height)
# 创建中央部件
central_widget = QWidget()
self.setCentralWidget(central_widget)
# 主布局
main_layout = QHBoxLayout(central_widget)
main_layout.setContentsMargins(0, 0, 0, 0)
main_layout.setSpacing(0)
# 创建分割器
self._splitter = QSplitter(Qt.Orientation.Horizontal)
main_layout.addWidget(self._splitter)
# 左侧面板：联系人列表
self._left_panel = self._create_left_panel()
self._splitter.addWidget(self._left_panel)
# 右侧面板：聊天窗口和功能区
self._right_panel = self._create_right_panel()
self._splitter.addWidget(self._right_panel)
# 设置分割比例
self._splitter.setSizes([250, 750])
self._splitter.setStretchFactor(0, 0)
self._splitter.setStretchFactor(1, 1)
def _create_left_panel(self) -> QWidget:
"""
创建左侧面板（联系人列表）
实现联系人列表面板 (需求 9.1)
"""
panel = QFrame()
panel.setFrameShape(QFrame.Shape.StyledPanel)
panel.setMinimumWidth(200)
panel.setMaximumWidth(400)
layout = QVBoxLayout(panel)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
# 用户信息区域
self._user_info_widget = self._create_user_info_widget()
layout.addWidget(self._user_info_widget)
# 联系人列表占位符（将由ContactListWidget替换）
self._contact_list_placeholder = QLabel("联系人列表")
self._contact_list_placeholder.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._contact_list_placeholder.setStyleSheet("""
QLabel {
background-color: #f5f5f5;
color: #666;
padding: 20px;
}
""")
layout.addWidget(self._contact_list_placeholder, 1)
return panel
def _create_user_info_widget(self) -> QWidget:
"""创建用户信息显示区域"""
widget = QFrame()
widget.setFrameShape(QFrame.Shape.StyledPanel)
widget.setStyleSheet("""
QFrame {
background-color: #4a90d9;
border: none;
border-bottom: 1px solid #3a80c9;
}
""")
layout = QHBoxLayout(widget)
layout.setContentsMargins(10, 10, 10, 10)
# 用户头像占位
self._avatar_label = QLabel("👤")
self._avatar_label.setFixedSize(40, 40)
self._avatar_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._avatar_label.setStyleSheet("""
QLabel {
background-color: white;
border-radius: 20px;
font-size: 20px;
}
""")
layout.addWidget(self._avatar_label)
# 用户名和状态
info_layout = QVBoxLayout()
info_layout.setSpacing(2)
self._username_label = QLabel("未登录")
self._username_label.setStyleSheet("color: white; font-weight: bold; font-size: 14px;")
info_layout.addWidget(self._username_label)
self._status_label = QLabel("离线")
self._status_label.setStyleSheet("color: #ddd; font-size: 12px;")
info_layout.addWidget(self._status_label)
layout.addLayout(info_layout, 1)
return widget
def _create_right_panel(self) -> QWidget:
"""
创建右侧面板（聊天窗口和功能区）
实现聊天窗口面板和功能按钮区域 (需求 9.1)
"""
panel = QFrame()
panel.setFrameShape(QFrame.Shape.StyledPanel)
layout = QVBoxLayout(panel)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
# 聊天窗口占位符（将由ChatWidget替换）
self._chat_placeholder = QLabel("选择联系人开始聊天")
self._chat_placeholder.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._chat_placeholder.setStyleSheet("""
QLabel {
background-color: #fafafa;
color: #999;
font-size: 16px;
}
""")
layout.addWidget(self._chat_placeholder, 1)
# 功能按钮区域
self._function_bar = self._create_function_bar()
layout.addWidget(self._function_bar)
return panel
def _create_function_bar(self) -> QWidget:
"""
创建功能按钮区域
实现功能按钮区域 (需求 9.1)
"""
bar = QFrame()
bar.setFrameShape(QFrame.Shape.StyledPanel)
bar.setFixedHeight(50)
bar.setStyleSheet("""
QFrame {
background-color: #f0f0f0;
border-top: 1px solid #ddd;
}
""")
layout = QHBoxLayout(bar)
layout.setContentsMargins(10, 5, 10, 5)
layout.setSpacing(10)
# 功能按钮将在后续子任务中添加
layout.addStretch()
return bar
def _setup_menu_bar(self) -> None:
"""设置菜单栏"""
menubar = self.menuBar()
# 文件菜单
file_menu = menubar.addMenu("文件(&F)")
self._login_action = QAction("登录(&L)", self)
self._login_action.setShortcut("Ctrl+L")
self._login_action.triggered.connect(self._on_login)
file_menu.addAction(self._login_action)
self._logout_action = QAction("注销(&O)", self)
self._logout_action.setEnabled(False)
self._logout_action.triggered.connect(self._on_logout)
file_menu.addAction(self._logout_action)
file_menu.addSeparator()
self._exit_action = QAction("退出(&X)", self)
self._exit_action.setShortcut("Ctrl+Q")
self._exit_action.triggered.connect(self.close)
file_menu.addAction(self._exit_action)
# 聊天菜单
chat_menu = menubar.addMenu("聊天(&C)")
self._send_file_action = QAction("发送文件(&F)", self)
self._send_file_action.setShortcut("Ctrl+Shift+F")
self._send_file_action.setEnabled(False)
chat_menu.addAction(self._send_file_action)
self._send_image_action = QAction("发送图片(&I)", self)
self._send_image_action.setShortcut("Ctrl+Shift+I")
self._send_image_action.setEnabled(False)
chat_menu.addAction(self._send_image_action)
chat_menu.addSeparator()
self._voice_call_action = QAction("语音通话(&V)", self)
self._voice_call_action.setShortcut("Ctrl+Shift+V")
self._voice_call_action.setEnabled(False)
chat_menu.addAction(self._voice_call_action)
# 视图菜单
view_menu = menubar.addMenu("视图(&V)")
self._show_contacts_action = QAction("显示联系人(&C)", self)
self._show_contacts_action.setCheckable(True)
self._show_contacts_action.setChecked(True)
self._show_contacts_action.triggered.connect(self._toggle_contacts_panel)
view_menu.addAction(self._show_contacts_action)
# 帮助菜单
help_menu = menubar.addMenu("帮助(&H)")
self._about_action = QAction("关于(&A)", self)
self._about_action.triggered.connect(self._show_about)
help_menu.addAction(self._about_action)
def _setup_tool_bar(self) -> None:
"""设置工具栏"""
toolbar = QToolBar("主工具栏")
toolbar.setMovable(False)
toolbar.setIconSize(QSize(24, 24))
self.addToolBar(toolbar)
# 工具栏按钮将在后续添加图标后完善
toolbar.addAction(self._login_action)
toolbar.addSeparator()
toolbar.addAction(self._send_file_action)
toolbar.addAction(self._send_image_action)
toolbar.addAction(self._voice_call_action)
def _setup_status_bar(self) -> None:
"""设置状态栏"""
self._statusbar = QStatusBar()
self.setStatusBar(self._statusbar)
# 连接状态标签
self._connection_status_label = QLabel("未连接")
self._connection_status_label.setStyleSheet("color: #999;")
self._statusbar.addPermanentWidget(self._connection_status_label)
self._statusbar.showMessage("就绪")
def _connect_signals(self) -> None:
"""连接信号和槽"""
self.message_received.connect(self._on_message_received)
self.user_status_changed.connect(self._on_user_status_changed)
self.connection_state_changed.connect(self._on_connection_state_changed)
# ==================== 事件处理 ====================
def closeEvent(self, event: QCloseEvent) -> None:
"""
窗口关闭事件
实现后台运行 (需求 9.5)
WHEN 应用程序最小化 THEN P2P_Client SHALL 保持后台运行并继续接收消息
"""
# 如果有系统托盘，最小化到托盘而不是关闭
if self._system_tray and self._system_tray.is_available():
event.ignore()
self.hide()
self._system_tray.show_message("P2P通信", "应用程序已最小化到系统托盘")
else:
# 确认退出
reply = QMessageBox.question(
self,
"确认退出",
"确定要退出应用程序吗？",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.Yes:
self._cleanup()
event.accept()
else:
event.ignore()
def resizeEvent(self, event) -> None:
"""
窗口大小调整事件
实现自适应调整界面布局 (需求 9.6)
WHEN 用户调整窗口大小 THEN P2P_Client SHALL 自适应调整界面布局
"""
super().resizeEvent(event)
# 布局会自动调整，无需额外处理
# ==================== 槽函数 ====================
def _on_login(self) -> None:
"""处理登录操作"""
from client.ui.login_dialog import LoginDialog
dialog = LoginDialog(self)
if dialog.exec():
user_info = dialog.get_user_info()
if user_info:
self.set_current_user(user_info)
self._statusbar.showMessage(f"已登录: {user_info.username}")
def _on_logout(self) -> None:
"""处理注销操作"""
reply = QMessageBox.question(
self,
"确认注销",
"确定要注销当前账户吗？",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.Yes:
self._current_user = None
self._update_user_display()
self._login_action.setEnabled(True)
self._logout_action.setEnabled(False)
self._statusbar.showMessage("已注销")
def _toggle_contacts_panel(self, checked: bool) -> None:
"""切换联系人面板显示"""
self._left_panel.setVisible(checked)
def _show_about(self) -> None:
"""显示关于对话框"""
QMessageBox.about(
self,
"关于 P2P 通信应用",
"P2P 网络通信应用程序\n\n"
"版本: 0.1.0\n\n"
"支持功能:\n"
"- 文本消息通信\n"
"- 文件传输\n"
"- 图片传输与显示\n"
"- 音视频播放\n"
"- 语音聊天"
)
def _on_message_received(self, message: Message) -> None:
"""处理接收到的消息"""
logger.debug(f"Message received: {message.msg_type.value}")
# 消息处理将在ChatWidget中实现
def _on_user_status_changed(self, user_id: str, status: UserStatus) -> None:
"""处理用户状态变化"""
logger.debug(f"User {user_id} status changed to {status.value}")
# 状态更新将在ContactListWidget中实现
def _on_connection_state_changed(self, state: str) -> None:
"""处理连接状态变化"""
self._connection_status_label.setText(state)
if state == "已连接":
self._connection_status_label.setStyleSheet("color: green;")
elif state == "连接中...":
self._connection_status_label.setStyleSheet("color: orange;")
else:
self._connection_status_label.setStyleSheet("color: red;")
# ==================== 公共方法 ====================
def set_current_user(self, user_info: UserInfo) -> None:
"""
设置当前用户
Args:
user_info: 用户信息
"""
self._current_user = user_info
self._update_user_display()
self._login_action.setEnabled(False)
self._logout_action.setEnabled(True)
logger.info(f"Current user set: {user_info.username}")
def _update_user_display(self) -> None:
"""更新用户显示"""
if self._current_user:
self._username_label.setText(self._current_user.display_name or self._current_user.username)
self._status_label.setText(self._current_user.status.value)
else:
self._username_label.setText("未登录")
self._status_label.setText("离线")
def set_contact_list_widget(self, widget: QWidget) -> None:
"""
设置联系人列表组件
Args:
widget: 联系人列表组件
"""
# 移除占位符
layout = self._left_panel.layout()
layout.removeWidget(self._contact_list_placeholder)
self._contact_list_placeholder.deleteLater()
# 添加实际组件
self._contact_list = widget
layout.addWidget(widget, 1)
def set_chat_widget(self, widget: QWidget) -> None:
"""
设置聊天组件
Args:
widget: 聊天组件
"""
# 移除占位符
layout = self._right_panel.layout()
layout.removeWidget(self._chat_placeholder)
self._chat_placeholder.deleteLater()
# 添加实际组件（在功能栏之前）
self._chat_widget = widget
layout.insertWidget(0, widget, 1)
def set_system_tray(self, tray_manager) -> None:
"""
设置系统托盘管理器
Args:
tray_manager: 系统托盘管理器
"""
self._system_tray = tray_manager
def show_notification(self, title: str, message: str) -> None:
"""
显示通知
实现新消息通知 (需求 9.3)
WHEN 有新消息到达 THEN P2P_Client SHALL 在系统托盘显示通知提醒
Args:
title: 通知标题
message: 通知内容
"""
if self._system_tray:
self._system_tray.show_message(title, message)
def enable_chat_actions(self, enabled: bool = True) -> None:
"""
启用/禁用聊天相关操作
Args:
enabled: 是否启用
"""
self._send_file_action.setEnabled(enabled)
self._send_image_action.setEnabled(enabled)
self._voice_call_action.setEnabled(enabled)
def set_current_chat_peer(self, peer_id: Optional[str]) -> None:
"""
设置当前聊天对象
Args:
peer_id: 对等端ID
"""
self._current_chat_peer = peer_id
self.enable_chat_actions(peer_id is not None)
def _cleanup(self) -> None:
"""清理资源"""
logger.info("MainWindow cleanup")
# 清理将在集成时实现
@property
def current_user(self) -> Optional[UserInfo]:
"""获取当前用户"""
return self._current_user
@property
def current_chat_peer(self) -> Optional[str]:
"""获取当前聊天对象"""
return self._current_chat_peer

335
client/ui/media_player_widget.py
Unescape View File

@ -0,0 +1,335 @@
# P2P Network Communication - Media Player Widget
"""
媒体播放器界面组件
实现音频和视频播放器界面
需求: 6.3, 6.4, 6.5, 6.7
"""
import logging
from typing import Optional
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton,
QSlider, QFrame, QStackedWidget, QSizePolicy
)
from PyQt6.QtCore import Qt, pyqtSignal, QTimer
from client.media_player import PlaybackState
logger = logging.getLogger(__name__)
class MediaPlayerWidget(QWidget):
"""
媒体播放器界面组件
实现音频播放器界面 (需求 6.3)
实现视频播放器界面 (需求 6.4)
实现播放控制按钮 (需求 6.5)
实现全屏模式 (需求 6.7)
"""
# 信号定义
play_requested = pyqtSignal()
pause_requested = pyqtSignal()
stop_requested = pyqtSignal()
seek_requested = pyqtSignal(float) # position in seconds
volume_changed = pyqtSignal(float) # volume 0.0-1.0
fullscreen_requested = pyqtSignal(bool)
def __init__(self, parent=None):
super().__init__(parent)
self._is_video = False
self._duration = 0.0
self._is_seeking = False
self._setup_ui()
self._connect_signals()
# 更新定时器
self._update_timer = QTimer()
self._update_timer.setInterval(100)
self._update_timer.timeout.connect(self._on_update_timer)
logger.info("MediaPlayerWidget initialized")
def _setup_ui(self) -> None:
"""设置UI"""
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
# 视频显示区域
self._video_frame = QFrame()
self._video_frame.setStyleSheet("background-color: black;")
self._video_frame.setMinimumHeight(200)
self._video_frame.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
video_layout = QVBoxLayout(self._video_frame)
video_layout.setContentsMargins(0, 0, 0, 0)
self._video_label = QLabel("无媒体")
self._video_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._video_label.setStyleSheet("color: #666; font-size: 16px;")
video_layout.addWidget(self._video_label)
layout.addWidget(self._video_frame, 1)
# 控制栏
self._control_bar = self._create_control_bar()
layout.addWidget(self._control_bar)
def _create_control_bar(self) -> QWidget:
"""创建控制栏"""
bar = QFrame()
bar.setFixedHeight(80)
bar.setStyleSheet("""
QFrame {
background-color: #2d2d2d;
}
""")
layout = QVBoxLayout(bar)
layout.setContentsMargins(10, 5, 10, 5)
layout.setSpacing(5)
# 进度条
progress_layout = QHBoxLayout()
progress_layout.setSpacing(10)
self._current_time_label = QLabel("00:00")
self._current_time_label.setStyleSheet("color: white; font-size: 12px;")
self._current_time_label.setFixedWidth(50)
progress_layout.addWidget(self._current_time_label)
self._progress_slider = QSlider(Qt.Orientation.Horizontal)
self._progress_slider.setRange(0, 1000)
self._progress_slider.setValue(0)
self._progress_slider.setStyleSheet("""
QSlider::groove:horizontal {
border: none;
height: 4px;
background: #555;
border-radius: 2px;
}
QSlider::handle:horizontal {
background: white;
width: 12px;
height: 12px;
margin: -4px 0;
border-radius: 6px;
}
QSlider::sub-page:horizontal {
background: #4a90d9;
border-radius: 2px;
}
""")
progress_layout.addWidget(self._progress_slider, 1)
self._total_time_label = QLabel("00:00")
self._total_time_label.setStyleSheet("color: white; font-size: 12px;")
self._total_time_label.setFixedWidth(50)
progress_layout.addWidget(self._total_time_label)
layout.addLayout(progress_layout)
# 控制按钮
button_layout = QHBoxLayout()
button_layout.setSpacing(10)
button_style = """
QPushButton {
background-color: transparent;
color: white;
border: none;
font-size: 18px;
padding: 5px 10px;
}
QPushButton:hover {
background-color: #444;
border-radius: 5px;
}
"""
self._play_btn = QPushButton("▶")
self._play_btn.setStyleSheet(button_style)
self._play_btn.setFixedSize(40, 40)
self._play_btn.clicked.connect(self._on_play_clicked)
button_layout.addWidget(self._play_btn)
self._stop_btn = QPushButton("⏹")
self._stop_btn.setStyleSheet(button_style)
self._stop_btn.setFixedSize(40, 40)
self._stop_btn.clicked.connect(self._on_stop_clicked)
button_layout.addWidget(self._stop_btn)
button_layout.addStretch()
# 音量控制
self._volume_btn = QPushButton("🔊")
self._volume_btn.setStyleSheet(button_style)
self._volume_btn.setFixedSize(40, 40)
button_layout.addWidget(self._volume_btn)
self._volume_slider = QSlider(Qt.Orientation.Horizontal)
self._volume_slider.setRange(0, 100)
self._volume_slider.setValue(100)
self._volume_slider.setFixedWidth(100)
self._volume_slider.setStyleSheet("""
QSlider::groove:horizontal {
border: none;
height: 4px;
background: #555;
border-radius: 2px;
}
QSlider::handle:horizontal {
background: white;
width: 10px;
height: 10px;
margin: -3px 0;
border-radius: 5px;
}
QSlider::sub-page:horizontal {
background: #4a90d9;
border-radius: 2px;
}
""")
button_layout.addWidget(self._volume_slider)
# 全屏按钮
self._fullscreen_btn = QPushButton("⛶")
self._fullscreen_btn.setStyleSheet(button_style)
self._fullscreen_btn.setFixedSize(40, 40)
self._fullscreen_btn.clicked.connect(self._on_fullscreen_clicked)
button_layout.addWidget(self._fullscreen_btn)
layout.addLayout(button_layout)
return bar
def _connect_signals(self) -> None:
"""连接信号"""
self._progress_slider.sliderPressed.connect(self._on_slider_pressed)
self._progress_slider.sliderReleased.connect(self._on_slider_released)
self._volume_slider.valueChanged.connect(self._on_volume_changed)
def _format_time(self, seconds: float) -> str:
"""格式化时间"""
minutes = int(seconds // 60)
secs = int(seconds % 60)
return f"{minutes:02d}:{secs:02d}"
def _on_play_clicked(self) -> None:
"""处理播放/暂停按钮点击"""
if self._play_btn.text() == "▶":
self.play_requested.emit()
else:
self.pause_requested.emit()
def _on_stop_clicked(self) -> None:
"""处理停止按钮点击"""
self.stop_requested.emit()
def _on_fullscreen_clicked(self) -> None:
"""处理全屏按钮点击"""
self.fullscreen_requested.emit(True)
def _on_slider_pressed(self) -> None:
"""进度条按下"""
self._is_seeking = True
def _on_slider_released(self) -> None:
"""进度条释放"""
self._is_seeking = False
if self._duration > 0:
position = (self._progress_slider.value() / 1000) * self._duration
self.seek_requested.emit(position)
def _on_volume_changed(self, value: int) -> None:
"""音量变化"""
volume = value / 100.0
self.volume_changed.emit(volume)
if value == 0:
self._volume_btn.setText("🔇")
elif value < 50:
self._volume_btn.setText("🔉")
else:
self._volume_btn.setText("🔊")
def _on_update_timer(self) -> None:
"""更新定时器回调"""
# 由外部调用update_position更新
pass
# ==================== 公共方法 ====================
def set_media_info(self, file_name: str, duration: float, is_video: bool = False) -> None:
"""
设置媒体信息
Args:
file_name: 文件名
duration: 时长（秒）
is_video: 是否是视频
"""
self._duration = duration
self._is_video = is_video
self._video_label.setText(file_name)
self._total_time_label.setText(self._format_time(duration))
self._progress_slider.setValue(0)
self._current_time_label.setText("00:00")
def update_position(self, position: float) -> None:
"""
更新播放位置
Args:
position: 当前位置（秒）
"""
if not self._is_seeking and self._duration > 0:
slider_value = int((position / self._duration) * 1000)
self._progress_slider.setValue(slider_value)
self._current_time_label.setText(self._format_time(position))
def update_state(self, state: PlaybackState) -> None:
"""
更新播放状态
Args:
state: 播放状态
"""
if state == PlaybackState.PLAYING:
self._play_btn.setText("⏸")
self._update_timer.start()
elif state == PlaybackState.PAUSED:
self._play_btn.setText("▶")
self._update_timer.stop()
elif state == PlaybackState.STOPPED:
self._play_btn.setText("▶")
self._update_timer.stop()
self._progress_slider.setValue(0)
self._current_time_label.setText("00:00")
def set_volume(self, volume: float) -> None:
"""
设置音量
Args:
volume: 音量 (0.0-1.0)
"""
self._volume_slider.setValue(int(volume * 100))
def clear(self) -> None:
"""清空播放器"""
self._duration = 0.0
self._video_label.setText("无媒体")
self._total_time_label.setText("00:00")
self._current_time_label.setText("00:00")
self._progress_slider.setValue(0)
self._play_btn.setText("▶")
self._update_timer.stop()

179
client/ui/system_tray.py
Unescape View File

@ -0,0 +1,179 @@
# P2P Network Communication - System Tray Manager
"""
系统托盘管理器
实现系统托盘图标、新消息通知和后台运行
需求: 9.3, 9.5
"""
import logging
from typing import Optional, Callable
from PyQt6.QtWidgets import QSystemTrayIcon, QMenu, QApplication
from PyQt6.QtGui import QIcon, QAction
from PyQt6.QtCore import pyqtSignal, QObject
logger = logging.getLogger(__name__)
class SystemTrayManager(QObject):
"""
系统托盘管理器
实现系统托盘图标 (需求 9.3)
实现新消息通知 (需求 9.3)
实现后台运行 (需求 9.5)
"""
# 信号定义
show_window_requested = pyqtSignal()
quit_requested = pyqtSignal()
def __init__(self, parent=None):
super().__init__(parent)
self._tray_icon: Optional[QSystemTrayIcon] = None
self._menu: Optional[QMenu] = None
self._unread_count: int = 0
self._setup_tray()
logger.info("SystemTrayManager initialized")
def _setup_tray(self) -> None:
"""设置系统托盘"""
if not QSystemTrayIcon.isSystemTrayAvailable():
logger.warning("System tray is not available")
return
self._tray_icon = QSystemTrayIcon(self.parent())
# 设置默认图标（使用应用程序图标或创建简单图标）
app = QApplication.instance()
if app and not app.windowIcon().isNull():
self._tray_icon.setIcon(app.windowIcon())
else:
# 创建一个简单的默认图标
from PyQt6.QtGui import QPixmap, QPainter, QColor
pixmap = QPixmap(32, 32)
pixmap.fill(QColor("#4a90d9"))
painter = QPainter(pixmap)
painter.setPen(QColor("white"))
painter.drawText(pixmap.rect(), 0x0084, "P2P") # AlignCenter
painter.end()
self._tray_icon.setIcon(QIcon(pixmap))
self._tray_icon.setToolTip("P2P 通信应用")
# 创建菜单
self._menu = QMenu()
show_action = QAction("显示主窗口", self._menu)
show_action.triggered.connect(self._on_show_window)
self._menu.addAction(show_action)
self._menu.addSeparator()
self._status_action = QAction("状态: 在线", self._menu)
self._status_action.setEnabled(False)
self._menu.addAction(self._status_action)
self._menu.addSeparator()
quit_action = QAction("退出", self._menu)
quit_action.triggered.connect(self._on_quit)
self._menu.addAction(quit_action)
self._tray_icon.setContextMenu(self._menu)
# 连接信号
self._tray_icon.activated.connect(self._on_tray_activated)
def _on_show_window(self) -> None:
"""显示主窗口"""
self.show_window_requested.emit()
def _on_quit(self) -> None:
"""退出应用"""
self.quit_requested.emit()
def _on_tray_activated(self, reason: QSystemTrayIcon.ActivationReason) -> None:
"""托盘图标激活"""
if reason == QSystemTrayIcon.ActivationReason.DoubleClick:
self.show_window_requested.emit()
# ==================== 公共方法 ====================
def is_available(self) -> bool:
"""
检查系统托盘是否可用
Returns:
是否可用
"""
return self._tray_icon is not None
def show(self) -> None:
"""显示托盘图标"""
if self._tray_icon:
self._tray_icon.show()
def hide(self) -> None:
"""隐藏托盘图标"""
if self._tray_icon:
self._tray_icon.hide()
def show_message(self, title: str, message: str,
icon: QSystemTrayIcon.MessageIcon = QSystemTrayIcon.MessageIcon.Information,
duration: int = 3000) -> None:
"""
显示通知消息
实现新消息通知 (需求 9.3)
WHEN 有新消息到达 THEN P2P_Client SHALL 在系统托盘显示通知提醒
Args:
title: 通知标题
message: 通知内容
icon: 图标类型
duration: 显示时长（毫秒）
"""
if self._tray_icon:
self._tray_icon.showMessage(title, message, icon, duration)
logger.debug(f"Tray notification: {title} - {message}")
def set_status(self, status: str) -> None:
"""
设置状态显示
Args:
status: 状态文本
"""
if self._status_action:
self._status_action.setText(f"状态: {status}")
if self._tray_icon:
self._tray_icon.setToolTip(f"P2P 通信应用 - {status}")
def set_unread_count(self, count: int) -> None:
"""
设置未读消息数
Args:
count: 未读消息数
"""
self._unread_count = count
if self._tray_icon:
if count > 0:
self._tray_icon.setToolTip(f"P2P 通信应用 - {count} 条未读消息")
else:
self._tray_icon.setToolTip("P2P 通信应用")
def increment_unread(self) -> None:
"""增加未读消息数"""
self.set_unread_count(self._unread_count + 1)
def clear_unread(self) -> None:
"""清除未读消息数"""
self.set_unread_count(0)

393
client/ui/voice_call_widget.py
Unescape View File

@ -0,0 +1,393 @@
# P2P Network Communication - Voice Call Widget
"""
语音通话界面组件
实现来电提示、通话中界面和通话状态显示
需求: 7.2, 7.8
"""
import logging
from typing import Optional
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton,
QFrame, QDialog
)
from PyQt6.QtCore import Qt, pyqtSignal, QTimer
from client.voice_chat import CallState, NetworkQuality
logger = logging.getLogger(__name__)
class IncomingCallDialog(QDialog):
"""
来电提示对话框
实现来电提示界面 (需求 7.2)
WHEN 收到语音通话邀请 THEN P2P_Client SHALL 显示来电提示并提供接听或拒绝选项
"""
accepted = pyqtSignal()
rejected = pyqtSignal()
def __init__(self, caller_name: str, parent=None):
super().__init__(parent)
self.caller_name = caller_name
self._setup_ui()
# 响铃动画定时器
self._ring_timer = QTimer()
self._ring_timer.setInterval(500)
self._ring_timer.timeout.connect(self._animate_ring)
self._ring_state = False
self._ring_timer.start()
def _setup_ui(self) -> None:
"""设置UI"""
self.setWindowTitle("来电")
self.setFixedSize(300, 200)
self.setModal(True)
self.setWindowFlags(self.windowFlags() | Qt.WindowType.WindowStaysOnTopHint)
layout = QVBoxLayout(self)
layout.setSpacing(20)
layout.setContentsMargins(30, 30, 30, 30)
# 来电图标
self._icon_label = QLabel("📞")
self._icon_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._icon_label.setStyleSheet("font-size: 48px;")
layout.addWidget(self._icon_label)
# 来电信息
info_label = QLabel(f"{self.caller_name}\n正在呼叫您...")
info_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
info_label.setStyleSheet("font-size: 16px;")
layout.addWidget(info_label)
# 按钮
button_layout = QHBoxLayout()
button_layout.setSpacing(20)
self._reject_btn = QPushButton("拒绝")
self._reject_btn.setFixedSize(80, 40)
self._reject_btn.setStyleSheet("""
QPushButton {
background-color: #f44336;
color: white;
border: none;
border-radius: 20px;
font-size: 14px;
}
QPushButton:hover {
background-color: #d32f2f;
}
""")
self._reject_btn.clicked.connect(self._on_reject)
button_layout.addWidget(self._reject_btn)
self._accept_btn = QPushButton("接听")
self._accept_btn.setFixedSize(80, 40)
self._accept_btn.setStyleSheet("""
QPushButton {
background-color: #4caf50;
color: white;
border: none;
border-radius: 20px;
font-size: 14px;
}
QPushButton:hover {
background-color: #388e3c;
}
""")
self._accept_btn.clicked.connect(self._on_accept)
button_layout.addWidget(self._accept_btn)
layout.addLayout(button_layout)
def _animate_ring(self) -> None:
"""响铃动画"""
self._ring_state = not self._ring_state
self._icon_label.setText("📞" if self._ring_state else "📱")
def _on_accept(self) -> None:
"""接听"""
self._ring_timer.stop()
self.accepted.emit()
self.accept()
def _on_reject(self) -> None:
"""拒绝"""
self._ring_timer.stop()
self.rejected.emit()
self.reject()
def closeEvent(self, event) -> None:
"""关闭事件"""
self._ring_timer.stop()
self.rejected.emit()
super().closeEvent(event)
class VoiceCallWidget(QWidget):
"""
语音通话界面组件
实现通话中界面 (需求 7.8)
实现通话状态显示 (需求 7.8)
WHEN 语音通话进行中 THEN P2P_Client SHALL 显示通话时长和网络质量指示
"""
# 信号定义
mute_toggled = pyqtSignal(bool)
end_call_requested = pyqtSignal()
def __init__(self, parent=None):
super().__init__(parent)
self._peer_name: str = ""
self._is_muted: bool = False
self._call_duration: int = 0
self._setup_ui()
# 通话计时器
self._duration_timer = QTimer()
self._duration_timer.setInterval(1000)
self._duration_timer.timeout.connect(self._update_duration)
logger.info("VoiceCallWidget initialized")
def _setup_ui(self) -> None:
"""设置UI"""
self.setStyleSheet("background-color: #2d2d2d;")
layout = QVBoxLayout(self)
layout.setSpacing(20)
layout.setContentsMargins(30, 30, 30, 30)
layout.addStretch()
# 通话对象头像
avatar_label = QLabel("👤")
avatar_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
avatar_label.setStyleSheet("""
font-size: 64px;
background-color: #4a90d9;
border-radius: 50px;
padding: 20px;
""")
avatar_label.setFixedSize(100, 100)
avatar_container = QHBoxLayout()
avatar_container.addStretch()
avatar_container.addWidget(avatar_label)
avatar_container.addStretch()
layout.addLayout(avatar_container)
# 通话对象名称
self._peer_name_label = QLabel("未知用户")
self._peer_name_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._peer_name_label.setStyleSheet("color: white; font-size: 24px; font-weight: bold;")
layout.addWidget(self._peer_name_label)
# 通话状态
self._status_label = QLabel("正在连接...")
self._status_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._status_label.setStyleSheet("color: #aaa; font-size: 14px;")
layout.addWidget(self._status_label)
# 通话时长
self._duration_label = QLabel("00:00")
self._duration_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._duration_label.setStyleSheet("color: white; font-size: 32px;")
layout.addWidget(self._duration_label)
# 网络质量
self._quality_label = QLabel("网络质量: --")
self._quality_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._quality_label.setStyleSheet("color: #aaa; font-size: 12px;")
layout.addWidget(self._quality_label)
layout.addStretch()
# 控制按钮
button_layout = QHBoxLayout()
button_layout.setSpacing(30)
button_layout.addStretch()
# 静音按钮
self._mute_btn = QPushButton("🎤")
self._mute_btn.setFixedSize(60, 60)
self._mute_btn.setStyleSheet("""
QPushButton {
background-color: #555;
color: white;
border: none;
border-radius: 30px;
font-size: 24px;
}
QPushButton:hover {
background-color: #666;
}
QPushButton:checked {
background-color: #f44336;
}
""")
self._mute_btn.setCheckable(True)
self._mute_btn.clicked.connect(self._on_mute_clicked)
button_layout.addWidget(self._mute_btn)
# 挂断按钮
self._end_btn = QPushButton("📞")
self._end_btn.setFixedSize(70, 70)
self._end_btn.setStyleSheet("""
QPushButton {
background-color: #f44336;
color: white;
border: none;
border-radius: 35px;
font-size: 28px;
}
QPushButton:hover {
background-color: #d32f2f;
}
""")
self._end_btn.clicked.connect(self._on_end_clicked)
button_layout.addWidget(self._end_btn)
# 扬声器按钮
self._speaker_btn = QPushButton("🔊")
self._speaker_btn.setFixedSize(60, 60)
self._speaker_btn.setStyleSheet("""
QPushButton {
background-color: #555;
color: white;
border: none;
border-radius: 30px;
font-size: 24px;
}
QPushButton:hover {
background-color: #666;
}
""")
button_layout.addWidget(self._speaker_btn)
button_layout.addStretch()
layout.addLayout(button_layout)
layout.addSpacing(30)
def _format_duration(self, seconds: int) -> str:
"""格式化时长"""
minutes = seconds // 60
secs = seconds % 60
return f"{minutes:02d}:{secs:02d}"
def _update_duration(self) -> None:
"""更新通话时长"""
self._call_duration += 1
self._duration_label.setText(self._format_duration(self._call_duration))
def _on_mute_clicked(self) -> None:
"""处理静音按钮点击"""
self._is_muted = self._mute_btn.isChecked()
self._mute_btn.setText("🔇" if self._is_muted else "🎤")
self.mute_toggled.emit(self._is_muted)
def _on_end_clicked(self) -> None:
"""处理挂断按钮点击"""
self.end_call_requested.emit()
# ==================== 公共方法 ====================
def start_call(self, peer_name: str) -> None:
"""
开始通话
Args:
peer_name: 通话对象名称
"""
self._peer_name = peer_name
self._peer_name_label.setText(peer_name)
self._status_label.setText("通话中")
self._call_duration = 0
self._duration_label.setText("00:00")
self._duration_timer.start()
logger.info(f"Call started with {peer_name}")
def end_call(self) -> None:
"""结束通话"""
self._duration_timer.stop()
self._status_label.setText("通话已结束")
logger.info("Call ended")
def update_state(self, state: CallState) -> None:
"""
更新通话状态
Args:
state: 通话状态
"""
state_texts = {
CallState.IDLE: "空闲",
CallState.CALLING: "正在呼叫...",
CallState.RINGING: "对方响铃中...",
CallState.CONNECTED: "通话中",
CallState.ENDING: "正在结束...",
}
self._status_label.setText(state_texts.get(state, "未知"))
if state == CallState.CONNECTED:
self._duration_timer.start()
elif state in (CallState.IDLE, CallState.ENDING):
self._duration_timer.stop()
def update_network_quality(self, quality: NetworkQuality) -> None:
"""
更新网络质量显示
实现网络质量指示 (需求 7.8)
WHEN 语音通话进行中 THEN P2P_Client SHALL 显示通话时长和网络质量指示
Args:
quality: 网络质量
"""
quality_texts = {
NetworkQuality.EXCELLENT: ("优秀", "#4caf50"),
NetworkQuality.GOOD: ("良好", "#8bc34a"),
NetworkQuality.FAIR: ("一般", "#ffeb3b"),
NetworkQuality.POOR: ("较差", "#ff9800"),
NetworkQuality.BAD: ("很差", "#f44336"),
}
text, color = quality_texts.get(quality, ("未知", "#999"))
self._quality_label.setText(f"网络质量: {text}")
self._quality_label.setStyleSheet(f"color: {color}; font-size: 12px;")
def get_call_duration(self) -> int:
"""
获取通话时长
Returns:
通话时长（秒）
"""
return self._call_duration
def reset(self) -> None:
"""重置界面"""
self._duration_timer.stop()
self._peer_name = ""
self._is_muted = False
self._call_duration = 0
self._peer_name_label.setText("未知用户")
self._status_label.setText("正在连接...")
self._duration_label.setText("00:00")
self._quality_label.setText("网络质量: --")
self._mute_btn.setChecked(False)
self._mute_btn.setText("🎤")

38
shared/__init__.py
Unescape View File

@ -27,6 +27,26 @@ from shared.message_handler import (
MessageRoutingError,
)
from shared.security import (
SecurityError,
EncryptionError,
DecryptionError,
KeyManagementError,
CertificateError,
EncryptedData,
AESCipher,
TLSManager,
MessageEncryptor,
FileEncryptor,
KeyManager,
LocalDataEncryptor,
create_message_encryptor,
create_file_encryptor,
create_local_data_encryptor,
encrypt_message,
decrypt_message,
)
__all__ = [
"MessageType",
"UserStatus",
@ -44,4 +64,22 @@ __all__ = [
"MessageValidationError",
"MessageSerializationError",
"MessageRoutingError",
# Security
"SecurityError",
"EncryptionError",
"DecryptionError",
"KeyManagementError",
"CertificateError",
"EncryptedData",
"AESCipher",
"TLSManager",
"MessageEncryptor",
"FileEncryptor",
"KeyManager",
"LocalDataEncryptor",
"create_message_encryptor",
"create_file_encryptor",
"create_local_data_encryptor",
"encrypt_message",
"decrypt_message",
]

1037
shared/security.py
View File

File diff suppressed because it is too large Load Diff

436
tests/test_security.py
Unescape View File

@ -0,0 +1,436 @@
# P2P Network Communication - Security Module Tests
"""
安全模块测试
测试传输加密、本地数据加密和密钥管理功能
需求: 10.1, 10.2, 10.3
"""
import os
import tempfile
import pytest
from pathlib import Path
from shared.security import (
AESCipher,
EncryptedData,
MessageEncryptor,
FileEncryptor,
KeyManager,
LocalDataEncryptor,
TLSManager,
EncryptionError,
DecryptionError,
create_message_encryptor,
create_file_encryptor,
create_local_data_encryptor,
encrypt_message,
decrypt_message,
)
class TestAESCipher:
"""AES-256-GCM 加密器测试"""
def test_generate_key(self):
"""测试密钥生成"""
key = AESCipher.generate_key()
assert len(key) == 32 # 256 bits
# 确保每次生成的密钥不同
key2 = AESCipher.generate_key()
assert key != key2
def test_generate_iv(self):
"""测试IV生成"""
iv = AESCipher.generate_iv()
assert len(iv) == 12 # 96 bits
def test_encrypt_decrypt(self):
"""测试加密解密往返"""
cipher = AESCipher()
plaintext = b"Hello, World! This is a test message."
encrypted = cipher.encrypt(plaintext)
decrypted = cipher.decrypt(encrypted)
assert decrypted == plaintext
def test_encrypt_decrypt_empty(self):
"""测试空数据加密解密"""
cipher = AESCipher()
plaintext = b""
encrypted = cipher.encrypt(plaintext)
decrypted = cipher.decrypt(encrypted)
assert decrypted == plaintext
def test_encrypt_decrypt_large_data(self):
"""测试大数据加密解密"""
cipher = AESCipher()
plaintext = os.urandom(1024 * 1024) # 1MB
encrypted = cipher.encrypt(plaintext)
decrypted = cipher.decrypt(encrypted)
assert decrypted == plaintext
def test_encrypt_decrypt_unicode(self):
"""测试Unicode数据加密解密"""
cipher = AESCipher()
plaintext = "你好，世界！这是一条测试消息。🎉".encode('utf-8')
encrypted = cipher.encrypt(plaintext)
decrypted = cipher.decrypt(encrypted)
assert decrypted == plaintext
def test_different_keys_produce_different_ciphertext(self):
"""测试不同密钥产生不同密文"""
cipher1 = AESCipher()
cipher2 = AESCipher()
plaintext = b"Same message"
encrypted1 = cipher1.encrypt(plaintext)
encrypted2 = cipher2.encrypt(plaintext)
assert encrypted1.ciphertext != encrypted2.ciphertext
def test_wrong_key_fails_decryption(self):
"""测试错误密钥解密失败"""
cipher1 = AESCipher()
cipher2 = AESCipher()
plaintext = b"Secret message"
encrypted = cipher1.encrypt(plaintext)
with pytest.raises(DecryptionError):
cipher2.decrypt(encrypted)
def test_derive_key_from_password(self):
"""测试从密码派生密钥"""
password = "my_secure_password"
key1, salt1 = AESCipher.derive_key_from_password(password)
assert len(key1) == 32
assert len(salt1) == 16
# 相同密码和盐应产生相同密钥
key2, _ = AESCipher.derive_key_from_password(password, salt1)
assert key1 == key2
# 不同盐应产生不同密钥
key3, salt3 = AESCipher.derive_key_from_password(password)
assert key1 != key3
def test_encrypt_with_password(self):
"""测试使用密码加密"""
cipher = AESCipher()
password = "test_password"
plaintext = b"Secret data"
encrypted = cipher.encrypt_with_password(plaintext, password)
assert encrypted.salt # 应包含盐值
decrypted = cipher.decrypt_with_password(encrypted, password)
assert decrypted == plaintext
def test_wrong_password_fails(self):
"""测试错误密码解密失败"""
cipher = AESCipher()
plaintext = b"Secret data"
encrypted = cipher.encrypt_with_password(plaintext, "correct_password")
with pytest.raises(DecryptionError):
cipher.decrypt_with_password(encrypted, "wrong_password")
class TestEncryptedData:
"""加密数据结构测试"""
def test_to_bytes_from_bytes(self):
"""测试序列化和反序列化"""
original = EncryptedData(
ciphertext=b"encrypted_content",
iv=b"123456789012",
tag=b"1234567890123456",
salt=b"salt_value_here!"
)
serialized = original.to_bytes()
restored = EncryptedData.from_bytes(serialized)
assert restored.ciphertext == original.ciphertext
assert restored.iv == original.iv
assert restored.tag == original.tag
assert restored.salt == original.salt
def test_to_base64_from_base64(self):
"""测试Base64编码和解码"""
original = EncryptedData(
ciphertext=b"encrypted_content",
iv=b"123456789012",
tag=b"1234567890123456"
)
base64_str = original.to_base64()
restored = EncryptedData.from_base64(base64_str)
assert restored.ciphertext == original.ciphertext
assert restored.iv == original.iv
assert restored.tag == original.tag
class TestMessageEncryptor:
"""消息加密器测试"""
def test_encrypt_decrypt_message(self):
"""测试消息加密解密"""
encryptor = MessageEncryptor()
message_data = b'{"type": "text", "content": "Hello!"}'
encrypted = encryptor.encrypt_message(message_data)
decrypted = encryptor.decrypt_message(encrypted)
assert decrypted == message_data
def test_set_key(self):
"""测试设置密钥"""
key = AESCipher.generate_key()
encryptor = MessageEncryptor()
encryptor.set_key(key)
assert encryptor.key == key
def test_shared_key_encryption(self):
"""测试共享密钥加密"""
key = AESCipher.generate_key()
encryptor1 = MessageEncryptor(AESCipher(key))
encryptor2 = MessageEncryptor(AESCipher(key))
message = b"Shared secret message"
encrypted = encryptor1.encrypt_message(message)
decrypted = encryptor2.decrypt_message(encrypted)
assert decrypted == message
class TestFileEncryptor:
"""文件加密器测试"""
def test_encrypt_decrypt_file(self):
"""测试文件加密解密"""
encryptor = FileEncryptor()
with tempfile.TemporaryDirectory() as tmpdir:
# 创建测试文件
input_file = Path(tmpdir) / "test.txt"
encrypted_file = Path(tmpdir) / "test.enc"
output_file = Path(tmpdir) / "test_decrypted.txt"
original_content = b"This is test file content.\n" * 100
input_file.write_bytes(original_content)
# 加密
encryptor.encrypt_file(str(input_file), str(encrypted_file))
assert encrypted_file.exists()
# 解密
encryptor.decrypt_file(str(encrypted_file), str(output_file))
assert output_file.exists()
# 验证内容
decrypted_content = output_file.read_bytes()
assert decrypted_content == original_content
def test_encrypt_decrypt_chunk(self):
"""测试文件块加密解密"""
encryptor = FileEncryptor()
chunk_data = os.urandom(64 * 1024) # 64KB
encrypted = encryptor.encrypt_chunk(chunk_data)
decrypted = encryptor.decrypt_chunk(encrypted)
assert decrypted == chunk_data
class TestKeyManager:
"""密钥管理器测试"""
def test_generate_key_pair(self):
"""测试生成密钥对"""
with tempfile.TemporaryDirectory() as tmpdir:
manager = KeyManager(tmpdir)
private_key, public_key = manager.generate_key_pair()
assert b"PRIVATE KEY" in private_key
assert b"PUBLIC KEY" in public_key
def test_save_load_key(self):
"""测试保存和加载密钥"""
with tempfile.TemporaryDirectory() as tmpdir:
manager = KeyManager(tmpdir)
key_data = AESCipher.generate_key()
# 保存
assert manager.save_key("test_key", key_data)
# 加载
loaded = manager.load_key("test_key")
assert loaded == key_data
def test_save_load_key_with_password(self):
"""测试使用密码保存和加载密钥"""
with tempfile.TemporaryDirectory() as tmpdir:
manager = KeyManager(tmpdir)
key_data = AESCipher.generate_key()
password = "secure_password"
# 保存
assert manager.save_key("protected_key", key_data, password)
# 加载
loaded = manager.load_key("protected_key", password)
assert loaded == key_data
def test_delete_key(self):
"""测试删除密钥"""
with tempfile.TemporaryDirectory() as tmpdir:
manager = KeyManager(tmpdir)
key_data = AESCipher.generate_key()
manager.save_key("to_delete", key_data)
assert manager.delete_key("to_delete")
# 验证已删除
assert manager.load_key("to_delete") is None
def test_session_key_management(self):
"""测试会话密钥管理"""
with tempfile.TemporaryDirectory() as tmpdir:
manager = KeyManager(tmpdir)
# 获取或创建会话密钥
key1 = manager.get_or_create_session_key("session1")
assert len(key1) == 32
# 再次获取应返回相同密钥
key2 = manager.get_or_create_session_key("session1")
assert key1 == key2
# 不同会话应有不同密钥
key3 = manager.get_or_create_session_key("session2")
assert key1 != key3
# 清除会话密钥
manager.clear_session_key("session1")
key4 = manager.get_or_create_session_key("session1")
assert key1 != key4 # 应该是新密钥
class TestLocalDataEncryptor:
"""本地数据加密器测试"""
def test_encrypt_decrypt_string(self):
"""测试字符串加密解密"""
encryptor = LocalDataEncryptor(password="test_password")
original = "Hello, World!"
encrypted = encryptor.encrypt_data(original)
decrypted = encryptor.decrypt_data(encrypted)
assert decrypted.decode('utf-8') == original
def test_encrypt_decrypt_dict(self):
"""测试字典加密解密"""
encryptor = LocalDataEncryptor(password="test_password")
original = {"name": "Test", "value": 123, "nested": {"key": "value"}}
encrypted = encryptor.encrypt_data(original)
decrypted = encryptor.decrypt_data(encrypted, as_json=True)
assert decrypted == original
def test_encrypt_decrypt_chat_history(self):
"""测试聊天记录加密解密"""
encryptor = LocalDataEncryptor(password="chat_password")
messages = [
{"sender": "user1", "content": "Hello", "timestamp": 1234567890},
{"sender": "user2", "content": "Hi there!", "timestamp": 1234567891},
]
encrypted = encryptor.encrypt_chat_history(messages)
decrypted = encryptor.decrypt_chat_history(encrypted)
assert decrypted == messages
def test_save_load_encrypted_file(self):
"""测试保存和加载加密文件"""
with tempfile.TemporaryDirectory() as tmpdir:
encryptor = LocalDataEncryptor(password="file_password")
file_path = Path(tmpdir) / "encrypted_data.enc"
original_data = {"key": "value", "list": [1, 2, 3]}
# 保存
assert encryptor.save_encrypted_file(original_data, str(file_path))
assert file_path.exists()
# 加载
loaded = encryptor.load_encrypted_file(str(file_path), as_json=True)
assert loaded == original_data
class TestConvenienceFunctions:
"""便捷函数测试"""
def test_create_message_encryptor(self):
"""测试创建消息加密器"""
encryptor = create_message_encryptor()
assert encryptor is not None
assert len(encryptor.key) == 32
def test_create_file_encryptor(self):
"""测试创建文件加密器"""
encryptor = create_file_encryptor()
assert encryptor is not None
assert len(encryptor.key) == 32
def test_create_local_data_encryptor(self):
"""测试创建本地数据加密器"""
encryptor = create_local_data_encryptor("password")
assert encryptor is not None
def test_encrypt_decrypt_message_functions(self):
"""测试快速加密解密函数"""
key = AESCipher.generate_key()
message = b"Quick encryption test"
encrypted = encrypt_message(message, key)
decrypted = decrypt_message(encrypted, key)
assert decrypted == message
class TestTLSManager:
"""TLS管理器测试"""
def test_create_client_ssl_context_no_verify(self):
"""测试创建客户端SSL上下文（不验证）"""
manager = TLSManager()
context = manager.create_client_ssl_context(verify=False)
assert context is not None
assert context.verify_mode.name == "CERT_NONE"
def test_create_client_ssl_context_with_verify(self):
"""测试创建客户端SSL上下文（验证）"""
manager = TLSManager()
context = manager.create_client_ssl_context(verify=True)
assert context is not None
assert context.verify_mode.name == "CERT_REQUIRED"
Write Preview
Loading…
Cancel
Save