# -*- coding: utf-8 -*- """可复用的 ODS 任务选择组件:按业务域分组显示,支持全选/反选。""" from typing import Dict, List, Optional, Set from PySide6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QCheckBox, QPushButton, QScrollArea, QFrame, QLabel, QSizePolicy ) from PySide6.QtCore import Signal, Qt from ..models.task_registry import ( TaskRegistry, TaskDefinition, BusinessDomain, DOMAIN_LABELS, task_registry, get_fact_ods_task_codes, get_dimension_ods_task_codes ) class TaskSelectorWidget(QWidget): """ODS 任务选择组件:按业务域分组显示""" # 选择变化信号 selection_changed = Signal(list) # 选中的任务编码列表 def __init__( self, parent: Optional[QWidget] = None, show_dimensions: bool = True, show_facts: bool = True, default_select_facts: bool = True, default_select_dimensions: bool = False, compact: bool = False, max_height: int = 0, ): """ 初始化任务选择器 Args: parent: 父组件 show_dimensions: 是否显示维度类任务 show_facts: 是否显示事实类任务 default_select_facts: 默认选中事实类任务 default_select_dimensions: 默认选中维度类任务 compact: 紧凑模式(更小的间距) max_height: 最大高度(0 表示不限制) """ super().__init__(parent) self.show_dimensions = show_dimensions self.show_facts = show_facts self.default_select_facts = default_select_facts self.default_select_dimensions = default_select_dimensions self.compact = compact self.max_height = max_height # 任务复选框映射:code -> QCheckBox self._checkboxes: Dict[str, QCheckBox] = {} # 业务域分组框映射:domain -> QGroupBox self._domain_groups: Dict[BusinessDomain, QGroupBox] = {} self._init_ui() self._apply_default_selection() def _init_ui(self): """初始化界面""" layout = QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) spacing = 4 if self.compact else 8 layout.setSpacing(spacing) # 顶部工具栏 toolbar = QHBoxLayout() toolbar.setSpacing(8) self.select_all_btn = QPushButton("全选") self.select_all_btn.setProperty("secondary", True) self.select_all_btn.setFixedWidth(60) self.select_all_btn.clicked.connect(self._select_all) toolbar.addWidget(self.select_all_btn) self.deselect_all_btn = QPushButton("全不选") self.deselect_all_btn.setProperty("secondary", True) self.deselect_all_btn.setFixedWidth(60) self.deselect_all_btn.clicked.connect(self._deselect_all) toolbar.addWidget(self.deselect_all_btn) self.select_facts_btn = QPushButton("选事实表") self.select_facts_btn.setProperty("secondary", True) self.select_facts_btn.setFixedWidth(70) self.select_facts_btn.setToolTip("选中所有事实类任务(需要时间窗口的任务)") self.select_facts_btn.clicked.connect(self._select_facts_only) toolbar.addWidget(self.select_facts_btn) toolbar.addStretch() self.selected_count_label = QLabel("已选: 0") self.selected_count_label.setProperty("subheading", True) toolbar.addWidget(self.selected_count_label) layout.addLayout(toolbar) # 滚动区域 scroll_area = QScrollArea() scroll_area.setWidgetResizable(True) scroll_area.setFrameShape(QFrame.NoFrame) if self.max_height > 0: scroll_area.setMaximumHeight(self.max_height) # 内容容器 content_widget = QWidget() content_layout = QVBoxLayout(content_widget) content_layout.setContentsMargins(0, 0, 0, 0) content_layout.setSpacing(spacing) # 按业务域分组创建复选框 grouped_tasks = task_registry.get_ods_tasks_grouped() # 定义业务域显示顺序 domain_order = [ BusinessDomain.MEMBER, BusinessDomain.SETTLEMENT, BusinessDomain.ASSISTANT, BusinessDomain.GOODS, BusinessDomain.TABLE, BusinessDomain.PROMOTION, BusinessDomain.INVENTORY, ] for domain in domain_order: if domain not in grouped_tasks: continue tasks = grouped_tasks[domain] # 过滤任务 filtered_tasks = [] for task in tasks: if task.is_dimension and not self.show_dimensions: continue if not task.is_dimension and not self.show_facts: continue filtered_tasks.append(task) if not filtered_tasks: continue # 创建业务域分组 group_box = self._create_domain_group(domain, filtered_tasks) self._domain_groups[domain] = group_box content_layout.addWidget(group_box) content_layout.addStretch() scroll_area.setWidget(content_widget) layout.addWidget(scroll_area, 1) def _create_domain_group(self, domain: BusinessDomain, tasks: List[TaskDefinition]) -> QGroupBox: """创建业务域分组框""" group_box = QGroupBox(DOMAIN_LABELS.get(domain, str(domain.value))) group_layout = QVBoxLayout(group_box) group_layout.setContentsMargins(8, 4, 8, 4) group_layout.setSpacing(2) for task in tasks: checkbox = QCheckBox(f"{task.name}") checkbox.setToolTip(f"{task.code}: {task.description}") checkbox.setProperty("task_code", task.code) checkbox.setProperty("is_dimension", task.is_dimension) checkbox.stateChanged.connect(self._on_selection_changed) self._checkboxes[task.code] = checkbox group_layout.addWidget(checkbox) return group_box def _apply_default_selection(self): """应用默认选择""" for code, checkbox in self._checkboxes.items(): is_dimension = checkbox.property("is_dimension") if is_dimension: checkbox.setChecked(self.default_select_dimensions) else: checkbox.setChecked(self.default_select_facts) self._update_count_label() def _on_selection_changed(self): """选择变化时""" self._update_count_label() self.selection_changed.emit(self.get_selected_codes()) def _update_count_label(self): """更新选中计数标签""" count = len(self.get_selected_codes()) total = len(self._checkboxes) self.selected_count_label.setText(f"已选: {count}/{total}") def _select_all(self): """全选""" for checkbox in self._checkboxes.values(): checkbox.blockSignals(True) checkbox.setChecked(True) checkbox.blockSignals(False) self._on_selection_changed() def _deselect_all(self): """全不选""" for checkbox in self._checkboxes.values(): checkbox.blockSignals(True) checkbox.setChecked(False) checkbox.blockSignals(False) self._on_selection_changed() def _select_facts_only(self): """只选事实表任务""" for code, checkbox in self._checkboxes.items(): checkbox.blockSignals(True) is_dimension = checkbox.property("is_dimension") checkbox.setChecked(not is_dimension) checkbox.blockSignals(False) self._on_selection_changed() def get_selected_codes(self) -> List[str]: """获取选中的任务编码列表""" selected = [] for code, checkbox in self._checkboxes.items(): if checkbox.isChecked(): selected.append(code) return selected def set_selected_codes(self, codes: List[str]): """设置选中的任务编码""" codes_set = set(codes) for code, checkbox in self._checkboxes.items(): checkbox.blockSignals(True) checkbox.setChecked(code in codes_set) checkbox.blockSignals(False) self._on_selection_changed() def get_all_codes(self) -> List[str]: """获取所有任务编码""" return list(self._checkboxes.keys()) def is_any_selected(self) -> bool: """是否有任何任务被选中""" return len(self.get_selected_codes()) > 0 class CompactTaskSelector(QWidget): """紧凑型任务选择器:单行显示业务域,点击展开选择""" selection_changed = Signal(list) def __init__( self, parent: Optional[QWidget] = None, show_dimensions: bool = True, show_facts: bool = True, default_select_facts: bool = True, default_select_dimensions: bool = False, ): super().__init__(parent) self.show_dimensions = show_dimensions self.show_facts = show_facts self.default_select_facts = default_select_facts self.default_select_dimensions = default_select_dimensions # 业务域复选框 self._domain_checkboxes: Dict[BusinessDomain, QCheckBox] = {} # 业务域下的任务编码 self._domain_tasks: Dict[BusinessDomain, List[str]] = {} self._init_ui() self._apply_default_selection() def _init_ui(self): """初始化界面""" layout = QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(4) # 工具栏 toolbar = QHBoxLayout() toolbar.setSpacing(8) self.select_all_btn = QPushButton("全选") self.select_all_btn.setProperty("secondary", True) self.select_all_btn.setFixedWidth(50) self.select_all_btn.clicked.connect(self._select_all) toolbar.addWidget(self.select_all_btn) self.deselect_all_btn = QPushButton("清空") self.deselect_all_btn.setProperty("secondary", True) self.deselect_all_btn.setFixedWidth(50) self.deselect_all_btn.clicked.connect(self._deselect_all) toolbar.addWidget(self.deselect_all_btn) toolbar.addStretch() self.count_label = QLabel("已选: 0") self.count_label.setProperty("subheading", True) toolbar.addWidget(self.count_label) layout.addLayout(toolbar) # 业务域复选框(横向排列) domains_layout = QHBoxLayout() domains_layout.setSpacing(12) grouped_tasks = task_registry.get_ods_tasks_grouped() domain_order = [ BusinessDomain.MEMBER, BusinessDomain.SETTLEMENT, BusinessDomain.ASSISTANT, BusinessDomain.GOODS, BusinessDomain.TABLE, BusinessDomain.PROMOTION, BusinessDomain.INVENTORY, ] for domain in domain_order: if domain not in grouped_tasks: continue tasks = grouped_tasks[domain] # 过滤任务 task_codes = [] for task in tasks: if task.is_dimension and not self.show_dimensions: continue if not task.is_dimension and not self.show_facts: continue task_codes.append(task.code) if not task_codes: continue self._domain_tasks[domain] = task_codes checkbox = QCheckBox(DOMAIN_LABELS.get(domain, str(domain.value))) checkbox.setToolTip(f"包含: {', '.join(task_codes)}") checkbox.stateChanged.connect(self._on_selection_changed) self._domain_checkboxes[domain] = checkbox domains_layout.addWidget(checkbox) domains_layout.addStretch() layout.addLayout(domains_layout) def _apply_default_selection(self): """应用默认选择""" # 默认选中所有业务域 for domain, checkbox in self._domain_checkboxes.items(): checkbox.setChecked(True) self._update_count_label() def _on_selection_changed(self): """选择变化时""" self._update_count_label() self.selection_changed.emit(self.get_selected_codes()) def _update_count_label(self): """更新计数标签""" count = len(self.get_selected_codes()) self.count_label.setText(f"已选: {count} 个任务") def _select_all(self): """全选所有业务域""" for checkbox in self._domain_checkboxes.values(): checkbox.blockSignals(True) checkbox.setChecked(True) checkbox.blockSignals(False) self._on_selection_changed() def _deselect_all(self): """取消全选""" for checkbox in self._domain_checkboxes.values(): checkbox.blockSignals(True) checkbox.setChecked(False) checkbox.blockSignals(False) self._on_selection_changed() def get_selected_codes(self) -> List[str]: """获取选中的任务编码""" selected = [] for domain, checkbox in self._domain_checkboxes.items(): if checkbox.isChecked(): selected.extend(self._domain_tasks.get(domain, [])) return selected def set_selected_domains(self, domains: List[BusinessDomain]): """设置选中的业务域""" domains_set = set(domains) for domain, checkbox in self._domain_checkboxes.items(): checkbox.blockSignals(True) checkbox.setChecked(domain in domains_set) checkbox.blockSignals(False) self._on_selection_changed() def is_any_selected(self) -> bool: """是否有任何任务被选中""" return len(self.get_selected_codes()) > 0