quanttide_work/
task.rs

1//! 任务聚合：工作流的一次执行实例。
2//!
3//! 指令是跑哪条工作流（`workflow_name`，**按名字**引用）与从哪开工（`start`）；
4//! 状态是流水（只增不改）、闸门项、产物落点；另带这次执行的运行上下文。
5//! 任务是运行数据，不是产物——程序只维护它，不往产物里写字。
6//!
7//! 不可变：`recorded` / `with_gates` 都返回新的任务，改动由调用方落盘。
8
9use std::collections::BTreeMap;
10
11use crate::workflow::Workflow;
12use crate::workflow::text_of;
13use serde_yaml::{Mapping, Value as Yaml};
14
15/// 流水里的一条：什么时候、哪一步、一句话、过没过。
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct JournalEvent {
18    pub at: String,
19    pub step: String,
20    pub detail: String,
21    pub ok: bool,
22}
23
24impl JournalEvent {
25    pub fn of(value: &Yaml) -> JournalEvent {
26        JournalEvent {
27            at: text_of(value, "at"),
28            step: text_of(value, "step"),
29            detail: text_of(value, "detail"),
30            ok: value.get("ok").and_then(|v| v.as_bool()).unwrap_or(false),
31        }
32    }
33
34    pub fn to_yaml(&self) -> Yaml {
35        let mut map = Mapping::new();
36        for (key, value) in [
37            ("at", &self.at),
38            ("step", &self.step),
39            ("detail", &self.detail),
40        ] {
41            map.insert(
42                Yaml::String(key.to_string()),
43                Yaml::String(value.to_string()),
44            );
45        }
46        map.insert(Yaml::String("ok".into()), Yaml::Bool(self.ok));
47        Yaml::Mapping(map)
48    }
49}
50
51/// 这次执行自带的运行上下文：工作区根、数据仓、工作流目录。
52///
53/// 三个字段怎么读、怎么写由各自的包定（工具箱只管托着它们）。
54#[derive(Debug, Clone, Default, PartialEq, Eq)]
55pub struct RunContext {
56    pub root: String,
57    pub data: String,
58    pub workflows: String,
59}
60
61impl RunContext {
62    pub fn of(value: &Yaml) -> RunContext {
63        RunContext {
64            root: text_of(value, "root"),
65            data: text_of(value, "data"),
66            workflows: text_of(value, "workflows"),
67        }
68    }
69
70    pub fn to_yaml(&self) -> Yaml {
71        let mut map = Mapping::new();
72        for (key, value) in [
73            ("root", &self.root),
74            ("data", &self.data),
75            ("workflows", &self.workflows),
76        ] {
77            map.insert(
78                Yaml::String(key.to_string()),
79                Yaml::String(value.to_string()),
80            );
81        }
82        Yaml::Mapping(map)
83    }
84}
85
86/// 任务聚合。
87#[derive(Debug, Clone, PartialEq, Eq)]
88pub struct Task {
89    pub name: String,
90    pub workflow_name: String,
91    pub start: String,
92    pub context: RunContext,
93    pub journal: Vec<JournalEvent>,
94    /// 等人拍板的事项。
95    pub gates: Vec<String>,
96    /// 这次执行往哪写产物（声明写成什么就是什么，相对工作区根）。
97    pub products: BTreeMap<String, String>,
98}
99
100impl Task {
101    /// 从任务文件里的字段读出；`name` 由调用方给（文件名即任务名）。
102    pub fn of(name: &str, payload: &Yaml) -> Task {
103        let journal = payload
104            .get("log")
105            .and_then(|v| v.as_sequence())
106            .map(|items| items.iter().map(JournalEvent::of).collect())
107            .unwrap_or_default();
108        let gates = payload
109            .get("gates")
110            .and_then(|v| v.as_sequence())
111            .map(|items| {
112                items
113                    .iter()
114                    .filter_map(|item| item.as_str().map(|text| text.to_string()))
115                    .collect()
116            })
117            .unwrap_or_default();
118        let products = payload
119            .get("products")
120            .and_then(|v| v.as_mapping())
121            .map(|mapping| {
122                mapping
123                    .iter()
124                    .filter_map(|(key, value)| {
125                        Some((key.as_str()?.to_string(), value.as_str()?.to_string()))
126                    })
127                    .collect()
128            })
129            .unwrap_or_default();
130        Task {
131            name: name.to_string(),
132            workflow_name: text_of(payload, "workflow"),
133            start: text_of(payload, "start"),
134            context: RunContext::of(payload),
135            journal,
136            gates,
137            products,
138        }
139    }
140
141    /// 这种产物声明了往哪写；没声明给 `None`（落哪是各自包的事）。
142    pub fn product(&self, kind: &str) -> Option<String> {
143        let written = self.products.get(kind)?.trim();
144        if written.is_empty() {
145            None
146        } else {
147            Some(written.to_string())
148        }
149    }
150
151    /// 走过哪几步。
152    ///
153    /// 带后缀的流水（`·审` / `·判`）给这一步**投票**，不带后缀的（重走一遍）
154    /// **把结论从头算**；工作流上没有的步骤名不算数。
155    pub fn done_steps(&self, workflow: &Workflow) -> Vec<String> {
156        let names = workflow.step_names();
157        let mut verdict: Vec<(String, bool)> = Vec::new();
158        for event in &self.journal {
159            let (step, extra) = match event.step.split_once('·') {
160                Some((step, rest)) => (step.to_string(), Some(rest.to_string())),
161                None => (event.step.clone(), None),
162            };
163            if !names.contains(&step) {
164                continue;
165            }
166            match verdict.iter_mut().find(|(name, _)| *name == step) {
167                Some((_, last)) => {
168                    if extra.is_some() {
169                        *last = *last && event.ok;
170                    } else {
171                        *last = event.ok;
172                    }
173                }
174                None => verdict.push((step, event.ok)),
175            }
176        }
177        verdict
178            .into_iter()
179            .filter(|(_, ok)| *ok)
180            .map(|(name, _)| name)
181            .collect()
182    }
183
184    /// 第一个没走到的步骤（按工作流里的顺序）。
185    pub fn next_step(&self, workflow: &Workflow) -> Option<String> {
186        let finished = self.done_steps(workflow);
187        workflow
188            .step_names()
189            .into_iter()
190            .find(|name| !finished.contains(name))
191    }
192
193    /// 状态行：下一步是谁，或者都走过了。
194    pub fn state_line(&self, workflow: &Workflow) -> String {
195        let names = workflow.step_names();
196        if names.is_empty() {
197            return format!(
198                "这条工作流没有步骤——在 workflows/{}.yaml 的 steps 里写步骤",
199                self.workflow_name
200            );
201        }
202        match self.next_step(workflow) {
203            Some(step) => format!("下一步：{step}"),
204            None => format!("{} 个步骤都走过了", names.len()),
205        }
206    }
207
208    /// 记一笔流水：流水只增不改，所以返回新的任务。
209    pub fn recorded(&self, at: &str, step: &str, detail: &str, ok: bool) -> Task {
210        let mut task = self.clone();
211        task.journal.push(JournalEvent {
212            at: at.to_string(),
213            step: step.to_string(),
214            detail: detail.to_string(),
215            ok,
216        });
217        task
218    }
219
220    /// 记下闸门项；已有的不重复记。同样返回新的任务。
221    pub fn with_gates(&self, notes: &[String]) -> Task {
222        let mut task = self.clone();
223        for note in notes {
224            if !task.gates.contains(note) {
225                task.gates.push(note.clone());
226            }
227        }
228        task
229    }
230
231    /// 写回任务文件的字段形状。
232    pub fn to_yaml(&self) -> Yaml {
233        let mut map = Mapping::new();
234        map.insert(Yaml::String("name".into()), Yaml::String(self.name.clone()));
235        map.insert(
236            Yaml::String("start".into()),
237            Yaml::String(self.start.clone()),
238        );
239        map.insert(
240            Yaml::String("workflow".into()),
241            Yaml::String(self.workflow_name.clone()),
242        );
243        map.insert(
244            Yaml::String("log".into()),
245            Yaml::Sequence(self.journal.iter().map(JournalEvent::to_yaml).collect()),
246        );
247        map.insert(
248            Yaml::String("gates".into()),
249            Yaml::Sequence(
250                self.gates
251                    .iter()
252                    .map(|note| Yaml::String(note.clone()))
253                    .collect(),
254            ),
255        );
256        let mut products = Mapping::new();
257        for (key, value) in &self.products {
258            products.insert(Yaml::String(key.clone()), Yaml::String(value.clone()));
259        }
260        map.insert(Yaml::String("products".into()), Yaml::Mapping(products));
261        if let Yaml::Mapping(context) = self.context.to_yaml() {
262            for (key, value) in context {
263                map.insert(key, value);
264            }
265        }
266        Yaml::Mapping(map)
267    }
268}
quanttide_work/task.rs

quanttide_work/
task.rs
