Files
feiqiu-ETL/etl_billiards/gui/widgets/task_selector.py
2026-02-01 22:04:15 +08:00

399 lines
14 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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