vin-decode 0.3.3

//! Build-time pipeline for converting vPIC tabular data into FST/rkyv lookup maps.
//!
//! Enabled via the `build` feature. End users normally don't need this — the
//! crate ships pre-built maps. This module is used by the monthly CI cron job
//! that regenerates the embedded data.

use std::collections::{BTreeMap, BTreeSet};
use std::fs::File;
use std::io::{BufWriter, Write};
use std::path::Path;

use fst::{MapBuilder, SetBuilder};
use rkyv::rancor::Error as RkyvError;

use crate::Error;
pub use crate::data::{
    BodyRow, EngineRow, EuModelRow, LookupRow, MakeRow, ModelRow, SchemaRow, VinRuleRow,
};
use crate::data::{RkyvSer, Saveable};
use crate::types::BodyType;

/// Write a typed `key → Vec<T>` map as paired `.fst` index + `.bin` rkyv blob.
///
/// `sorted_pairs` MUST be sorted by key (FST builders require lex-ordered insertion).
pub fn write_grouped<T>(
    sorted_pairs: &[(String, Vec<T>)],
    fst_path: &Path,
    values_path: &Path,
) -> crate::Result<()>
where
    T: RkyvSer,
{
    let fst_writer = BufWriter::new(File::create(fst_path)?);
    let mut values_writer = BufWriter::new(File::create(values_path)?);
    let mut builder = MapBuilder::new(fst_writer).map_err(|e| Error::MissingData(e.to_string()))?;
    let mut offset: u64 = 0;
    for (key, values) in sorted_pairs {
        let bytes =
            rkyv::to_bytes::<RkyvError>(values).map_err(|e| Error::MissingData(e.to_string()))?;
        values_writer.write_all(&bytes)?;
        let combined = (offset << 32) | (bytes.len() as u64);
        builder
            .insert(key, combined)
            .map_err(|e| Error::MissingData(e.to_string()))?;
        offset += bytes.len() as u64;
    }
    builder
        .finish()
        .map_err(|e| Error::MissingData(e.to_string()))?;
    values_writer.flush()?;
    Ok(())
}

/// Write a sorted set of keys as a value-less `.fst` (used for the makes index).
pub fn write_set(sorted_keys: &[String], fst_path: &Path) -> crate::Result<()> {
    let writer = BufWriter::new(File::create(fst_path)?);
    let mut builder = SetBuilder::new(writer).map_err(|e| Error::MissingData(e.to_string()))?;
    for key in sorted_keys {
        builder
            .insert(key)
            .map_err(|e| Error::MissingData(e.to_string()))?;
    }
    builder
        .finish()
        .map_err(|e| Error::MissingData(e.to_string()))?;
    Ok(())
}

/// Derive a make→models reverse index by joining WMI/schema/lookup tables.
///
/// Walks every WMI's schemas, collects every lookup row with `element == "Model"`,
/// and bucket-sorts the resulting model strings under their make.
pub fn derive_make_models(
    wmi_make: &[(String, Vec<MakeRow>)],
    wmi_schema: &[(String, Vec<SchemaRow>)],
    schema_lookup: &[(String, Vec<LookupRow>)],
) -> Vec<(String, Vec<ModelRow>)> {
    let make_by_wmi: BTreeMap<&str, &str> = wmi_make
        .iter()
        .filter_map(|(wmi, rows)| rows.first().map(|r| (wmi.as_str(), r.name.as_str())))
        .collect();

    let schemas_by_wmi: BTreeMap<&str, Vec<&str>> = wmi_schema
        .iter()
        .map(|(wmi, rows)| {
            (
                wmi.as_str(),
                rows.iter().map(|r| r.id.as_str()).collect::<Vec<_>>(),
            )
        })
        .collect();

    let lookups_by_schema: BTreeMap<&str, &[LookupRow]> = schema_lookup
        .iter()
        .map(|(id, rows)| (id.as_str(), rows.as_slice()))
        .collect();

    let mut acc: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
    for (wmi, make) in &make_by_wmi {
        let Some(schemas) = schemas_by_wmi.get(wmi) else {
            continue;
        };
        for sid in schemas {
            let Some(lookups) = lookups_by_schema.get(sid) else {
                continue;
            };
            for lk in lookups.iter() {
                if lk.element == "Model" {
                    acc.entry(make.to_ascii_uppercase())
                        .or_default()
                        .insert(lk.value.clone());
                }
            }
        }
    }

    acc.into_iter()
        .map(|(make, models)| {
            (
                make,
                models.into_iter().map(|name| ModelRow { name }).collect(),
            )
        })
        .collect()
}

/// Collect the unique sorted set of uppercase make names from a wmi_make table.
pub fn collect_makes(wmi_make: &[(String, Vec<MakeRow>)]) -> Vec<String> {
    let mut set: BTreeSet<String> = BTreeSet::new();
    for (_, rows) in wmi_make {
        for r in rows {
            set.insert(r.name.to_ascii_uppercase());
        }
    }
    set.into_iter().collect()
}

/// Curated supplement of real passenger-car / light-commercial (van) makes that
/// are **not** derived from any WMI/VIN row.
///
/// The make index is normally built from WMI data ([`collect_makes`]), so a
/// brand only exists if some VIN row carries it. That excludes recent / niche /
/// EV / microcar marques for which we hold no WMI data but which are real cars
/// on the EU/RO market. This list decouples make membership from VIN data: every
/// entry here is folded into `makes.fst` by [`merge_makes`] so
/// [`crate::Catalog::has_make`] / [`crate::Catalog::all_makes`] recognise them.
///
/// Entries are uppercased on merge, so casing here is cosmetic. Keep this list
/// to **real passenger cars / vans only** — no motorcycles, trailers, campers,
/// boats, or junk tokens. When unsure whether a token is a real car make, leave
/// it out (a downstream canonicalizer should reject + surface unknowns rather
/// than silently accept a wrong make).
pub const EXTRA_MAKES: &[&str] = &[
    // --- Chinese EV / new-energy brands now on the EU/RO market ---
    "DENZA",     // BYD premium sub-brand
    "JAC",       // JAC Motors (e.g. e-JS, iEV, vans)
    "LEAPMOTOR", // Leapmotor (Stellantis JV, sold in EU)
    "AION",      // GAC Aion (EV)
    "AIWAYS",    // Aiways (U5/U6, was sold in EU)
    "ZEEKR",     // Geely premium EV
    "OMODA",     // Chery sub-brand (Omoda 5 etc.)
    "JAECOO",    // Chery off-road sub-brand (sister of Omoda, EU launch)
    "SERES",     // Seres / DFSK EV (sold in EU)
    "VOYAH",     // Dongfeng premium EV (sold in EU, e.g. Free/Courage)
    "HONGQI",    // FAW premium (sold in EU, e.g. E-HS9)
    "MAXUS",     // SAIC Maxus (LCV/vans + EVs) — already WMI-covered, kept for safety
    "JETOUR",    // Chery SUV sub-brand (Dashing/T2), EU/RO launch
    "AVATR",     // Changan/Huawei/CATL premium EV (canonical "Avatr", NOT "Avatar")
    "BAIC",      // Beijing Automotive (X55/X7/B-series), sold in EU/RO
    "GWM",       // Great Wall Motor — umbrella consumer marque in Europe
    "TANK",      // GWM standalone off-road SUV marque (Tank 300/500/700)
    "ORA",       // GWM standalone EV marque (Funky/Good Cat), in EU since 2022
    "FORTHING",  // Dongfeng Forthing passenger SUVs/sedans (T5 etc.)
    "BESTUNE",   // FAW Bestune passenger cars (B70 sedan, T-series SUVs)
    "SKYWELL",   // Skywell EV SUVs (BE11), sold in UK/EU
    "DFSK",      // Dongfeng Sokon — passenger SUVs/MPVs (Glory/Fengon) + vans
    "XIAOMI", // Xiaomi Auto / Xiaomi EV (SU7 sedan, YU7 SUV) — car maker since 2024, EU-registered
    "IM",     // IM Motors (SAIC "Intelligence in Motion") — IM5/IM6 sold in EU via MG network
    // --- Geely/Volvo orbit ---
    "LYNK & CO", // Lynk & Co (Geely/Volvo) — canonical spelling
    // --- Korean ---
    "KGM", // KG Mobility — the renamed SsangYong
    // --- European microcars / quadricycles / small EVs (real cars) ---
    "AIXAM",     // French microcar/quadricycle
    "LIGIER",    // French microcar/quadricycle
    "MICROLINO", // Swiss bubble-car EV
    "TAZZARI",   // Italian EV microcar (Zero)
    "XEV",       // XEV (YoYo) EV microcar
    "ESTRIMA",   // Estrima Birò EV quadricycle
    "CHATENET",  // French microcar/quadricycle
    "DR",        // DR Automobiles (Italy, rebadged Chery)
    "EVO",       // EVO (DR's value sub-brand, Italy)
    "SWM",       // SWM (Italy, SUVs)
    // --- More Chinese makes (2024–2025 EU/RO entrants), verified identities ---
    "DONGFENG",       // Dongfeng — Chinese "Big Four" OEM (raw alias: DFM)
    "M-HERO",         // Dongfeng Mengshi luxury off-road EV (raw: MHERO, Mengshi)
    "MAEXTRO",        // JAC × Huawei (HIMA) ultra-luxury (Chinese name: Zunjie)
    "FANGCHENGBAO",   // BYD off-road sub-brand, marketed "Bao" / "Leopard"
    "LINKTOUR",       // Weiqiao microcar/quadricycle (Alumi), EU launch
    "TODAY SUNSHINE", // Zhejiang Today Sunshine microcar — the MAKE (M1/M2 = models)
];

/// Fold the curated [`EXTRA_MAKES`] supplement into a WMI-derived make list.
///
/// Takes the makes already collected from WMI data and returns the union with
/// the curated supplement, uppercased, deduped and lexicographically sorted
/// (the ordering [`write_set`] requires for FST insertion). This is the seam
/// that makes the catalog the single source of truth for car makes regardless
/// of whether a brand has any VIN/WMI rows.
pub fn merge_makes(wmi_makes: &[String]) -> Vec<String> {
    // Drop every alias source so a zero-model duplicate make ("DS",
    // "MERCEDES-AMG", "ROLLS ROYCE") can't shadow its populated canonical
    // sibling on resolve_make's exact-match fast path. Non-make sources (raw
    // dealer tokens like "KG") are simply never present — the filter is a no-op.
    let drop: BTreeSet<String> = crate::catalog::MAKE_ALIASES
        .iter()
        .map(|(from, _)| from.to_ascii_uppercase())
        .collect();
    let mut set: BTreeSet<String> = wmi_makes
        .iter()
        .map(|m| m.to_ascii_uppercase())
        .filter(|m| !drop.contains(m))
        .collect();
    for extra in EXTRA_MAKES {
        set.insert(extra.to_ascii_uppercase());
    }
    // Force every alias target present so the redirect always lands a real key.
    for (_, to) in crate::catalog::MAKE_ALIASES {
        set.insert(to.to_ascii_uppercase());
    }
    set.into_iter().collect()
}

/// Curated supplement of real models for the [`EXTRA_MAKES`] cohort.
///
/// `derive_make_models` only buckets models under makes that carry WMI/VIN rows,
/// so the curated (mostly Chinese-EV / EU-microcar) makes — which have no WMI
/// data — ship an empty model list. That breaks a downstream resolver that bails
/// the instant `models_for_make(make)` is empty. This list decouples model
/// membership from VIN data, mirroring how [`EXTRA_MAKES`] decouples make
/// membership.
///
/// Contract: each entry's make must be a canonical key `resolve_make` can emit
/// (WMI/rip-derived or in [`EXTRA_MAKES`]), and each model must be stored in the form
/// that survives stripping the make token off a listing slug. A listing
/// `"DFSK SERES 3"` resolves to make `DFSK` (trailing tokens dropped) leaving
/// residual `"SERES 3"`, so `SERES 3` is filed under `DFSK`; a bare-tagged
/// `"Seres 3"` resolves to `SERES` leaving residual `"3"`, so `3` is filed under
/// `SERES`. When a brand is tagged both ways, both forms are filed.
///
/// A trailing section also tops up **mainstream** makes (which do carry WMI
/// data) with real models the vPIC pattern index and the EU rip miss — mostly
/// EU light-commercial/van and recent EV/SUV nameplates. Those merge into the
/// make's existing model set; `models_for_make` dedups, so a model already
/// present via another source is a harmless no-op.
///
/// Models are real/verified — no invented trims. Values are uppercased on merge.
pub const EXTRA_MODELS: &[(&str, &[&str])] = &[
    // --- Dongfeng Sokon: Seres-derived SUVs badged DFSK, plus Glory/Fengon/vans ---
    (
        "DFSK",
        &[
            "SERES 3", "SERES 5", "SERES 7", "GLORY 500", "GLORY 560", "GLORY 580", "FENGON",
            "EC35",
        ],
    ),
    // bare residual when a dealer tags the make as "Seres"
    ("SERES", &["3", "5", "7"]),
    ("ZEEKR", &["001", "007", "009", "X", "7X", "8X", "9X"]),
    ("OMODA", &["5", "7", "9", "E5"]),
    ("JAECOO", &["5", "7", "8"]),
    ("LEAPMOTOR", &["C10", "C11", "C16", "T03", "B10"]),
    ("DENZA", &["D9", "N7", "N8", "Z9", "Z9 GT"]),
    ("VOYAH", &["FREE", "DREAMER", "COURAGE", "PASSION"]),
    ("HONGQI", &["E-HS9", "H9", "HS5", "H5", "EH7"]),
    ("MAXUS", &["MIFA 9", "DELIVER 9", "EUNIQ 5", "EUNIQ 6", "T90", "D90", "EV30"]),
    ("JETOUR", &["DASHING", "T2", "X70", "X90", "G700"]),
    ("AVATR", &["11", "12", "07", "06"]),
    ("BAIC", &["X55", "X7", "X35", "B40", "BJ40"]),
    // GWM umbrella marque: cross-sub-brand residuals ("GWM Ora 03" -> GWM / "ORA 03")
    (
        "GWM",
        &[
            "HAVAL H6", "HAVAL JOLION", "TANK 300", "TANK 500", "ORA 03", "ORA 07", "POER",
            "WINGLE 5",
        ],
    ),
    ("TANK", &["300", "400", "500", "700"]),
    ("ORA", &["03", "07", "FUNKY CAT", "GOOD CAT", "LIGHTNING CAT"]),
    ("FORTHING", &["T5", "T5 EVO", "S7", "FRIDAY", "Z3"]),
    ("BESTUNE", &["B70", "B70S", "T77", "T99", "T55"]),
    ("SKYWELL", &["BE11", "ET5"]),
    ("AION", &["Y", "S", "V", "ES", "UT"]),
    ("AIWAYS", &["U5", "U6"]),
    ("JAC", &["E-JS4", "JS4", "IEV7S", "T8", "T9"]),
    ("DONGFENG", &["BOX", "NAMMI", "E70", "RICH 6", "AX7"]),
    ("M-HERO", &["917"]),
    ("MAEXTRO", &["S800"]),
    // BYD off-road sub-brand: numeric residual + "Bao"/"Leopard" badge forms
    (
        "FANGCHENGBAO",
        &[
            "3", "5", "8", "TI7", "BAO 3", "BAO 5", "BAO 8", "LEOPARD 5", "LEOPARD 8",
        ],
    ),
    ("LINKTOUR", &["ALUMI", "ALUMI PLUS"]),
    ("TODAY SUNSHINE", &["M1", "M2"]),
    ("XIAOMI", &["SU7", "YU7"]),
    ("IM", &["IM5", "IM6", "L6", "L7", "LS6", "LS7", "5", "6"]),
    // --- Geely/Volvo orbit ---
    ("LYNK & CO", &["01", "02", "03", "05", "06", "08", "09"]),
    // --- Korean ---
    (
        "KGM",
        &[
            "TORRES", "TORRES EVX", "KORANDO", "TIVOLI", "REXTON", "MUSSO", "ACTYON",
        ],
    ),
    // --- European microcars / quadricycles / small EVs ---
    (
        "AIXAM",
        &["CITY", "CROSSLINE", "COUPE", "MINAUTO", "CROSSOVER", "E-CITY"],
    ),
    ("LIGIER", &["JS50", "JS60", "MYLI", "JS50L"]),
    ("MICROLINO", &["LITE", "DOLCE", "PIONEER", "COMPETIZIONE"]),
    ("TAZZARI", &["ZERO"]),
    ("XEV", &["YOYO"]),
    ("ESTRIMA", &["BIRO"]),
    ("CHATENET", &["CH26", "CH32", "CH40", "SPORTEVO"]),
    ("DR", &["1", "3", "4", "5", "6", "7", "F35"]),
    ("EVO", &["3", "4", "5", "6"]),
    ("SWM", &["G01", "G03", "G05", "DOLCEVITA"]),
    // --- Mainstream-make top-ups: real models (mostly EU LCV/van + recent
    // EV/SUV) that the vPIC pattern index and the EU rip miss. Folded into
    // whatever the make already carries; models_for_make dedups against the EU
    // catalog, so a model already present via another source is a no-op. Keys
    // are canonical makes resolve_make already emits (WMI/rip-derived). ---
    (
        "ROLLS-ROYCE",
        &["PHANTOM", "GHOST", "WRAITH", "DAWN", "CULLINAN", "SPECTRE"],
    ),
    ("IVECO", &["DAILY"]),
    ("RENAULT", &["MASTER", "TRAFIC"]),
    (
        "CITROEN",
        &["C5", "C1", "JUMPER", "JUMPY", "SPACETOURER", "BERLINGO"],
    ),
    (
        "FORD",
        &[
            "TOURNEO CONNECT",
            "TOURNEO COURIER",
            "TOURNEO CUSTOM",
            "TRANSIT CUSTOM",
            "TRANSIT CONNECT",
        ],
    ),
    (
        "VOLKSWAGEN",
        &["CARAVELLE", "TRANSPORTER", "CRAFTER", "LT", "TAYRON"],
    ),
    ("FIAT", &["TALENTO"]),
    ("NISSAN", &["INTERSTAR"]),
    ("HYUNDAI", &["IX35", "STARIA", "INSTER"]),
    ("DACIA", &["BIGSTER"]),
    ("SKODA", &["ELROQ"]),
    ("NIO", &["EL6", "EL8"]),
    ("MG", &["EHS"]),
    ("GEELY", &["CITYRAY", "STARRAY", "GALAXY"]),
    ("BYD", &["SEALION 07"]),
];

/// Fold the curated [`EXTRA_MODELS`] supplement into a derived make→models index.
///
/// Keys are normalised the same way [`crate::Catalog::models_for_make`] looks
/// them up (`normalize_make`: uppercase, `-`/`_` → space, diacritics folded) so a
/// make like `"M-HERO"` files under `"M HERO"` and resolves on lookup. Models are
/// uppercased, deduped against any WMI-derived models, and the whole map is
/// returned key-sorted as FST insertion requires.
pub fn merge_make_models(base: Vec<(String, Vec<ModelRow>)>) -> Vec<(String, Vec<ModelRow>)> {
    let mut acc: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
    for (make, rows) in base {
        let set = acc.entry(make).or_default();
        for r in rows {
            set.insert(r.name);
        }
    }
    for (make, models) in EXTRA_MODELS {
        let key = crate::decoder::normalize_make(make);
        let set = acc.entry(key).or_default();
        for m in *models {
            set.insert(m.to_ascii_uppercase());
        }
    }
    acc.into_iter()
        .map(|(make, models)| {
            (
                make,
                models.into_iter().map(|name| ModelRow { name }).collect(),
            )
        })
        .collect()
}

/// Build the full set of FST/bin files plus the makes index and make_models reverse index.
pub fn build_all(
    wmi_make: &[(String, Vec<MakeRow>)],
    wmi_schema: &[(String, Vec<SchemaRow>)],
    schema_lookup: &[(String, Vec<LookupRow>)],
    out_dir: &Path,
) -> crate::Result<()> {
    std::fs::create_dir_all(out_dir)?;
    write_grouped(
        wmi_make,
        &out_dir.join(format!("{}.fst", MakeRow::base_name())),
        &out_dir.join(format!("{}.bin", MakeRow::base_name())),
    )?;
    write_grouped(
        wmi_schema,
        &out_dir.join(format!("{}.fst", SchemaRow::base_name())),
        &out_dir.join(format!("{}.bin", SchemaRow::base_name())),
    )?;
    write_grouped(
        schema_lookup,
        &out_dir.join(format!("{}.fst", LookupRow::base_name())),
        &out_dir.join(format!("{}.bin", LookupRow::base_name())),
    )?;

    let makes = collect_makes(wmi_make);
    write_set(&makes, &out_dir.join("makes.fst"))?;

    let make_models = merge_make_models(derive_make_models(wmi_make, wmi_schema, schema_lookup));
    write_grouped(
        &make_models,
        &out_dir.join(format!("{}.fst", ModelRow::base_name())),
        &out_dir.join(format!("{}.bin", ModelRow::base_name())),
    )?;
    Ok(())
}

/// Stream-build a typed map directly from a CSV file grouped by its leading column.
///
/// Avoids buffering the full CSV in memory — emits FST entries as soon as the
/// leading key changes. Caller must ensure the CSV is already sorted by the key column.
pub fn write_csv_grouped<T, F>(
    csv_path: &Path,
    fst_path: &Path,
    values_path: &Path,
    mut row_fn: F,
) -> crate::Result<()>
where
    T: RkyvSer,
    F: FnMut(&csv::StringRecord) -> Option<T>,
{
    let mut reader = csv::ReaderBuilder::new()
        .has_headers(true)
        .delimiter(b'\t')
        .quoting(false)
        .from_path(csv_path)?;
    let fst_writer = BufWriter::new(File::create(fst_path)?);
    let mut values_writer = BufWriter::new(File::create(values_path)?);
    let mut builder = MapBuilder::new(fst_writer).map_err(|e| Error::MissingData(e.to_string()))?;

    let mut offset: u64 = 0;
    let mut current_key: Option<String> = None;
    let mut group: Vec<T> = Vec::new();

    let flush = |builder: &mut MapBuilder<BufWriter<File>>,
                 values_writer: &mut BufWriter<File>,
                 offset: &mut u64,
                 key: &str,
                 group: &Vec<T>|
     -> crate::Result<()> {
        let bytes =
            rkyv::to_bytes::<RkyvError>(group).map_err(|e| Error::MissingData(e.to_string()))?;
        values_writer.write_all(&bytes)?;
        let combined = (*offset << 32) | (bytes.len() as u64);
        builder
            .insert(key, combined)
            .map_err(|e| Error::MissingData(format!("FST insert failed at key {key:?}: {e}")))?;
        *offset += bytes.len() as u64;
        Ok(())
    };

    for rec in reader.records() {
        let rec = rec.map_err(|e| Error::MissingData(e.to_string()))?;
        let Some(key) = rec.get(0) else { continue };
        let Some(item) = row_fn(&rec) else { continue };
        match &current_key {
            Some(k) if k == key => group.push(item),
            Some(k) => {
                flush(&mut builder, &mut values_writer, &mut offset, k, &group)?;
                group.clear();
                group.push(item);
                current_key = Some(key.to_string());
            }
            None => {
                current_key = Some(key.to_string());
                group.push(item);
            }
        }
    }
    if let Some(k) = current_key {
        flush(&mut builder, &mut values_writer, &mut offset, &k, &group)?;
    }
    builder
        .finish()
        .map_err(|e| Error::MissingData(e.to_string()))?;
    values_writer.flush()?;
    Ok(())
}

/// Build the full map suite from a directory of vPIC CSVs.
///
/// Expects `wmi_make.csv`, `wmi_schema_id.csv`, `schema_id_lookup.csv`, each
/// sorted by their first column. Optional EU rip files (`wmi_merged.tsv`,
/// `brands.tsv`, `brand_models.tsv`, `engines.tsv`) extend the output with
/// merged WMI metadata, the canonical brand set, and per-brand engine specs.
pub fn build_from_csv(csv_dir: &Path, out_dir: &Path) -> crate::Result<()> {
    std::fs::create_dir_all(out_dir)?;
    write_csv_grouped::<MakeRow, _>(
        &csv_dir.join("wmi_make.csv"),
        &out_dir.join(format!("{}.fst", MakeRow::base_name())),
        &out_dir.join(format!("{}.bin", MakeRow::base_name())),
        |rec| {
            rec.get(1).map(|s| MakeRow {
                name: s.to_string(),
                country: rec.get(2).unwrap_or("").to_string(),
                region: rec.get(3).unwrap_or("").to_string(),
            })
        },
    )?;
    write_csv_grouped::<SchemaRow, _>(
        &csv_dir.join("wmi_schema_id.csv"),
        &out_dir.join(format!("{}.fst", SchemaRow::base_name())),
        &out_dir.join(format!("{}.bin", SchemaRow::base_name())),
        |rec| rec.get(1).map(|s| SchemaRow { id: s.to_string() }),
    )?;
    write_csv_grouped::<LookupRow, _>(
        &csv_dir.join("schema_id_lookup.csv"),
        &out_dir.join(format!("{}.fst", LookupRow::base_name())),
        &out_dir.join(format!("{}.bin", LookupRow::base_name())),
        |rec| {
            let pattern = rec.get(1)?.to_string();
            let element = rec.get(2)?.to_string();
            let value = rec.get(3)?.to_string();
            let weight = rec.get(4).and_then(|s| s.parse().ok()).unwrap_or(0);
            Some(LookupRow {
                pattern,
                element,
                value,
                weight,
            })
        },
    )?;

    let wmi_make = read_csv_grouped(&csv_dir.join("wmi_make.csv"), |rec| {
        rec.get(1).map(|s| MakeRow {
            name: s.to_string(),
            country: rec.get(2).unwrap_or("").to_string(),
            region: rec.get(3).unwrap_or("").to_string(),
        })
    })?;
    let wmi_schema = read_csv_grouped(&csv_dir.join("wmi_schema_id.csv"), |rec| {
        rec.get(1).map(|s| SchemaRow { id: s.to_string() })
    })?;
    let schema_lookup = read_csv_grouped(&csv_dir.join("schema_id_lookup.csv"), |rec| {
        let pattern = rec.get(1)?.to_string();
        let element = rec.get(2)?.to_string();
        let value = rec.get(3)?.to_string();
        let weight = rec.get(4).and_then(|s| s.parse().ok()).unwrap_or(0);
        Some(LookupRow {
            pattern,
            element,
            value,
            weight,
        })
    })?;

    let makes = merge_makes(&collect_makes(&wmi_make));
    write_set(&makes, &out_dir.join("makes.fst"))?;
    let make_models = derive_make_models(&wmi_make, &wmi_schema, &schema_lookup);
    write_grouped(
        &make_models,
        &out_dir.join(format!("{}.fst", ModelRow::base_name())),
        &out_dir.join(format!("{}.bin", ModelRow::base_name())),
    )?;
    Ok(())
}

fn read_csv_grouped<T, F>(csv_path: &Path, mut row_fn: F) -> crate::Result<Vec<(String, Vec<T>)>>
where
    F: FnMut(&csv::StringRecord) -> Option<T>,
{
    let mut reader = csv::ReaderBuilder::new()
        .has_headers(true)
        .delimiter(b'\t')
        .quoting(false)
        // tolerate rows that omit trailing empty columns (e.g. make-only
        // catch-all rules `WMI\t\tMAKE` with no model) — get(n) yields None.
        .flexible(true)
        .from_path(csv_path)?;
    let mut acc: BTreeMap<String, Vec<T>> = BTreeMap::new();
    for rec in reader.records() {
        let rec = rec.map_err(|e| Error::MissingData(e.to_string()))?;
        let Some(key) = rec.get(0) else { continue };
        let Some(item) = row_fn(&rec) else { continue };
        acc.entry(key.to_string()).or_default().push(item);
    }
    Ok(acc.into_iter().collect())
}

impl From<csv::Error> for Error {
    fn from(e: csv::Error) -> Self {
        Error::MissingData(e.to_string())
    }
}

/// Build the EU/global rip output: merged WMI metadata, brand set, brand→model
/// index, and per-brand engine variants.
///
/// `rip_dir` must contain `wmi_merged.tsv`, `brands.tsv`, `brand_models.tsv`,
/// `engines.tsv` produced by the offline rip scripts. Output filenames match
/// the existing FST conventions (`wmi_make.{fst,bin}`, `makes.fst`,
/// `eu_brand_models.{fst,bin}`, `eu_engines.{fst,bin}`).
pub fn build_from_rip(rip_dir: &Path, out_dir: &Path) -> crate::Result<()> {
    std::fs::create_dir_all(out_dir)?;

    write_csv_grouped::<MakeRow, _>(
        &rip_dir.join("wmi_merged.tsv"),
        &out_dir.join(format!("{}.fst", MakeRow::base_name())),
        &out_dir.join(format!("{}.bin", MakeRow::base_name())),
        |rec| {
            rec.get(1).map(|s| MakeRow {
                name: s.to_string(),
                country: rec.get(2).unwrap_or("").to_string(),
                region: rec.get(3).unwrap_or("").to_string(),
            })
        },
    )?;

    let mut brands: Vec<String> = Vec::new();
    let mut brand_reader = csv::ReaderBuilder::new()
        .has_headers(true)
        .delimiter(b'\t')
        .quoting(false)
        .from_path(rip_dir.join("brands.tsv"))?;
    for rec in brand_reader.records() {
        let rec = rec.map_err(|e| Error::MissingData(e.to_string()))?;
        if let Some(b) = rec.get(0) {
            brands.push(b.to_string());
        }
    }
    let brands = merge_makes(&brands);
    write_set(&brands, &out_dir.join("makes.fst"))?;

    write_csv_grouped::<EuModelRow, _>(
        &rip_dir.join("brand_models.tsv"),
        &out_dir.join(format!("{}.fst", EuModelRow::base_name())),
        &out_dir.join(format!("{}.bin", EuModelRow::base_name())),
        |rec| {
            let name = rec.get(1)?.to_string();
            let first_year: u16 = rec.get(2).and_then(|s| s.parse().ok()).unwrap_or(0);
            let last_year: u16 = rec.get(3).and_then(|s| s.parse().ok()).unwrap_or(0);
            Some(EuModelRow {
                name,
                first_year,
                last_year,
            })
        },
    )?;

    write_csv_grouped::<EngineRow, _>(
        &rip_dir.join("engines.tsv"),
        &out_dir.join(format!("{}.fst", EngineRow::base_name())),
        &out_dir.join(format!("{}.bin", EngineRow::base_name())),
        engine_row_from_record,
    )?;

    let wmi_rules_path = rip_dir.join("wmi_rules.tsv");
    if wmi_rules_path.exists() {
        // Read rules grouped by 3-char WMI, sort each group by remainder
        // length DESC so the decoder can do longest-prefix-match.
        let rule_rows = read_csv_grouped(&wmi_rules_path, |rec| {
            let remainder = rec.get(1).unwrap_or("").to_string();
            let make = rec.get(2).unwrap_or("").to_string();
            let model = rec.get(3).unwrap_or("").to_string();
            Some(VinRuleRow {
                remainder,
                make,
                model,
            })
        })?;
        let mut sorted: Vec<(String, Vec<VinRuleRow>)> = rule_rows
            .into_iter()
            .map(|(wmi, mut rows)| {
                rows.sort_by_key(|r| std::cmp::Reverse(r.remainder.len()));
                (wmi, rows)
            })
            .collect();
        sorted.sort_by(|a, b| a.0.cmp(&b.0));
        write_grouped(
            &sorted,
            &out_dir.join(format!("{}.fst", VinRuleRow::base_name())),
            &out_dir.join(format!("{}.bin", VinRuleRow::base_name())),
        )?;
    }

    // Optional make+model → body-type map (header `brand\tmodel\tbody`). Keyed
    // by the raw col-0 brand, exactly like `eu_brand_models`. Within a brand we
    // collapse rows into one `BodyRow` per distinct model, whose `bodies` is the
    // sorted-deduped set of parsed EU type-approval codes.
    let body_models_path = rip_dir.join("body_models.tsv");
    if body_models_path.exists() {
        let mut acc: BTreeMap<String, BTreeMap<String, BTreeSet<String>>> = BTreeMap::new();
        let mut reader = csv::ReaderBuilder::new()
            .has_headers(true)
            .delimiter(b'\t')
            .quoting(false)
            .from_path(&body_models_path)?;
        for rec in reader.records() {
            let rec = rec.map_err(|e| Error::MissingData(e.to_string()))?;
            let (Some(brand), Some(model), Some(code)) = (rec.get(0), rec.get(1), rec.get(2)) else {
                continue;
            };
            let Some(body) = BodyType::from_code(code) else {
                continue;
            };
            acc.entry(brand.to_string())
                .or_default()
                .entry(model.to_string())
                .or_default()
                .insert(body.code().to_string());
        }
        let grouped: Vec<(String, Vec<BodyRow>)> = acc
            .into_iter()
            .map(|(brand, models)| {
                let rows = models
                    .into_iter()
                    .map(|(model, codes)| BodyRow {
                        model,
                        bodies: codes.iter().filter_map(|c| BodyType::from_code(c)).collect(),
                    })
                    .collect();
                (brand, rows)
            })
            .collect();
        write_grouped(
            &grouped,
            &out_dir.join(format!("{}.fst", BodyRow::base_name())),
            &out_dir.join(format!("{}.bin", BodyRow::base_name())),
        )?;
    }

    Ok(())
}

/// Map an `engines.tsv` record onto an `EngineRow`. The TSV is keyed by
/// `BRAND` (column 0) and embeds `MODEL` (column 1) inside the row so callers
/// can `engines_for(brand)` then filter by model.
fn engine_row_from_record(rec: &csv::StringRecord) -> Option<EngineRow> {
    let model = rec.get(1)?.to_string();
    let year: u16 = rec.get(2).and_then(|s| s.parse().ok()).unwrap_or(0);
    let name = rec.get(3).unwrap_or("").to_string();
    let cylinders = rec.get(4).unwrap_or("").to_string();
    let displacement_cm3: u32 = rec
        .get(5)
        .and_then(|s| s.parse::<f64>().ok())
        .map(|v| v.round() as u32)
        .unwrap_or(0);
    let power_kw: u32 = rec
        .get(6)
        .and_then(|s| s.parse::<f64>().ok())
        .map(|v| v.round() as u32)
        .unwrap_or(0);
    let power_hp: u32 = rec
        .get(7)
        .and_then(|s| s.parse::<f64>().ok())
        .map(|v| v.round() as u32)
        .unwrap_or(0);
    let torque_nm: u32 = rec
        .get(8)
        .and_then(|s| s.parse::<f64>().ok())
        .map(|v| v.round() as u32)
        .unwrap_or(0);
    let fuel = rec.get(9).unwrap_or("").to_string();
    let drive = rec.get(11).unwrap_or("").to_string();
    let gearbox = rec.get(12).unwrap_or("").to_string();
    Some(EngineRow {
        model,
        year,
        name,
        cylinders,
        displacement_cm3,
        power_kw,
        power_hp,
        torque_nm,
        fuel,
        drive,
        gearbox,
    })
}