//! **CSV (RFC 4180 + real-world dialects)** collector — feature `csv`.
//!
//! RFC 4180 is the baseline (CRLF records, doubled-quote `""` escaping, quoted
//! fields may contain the delimiter / CR / LF), but real Swedish + EU open-data
//! CSV routinely violates it, so this parser also tolerates: `;`/TAB delimiters
//! (EU locales use `,` as the decimal separator), a **UTF-8 BOM**, **Latin-1 /
//! ISO-8859-1** legacy encoding (åäö), **decimal-comma** (`1234,56`) with space /
//! NBSP thousands, bare `\n`/`\r` line ends, embedded newlines in quoted fields,
//! and ragged rows.
//!
//! CSV has **no external schema**, so korp adds the value: a typed **two-pass
//! sniff** — pass 1 samples the cells and picks, per column, the *narrowest* type
//! that parses every non-empty value (`Boolean → Int64 → Float64 → Date32 →
//! Timestamp(µs) → Utf8`); pass 2 builds the Arrow columns. Every column is
//! nullable; an empty cell is a null.
//!
//! The parse core [`parse_csv`] is pure (bytes in, Arrow out — no I/O), so the
//! fixture tests assert the round-tripped Arrow as DATA.
use std::path::Path;
use anyhow::{Result, anyhow};
use arrow_array::{ArrayRef, RecordBatch};
use arrow_schema::{DataType, Schema};
use super::arrow_util::{
batch, bool_col, date32_col, f64_col, i64_col, schema as build_schema, str_col, timestamp_col,
timestamp_type,
};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
// ── dialect ───────────────────────────────────────────────────────────────────
/// Source text encoding of a CSV file.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CsvEncoding {
/// UTF-8 (BOM stripped if present).
Utf8,
/// Latin-1 / ISO-8859-1 (decoded via `encoding_rs` WINDOWS-1252, an ISO-8859-1
/// superset that agrees on åäö) → transcoded to UTF-8.
Latin1,
/// **UTF-16, little-endian** — with or without a BOM.
///
/// Finansinspektionen's insider-register export is this, and serves **no BOM**
/// (measured live 2026-09-04). It is the encoding that most needs naming, because
/// `NUL` is a VALID UTF-8 byte: `std::str::from_utf8` SUCCEEDS on UTF-16LE ASCII
/// text, so a validity check answers "UTF-8", every character arrives followed by a
/// NUL, the first line's delimiter count is unchanged, and the file parses — into
/// column names like `P\0u\0b\0…`, with no error anywhere. A silent corruption,
/// not an unsupported format.
Utf16Le,
/// **UTF-16, big-endian** — with or without a BOM. Rarer, and the same trap in
/// mirror image (the NUL leads instead of trailing).
Utf16Be,
}
/// The dialect knobs a real-world CSV needs — all [`sniff_dialect`]-detectable and
/// overridable via `CollectorCfg.params`.
#[derive(Debug, Clone, Copy)]
pub struct CsvDialect {
/// Field delimiter (`,` / `;` / `\t`).
pub delimiter: u8,
/// Quote char (RFC-4180 `"`).
pub quote: u8,
/// Is row 0 a header (column names) rather than data?
pub has_headers: bool,
/// Source text encoding.
pub encoding: CsvEncoding,
/// EU decimal-comma dialect: `1234,56` is a float and space/NBSP are thousands.
pub decimal_comma: bool,
}
impl Default for CsvDialect {
fn default() -> Self {
Self {
delimiter: b',',
quote: b'"',
has_headers: true,
encoding: CsvEncoding::Utf8,
decimal_comma: false,
}
}
}
/// Strip a leading UTF-8 BOM (`EF BB BF`) if present.
fn strip_utf8_bom(bytes: &[u8]) -> &[u8] {
bytes.strip_prefix(&[0xEF, 0xBB, 0xBF]).unwrap_or(bytes)
}
/// Decode raw bytes to a UTF-8 `String` per the chosen encoding, stripping a BOM.
fn decode(bytes: &[u8], enc: CsvEncoding) -> String {
match enc {
CsvEncoding::Utf8 => String::from_utf8_lossy(strip_utf8_bom(bytes)).into_owned(),
CsvEncoding::Latin1 => {
encoding_rs::WINDOWS_1252.decode(strip_utf8_bom(bytes)).0.into_owned()
}
// `encoding_rs` strips the UTF-16 BOM itself and handles an odd trailing byte,
// so the raw bytes go in — stripping a UTF-8 BOM first would eat the first
// code unit of a BOM-less UTF-16 file.
CsvEncoding::Utf16Le => encoding_rs::UTF_16LE.decode(bytes).0.into_owned(),
CsvEncoding::Utf16Be => encoding_rs::UTF_16BE.decode(bytes).0.into_owned(),
}
}
/// **UTF-16 first, then UTF-8, then Latin-1** — and the order is the whole point.
///
/// A validity check cannot come first, because `NUL` is a valid UTF-8 byte:
/// `from_utf8` SUCCEEDS on UTF-16LE ASCII text, so "valid UTF-8 ⇒ UTF-8" reads
/// Finansinspektionen's insider export as UTF-8 and silently produces column names like
/// `P\0u\0b\0…`. So UTF-16 is decided by SHAPE before validity is consulted:
///
/// * an explicit BOM (`FF FE` / `FE FF`) settles it outright;
/// * otherwise, a run of ASCII text in UTF-16 is half NUL bytes at ALTERNATING
/// positions, which no real UTF-8 CSV is. The byte order follows from which side of
/// each pair the NUL sits on.
///
/// A file with no NULs at all never reaches either branch, so plain UTF-8 and Latin-1
/// are decided exactly as before.
fn sniff_encoding(bytes: &[u8]) -> CsvEncoding {
match bytes {
[0xFF, 0xFE, ..] => return CsvEncoding::Utf16Le,
[0xFE, 0xFF, ..] => return CsvEncoding::Utf16Be,
_ => {}
}
// Look at the head only: a 500 KB export is decided by its first line like every
// other property here, and a NUL that appears only deep inside a UTF-8 file is a
// corrupt byte rather than an encoding.
let head = &bytes[..bytes.len().min(1024)];
if head.len() >= 4 && head.iter().any(|b| *b == 0) {
let (mut even_nul, mut odd_nul) = (0usize, 0usize);
for (i, b) in head.iter().enumerate() {
if *b == 0 {
if i % 2 == 0 {
even_nul += 1;
} else {
odd_nul += 1;
}
}
}
let pairs = head.len() / 2;
// Two thirds of the pairs, not all of them: a Swedish export is mostly ASCII
// but `ä` is `E4 00` in UTF-16LE and contributes a NUL like everything else,
// while a genuinely non-Latin field would not — the threshold is what keeps
// this from needing the whole file to be ASCII.
let want = pairs * 2 / 3;
if odd_nul >= want && odd_nul > even_nul {
return CsvEncoding::Utf16Le;
}
if even_nul >= want && even_nul > odd_nul {
return CsvEncoding::Utf16Be;
}
}
if std::str::from_utf8(strip_utf8_bom(bytes)).is_ok() {
CsvEncoding::Utf8
} else {
CsvEncoding::Latin1
}
}
/// Pick the delimiter with the most *unquoted* occurrences on the first line;
/// ties fall back to `,`.
fn sniff_delimiter(first_line: &str) -> u8 {
let mut counts = [0usize; 3]; // comma, semicolon, tab
let mut in_quote = false;
for ch in first_line.chars() {
match ch {
'"' => in_quote = !in_quote,
',' if !in_quote => counts[0] += 1,
';' if !in_quote => counts[1] += 1,
'\t' if !in_quote => counts[2] += 1,
_ => {}
}
}
let best = counts
.iter()
.enumerate()
.max_by_key(|(_, c)| **c)
.unwrap()
.0;
if counts[best] == 0 {
b','
} else {
[b',', b';', b'\t'][best]
}
}
/// Auto-detect a dialect from the bytes (encoding → delimiter → decimal-comma →
/// header heuristic). Overridable per-field afterwards.
pub fn sniff_dialect(bytes: &[u8]) -> CsvDialect {
let encoding = sniff_encoding(bytes);
let text = decode(bytes, encoding);
let first_line = text.lines().next().unwrap_or("");
let delimiter = sniff_delimiter(first_line);
// EU convention: `;`-delimited files use decimal comma.
let decimal_comma = delimiter == b';';
let has_headers = infer_has_headers(&text, delimiter, decimal_comma);
CsvDialect {
delimiter,
quote: b'"',
has_headers,
encoding,
decimal_comma,
}
}
/// Read the raw string cells (no typing) from already-decoded text.
fn read_raw(text: &str, delimiter: u8) -> Result<Vec<Vec<Option<String>>>> {
let mut rdr = ::csv::ReaderBuilder::new()
.delimiter(delimiter)
.has_headers(false)
.flexible(true)
.from_reader(text.as_bytes());
let mut raw = Vec::new();
for rec in rdr.records() {
let rec = rec?;
raw.push(
rec.iter()
.map(|c| {
if c.is_empty() {
None
} else {
Some(c.to_string())
}
})
.collect(),
);
}
Ok(raw)
}
/// Header heuristic: infer the data rows' column types (skipping row 0); row 0 is
/// a **header** if, for some column whose data is a *specific* (non-Utf8) type,
/// row 0's cell does NOT parse as that type (i.e. it's a label). If no column
/// disambiguates (all-string table), assume a header is present (the common case).
fn infer_has_headers(text: &str, delimiter: u8, decimal_comma: bool) -> bool {
let raw = match read_raw(text, delimiter) {
Ok(r) => r,
Err(_) => return true,
};
if raw.len() < 2 {
return true;
}
let ncols = raw[0].len();
let data = &raw[1..];
let mut disambiguated = false;
for c in 0..ncols {
let ty = infer_col_type(data, c, decimal_comma);
if ty == DataType::Utf8 {
continue;
}
disambiguated = true;
let cell = raw[0].get(c).and_then(|v| v.as_ref());
let row0_is_data = cell
.map(|s| parses_as(s.trim(), &ty, decimal_comma))
.unwrap_or(false);
if !row0_is_data {
return true; // row 0 is a label → header
}
}
// all columns Utf8 (can't tell) → default to header present
!disambiguated
}
// ── typed sniff + parse primitives ────────────────────────────────────────────
fn parse_bool(s: &str) -> Option<bool> {
match s.to_ascii_lowercase().as_str() {
"true" | "1" | "yes" | "ja" | "y" => Some(true),
"false" | "0" | "no" | "nej" | "n" => Some(false),
_ => None,
}
}
fn parse_int(s: &str, decimal_comma: bool) -> Option<i64> {
let cleaned;
let t = if decimal_comma {
cleaned = s.replace('\u{00A0}', "").replace(' ', "");
cleaned.as_str()
} else {
s
};
t.parse::<i64>().ok()
}
fn parse_float(s: &str, decimal_comma: bool) -> Option<f64> {
if decimal_comma {
let t = s.replace('\u{00A0}', "").replace(' ', "").replace(',', ".");
t.parse::<f64>().ok()
} else {
s.parse::<f64>().ok()
}
}
/// Days since the Unix epoch (1970-01-01) for a civil date — Howard Hinnant's
/// algorithm, no `chrono` dep.
fn days_from_civil(y: i64, m: u32, d: u32) -> i32 {
let y = if m <= 2 { y - 1 } else { y };
let era = if y >= 0 { y } else { y - 399 } / 400;
let yoe = (y - era * 400) as i64;
let mp = if m > 2 { m - 3 } else { m + 9 } as i64;
let doy = (153 * mp + 2) / 5 + (d as i64 - 1);
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
(era * 146097 + doe - 719468) as i32
}
/// Parse `YYYY-MM-DD` or `DD/MM/YYYY` → days since epoch.
fn parse_date(s: &str) -> Option<i32> {
let (y, m, d) = if let Some((y, rest)) = s.split_once('-') {
let (mo, da) = rest.split_once('-')?;
(
y.parse::<i64>().ok()?,
mo.parse::<u32>().ok()?,
da.parse::<u32>().ok()?,
)
} else if let Some((da, rest)) = s.split_once('/') {
let (mo, y) = rest.split_once('/')?;
(
y.parse::<i64>().ok()?,
mo.parse::<u32>().ok()?,
da.parse::<u32>().ok()?,
)
} else {
return None;
};
if !(1..=12).contains(&m) || !(1..=31).contains(&d) {
return None;
}
Some(days_from_civil(y, m, d))
}
/// Parse `YYYY-MM-DDThh:mm:ss` (or a space separator, optional `Z`/fraction) →
/// microseconds since epoch (naive).
fn parse_timestamp(s: &str) -> Option<i64> {
let s = s.trim_end_matches('Z');
let (date, time) = s.split_once(['T', ' '])?;
let days = parse_date(date)? as i64;
let mut it = time.split(':');
let hh: i64 = it.next()?.parse().ok()?;
let mm: i64 = it.next()?.parse().ok()?;
let ss_frac = it.next().unwrap_or("0");
let (ss, frac) = ss_frac.split_once('.').unwrap_or((ss_frac, "0"));
let ss: i64 = ss.parse().ok()?;
// pad/truncate the fraction to microseconds (6 digits)
let mut micros_frac = format!("{frac:0<6}");
micros_frac.truncate(6);
let micros_frac: i64 = micros_frac.parse().ok()?;
if !(0..24).contains(&hh) || !(0..60).contains(&mm) || !(0..60).contains(&ss) {
return None;
}
let secs = days * 86_400 + hh * 3600 + mm * 60 + ss;
Some(secs * 1_000_000 + micros_frac)
}
fn parses_as(s: &str, ty: &DataType, decimal_comma: bool) -> bool {
match ty {
DataType::Boolean => parse_bool(s).is_some(),
DataType::Int64 => parse_int(s, decimal_comma).is_some(),
DataType::Float64 => parse_float(s, decimal_comma).is_some(),
DataType::Date32 => parse_date(s).is_some(),
DataType::Timestamp(_, _) => parse_timestamp(s).is_some(),
_ => true,
}
}
/// The narrowest Arrow type that parses every non-empty cell in column `col`.
fn infer_col_type(data: &[Vec<Option<String>>], col: usize, decimal_comma: bool) -> DataType {
let (mut is_bool, mut is_int, mut is_float, mut is_date, mut is_ts) =
(true, true, true, true, true);
let mut saw_value = false;
for row in data {
let Some(Some(raw)) = row.get(col) else {
continue;
};
let s = raw.trim();
if s.is_empty() {
continue;
}
saw_value = true;
is_bool &= parse_bool(s).is_some();
is_int &= parse_int(s, decimal_comma).is_some();
is_float &= parse_float(s, decimal_comma).is_some();
is_date &= parse_date(s).is_some();
is_ts &= parse_timestamp(s).is_some();
if !(is_bool || is_int || is_float || is_date || is_ts) {
break;
}
}
if !saw_value {
return DataType::Utf8; // all-null / all-empty column stays Utf8
}
if is_bool {
DataType::Boolean
} else if is_int {
DataType::Int64
} else if is_float {
DataType::Float64
} else if is_date {
DataType::Date32
} else if is_ts {
timestamp_type()
} else {
DataType::Utf8
}
}
/// Build one typed Arrow column from the raw cells.
fn build_column(
data: &[Vec<Option<String>>],
col: usize,
ty: &DataType,
decimal_comma: bool,
) -> ArrayRef {
let cell = |row: &Vec<Option<String>>| -> Option<String> {
row.get(col)
.and_then(|v| v.clone())
.filter(|s| !s.trim().is_empty())
};
match ty {
DataType::Boolean => {
let v: Vec<Option<bool>> = data
.iter()
.map(|r| cell(r).and_then(|s| parse_bool(s.trim())))
.collect();
bool_col(&v)
}
DataType::Int64 => {
let v: Vec<Option<i64>> = data
.iter()
.map(|r| cell(r).and_then(|s| parse_int(s.trim(), decimal_comma)))
.collect();
i64_col(&v)
}
DataType::Float64 => {
let v: Vec<Option<f64>> = data
.iter()
.map(|r| cell(r).and_then(|s| parse_float(s.trim(), decimal_comma)))
.collect();
f64_col(&v)
}
DataType::Date32 => {
let v: Vec<Option<i32>> = data
.iter()
.map(|r| cell(r).and_then(|s| parse_date(s.trim())))
.collect();
date32_col(&v)
}
DataType::Timestamp(_, _) => {
let v: Vec<Option<i64>> = data
.iter()
.map(|r| cell(r).and_then(|s| parse_timestamp(s.trim())))
.collect();
timestamp_col(&v)
}
_ => {
let v: Vec<Option<String>> = data.iter().map(cell).collect();
str_col(&v)
}
}
}
/// **The pure parse core**: bytes + dialect → (inferred Arrow schema, batches).
/// No I/O. A file with no data rows returns the schema and an empty batch list
/// (the collector turns that into a RAGNARÖK `reason`).
pub fn parse_csv(bytes: &[u8], dialect: &CsvDialect) -> Result<(Schema, Vec<RecordBatch>)> {
let text = decode(bytes, dialect.encoding);
let raw = read_raw(&text, dialect.delimiter)?;
if raw.is_empty() {
return Ok((Schema::empty(), vec![]));
}
let (header, data_start) = if dialect.has_headers {
(Some(&raw[0]), 1)
} else {
(None, 0)
};
let data = &raw[data_start..];
let ncols = match &header {
Some(h) => h.len(),
None => data.iter().map(|r| r.len()).max().unwrap_or(0),
};
let names: Vec<String> = (0..ncols)
.map(|i| match &header {
Some(h) => h
.get(i)
.and_then(|c| c.clone())
.filter(|s| !s.is_empty())
.unwrap_or_else(|| format!("col_{i}")),
None => format!("col_{i}"),
})
.collect();
let types: Vec<DataType> = (0..ncols)
.map(|c| infer_col_type(data, c, dialect.decimal_comma))
.collect();
let fdefs: Vec<(&str, DataType, bool)> = names
.iter()
.zip(&types)
.map(|(n, t)| (n.as_str(), t.clone(), true))
.collect();
let schema = build_schema(&fdefs);
if data.is_empty() {
return Ok((schema, vec![])); // header-only → 0 rows
}
let columns: Vec<ArrayRef> = (0..ncols)
.map(|c| build_column(data, c, &types[c], dialect.decimal_comma))
.collect();
let rb = batch(&schema, columns)?;
Ok((schema, vec![rb]))
}
// ── dialect from cfg overrides ────────────────────────────────────────────────
fn dialect_from_cfg(bytes: &[u8], cfg: &CollectorCfg) -> CsvDialect {
let mut d = sniff_dialect(bytes);
let p = &cfg.params;
if let Some(delim) = p.get("delimiter").and_then(|v| v.as_str()) {
d.delimiter = match delim {
"tab" | "\\t" | "\t" => b'\t',
other => other.bytes().next().unwrap_or(d.delimiter),
};
d.decimal_comma = d.delimiter == b';';
}
if let Some(enc) = p.get("encoding").and_then(|v| v.as_str()) {
d.encoding = match enc.to_ascii_lowercase().as_str() {
"latin1" | "iso-8859-1" | "iso8859-1" | "windows-1252" => CsvEncoding::Latin1,
_ => CsvEncoding::Utf8,
};
}
if let Some(h) = p.get("has_headers").and_then(|v| v.as_bool()) {
d.has_headers = h;
}
if let Some(dc) = p.get("decimal_comma").and_then(|v| v.as_bool()) {
d.decimal_comma = dc;
}
d
}
// ── collector ─────────────────────────────────────────────────────────────────
/// The RFC-4180 (+ dialects) CSV collector. `cfg.base_url` is the file source (a
/// local path or an `http(s)://` URL); `cfg.params` may override the sniffed
/// dialect. `gdpr` is per-collector (default [`GdprClass::Open`]; construct with
/// [`CsvCollector::personal`] for a personnummer/owner CSV).
pub struct CsvCollector {
gdpr: GdprClass,
/// The registry name. `"csv"` for the generic reader; a NAMED preset overrides it,
/// which is what makes a specific published CSV a first-class collector — with a
/// business key, a graph mapping and a human card — instead of an anonymous config
/// row against `"csv"`.
name: String,
/// The description the roster shows.
description: String,
/// The licence, when the preset knows it (`"per-source"` for the generic reader,
/// which genuinely cannot know).
license: String,
/// The GDPR-relevant join axes this file carries.
join_keys: Vec<String>,
/// The declared identity of a row, when the preset knows the file's own key.
business_key: Option<crate::history::BusinessKey>,
/// The declared graph, when this file's columns name a relationship (lane EC O7).
graph: Option<crate::graph_mapping::GraphMapping>,
/// The endpoint the preset reads, for `source_target()`.
endpoint: String,
/// The warehouse table it lands in.
table: String,
}
impl Default for CsvCollector {
fn default() -> Self {
Self {
gdpr: GdprClass::Open,
name: "csv".into(),
description: "RFC 4180 (+ dialects: ;/TAB, BOM, Latin-1, UTF-16, \
decimal-comma) CSV → typed Arrow via a two-pass schema sniff."
.into(),
license: "per-source".into(),
join_keys: Vec::new(),
business_key: None,
graph: None,
endpoint: String::new(),
table: "csv_import".into(),
}
}
}
impl CsvCollector {
/// A CSV collector whose payload carries personal data (legal-basis gated).
pub fn personal() -> Self {
Self {
gdpr: GdprClass::Personal,
..Default::default()
}
}
/// **Finansinspektionen's insider register (`Insyn`)** — every reported
/// transaction by a person in a leading position at a Swedish listed issuer, and
/// the second half of the join chain `gleif-lei` was built for (lane EC O6).
///
/// `fi_insyn.lei` → `gleif_lei.lei` → `gleif_lei.orgnr` → the `Company` node
/// Bolagsverket keys on `orgnr`. FI names the issuer by **LEI only** — no orgnr
/// anywhere in the export — which is exactly why the bridge had to exist first.
///
/// # Measured live 2026-09-04
///
/// `marknadssok.fi.se/publiceringsklient/sv-SE/Search/Search?SearchFunctionType=Insyn&button=export`
/// answers **HTTP 200, `text/csv`, 524 096 B, 1 000 rows + header**, `;`-delimited,
/// **23 columns** (the lane note said 22; the 23rd is the empty field after a
/// trailing `;`). The bytes are **UTF-16LE with NO byte-order mark** — the encoding
/// that made [`sniff_encoding`] a silent corruption until this row: `NUL` is valid
/// UTF-8, so the file parsed into columns named `P\0u\0b\0…` and one extra
/// phantom row, with nothing red.
///
/// # GDPR
///
/// `Personal`. `Anmälningsskyldig` and `Person i ledande ställning` are named
/// natural persons. It is a public register by law (MAR Article 19), which is a
/// legal basis and not an absence of one.
pub fn fi_insyn() -> Self {
use crate::graph_mapping::GraphMapping;
Self {
gdpr: GdprClass::Personal,
name: "fi-insyn".into(),
description:
"Finansinspektionen insynsregistret (MAR art. 19) — transactions by \
persons discharging managerial responsibility at Swedish listed \
issuers, keyed to the issuer by LEI."
.into(),
license: "Public register under MAR Article 19 / FFFS; Finansinspektionen \
publishes it openly, attribution expected"
.into(),
// `lei` is the join axis, and it is the ONLY identifier FI gives the issuer.
join_keys: vec!["LEI-kod".into()],
business_key: Some(crate::history::BusinessKey::of(
&["Transaktionsdatum", "LEI-kod", "Anmälningsskyldig", "ISIN", "Volym"],
"one reported transaction — FI assigns no id, so the identity is the \
report's own content: when, whose issuer, who reported, which \
instrument, how much. `Publiceringsdatum` is EXCLUDED: a correction \
(`Korrigering = Ja`) republishes the same transaction on a later date, \
and keying on it would land the corrected row beside the wrong one \
instead of over it",
)),
graph: Some(
GraphMapping::new(
"fi_insyn",
"Finansinspektionen — insynshandel",
"fi_insyn",
"FI names the issuer by LEI and by name, never by orgnummer — so the \
edge to a Company exists only through `gleif_lei`, and an issuer \
whose LEI is not in GLEIF's Swedish slice reaches no Company node. \
`Person` is keyed on a NAME, so two people who share one are one \
node; FI publishes no person identifier and korp will not invent \
one.",
)
.vertex(
"InsiderTrade",
"trade_id",
&[
"Transaktionsdatum",
"Publiceringsdatum",
"Karaktär",
"Instrumenttyp",
"Instrumentnamn",
"ISIN",
"Volym",
"Volymsenhet",
"Pris",
"Valuta",
"Handelsplats",
"Korrigering",
],
)
.derive_id(crate::graph_mapping::DerivedId::from_columns(
"trade_id",
&["Transaktionsdatum", "LEI-kod", "Anmälningsskyldig", "ISIN", "Volym"],
))
// The bridge: the edge lands on the `Lei` node `gleif-lei` declares,
// keyed the SAME way (`lei`), so the two collectors MERGE onto one node.
.edge("OF_ISSUER", ("InsiderTrade", "trade_id", "trade_id"), ("Lei", "LEI-kod", "lei"))
.edge("REPORTED_BY", ("InsiderTrade", "trade_id", "trade_id"), ("Person", "Anmälningsskyldig", "name"))
.edge("BY_INSIDER", ("InsiderTrade", "trade_id", "trade_id"), ("Person", "Person i ledande ställning", "name")),
),
endpoint: FI_INSYN_URL.to_string(),
table: "fi_insyn".into(),
..Default::default()
}
}
}
/// **Finansinspektionen's `Insyn` CSV export**, as one url.
///
/// Measured 2026-09-04: HTTP 200, `text/csv`, 524 096 B, 1 000 rows + header,
/// `;`-delimited, 23 columns, **UTF-16LE with no BOM**. `Page=1` is the first page of
/// FI's own paging — the export is not the whole register, and that is a limit this
/// collector states rather than hides.
pub const FI_INSYN_URL: &str = "https://marknadssok.fi.se/publiceringsklient/sv-SE/\
Search/Search?SearchFunctionType=Insyn&button=export&Page=1";
/// Read a local path (sync). Remote fetch is async and lives in `collect`.
fn read_local(src: &str) -> Result<Vec<u8>> {
if src.starts_with("http://") || src.starts_with("https://") {
return Err(anyhow!(
"schema() needs local bytes; remote CSV schema is derived in collect()"
));
}
Ok(std::fs::read(Path::new(src))?)
}
#[async_trait::async_trait]
impl Collector for CsvCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: self.name.clone(),
direction: Direction::Source,
source_kind: SourceKind::FlatFile,
path: CollectorPath::RustNative,
formats: vec!["text/csv".into(), "tsv".into()],
license: self.license.clone(),
gdpr: self.gdpr,
auth: AuthKind::None,
join_keys: self.join_keys.clone(),
description: self.description.clone(),
}
}
/// The preset's declared key, or the generic reader's honest append-only.
fn business_key(&self) -> crate::history::BusinessKey {
self.business_key.clone().unwrap_or_else(|| {
crate::history::BusinessKey::append_only(
"a CSV the operator supplied — korp does not know what identifies a row \
in someone else's file, and guessing one would collapse rows silently",
)
})
}
/// The preset's declared graph, or `None` for the generic reader.
fn graph_mapping(&self) -> Option<crate::graph_mapping::GraphMapping> {
self.graph.clone()
}
/// **There is no authority to name, and the two things that WILL bite are both
/// guesses this collector makes on the operator's behalf.**
///
/// CSV has no external schema and no publisher, so the card cannot answer "who
/// runs this" — the operator points `base_url` at their own file or URL. What it
/// can do, and what earns its place, is name the two sniff decisions that are
/// silently wrong often enough to cost an afternoon: `;` implying decimal-comma,
/// and an all-text table being assumed to have a header. Both are documented in
/// [`sniff_dialect`] and [`infer_has_headers`]; neither raises anything.
///
/// `terms_url` is korp-collectors' own repository, deliberately the narrower claim:
/// MIT-OR-Apache-2.0 covers the READER, and the data's licence came with the file.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit};
Some(
HumanDoc::new(
"no external authority — the operator supplies the file, so whoever \
published it is the authority and korp cannot name them",
"whoever supplied the file",
"https://git.gunnar.rs/nordisk/korp-collectors",
"https://www.rfc-editor.org/rfc/rfc4180",
)
.with_contact(Contact::NonePublished)
.with_rate_limit(RateLimit::Documented(
"none — the collector reads the whole file once per run. `base_url` may \
also be an `http(s)` URL, in which case the only limit is the hosting \
site's and korp does not know it."
.into(),
))
.with_quirk(
"THE DATA'S LICENCE TRAVELS WITH THE FILE. `terms_url` above is \
korp-collectors' own repository (MIT OR Apache-2.0) and covers the \
READER only. RFC 4180 in `docs_url` is the FORMAT baseline, not a \
permission — for either question, ask whoever published the CSV.",
)
.with_quirk(
"A `;` delimiter FORCES decimal-comma: `sniff_dialect` sets \
`decimal_comma = (delimiter == b';')` because that is the EU convention, \
and setting `params.delimiter` re-applies the same rule. A `;`-delimited \
file that uses `.` decimals therefore needs an explicit \
`params.decimal_comma: false`, which is read AFTER `params.delimiter` and \
so wins.",
)
.with_quirk(
"A table whose columns are ALL text is assumed to have a header row. \
`infer_has_headers` can only decide by finding a column with a specific \
(non-Utf8) type whose row 0 does not parse as that type; with nothing to \
disambiguate it defaults to \"header present\", and a headerless \
all-string CSV silently loses its first record. Set \
`params.has_headers: false` when you know.",
)
.with_quirk(
"The type sniff is per-column and picks the NARROWEST type that parses \
every non-empty value (`Boolean → Int64 → Float64 → Date32 → \
Timestamp(µs) → Utf8`) across EVERY data row. One stray cell decides the \
whole column, so a single `\"n/a\"` or `\"-\"` in an otherwise numeric \
column lands it as Utf8 — and a value that then fails to parse becomes a \
null, not an error.",
)
.with_quirk(
"Every column is nullable and an EMPTY CELL IS A NULL — there is no \
empty-string in the output. Ragged rows are tolerated (`flexible(true)`), \
so a short row yields nulls for the missing columns rather than an \
error, and a file that is header-only lands zero rows and comes back as \
the reason `CSV … had no data rows (empty or header-only)`.",
)
.with_quirk(
"`base_url` is EMPTY by default (`SourceTarget::warehouse(FlatFile, \"\", \
…)`), so an unconfigured collector reports `Missing::Source`. Note also \
that `schema()` refuses a remote URL — a remote CSV's schema is derived \
inside `collect()`, once the bytes are in hand.",
),
)
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(
SourceEndpoint::FlatFile,
self.endpoint.clone(),
self.table.clone(),
)
}
fn schema(&self, cfg: &CollectorCfg) -> Result<Schema> {
let bytes = read_local(&cfg.base_url)?;
let dialect = dialect_from_cfg(&bytes, cfg);
Ok(parse_csv(&bytes, &dialect)?.0)
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let bytes = crate::http::read_source(cfg, &cfg.base_url).await?;
let dialect = dialect_from_cfg(&bytes, cfg);
let (_schema, batches) = parse_csv(&bytes, &dialect)?;
let rows: u64 = batches.iter().map(|b| b.num_rows() as u64).sum();
if rows == 0 {
return Ok(CollectStats {
reason: Some(format!(
"CSV `{}` had no data rows (empty or header-only)",
cfg.base_url
)),
..Default::default()
});
}
let nbatches = batches.len() as u64;
for b in batches {
sink.push(b)?;
}
Ok(CollectStats {
rows,
batches: nbatches,
table: cfg.table.clone(),
reason: None,
})
}
}