|
|
# 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._connection_mode: str = "relay" # 默认中转模式
|
|
|
self._update_display()
|
|
|
|
|
|
def _update_display(self) -> None:
|
|
|
"""更新显示"""
|
|
|
status_icons = {
|
|
|
UserStatus.ONLINE: "🟢",
|
|
|
UserStatus.OFFLINE: "⚫",
|
|
|
UserStatus.BUSY: "🔴",
|
|
|
UserStatus.AWAY: "🟡",
|
|
|
}
|
|
|
|
|
|
# 连接模式图标
|
|
|
mode_icon = "🔗" if self._connection_mode == "p2p" else ""
|
|
|
|
|
|
icon = status_icons.get(self.user_info.status, "⚫")
|
|
|
display_name = self.user_info.display_name or self.user_info.username
|
|
|
|
|
|
if mode_icon:
|
|
|
self.setText(f"{icon} {display_name} {mode_icon}")
|
|
|
else:
|
|
|
self.setText(f"{icon} {display_name}")
|
|
|
|
|
|
# 设置提示信息
|
|
|
mode_text = "P2P直连" if self._connection_mode == "p2p" else "服务器中转"
|
|
|
self.setToolTip(
|
|
|
f"用户名: {self.user_info.username}\n"
|
|
|
f"状态: {self.user_info.status.value}\n"
|
|
|
f"连接模式: {mode_text}\n"
|
|
|
f"ID: {self.user_info.user_id}"
|
|
|
)
|
|
|
|
|
|
def update_status(self, status: UserStatus) -> None:
|
|
|
"""更新用户状态"""
|
|
|
self.user_info.status = status
|
|
|
self._update_display()
|
|
|
|
|
|
def set_connection_mode(self, mode: str) -> None:
|
|
|
"""设置连接模式 ('p2p' 或 'relay')"""
|
|
|
self._connection_mode = mode
|
|
|
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() # 请求刷新联系人列表
|
|
|
lan_discovery_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)
|
|
|
|
|
|
self._lan_btn = QPushButton("📡")
|
|
|
self._lan_btn.setFixedSize(30, 30)
|
|
|
self._lan_btn.setToolTip("发现局域网用户")
|
|
|
self._lan_btn.clicked.connect(self._on_lan_discovery_clicked)
|
|
|
search_layout.addWidget(self._lan_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 _on_lan_discovery_clicked(self) -> None:
|
|
|
"""处理局域网发现按钮点击"""
|
|
|
self.lan_discovery_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])
|
|
|
|
|
|
def update_connection_mode(self, user_id: str, mode: str) -> None:
|
|
|
"""
|
|
|
更新联系人的连接模式
|
|
|
|
|
|
Args:
|
|
|
user_id: 用户ID
|
|
|
mode: 连接模式 ('p2p' 或 'relay')
|
|
|
"""
|
|
|
if user_id in self._contacts:
|
|
|
self._contacts[user_id].set_connection_mode(mode)
|
|
|
logger.debug(f"Contact {user_id} connection mode updated to {mode}")
|
