plot/lib/pipelines/
step1.rs

1use std::{fs, path::Path};
2
3use anyhow::{Context, Result};
4use chrono::{DateTime, Local};
5
6use crate::lib::{
7	logic::{
8		self,
9		ScanMatchLabel,
10		ScanNodeDirScanned,
11		ScanNodeIs,
12		ScanNodeScanned,
13		ScanPartition,
14		ScanPartitionScanned,
15	},
16	schema::{PipelineJobRow, PipelineJobTab, ReadyJob, SharedJobPart},
17};
18
19// ============================================================================
20// TRAIT POMOCNICZY (Unifikacja typów węzłów)
21// ============================================================================
22
23/// Pozwala na bezproblemową konwersję specjalistycznych węzłów na węzeł uniwersalny.
24pub trait AsScannedNode {
25	fn as_scanned_node(&self) -> logic::ScanNodeScanned;
26}
27
28impl AsScannedNode for logic::ScanNodeFileScanned {
29	fn as_scanned_node(&self) -> logic::ScanNodeScanned {
30		logic::ScanNodeScanned {
31			name: self.name.clone(),
32			path: self.path.clone(),
33			node: ScanNodeIs::File,
34			tier: self.tier,
35			id_self: self.id_self,
36			id_path: self.id_path.clone(),
37			dir_has_subdirs: None,
38			dir_has_files_binary: None,
39			dir_has_files_text: None,
40			dir_has_symlinks: None,
41			file_is_binary: Some(self.is_binary),
42			file_is_empty: Some(self.is_empty),
43		}
44	}
45}
46
47impl AsScannedNode for ScanNodeDirScanned {
48	fn as_scanned_node(&self) -> ScanNodeScanned {
49		ScanNodeScanned {
50			name: self.name.clone(),
51			path: self.path.clone(),
52			node: ScanNodeIs::Dir,
53			tier: self.tier,
54			id_self: self.id_self,
55			id_path: self.id_path.clone(),
56			dir_has_subdirs: Some(self.has_subdirs),
57			dir_has_files_binary: Some(self.has_files_binary),
58			dir_has_files_text: Some(self.has_files_text),
59			dir_has_symlinks: Some(self.has_symlinks),
60			file_is_binary: None,
61			file_is_empty: None,
62		}
63	}
64}
65
66// ============================================================================
67// GATHER & INSPECT (Faza Hydracji z Dysku)
68// ============================================================================
69
70fn piece_gather<L>(job: &ReadyJob, p: &ScanPartition<L>) -> Vec<PipelineJobRow>
71where
72	L: ScanMatchLabel,
73	L::Node: AsScannedNode, {
74	p.nodes
75		.iter()
76		.filter_map(|n| {
77			let unified_node = n.as_scanned_node();
78			data_inspect(job, unified_node).ok()
79		})
80		.collect()
81}
82
83fn data_inspect(job: &ReadyJob, scanned_node: ScanNodeScanned) -> Result<PipelineJobRow> {
84	let clean_rel = scanned_node.path.str.strip_prefix("./").unwrap_or(&scanned_node.path.str);
85
86	// Budujemy pełną ścieżkę fizyczną z workspace_dir zdefiniowanego w ReadyJob
87	let absolute_path = job.explorer().workspace_dir().join(clean_rel);
88
89	// 2. Pobranie metadanych systemu plików
90	let metadata = fs::metadata(&absolute_path).context("Nie można odczytać metadanych")?;
91
92	// Konwersja czasu systemowego na czas lokalny, bezpieczna dla potoku
93	let dt_modified: DateTime<Local> = metadata.modified()?.into();
94	let name_with_ext = absolute_path.file_name().unwrap_or_default().to_string_lossy().to_string();
95	let size_real = if metadata.is_dir() { get_dir_size(&absolute_path) } else { metadata.len() };
96
97	Ok(PipelineJobRow { dt_modified, name_with_ext, size_real, node: scanned_node })
98}
99
100/// Oblicza faktyczny (fizyczny) rozmiar katalogu sumując wszystkie pliki wewnątrz.
101fn get_dir_size(path: &Path) -> u64 {
102	walkdir::WalkDir::new(path)
103		.into_iter()
104		.filter_map(|e| e.ok())
105		.filter(|e| e.file_type().is_file())
106		.filter_map(|e| e.metadata().ok())
107		.map(|m| m.len())
108		.sum()
109}
110
111// ============================================================================
112// MAIN PIPELINE
113// ============================================================================
114
115pub fn data_gather(job: &ReadyJob, b: &ScanPartitionScanned) -> PipelineJobTab {
116	let mut final_rows = Vec::new();
117	let mut t_max = 0;
118	let mut n_max = 0;
119	let mut p_max = 0;
120
121	let parts = job.explorer().parts();
122
123	// Leniwe pobieranie danych - z dysku tylko dla wybranych partycji (MD, MF, XD, XF)
124	if parts.contains(&SharedJobPart::MD) {
125		final_rows.extend(piece_gather(job, &b.m_d));
126		t_max = t_max.max(b.m_d.tier_max);
127		n_max = n_max.max(b.m_d.name_len_max);
128		p_max = p_max.max(b.m_d.path_len_max);
129	}
130	if parts.contains(&SharedJobPart::MF) {
131		final_rows.extend(piece_gather(job, &b.m_f));
132		t_max = t_max.max(b.m_f.tier_max);
133		n_max = n_max.max(b.m_f.name_len_max);
134		p_max = p_max.max(b.m_f.path_len_max);
135	}
136	if parts.contains(&SharedJobPart::XD) {
137		final_rows.extend(piece_gather(job, &b.x_d));
138		t_max = t_max.max(b.x_d.tier_max);
139		n_max = n_max.max(b.x_d.name_len_max);
140		p_max = p_max.max(b.x_d.path_len_max);
141	}
142	if parts.contains(&SharedJobPart::XF) {
143		final_rows.extend(piece_gather(job, &b.x_f));
144		t_max = t_max.max(b.x_f.tier_max);
145		n_max = n_max.max(b.x_f.name_len_max);
146		p_max = p_max.max(b.x_f.path_len_max);
147	}
148
149	// Sortujemy złączony wynik, aby przywrócić spójną strukturę drzewa po ścieżkach
150	final_rows.sort_unstable_by(|row_a, row_b| row_a.node.path.str.cmp(&row_b.node.path.str));
151
152	PipelineJobTab { rows: final_rows, tier_max: t_max, name_len_max: n_max, path_len_max: p_max }
153}
154//============================================================================
155/// [Step 1] Skaner (Punkt wejściowy pierwszego etapu)
156pub fn engine_step1_scanner(job: &ReadyJob) -> PipelineJobTab {
157	// Odpalamy skaner systemu plików używając Explorera
158	let partition_scanned = ScanPartitionScanned::scan(job.explorer());
159
160	// Pobieramy metadane z dysku dla pomyślnie dopasowanych węzłów
161	data_gather(job, &partition_scanned)
162}
plot/lib/pipelines/step1.rs

plot/lib/pipelines/
step1.rs
