# certflow/views/correction_queue_view.py
"""校正队列界面
#30 P1 校正队列（替代 VBA InputBox/Stop 非阻塞校正）：
- 黄标页：列出 flag=True 的销售计划记录，人工校正口径/压力/标准号 →
upsert 进字典表自学习（下次导入自动命中）+ 清除黄标；
- 隔离页：列出 sale_plans_quarantine 未释放记录，补全要货单号后释放重建为 SalePlan，
或直接删除（确认无效）。
从主窗口导航栏打开，模态对话框，内部通过 CorrectionQueueController 完成读写。
"""
from __future__ import annotations
import contextlib
from typing import Any, override
from PySide6.QtCore import Qt
from PySide6.QtWidgets import (
QDialog,
QFormLayout,
QLabel,
QLineEdit,
QMenu,
QMessageBox,
QTableWidget,
QTableWidgetItem,
QTabWidget,
QVBoxLayout,
QWidget,
)
from certflow.actions import dispatch, get_spec
from certflow.config.settings import get_ui_config
from certflow.controllers.correction_queue_controller import CorrectionQueueController
from certflow.utils.logger import logger
from certflow.widgets import ActionToolbar, ConfirmDialog, DialogButtonBar
from certflow.widgets.enhanced_table import EnhancedTable
# 校正队列隔离表工具栏 / 右键菜单：唯一来源为 ui.yaml 的 correction_queue 段
# （Phase 2.2 配置驱动，经 certflow.actions.registry dispatch 驱动）；本视图不再
# 保留硬编码回退，落实 UI 域单入口（ui.yaml 缺失时返回空列表，由 UI 显示空白）。
# 黄标表列定义已迁移到 ``CorrectionDataSource.get_columns``（见
# ``views/bases/datasources/correction.py``）；本视图黄标 tab 由
# ``CorrectionFlagQueryView``（BaseQueryView + CorrectionDataSource）承载。
# 隔离表列定义
_QUARANTINE_COLUMNS = [
("ID", "id"),
("计划单号", "plan_no"),
("销售订单号", "sales_order_no"),
("生产令号", "production_order_no"),
("合同号", "contract_no"),
("订货单位", "customer"),
("型号", "product_model"),
("规格", "product_spec"),
("隔离原因", "quarantine_reason"),
("隔离时间", "quarantined_at"),
]
[文档]
class CorrectionQueueView(QDialog):
"""校正队列对话框（黄标校正 + 隔离释放）"""
def __init__(
self,
db_manager: Any = None,
session: Any = None,
parent: QWidget | None = None,
) -> None:
super().__init__(parent)
self.setWindowTitle("校正队列（黄标校正 + 隔离释放）")
self.setMinimumSize(1100, 640)
self.resize(1280, 720)
self._session = session
self._db_manager = db_manager
if db_manager is not None:
self._controller = CorrectionQueueController(db_manager=db_manager)
else:
self._controller = CorrectionQueueController(db_session=session)
self._init_ui()
self._shown_once = False
# ============================================================
# UI 构建
# ============================================================
def _init_ui(self) -> None:
layout = QVBoxLayout(self)
layout.setContentsMargins(12, 12, 12, 12)
layout.setSpacing(10)
# 统计标签
self._stats_label = QLabel("加载中…")
self._stats_label.setStyleSheet("font-size: 12px; color: #666;")
layout.addWidget(self._stats_label)
# 标签页
self._tabs = QTabWidget()
self._tabs.addTab(self._build_flag_tab(), "🟡 黄标校正")
self._tabs.addTab(self._build_quarantine_tab(), "🔒 隔离释放")
layout.addWidget(self._tabs)
# 底部提示
hint = QLabel(
"黄标校正：人工复核口径/压力/标准号，保存即 upsert 进字典表（下次导入自动命中，等价 VBA 自学习回写，非阻塞）；"
"隔离释放：补全要货单号后重建为销售计划，或删除确认无效行。"
)
hint.setStyleSheet("font-size: 11px; color: #666;")
hint.setWordWrap(True)
layout.addWidget(hint)
def _build_flag_tab(self) -> QWidget:
# 黄标校正表迁移到 BaseQueryView + CorrectionDataSource（任务 2.1 首域）：
# 复用统一表格/筛选/排序/分页，校正动作以信号回传宿主处理。
from certflow.views.correction_flag_query_view import CorrectionFlagQueryView
self._flag_query = CorrectionFlagQueryView(
session=self._session, db_manager=self._db_manager, parent=self
)
self._flag_query.correct_requested.connect(self.on_correct_flagged)
self._flag_query.clear_requested.connect(self.on_clear_flag)
self._flag_query.row_activated.connect(self.on_correct_flagged)
return self._flag_query
def _build_quarantine_tab(self) -> QWidget:
tab = QWidget()
v = QVBoxLayout(tab)
toolbar = ActionToolbar()
for act in self._load_correction_queue_actions():
action_name = act.get("action", "")
spec = get_spec(action_name)
if spec is None or getattr(self, spec.method, None) is None:
logger.warning("隔离表动作 {} 未注册或未找到处理方法，已跳过", action_name)
continue
toolbar.add_button(
act["label"],
lambda _=False, a=act: dispatch(self, a["action"], a),
danger=act.get("danger", False),
)
toolbar.add_stretch()
toolbar.add_button("🔄 刷新", self.refresh)
v.addWidget(toolbar)
self._q_table = EnhancedTable()
headers = [col[0] for col in _QUARANTINE_COLUMNS]
self._q_table.setColumnCount(len(headers))
self._q_table.setHorizontalHeaderLabels(headers)
for idx, (_, field) in enumerate(_QUARANTINE_COLUMNS):
self._q_table.set_header_field(idx, field)
self._q_table.setSelectionMode(self._q_table.selectionMode().__class__.SingleSelection)
self._q_table.horizontalHeader().setStretchLastSection(True)
# 隔离表是全量本地数据，启用客户端排序
self._q_table.setSortingEnabled(True)
self._q_table.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self._q_table.customContextMenuRequested.connect(self._on_quarantine_context_menu)
self._q_table.doubleClicked.connect(self.on_release_quarantine)
v.addWidget(self._q_table)
return tab
def _load_correction_queue_actions(self) -> list[dict]:
"""隔离表工具栏动作（配置驱动，唯一来源为 ui.yaml correction_queue.actions）。
UIConfig 不可用时返回空列表（不再回退硬编码默认，落实 UI 域单入口）。
"""
cfg = get_ui_config()
if cfg is False or cfg is None:
return []
return cfg.get_correction_queue_actions() or []
def _on_quarantine_context_menu(self, pos: Any) -> None:
"""隔离表右键菜单（配置驱动，见 ui.yaml correction_queue.context_menu）。"""
cfg = self._load_correction_queue_context_menu()
if not cfg:
return
menu = QMenu(self)
for act in cfg:
action_name = act.get("action", "")
if get_spec(action_name) is None:
logger.warning("隔离表右键动作 {} 未注册，已跳过", action_name)
continue
qact = menu.addAction(act["label"])
qact.setData(act)
action = menu.exec(self._q_table.viewport().mapToGlobal(pos))
if action is None:
return
item = action.data()
dispatch(self, item.get("action", ""), item)
def _load_correction_queue_context_menu(self) -> list[dict]:
"""隔离表右键菜单（配置驱动，唯一来源为 ui.yaml correction_queue.context_menu）。
UIConfig 不可用时返回空列表（不再回退硬编码默认，落实 UI 域单入口）。
"""
cfg = get_ui_config()
if cfg is False or cfg is None:
return []
return cfg.get_correction_queue_context_menu() or []
# ============================================================
# 数据加载
# ============================================================
[文档]
@override
def showEvent(self, event: Any) -> None:
"""首次显示时加载数据（避免构造期在无事件循环的测试环境中阻塞）。"""
super().showEvent(event)
if not self._shown_once:
self._shown_once = True
self.refresh()
[文档]
def refresh(self) -> None:
"""重新加载黄标表（BaseQueryView 查询）+ 隔离表 + 统计。"""
try:
if self._flag_query.data_source is None:
from certflow.controllers.query_controller import QueryController
from certflow.views.bases.datasources.correction import CorrectionDataSource
ds = CorrectionDataSource(QueryController(self._session, self._db_manager))
self._flag_query.set_data_source(ds)
else:
self._flag_query.refresh()
quarantined = self._controller.list_quarantined()
stats = self._controller.quarantine_stats()
except Exception as e:
logger.error(f"加载校正队列失败: {e}")
QMessageBox.critical(self, "错误", f"加载失败: {e}")
return
self._fill_table(self._q_table, quarantined, _QUARANTINE_COLUMNS)
self._stats_label.setText(
f"黄标待校正：{self._flag_query.total_count} 条 ｜ 隔离待处理：{stats['pending']} 条"
f"（已释放 {stats['resolved']} / 共 {stats['total']}）"
)
@staticmethod
def _fill_table(table: QTableWidget, rows: list[Any], columns: list[tuple[str, str]]) -> None:
table.setRowCount(len(rows))
for r, obj in enumerate(rows):
for c, (_, field) in enumerate(columns):
value = getattr(obj, field, "")
if value is None:
value = ""
item = QTableWidgetItem(str(value))
item.setData(Qt.ItemDataRole.UserRole, getattr(obj, "id", None))
table.setItem(r, c, item)
table.resizeColumnsToContents()
# ============================================================
# 黄标校正
# ============================================================
[文档]
def on_correct_flagged(self) -> None:
"""校正选中黄标行：弹编辑框 → 写字段 + upsert 字典 + 清 flag。"""
sp_id = self._flag_query.selected_sale_plan_id()
if sp_id is None:
QMessageBox.information(self, "提示", "请先选择一行黄标记录")
return
sp = self._controller.get_sale_plan(sp_id)
if sp is None:
return
dlg = FlagCorrectDialog(sp, parent=self)
if dlg.exec() != QDialog.DialogCode.Accepted:
return
try:
_sp, remaining = self._controller.correct_sale_plan(sp_id, dlg.get_data(), learn=True)
self.refresh()
if remaining:
QMessageBox.information(
self,
"已补录，黄标保留",
f"该记录仍有 {len(remaining)} 处待补录："
f"{'、'.join(remaining)}。\n补齐后黄标将自动取消。",
)
except Exception as e:
QMessageBox.warning(self, "校正失败", str(e))
[文档]
def on_clear_flag(self) -> None:
"""仅清除黄标（不写字典，标记已复核）。"""
sp_id = self._flag_query.selected_sale_plan_id()
if sp_id is None:
QMessageBox.information(self, "提示", "请先选择一行黄标记录")
return
try:
self._controller.correct_sale_plan(sp_id, {}, learn=False, recheck=False)
self.refresh()
except Exception as e:
QMessageBox.warning(self, "操作失败", str(e))
# ============================================================
# 隔离释放 / 删除
# ============================================================
[文档]
def on_release_quarantine(self) -> None:
"""释放选中隔离行：补全要货单号 → 重建 SalePlan。"""
q_id = self._selected_id(self._q_table)
if q_id is None:
QMessageBox.information(self, "提示", "请先选择一行隔离记录")
return
q = self._controller.get_quarantine(q_id)
if q is None:
return
dlg = QuarantineReleaseDialog(q, parent=self)
if dlg.exec() != QDialog.DialogCode.Accepted:
return
try:
sp = self._controller.release_quarantine(q_id, dlg.get_data())
if sp is None:
QMessageBox.warning(self, "提示", "该记录已释放或不存在")
return
QMessageBox.information(self, "成功", f"已释放并重建销售计划 #{sp.id}")
self.refresh()
except Exception as e:
QMessageBox.warning(self, "释放失败", str(e))
[文档]
def on_delete_quarantine(self) -> None:
"""删除选中隔离记录（确认无效）。"""
q_id = self._selected_id(self._q_table)
if q_id is None:
QMessageBox.information(self, "提示", "请先选择一行隔离记录")
return
confirmed = ConfirmDialog.danger(
self,
"确认删除",
"确定删除该隔离记录？此操作不可撤销（不会重建销售计划）。",
)
if not confirmed:
return
try:
if self._controller.delete_quarantine(q_id):
self.refresh()
except Exception as e:
QMessageBox.warning(self, "删除失败", str(e))
[文档]
def on_edit_and_release_quarantine(self) -> None:
"""右键菜单「补全后释放」：打开隔离释放编辑框（等价于双击/释放）。"""
self.on_release_quarantine()
# ============================================================
# 工具
# ============================================================
@staticmethod
def _selected_id(table: QTableWidget) -> int | None:
row = table.currentRow()
if row < 0:
return None
item = table.item(row, 0)
if item is None:
return None
return item.data(Qt.ItemDataRole.UserRole)
[文档]
@override
def closeEvent(self, event: Any) -> None:
with contextlib.suppress(Exception): # pragma: no cover - 防御性
self._controller.close()
super().closeEvent(event)
[文档]
class FlagCorrectDialog(QDialog):
"""黄标校正编辑子对话框"""
def __init__(self, sale_plan: Any, parent: QWidget | None = None) -> None:
super().__init__(parent)
self.setWindowTitle("校正黄标记录（校正后 upsert 字典自学习）")
self.setMinimumWidth(460)
self._edits: dict[str, QLineEdit] = {}
self._build(sale_plan)
def _build(self, sp: Any) -> None:
layout = QVBoxLayout(self)
form = QFormLayout()
form.setSpacing(8)
def add_ref(label: str, value: str = "") -> None:
"""只读参考字段（型号/合同口径不可改，保护重导入唯一键）。"""
ref = QLineEdit()
ref.setText(str(value or ""))
ref.setReadOnly(True)
ref.setStyleSheet("color: #888; background: #f5f5f5;")
form.addRow(label, ref)
def add_line(field: str, label: str, value: str = "") -> None:
edit = QLineEdit()
edit.setText(str(value or ""))
self._edits[field] = edit
form.addRow(label, edit)
# 参考区（只读）：型号 + 合同口径供补录对照，均不可改
add_ref("型号(参考·不可改)", getattr(sp, "product_model", ""))
add_ref("合同规格(参考·不可改)", getattr(sp, "product_spec", ""))
# 补录区（可写）：合同口径空白时把推得的口径写入合格证字段，不动合同口径
add_line("cert_product_spec", "合格证规格(口径串)", getattr(sp, "cert_product_spec", ""))
add_line("spec_norm", "规范化口径(数字)", getattr(sp, "spec_norm", ""))
add_line("pressure_value", "公称压力", getattr(sp, "pressure_value", ""))
add_line("test_standard", "试压标准号", getattr(sp, "test_standard", ""))
layout.addLayout(form)
tip = QLabel(
"型号 / 合同规格为参考（不可改，保护重导入唯一键）；\n"
"合同口径为空时，从型号等推得的口径串填「合格证规格」、数字填「规范化口径」，\n"
"仅写合格证字段与规范化列，不改动空白的合同口径；\n"
"校正压力/标准号：填「公称压力/试压标准」→ 学习到型号参数字典；\n"
"保存后自动核对：口径与公称压力均补齐才取消黄标，未齐则保留待续补。"
)
tip.setStyleSheet("font-size: 11px; color: #666;")
layout.addWidget(tip)
buttons = DialogButtonBar(confirm_text="校正并学习")
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
layout.addWidget(buttons)
[文档]
def get_data(self) -> dict[str, str]:
return {f: e.text().strip() for f, e in self._edits.items()}
[文档]
class QuarantineReleaseDialog(QDialog):
"""隔离释放编辑子对话框（补全要货单号）"""
def __init__(self, quarantine: Any, parent: QWidget | None = None) -> None:
super().__init__(parent)
self.setWindowTitle("释放隔离记录（补全要货单号后重建销售计划）")
self.setMinimumWidth(460)
self._edits: dict[str, QLineEdit] = {}
self._build(quarantine)
def _build(self, q: Any) -> None:
layout = QVBoxLayout(self)
form = QFormLayout()
form.setSpacing(8)
def add_line(field: str, label: str, value: str = "") -> None:
edit = QLineEdit()
edit.setText(str(value or ""))
self._edits[field] = edit
form.addRow(label, edit)
add_line("plan_no", "计划单号", getattr(q, "plan_no", ""))
add_line("sales_order_no", "销售订单号", getattr(q, "sales_order_no", ""))
add_line("production_order_no", "生产令号", getattr(q, "production_order_no", ""))
add_line("contract_no", "合同号", getattr(q, "contract_no", ""))
add_line("product_model", "型号", getattr(q, "product_model", ""))
add_line("product_spec", "规格", getattr(q, "product_spec", ""))
layout.addLayout(form)
tip = QLabel(
"补全要货单号（销售订单号/生产令号/计划单号 任一非空）后释放，"
"将用原始快照 + 此处校正重建为销售计划记录。"
)
tip.setStyleSheet("font-size: 11px; color: #666;")
tip.setWordWrap(True)
layout.addWidget(tip)
buttons = DialogButtonBar(confirm_text="释放并导入")
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
layout.addWidget(buttons)
[文档]
def get_data(self) -> dict[str, str]:
return {f: e.text().strip() for f, e in self._edits.items()}