端到端新架构骨架:src/new 包入口 + BaseView 接口契约(dispatch_execute 沿父链上溯 run_single_step)

This commit is contained in:
DXC
2026-06-16 17:53:01 +08:00
parent 15547bddfb
commit 1e0e7d1973
3 changed files with 101 additions and 0 deletions

19
src/new/__init__.py Normal file
View File

@ -0,0 +1,19 @@
# -*- coding: utf-8 -*-
"""
WQ_GUI 端到端模块化新架构
============================
四层一对一分离:
* ``core/`` 基础通讯接口BaseView 等契约)
* ``views/`` 前端皮囊 —— 仅 PyQt UI零业务
* ``services/`` 后端大脑 —— 纯函数计算,零 PyQt
* ``main_router.py`` 路由与调度壳,连接前端按钮与后端服务
调用链:
Step1View._on_run_clicked
→ BaseView.dispatch_execute(step_id, config)
→ MainRouter.run_single_step(step_id, config)
→ TaskWorker 线程调用 services.step1_service.execute_step1(config)
→ 返回 dict 结果 → 日志区
"""

5
src/new/core/__init__.py Normal file
View File

@ -0,0 +1,5 @@
# -*- coding: utf-8 -*-
"""core —— 基础通讯接口包"""
from .base_view import BaseView
__all__ = ["BaseView"]

77
src/new/core/base_view.py Normal file
View File

@ -0,0 +1,77 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
BaseView —— 模块化新页面的标准接口基类
设计目标:
1. 继承自 QWidget保留 PyQt 控件的全部能力。
2. 剥离对全局 main_window 的直接依赖——所有跨层通讯通过
``dispatch_execute`` 沿 ``self.parent()`` 父链向上查找具备
``run_single_step`` 方法的祖先容器完成。
3. 声明一组可重写的生命周期空方法init_ui / get_config /
set_config / update_work_directory / update_from_config
子类按需覆盖,避免基类强制绑定具体 UI 形态。
"""
from PyQt5.QtWidgets import QWidget
class BaseView(QWidget):
"""所有新页面的标准基类"""
def __init__(self, parent=None):
super().__init__(parent)
# 全局推送状态的本地缓存——主窗口通过 update_from_config 写入
self.work_dir = None
self.pipeline = None
# 子类应重写 init_ui 以构建自身 UI
self.init_ui()
# ------------------------------------------------------------------
# 生命周期空方法(子类按需重写)
# ------------------------------------------------------------------
def init_ui(self):
"""构建 UI——子类必须重写"""
pass
def get_config(self) -> dict:
"""返回当前步骤的配置字典——子类重写"""
return {}
def set_config(self, config: dict):
"""根据 config 恢复 UI 状态——子类重写"""
pass
def update_work_directory(self, work_dir: str):
"""主窗口推送工作目录变更——子类按需重写以联动自动填充"""
self.work_dir = work_dir
def update_from_config(self, work_dir: str, pipeline=None):
"""主窗口推送全局状态——子类可重写以联动 pipeline 状态"""
self.work_dir = work_dir
self.pipeline = pipeline
# ------------------------------------------------------------------
# 核心通讯:沿父链向上查找 run_single_step 容器
# ------------------------------------------------------------------
def dispatch_execute(self, step_id: str, config: dict):
"""
向主窗口请求执行某个步骤。
实现要点:
- 沿 ``self.parent()`` 一路向上爬,直到找到具备
``run_single_step`` 方法的祖先容器。
- 找不到则抛出 ``RuntimeError``,便于在迁移早期定位未挂载的面板。
"""
ancestor = self.parent()
while ancestor is not None and not hasattr(ancestor, "run_single_step"):
ancestor = ancestor.parent()
if ancestor is not None and hasattr(ancestor, "run_single_step"):
ancestor.run_single_step(step_id, config)
else:
raise RuntimeError(
"[BaseView] dispatch_execute 失败:未在父链中找到 "
"具备 run_single_step 方法的主窗口容器 "
f"step_id={step_id}"
)