face_core/input/
parse.rs

1//! Format-specific input parsing.
2//!
3//! [`parse`] takes a [`BufRead`] positioned past any UTF-8 BOM and the
4//! [`InputFormat`] picked by sniffing, and returns a [`ParsedInput`]:
5//! the items to cluster, optional sidecar metadata, and any per-record
6//! skip reports (§11.1).
7
8use std::io::BufRead;
9
10use serde_json::{Map, Value};
11
12use crate::InputFormat;
13use crate::error::{FaceError, SkipReason, SkipReport};
14use crate::input::items::{ItemsOptions, detect_items_with_options};
15use crate::path;
16
17/// One parsed input, ready for clustering.
18///
19/// `PartialEq` is intentionally not derived because [`SkipReport`] does
20/// not implement it; compare fields individually in tests.
21#[derive(Debug, Clone)]
22#[non_exhaustive]
23pub struct ParsedInput {
24    /// Records to cluster, in input order.
25    pub items: Vec<Value>,
26    /// Sidecar metadata: present when the JSON root was an object and
27    /// items detection collected leftover non-array fields. `None` for
28    /// JSONL, JSON-array roots, and CSV/TSV inputs.
29    pub meta: Option<Map<String, Value>>,
30    /// jq-style path that produced [`Self::items`].
31    ///
32    /// - `"."` — top-level array, JSONL stream, or face envelope.
33    /// - `".items"` / `".results"` / etc. — the named candidate
34    ///   selected by §4.2 items detection on a JSON object root.
35    /// - The user's `--items=PATH` value when supplied via the
36    ///   `items_path` argument to [`parse`].
37    ///
38    /// Surfaced into the envelope's `detection.items_path` field per
39    /// §7.
40    pub items_path: String,
41    /// Per-record skip events surfaced during parsing.
42    pub skips: Vec<SkipReport>,
43}
44
45/// Optional knobs for format parsers.
46#[derive(Debug, Clone, PartialEq, Eq, Default)]
47#[non_exhaustive]
48pub struct ParseOptions {
49    /// Candidate field names for JSON object items-array detection.
50    pub items: ItemsOptions,
51}
52
53/// Parse the input as the chosen format and locate the items array.
54///
55/// The caller has already run [`crate::input::sniff::sniff_format`] and
56/// consumed the BOM. `items_path` overrides automatic items detection
57/// per §4.2 — when `Some(path)`, the parser resolves it via
58/// [`path::resolve`] before returning items.
59///
60/// # Errors
61///
62/// - [`FaceError::Io`] for I/O failure on the reader.
63/// - [`FaceError::InputParse`] for parse failure or unsupported raw-parser
64///   routes (face-envelope re-processing is handled by the CLI pipeline
65///   before raw parsing).
66/// - [`FaceError::UnknownItemsPath`] when `items_path` was supplied but
67///   does not resolve.
68/// - [`FaceError::AmbiguousDetection`] when auto-detection cannot pick
69///   a unique items array.
70///
71/// # Examples
72///
73/// ```
74/// use std::io::BufReader;
75/// use face_core::input::parse;
76/// use face_core::InputFormat;
77///
78/// let mut r = BufReader::new(b"[1, 2, 3]" as &[u8]);
79/// let parsed = parse(&mut r, InputFormat::Json, None).unwrap();
80/// assert_eq!(parsed.items.len(), 3);
81/// assert!(parsed.skips.is_empty());
82/// ```
83pub fn parse(
84    reader: &mut impl BufRead,
85    format: InputFormat,
86    items_path: Option<&str>,
87) -> Result<ParsedInput, FaceError> {
88    parse_with_columns(reader, format, items_path, None)
89}
90
91/// Parse input, optionally supplying column names for headerless
92/// CSV/TSV data.
93pub fn parse_with_columns(
94    reader: &mut impl BufRead,
95    format: InputFormat,
96    items_path: Option<&str>,
97    columns: Option<&[String]>,
98) -> Result<ParsedInput, FaceError> {
99    parse_with_columns_and_options(
100        reader,
101        format,
102        items_path,
103        columns,
104        &ParseOptions::default(),
105    )
106}
107
108/// Parse input with explicit delimited-column and detection options.
109pub fn parse_with_columns_and_options(
110    reader: &mut impl BufRead,
111    format: InputFormat,
112    items_path: Option<&str>,
113    columns: Option<&[String]>,
114    options: &ParseOptions,
115) -> Result<ParsedInput, FaceError> {
116    match format {
117        InputFormat::Json => parse_json(reader, items_path, options),
118        InputFormat::Jsonl => parse_jsonl(reader, items_path),
119        InputFormat::Csv | InputFormat::Tsv => parse_delimited(reader, format, items_path, columns),
120        InputFormat::FaceEnvelope => parse_face_envelope(reader),
121    }
122}
123
124/// Parse a JSON document into a [`ParsedInput`].
125fn parse_json(
126    reader: &mut impl BufRead,
127    items_path: Option<&str>,
128    options: &ParseOptions,
129) -> Result<ParsedInput, FaceError> {
130    let value: Value = serde_json::from_reader(reader).map_err(|e| FaceError::InputParse {
131        format: InputFormat::Json,
132        message: e.to_string(),
133    })?;
134
135    if let Some(path) = items_path {
136        let resolved = path::resolve(&value, path)?.clone();
137        let items = match resolved {
138            Value::Array(a) => a,
139            _ => {
140                return Err(FaceError::InputParse {
141                    format: InputFormat::Json,
142                    message: format!("items path `{path}` did not resolve to an array"),
143                });
144            }
145        };
146        // When the user pinned the items path, no sidecar meta — they
147        // told us exactly where the records live; everything else is
148        // theirs to handle (§4.2's `meta` is only for auto-detection).
149        return Ok(ParsedInput {
150            items,
151            meta: None,
152            items_path: path.to_string(),
153            skips: Vec::new(),
154        });
155    }
156
157    let detection = detect_items_with_options(value, &options.items)?;
158    Ok(ParsedInput {
159        items: detection.items,
160        meta: detection.meta,
161        items_path: detection.items_path,
162        skips: Vec::new(),
163    })
164}
165
166/// Parse a JSONL stream into a [`ParsedInput`], skipping malformed
167/// records per §11.1.
168///
169/// Empty lines are silently skipped (no [`SkipReport`]).
170/// `items_path` is respected per record: if set, each line is parsed,
171/// then `path::resolve` is applied; failure to resolve is a
172/// [`SkipReason::MissingField`].
173fn parse_jsonl(
174    reader: &mut impl BufRead,
175    items_path: Option<&str>,
176) -> Result<ParsedInput, FaceError> {
177    let mut items = Vec::new();
178    let mut skips = Vec::new();
179    let mut record_index = 0usize;
180    // The detected items path: when the user supplies `--items`, echo
181    // it back; otherwise the JSONL stream's "items path" is `.`
182    // (each line is a record).
183    let resolved_items_path = items_path
184        .map(str::to_string)
185        .unwrap_or_else(|| ".".to_string());
186
187    let mut line = String::new();
188    loop {
189        line.clear();
190        let read = reader.read_line(&mut line)?;
191        if read == 0 {
192            break;
193        }
194        // Trim a trailing `\n` or `\r\n` for parser-friendly slicing,
195        // but preserve column offsets for the SkipReport.
196        let trimmed = line.trim_end_matches(['\n', '\r']);
197        if trimmed.trim().is_empty() {
198            continue;
199        }
200
201        match serde_json::from_str::<Value>(trimmed) {
202            Ok(value) => {
203                if let Some(path) = items_path {
204                    match path::resolve(&value, path) {
205                        Ok(v) => items.push(v.clone()),
206                        Err(_) => skips.push(SkipReport {
207                            record_index,
208                            reason: SkipReason::MissingField {
209                                field: path.to_string(),
210                            },
211                        }),
212                    }
213                } else {
214                    items.push(value);
215                }
216            }
217            Err(e) => skips.push(SkipReport {
218                record_index,
219                reason: SkipReason::InvalidJson {
220                    column: e.column(),
221                    message: e.to_string(),
222                },
223            }),
224        }
225
226        record_index += 1;
227    }
228
229    Ok(ParsedInput {
230        items,
231        meta: None,
232        items_path: resolved_items_path,
233        skips,
234    })
235}
236
237fn parse_delimited(
238    reader: &mut impl BufRead,
239    format: InputFormat,
240    items_path: Option<&str>,
241    columns: Option<&[String]>,
242) -> Result<ParsedInput, FaceError> {
243    if let Some(path) = items_path {
244        return Err(FaceError::InputParse {
245            format,
246            message: format!("--items={path} is not supported for CSV/TSV input"),
247        });
248    }
249
250    let mut bytes = Vec::new();
251    reader.read_to_end(&mut bytes)?;
252    if bytes.is_empty() {
253        return Err(FaceError::InputParse {
254            format,
255            message: "input is empty (no rows to parse)".to_string(),
256        });
257    }
258
259    let delimiter = match format {
260        InputFormat::Csv => detect_csv_delimiter(&bytes).unwrap_or(b','),
261        InputFormat::Tsv => b'\t',
262        _ => unreachable!("parse_delimited called only for CSV/TSV"),
263    };
264
265    let mut csv_reader = csv::ReaderBuilder::new()
266        .has_headers(false)
267        .delimiter(delimiter)
268        .flexible(false)
269        .from_reader(bytes.as_slice());
270
271    let mut rows = Vec::new();
272    for record in csv_reader.byte_records() {
273        let record = record.map_err(|e| FaceError::InputParse {
274            format,
275            message: e.to_string(),
276        })?;
277        rows.push(record.iter().map(decode_field).collect::<Vec<_>>());
278    }
279
280    if rows.is_empty() {
281        return Err(FaceError::InputParse {
282            format,
283            message: "input is empty (no rows to parse)".to_string(),
284        });
285    }
286
287    let (headers, data_start) = resolve_delimited_headers(&rows, columns, format)?;
288    let expected_width = headers.len();
289    let mut items = Vec::new();
290    for (row_index, row) in rows.iter().enumerate().skip(data_start) {
291        if row.len() != expected_width {
292            return Err(FaceError::InputParse {
293                format,
294                message: format!(
295                    "row {} has {} columns but header has {expected_width}",
296                    row_index + 1,
297                    row.len()
298                ),
299            });
300        }
301        let mut object = Map::new();
302        for (name, value) in headers.iter().zip(row) {
303            object.insert(name.clone(), parse_delimited_scalar(value));
304        }
305        items.push(Value::Object(object));
306    }
307
308    Ok(ParsedInput {
309        items,
310        meta: None,
311        items_path: ".".to_string(),
312        skips: Vec::new(),
313    })
314}
315
316fn resolve_delimited_headers(
317    rows: &[Vec<String>],
318    columns: Option<&[String]>,
319    format: InputFormat,
320) -> Result<(Vec<String>, usize), FaceError> {
321    if let Some(columns) = columns {
322        if columns.is_empty() {
323            return Err(FaceError::InputParse {
324                format,
325                message: "--columns must include at least one column name".to_string(),
326            });
327        }
328        return Ok((normalize_headers(columns.iter().map(String::as_str)), 0));
329    }
330
331    if looks_like_header(rows) {
332        Ok((normalize_headers(rows[0].iter().map(String::as_str)), 1))
333    } else {
334        let width = rows.first().map(Vec::len).unwrap_or(0);
335        let generated = (1..=width).map(|idx| format!("column{idx}"));
336        Ok((generated.collect(), 0))
337    }
338}
339
340fn looks_like_header(rows: &[Vec<String>]) -> bool {
341    let Some(first) = rows.first() else {
342        return false;
343    };
344    if first.is_empty() || !first.iter().all(|field| !field.trim().is_empty()) {
345        return false;
346    }
347
348    let first_text = first.iter().filter(|field| !is_scalarish(field)).count();
349    if first_text == 0 {
350        return false;
351    }
352
353    let Some(second) = rows.get(1) else {
354        return true;
355    };
356    let second_text = second.iter().filter(|field| !is_scalarish(field)).count();
357    first_text > second_text || first.iter().any(|field| is_common_header_name(field))
358}
359
360fn is_common_header_name(value: &str) -> bool {
361    matches!(
362        value.trim().to_ascii_lowercase().as_str(),
363        "id" | "kind"
364            | "type"
365            | "status"
366            | "severity"
367            | "score"
368            | "rank"
369            | "path"
370            | "file"
371            | "module"
372            | "repo"
373            | "name"
374            | "title"
375            | "category"
376            | "value"
377            | "count"
378    )
379}
380
381fn normalize_headers<'a>(headers: impl IntoIterator<Item = &'a str>) -> Vec<String> {
382    use std::collections::BTreeMap;
383
384    let mut seen: BTreeMap<String, usize> = BTreeMap::new();
385    headers
386        .into_iter()
387        .enumerate()
388        .map(|(idx, raw)| {
389            let base = if raw.trim().is_empty() {
390                format!("column{}", idx + 1)
391            } else {
392                raw.trim().to_string()
393            };
394            let count = seen.entry(base.clone()).or_default();
395            *count += 1;
396            if *count == 1 {
397                base
398            } else {
399                format!("{base}_{}", *count)
400            }
401        })
402        .collect()
403}
404
405fn parse_delimited_scalar(value: &str) -> Value {
406    let trimmed = value.trim();
407    if trimmed.is_empty() {
408        return Value::Null;
409    }
410    if trimmed.eq_ignore_ascii_case("true") {
411        return Value::Bool(true);
412    }
413    if trimmed.eq_ignore_ascii_case("false") {
414        return Value::Bool(false);
415    }
416    if is_integer_literal(trimmed)
417        && let Ok(n) = trimmed.parse::<i64>()
418    {
419        return Value::Number(n.into());
420    }
421    if let Ok(n) = trimmed.parse::<f64>()
422        && n.is_finite()
423        && let Some(number) = serde_json::Number::from_f64(n)
424    {
425        return Value::Number(number);
426    }
427    Value::String(value.to_string())
428}
429
430fn is_scalarish(value: &str) -> bool {
431    let trimmed = value.trim();
432    trimmed.is_empty()
433        || trimmed.eq_ignore_ascii_case("true")
434        || trimmed.eq_ignore_ascii_case("false")
435        || trimmed.parse::<f64>().is_ok()
436}
437
438fn is_integer_literal(value: &str) -> bool {
439    let rest = value.strip_prefix(['+', '-']).unwrap_or(value);
440    !rest.is_empty() && rest.bytes().all(|b| b.is_ascii_digit())
441}
442
443fn decode_field(bytes: &[u8]) -> String {
444    match std::str::from_utf8(bytes) {
445        Ok(s) => s.to_string(),
446        Err(_) => bytes.iter().map(|&b| char::from(b)).collect(),
447    }
448}
449
450fn detect_csv_delimiter(bytes: &[u8]) -> Option<u8> {
451    const CANDIDATES: [u8; 3] = [b',', b';', b'|'];
452    let lines = sample_nonempty_lines(bytes);
453    let mut best = None;
454    for delimiter in CANDIDATES {
455        let counts = lines
456            .iter()
457            .map(|line| count_delimiter_outside_quotes(line, delimiter))
458            .collect::<Vec<_>>();
459        let Some((&first, rest)) = counts.split_first() else {
460            continue;
461        };
462        if first == 0 || rest.iter().any(|count| *count != first) {
463            continue;
464        }
465        if best.is_none_or(|(_, best_count)| first > best_count) {
466            best = Some((delimiter, first));
467        }
468    }
469    best.map(|(delimiter, _)| delimiter)
470}
471
472fn sample_nonempty_lines(bytes: &[u8]) -> Vec<&[u8]> {
473    bytes
474        .split(|b| *b == b'\n')
475        .map(|line| line.strip_suffix(b"\r").unwrap_or(line))
476        .filter(|line| line.iter().any(|b| !b.is_ascii_whitespace()))
477        .take(8)
478        .collect()
479}
480
481fn count_delimiter_outside_quotes(line: &[u8], delimiter: u8) -> usize {
482    let mut count = 0usize;
483    let mut in_quotes = false;
484    let mut i = 0usize;
485    while let Some(&byte) = line.get(i) {
486        if byte == b'"' {
487            if in_quotes && line.get(i + 1) == Some(&b'"') {
488                i += 2;
489                continue;
490            }
491            in_quotes = !in_quotes;
492        } else if byte == delimiter && !in_quotes {
493            count += 1;
494        }
495        i += 1;
496    }
497    count
498}
499
500/// Parse a face envelope through the raw input parser.
501///
502/// The CLI handles §9 re-processing before it reaches this parser
503/// because a full [`Envelope`](crate::Envelope) is not a [`ParsedInput`].
504/// Calling the raw parser directly with [`InputFormat::FaceEnvelope`]
505/// is therefore a routing error.
506fn parse_face_envelope(_reader: &mut impl BufRead) -> Result<ParsedInput, FaceError> {
507    Err(FaceError::InputParse {
508        format: InputFormat::FaceEnvelope,
509        message: "face envelope input is handled by the CLI re-processing path".to_string(),
510    })
511}
512
513#[cfg(test)]
514mod tests {
515    use super::*;
516    use serde_json::json;
517    use std::io::BufReader;
518
519    #[test]
520    fn parses_json_array() {
521        let mut r = BufReader::new(b"[1, 2, 3]" as &[u8]);
522        let p = parse(&mut r, InputFormat::Json, None).unwrap();
523        assert_eq!(p.items, vec![json!(1), json!(2), json!(3)]);
524        assert!(p.meta.is_none());
525        assert!(p.skips.is_empty());
526    }
527
528    #[test]
529    fn parses_json_object_with_items_field() {
530        let mut r = BufReader::new(br#"{"items": [10, 20], "took_ms": 5}"# as &[u8]);
531        let p = parse(&mut r, InputFormat::Json, None).unwrap();
532        assert_eq!(p.items, vec![json!(10), json!(20)]);
533        let meta = p.meta.unwrap();
534        assert_eq!(meta.get("took_ms"), Some(&json!(5)));
535    }
536
537    #[test]
538    fn parses_jsonl_skips_malformed() {
539        let input = b"{\"a\":1}\n{this is not json}\n{\"a\":3}\n";
540        let mut r = BufReader::new(&input[..]);
541        let p = parse(&mut r, InputFormat::Jsonl, None).unwrap();
542        assert_eq!(p.items, vec![json!({"a": 1}), json!({"a": 3})]);
543        assert_eq!(p.skips.len(), 1);
544        assert_eq!(p.skips[0].record_index, 1);
545        match &p.skips[0].reason {
546            SkipReason::InvalidJson { .. } => {}
547            other => panic!("unexpected reason: {other:?}"),
548        }
549    }
550
551    #[test]
552    fn jsonl_skips_empty_lines_silently() {
553        let input = b"{\"a\":1}\n\n{\"a\":2}\n\n";
554        let mut r = BufReader::new(&input[..]);
555        let p = parse(&mut r, InputFormat::Jsonl, None).unwrap();
556        assert_eq!(p.items, vec![json!({"a": 1}), json!({"a": 2})]);
557        assert!(p.skips.is_empty());
558    }
559
560    #[test]
561    fn json_with_items_path_override() {
562        let mut r = BufReader::new(br#"{"hits": {"records": [1, 2, 3]}, "extra": 9}"# as &[u8]);
563        let p = parse(&mut r, InputFormat::Json, Some(".hits.records")).unwrap();
564        assert_eq!(p.items, vec![json!(1), json!(2), json!(3)]);
565        assert!(p.meta.is_none());
566    }
567
568    #[test]
569    fn json_unknown_items_path_errors() {
570        let mut r = BufReader::new(br#"{"hits": [1]}"# as &[u8]);
571        let err = parse(&mut r, InputFormat::Json, Some(".missing")).unwrap_err();
572        match err {
573            FaceError::UnknownItemsPath { path } => assert_eq!(path, ".missing"),
574            other => panic!("unexpected: {other:?}"),
575        }
576    }
577
578    #[test]
579    fn parses_csv_with_header() {
580        let mut r = BufReader::new(b"a,b\n1,2\n" as &[u8]);
581        let parsed = parse(&mut r, InputFormat::Csv, None).unwrap();
582        assert_eq!(parsed.items, vec![json!({"a": 1, "b": 2})]);
583        assert_eq!(parsed.items_path, ".");
584    }
585
586    #[test]
587    fn face_envelope_path_returns_input_parse() {
588        let mut r =
589            BufReader::new(br#"{"result":{"input_total":0,"skipped":0,"axes":[],"detection":{"format":"json","items_path":".","score_path":null,"preset":null,"fallback_reason":null}},"meta":{},"clusters":[],"page":{"cluster_id":null,"page":0,"per_page":0,"total_items":0,"items":[]}}"# as &[u8]);
590        let err = parse(&mut r, InputFormat::FaceEnvelope, None).unwrap_err();
591        match err {
592            FaceError::InputParse {
593                format: InputFormat::FaceEnvelope,
594                ..
595            } => {}
596            other => panic!("unexpected: {other:?}"),
597        }
598    }
599}
face_core/input/parse.rs

face_core/input/
parse.rs
