Turn a messy CSV into usable data

"""
Read a messy CSV: detect its dialect, normalise its encoding, coerce its
types, and write down every row it refuses.

Rung N0. Standard library only.

Three jobs, in that order, because each one needs the previous one done.
Bytes have to become text before a delimiter can be counted, and a delimiter
has to be known before a column can be typed.

The third job is the point of the whole snippet. A cleaner that silently drops
the rows it does not understand is worse than no cleaner at all: it hands back
a tidy file and hides the part you needed to look at. Every refusal here says
which line, which column, and why.
"""

import codecs
import csv
import io
import re
from datetime import date

DELIMITERS = (",", ";", "\t", "|")
SAMPLE_LINES = 20

# ---------------------------------------------------------------------------
# 1. Encoding
# ---------------------------------------------------------------------------

BOMS = (
    (codecs.BOM_UTF8, "utf-8-sig"),
    (codecs.BOM_UTF16_LE, "utf-16"),
    (codecs.BOM_UTF16_BE, "utf-16"),
)


def decode_text(data: bytes) -> str:
    """
    Turn bytes into text, guessing only when there is nothing else to go on.

    A byte order mark is a statement about the file, so it wins. Failing that,
    strict UTF-8 either succeeds, and then it is almost certainly right, or
    fails, and the file is one of the single-byte encodings a spreadsheet
    still exports. cp1252 is by far the most common of those.

    UTF-32 is left out on purpose: no spreadsheet writes it, and pretending to
    support an encoding you have never seen in a real file is how a cleaner
    acquires code nobody can test.
    """
    for bom, encoding in BOMS:
        if data.startswith(bom):
            return data.decode(encoding)
    try:
        return data.decode("utf-8")
    except UnicodeDecodeError:
        # Five byte values are undefined in cp1252; replacing them keeps the
        # rest of the file readable and marks the damage where it happened.
        return data.decode("cp1252", errors="replace")


# ---------------------------------------------------------------------------
# 2. Dialect
# ---------------------------------------------------------------------------


def _count_outside_quotes(line: str, delimiter: str, quote: str) -> int:
    """Count delimiters that separate fields, not those sitting inside one."""
    count, inside = 0, False
    for char in line:
        if char == quote:
            inside = not inside
        elif char == delimiter and not inside:
            count += 1
    return count


def detect_dialect(text: str) -> tuple[str, str]:
    """
    Guess the delimiter first, then the quote character.

    In that order, and not the other way round: counting delimiters needs to
    know what a quoted field looks like, but a quote character cannot be
    recognised without a delimiter to anchor it against. So the count assumes
    the usual double quote, and the quote character is then looked for with
    the delimiter in hand.

    The candidate that wins is the one whose count is the same on the most
    lines. A file separated by semicolons whose free-text column is full of
    commas still lands on its feet, because the comma count varies from line
    to line while the semicolon count does not.
    """
    sample = [line for line in text.split("\n")[:SAMPLE_LINES] if line.strip()]
    delimiter, best = ",", (-1.0, -1)
    for candidate in DELIMITERS:
        counts = [_count_outside_quotes(line, candidate, '"') for line in sample]
        if not counts or counts[0] == 0:
            continue  # absent from the header, so it separates nothing
        score = (sum(1 for c in counts if c == counts[0]) / len(counts), counts[0])
        if score > best:
            delimiter, best = candidate, score

    # A quote character only counts when it opens a field: at the start of a
    # line, or straight after the delimiter.
    opens = {}
    for quote in ('"', "'"):
        starts = sum(1 for line in sample if line.startswith(quote))
        opens[quote] = starts + sum(line.count(delimiter + quote) for line in sample)
    return delimiter, "'" if opens["'"] > opens['"'] else '"'


# ---------------------------------------------------------------------------
# 3. Types, and the journal of what did not fit
# ---------------------------------------------------------------------------

_SPACES = re.compile(r"[\s\u00a0\u202f]")
_INTEGER = re.compile(r"[+-]?\d+")
_NUMBER = re.compile(r"[+-]?(?:\d+\.?\d*|\.\d+)")
_ISO_DATE = re.compile(r"(\d{4})-(\d{2})-(\d{2})")
_DAY_FIRST = re.compile(r"(\d{2})[/.](\d{2})[/.](\d{4})")

TRUE_WORDS = frozenset({"true", "yes", "y", "1", "vrai", "oui", "o"})
FALSE_WORDS = frozenset({"false", "no", "n", "0", "faux", "non"})


def _to_integer(raw: str) -> int:
    text = _SPACES.sub("", raw)
    if not _INTEGER.fullmatch(text):
        raise ValueError("not an integer")
    return int(text)


def _to_number(raw: str) -> float:
    """
    Accept the decimal marks a European spreadsheet actually writes.

    When a comma and a dot are both present, the last one is the decimal mark
    and the other groups the thousands. When only a comma is present it is the
    decimal mark, which is the convention across most of the continent.
    """
    text = _SPACES.sub("", raw)
    if "," in text and "." in text:
        grouping = "," if text.rindex(".") > text.rindex(",") else "."
        text = text.replace(grouping, "")
    text = text.replace(",", ".")
    if not _NUMBER.fullmatch(text):
        raise ValueError("not a number")
    return float(text)


def _to_date(raw: str) -> str:
    """Return an ISO date, or refuse. Day-first is assumed outside ISO form."""
    text = raw.strip()
    iso, day_first = _ISO_DATE.fullmatch(text), _DAY_FIRST.fullmatch(text)
    if iso:
        year, month, day = iso.groups()
    elif day_first:
        day, month, year = day_first.groups()
    else:
        raise ValueError("not a date")
    try:
        date(int(year), int(month), int(day))  # rejects 31 February and month 13
    except ValueError as error:
        raise ValueError("not a real date") from error
    return f"{year}-{month}-{day}"


def _to_boolean(raw: str) -> bool:
    text = raw.strip().lower()
    if text in TRUE_WORDS:
        return True
    if text in FALSE_WORDS:
        return False
    raise ValueError("not a true or false value")


COERCERS = {
    "text": lambda raw: raw.strip(),
    "integer": _to_integer,
    "number": _to_number,
    "date": _to_date,
    "boolean": _to_boolean,
}


def _journal(line: int, column: str, reason: str, fields: list[str]) -> dict:
    """One entry of the journal: where, which column, why, and what was read."""
    return {"line": line, "column": column, "reason": reason, "fields": fields}


class Rejected(Exception):
    """One value the schema refuses, carrying the column and the reason."""

    def __init__(self, column: str, reason: str) -> None:
        super().__init__(f"{column}: {reason}")
        self.column = column
        self.reason = reason


def coerce_row(header: list[str], fields: list[str], schema: dict) -> dict:
    """
    Coerce one row, or refuse it at the first value that does not fit.

    First failure wins: a row is refused once, naming the column that caused
    it. Whoever repairs the file then has one thing to look at rather than a
    list of consequences.
    """
    row = {}
    for name, raw in zip(header, fields):
        value = raw.strip()
        if value == "":
            row[name] = None  # an empty cell is missing, not malformed
            continue
        try:
            row[name] = COERCERS[schema.get(name, "text")](value)
        except ValueError as error:
            raise Rejected(name, str(error)) from error
    return row


def clean_csv(data: bytes, schema: dict) -> dict:
    """
    Return the rows that survived, and a journal of everything refused.

    `data` is the raw bytes of the file and `schema` maps a column name to one
    of the keys of COERCERS. A column absent from the schema is kept as text.

    The journal is the whole point. Each entry carries the line, the column
    and the reason, plus the fields as they were read, so the refusal can be
    acted on without opening the file again. Hand it back to whoever produced
    the file, or to rung N3, which repairs those rows and only those.
    """
    text = decode_text(data)
    delimiter, quote = detect_dialect(text)
    reader = csv.reader(io.StringIO(text, newline=""), delimiter=delimiter, quotechar=quote)

    header, rows, rejects, previous = None, [], [], 0
    for fields in reader:
        line, previous = previous + 1, reader.line_num
        if not fields or fields == [""]:
            continue  # a blank line carries nothing, in any dialect
        if header is None:
            header = [name.strip() for name in fields]
            continue
        if len(fields) != len(header):
            plural = "" if len(header) == 1 else "s"
            reason = f"expected {len(header)} field{plural}, found {len(fields)}"
            rejects.append(_journal(line, "", reason, fields))
            continue
        try:
            rows.append(coerce_row(header, fields, schema))
        except Rejected as refusal:
            rejects.append(_journal(line, refusal.column, refusal.reason, fields))
    return {
        "columns": header or [],
        "delimiter": delimiter,
        "quote": quote,
        "rows": rows,
        "rejects": rejects,
    }

/**
 * Read a messy CSV: detect its dialect, normalise its encoding, coerce its
 * types, and write down every row it refuses.
 *
 * Rung N0. Node's standard library only, parser included.
 *
 * Three jobs, in that order, because each one needs the previous one done.
 * Bytes have to become text before a delimiter can be counted, and a delimiter
 * has to be known before a column can be typed.
 *
 * The third job is the point of the whole snippet. A cleaner that silently
 * drops the rows it does not understand is worse than no cleaner at all: it
 * hands back a tidy file and hides the part you needed to look at. Every
 * refusal here says which line, which column, and why.
 */

const DELIMITERS = [',', ';', '\t', '|'];
const SAMPLE_LINES = 20;

// ---------------------------------------------------------------------------
// 1. Encoding
// ---------------------------------------------------------------------------

const BOMS = [
  [[0xef, 0xbb, 0xbf], 'utf-8'],
  [[0xff, 0xfe], 'utf-16le'],
  [[0xfe, 0xff], 'utf-16be'],
];

// The five byte values cp1252 leaves undefined. Python's codec refuses them,
// so they are mapped to the replacement character here too: both versions of
// this snippet have to return the same text for the same bytes.
const UNDEFINED_IN_CP1252 = /[\u0081\u008d\u008f\u0090\u009d]/g;

/**
 * Turn bytes into text, guessing only when there is nothing else to go on.
 *
 * A byte order mark is a statement about the file, so it wins. Failing that,
 * strict UTF-8 either succeeds, and then it is almost certainly right, or
 * fails, and the file is one of the single-byte encodings a spreadsheet still
 * exports. cp1252 is by far the most common of those.
 *
 * UTF-32 is left out on purpose: no spreadsheet writes it, and pretending to
 * support an encoding you have never seen in a real file is how a cleaner
 * acquires code nobody can test.
 *
 * @param {Uint8Array} data
 * @returns {string}
 */
export function decodeText(data) {
  for (const [bom, encoding] of BOMS) {
    if (bom.every((byte, i) => data[i] === byte)) {
      // The decoder drops the mark itself, which is what we want.
      return new TextDecoder(encoding).decode(data);
    }
  }
  try {
    return new TextDecoder('utf-8', { fatal: true }).decode(data);
  } catch {
    return new TextDecoder('windows-1252').decode(data).replace(UNDEFINED_IN_CP1252, '\ufffd');
  }
}

// ---------------------------------------------------------------------------
// 2. Dialect
// ---------------------------------------------------------------------------

/** Count delimiters that separate fields, not those sitting inside one. */
function countOutsideQuotes(line, delimiter, quote) {
  let count = 0;
  let inside = false;
  for (const char of line) {
    if (char === quote) inside = !inside;
    else if (char === delimiter && !inside) count += 1;
  }
  return count;
}

/**
 * Guess the delimiter first, then the quote character.
 *
 * In that order, and not the other way round: counting delimiters needs to
 * know what a quoted field looks like, but a quote character cannot be
 * recognised without a delimiter to anchor it against. So the count assumes
 * the usual double quote, and the quote character is then looked for with the
 * delimiter in hand.
 *
 * The candidate that wins is the one whose count is the same on the most
 * lines. A file separated by semicolons whose free-text column is full of
 * commas still lands on its feet, because the comma count varies from line to
 * line while the semicolon count does not.
 *
 * @returns {{delimiter: string, quote: string}}
 */
export function detectDialect(text) {
  const sample = text.split('\n').slice(0, SAMPLE_LINES).filter((line) => line.trim() !== '');
  let delimiter = ',';
  let best = [-1, -1];
  for (const candidate of DELIMITERS) {
    const counts = sample.map((line) => countOutsideQuotes(line, candidate, '"'));
    if (counts.length === 0 || counts[0] === 0) continue; // separates nothing
    const agree = counts.filter((c) => c === counts[0]).length / counts.length;
    const score = [agree, counts[0]];
    if (score[0] > best[0] || (score[0] === best[0] && score[1] > best[1])) {
      delimiter = candidate;
      best = score;
    }
  }

  // A quote character only counts when it opens a field: at the start of a
  // line, or straight after the delimiter.
  const opens = {};
  for (const quote of ['"', "'"]) {
    opens[quote] = sample.reduce(
      (total, line) => total + (line.startsWith(quote) ? 1 : 0) + (line.split(delimiter + quote).length - 1),
      0,
    );
  }
  return { delimiter, quote: opens["'"] > opens['"'] ? "'" : '"' };
}

// ---------------------------------------------------------------------------
// 3. The parser
// ---------------------------------------------------------------------------

/**
 * Read records, with the line each one starts on.
 *
 * Quoted fields, doubled quotes, delimiters and newlines inside a quoted
 * field: written out because a CSV parser is a thirty-line state machine, not
 * a dependency. A quote only opens a field at the start of one, which is what
 * lets `a"b` stay the three characters somebody actually typed.
 *
 * @returns {Array<[number, string[]]>}
 */
export function parseRecords(text, delimiter, quote) {
  const records = [];
  let fields = [];
  let field = '';
  let quoted = false;
  let atFieldStart = true;
  let line = 1;
  let start = 1;

  for (let i = 0; i < text.length; i += 1) {
    const char = text[i];
    if (quoted) {
      if (char === quote && text[i + 1] === quote) {
        field += quote; // a doubled quote is one literal quote
        i += 1;
      } else if (char === quote) {
        quoted = false;
      } else {
        if (char === '\n') line += 1;
        field += char;
      }
    } else if (char === quote && atFieldStart) {
      quoted = true;
      atFieldStart = false;
    } else if (char === delimiter) {
      fields.push(field);
      field = '';
      atFieldStart = true;
    } else if (char === '\n' || char === '\r') {
      if (char === '\r' && text[i + 1] === '\n') i += 1;
      fields.push(field);
      records.push([start, fields]);
      fields = [];
      field = '';
      atFieldStart = true;
      line += 1;
      start = line;
    } else {
      field += char;
      atFieldStart = false;
    }
  }
  // A file that does not end with a newline still ends with a record.
  if (field !== '' || fields.length > 0) {
    fields.push(field);
    records.push([start, fields]);
  }
  return records;
}

// ---------------------------------------------------------------------------
// 4. Types, and the journal of what did not fit
// ---------------------------------------------------------------------------

const SPACES = /[\s\u00a0\u202f]/g;
const INTEGER = /^[+-]?\d+$/;
const NUMBER = /^[+-]?(?:\d+\.?\d*|\.\d+)$/;
const ISO_DATE = /^(\d{4})-(\d{2})-(\d{2})$/;
const DAY_FIRST = /^(\d{2})[/.](\d{2})[/.](\d{4})$/;
const MONTH_LENGTHS = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];

const TRUE_WORDS = new Set(['true', 'yes', 'y', '1', 'vrai', 'oui', 'o']);
const FALSE_WORDS = new Set(['false', 'no', 'n', '0', 'faux', 'non']);

function toInteger(raw) {
  const text = raw.replace(SPACES, '');
  if (!INTEGER.test(text)) throw new TypeError('not an integer');
  return Number(text);
}

/**
 * Accept the decimal marks a European spreadsheet actually writes.
 *
 * When a comma and a dot are both present, the last one is the decimal mark
 * and the other groups the thousands. When only a comma is present it is the
 * decimal mark, which is the convention across most of the continent.
 */
function toNumber(raw) {
  let text = raw.replace(SPACES, '');
  if (text.includes(',') && text.includes('.')) {
    const grouping = text.lastIndexOf('.') > text.lastIndexOf(',') ? ',' : '.';
    text = text.split(grouping).join('');
  }
  text = text.split(',').join('.');
  if (!NUMBER.test(text)) throw new TypeError('not a number');
  return Number(text);
}

function daysInMonth(year, month) {
  const leap = (year % 4 === 0 && year % 100 !== 0) || year % 400 === 0;
  return month === 2 && leap ? 29 : MONTH_LENGTHS[month - 1];
}

/** Return an ISO date, or refuse. Day-first is assumed outside ISO form. */
function toDate(raw) {
  const iso = ISO_DATE.exec(raw.trim());
  const dayFirst = DAY_FIRST.exec(raw.trim());
  if (!iso && !dayFirst) throw new TypeError('not a date');
  const [year, month, day] = iso ? iso.slice(1) : [dayFirst[3], dayFirst[2], dayFirst[1]];
  // Built-in date objects roll 31 February over to 2 March instead of refusing
  // it, so the calendar is checked by hand.
  const [y, m, d] = [year, month, day].map(Number);
  if (y < 1 || m < 1 || m > 12 || d < 1 || d > daysInMonth(y, m)) {
    throw new TypeError('not a real date');
  }
  return `${year}-${month}-${day}`;
}

function toBoolean(raw) {
  const text = raw.trim().toLowerCase();
  if (TRUE_WORDS.has(text)) return true;
  if (FALSE_WORDS.has(text)) return false;
  throw new TypeError('not a true or false value');
}

export const COERCERS = {
  text: (raw) => raw.trim(),
  integer: toInteger,
  number: toNumber,
  date: toDate,
  boolean: toBoolean,
};

/** One value the schema refuses, carrying the column and the reason. */
export class Rejected extends Error {
  constructor(column, reason) {
    super(`${column}: ${reason}`);
    this.column = column;
    this.reason = reason;
  }
}

/**
 * Coerce one row, or refuse it at the first value that does not fit.
 *
 * First failure wins: a row is refused once, naming the column that caused it.
 * Whoever repairs the file then has one thing to look at rather than a list of
 * consequences.
 */
export function coerceRow(header, fields, schema) {
  const row = {};
  header.forEach((name, i) => {
    const value = (fields[i] ?? '').trim();
    const coerce = COERCERS[schema[name] ?? 'text'];
    if (!coerce) throw new Error(`unknown type ${schema[name]} for column ${name}`);
    // An empty cell is missing, not malformed.
    if (value === '') row[name] = null;
    else {
      try {
        row[name] = coerce(value);
      } catch (error) {
        throw new Rejected(name, error.message);
      }
    }
  });
  return row;
}

/**
 * Return the rows that survived, and a journal of everything refused.
 *
 * `data` is the raw bytes of the file and `schema` maps a column name to one
 * of the keys of COERCERS. A column absent from the schema is kept as text.
 *
 * The journal is the whole point. Each entry carries the line, the column and
 * the reason, plus the fields as they were read, so the refusal can be acted
 * on without opening the file again. Hand it back to whoever produced the
 * file, or to rung N3, which repairs those rows and only those.
 *
 * @param {Uint8Array} data
 * @param {Record<string,string>} schema
 */
export function cleanCsv(data, schema) {
  const text = decodeText(data);
  const { delimiter, quote } = detectDialect(text);

  let header = null;
  const rows = [];
  const rejects = [];
  for (const [line, fields] of parseRecords(text, delimiter, quote)) {
    // A blank line carries nothing, in any dialect.
    if (fields.length === 0 || (fields.length === 1 && fields[0] === '')) continue;
    if (header === null) {
      header = fields.map((name) => name.trim());
    } else if (fields.length !== header.length) {
      const plural = header.length === 1 ? '' : 's';
      const reason = `expected ${header.length} field${plural}, found ${fields.length}`;
      rejects.push({ line, column: '', reason, fields });
    } else {
      try {
        rows.push(coerceRow(header, fields, schema));
      } catch (refusal) {
        if (!(refusal instanceof Rejected)) throw refusal;
        rejects.push({ line, column: refusal.column, reason: refusal.reason, fields });
      }
    }
  }
  return { columns: header ?? [], delimiter, quote, rows, rejects };
}

"""
Guess what each column holds, instead of writing the schema by hand.

Rung N1. Rung N0 needs a schema: somebody has to declare that `joined` holds
dates and `amount` holds numbers. On a file with three columns that takes a
minute. On the two hundred columns of an export nobody documented, it does not
happen, and the file gets loaded as text.

The classifier never looks at a value on its own. It looks at eight traits of
a column taken as a whole — how often a value is all digits, how often it
carries a decimal mark, how many distinct values there are — and learns which
combination goes with which type. The training set is a few dozen columns
labelled by hand, which is an afternoon rather than a project.

The output is exactly the schema `clean_csv` of rung N0 takes as its second
argument. This rung replaces the typing, not the cleaning.
"""

import re

import numpy as np
from sklearn.linear_model import LogisticRegression

_SPACES = re.compile(r"[\s\u00a0\u202f]")
_ALL_DIGITS = re.compile(r"[+-]?\d+")
_DECIMAL = re.compile(r"[+-]?\d*[.,]\d+")
_DATE_SHAPE = re.compile(r"\d{2,4}[-/.]\d{1,2}[-/.]\d{2,4}")
_LETTER = re.compile(r"[^\W\d_]")

BOOLEAN_WORDS = frozenset(
    {"true", "false", "yes", "no", "y", "n", "0", "1", "vrai", "faux", "oui", "non"}
)
LONG_VALUE = 20  # a length past which a column reads as free text

FEATURE_NAMES = (
    "all digits",
    "decimal mark",
    "date shape",
    "contains a letter",
    "boolean word",
    "distinct ratio",
    "mean length",
    "empty ratio",
)


def column_features(values: list[str]) -> list[float]:
    """
    Describe one column with eight numbers, all between zero and one.

    Every trait is a proportion rather than a count, so a column sampled from
    ten rows and one sampled from ten thousand are described on the same
    scale. Blank values are set aside before the proportions are taken, and
    counted separately: a column that is mostly empty is a fact about the
    file, not about the type.
    """
    filled = [value.strip() for value in values if value.strip()]
    if not filled:
        return [0.0] * len(FEATURE_NAMES)
    bare = [_SPACES.sub("", value) for value in filled]
    count = len(filled)
    return [
        sum(1 for v in bare if _ALL_DIGITS.fullmatch(v)) / count,
        sum(1 for v in bare if _DECIMAL.fullmatch(v)) / count,
        sum(1 for v in bare if _DATE_SHAPE.fullmatch(v)) / count,
        sum(1 for v in filled if _LETTER.search(v)) / count,
        sum(1 for v in filled if v.lower() in BOOLEAN_WORDS) / count,
        len(set(filled)) / count,
        min(sum(len(v) for v in filled) / count, LONG_VALUE) / LONG_VALUE,
        (len(values) - count) / len(values),
    ]


def train(columns: list[list[str]], labels: list[str]):
    """
    `columns` is a list of column samples, `labels` the type name of each.

    The type names are the ones rung N0 coerces to: text, integer, number,
    date, boolean.
    """
    model = LogisticRegression(max_iter=1000, C=10.0)
    model.fit(np.array([column_features(column) for column in columns]), labels)
    return model


def classify(model, values: list[str]) -> str:
    """
    The type of one column.

    A column with nothing in it is text, and the model is not consulted:
    there is no evidence to weigh, and text is the type that loses nothing.
    """
    if not any(value.strip() for value in values):
        return "text"
    return str(model.predict(np.array([column_features(values)]))[0])


def infer_schema(model, header: list[str], rows: list[list[str]], sample: int = 200) -> dict:
    """
    Build the schema that `clean_csv` of rung N0 asks for.

    The whole file is not needed. A couple of hundred rows say as much about
    the shape of a column as a million do, and reading them costs nothing.
    """
    schema = {}
    for index, name in enumerate(header):
        values = [row[index] if index < len(row) else "" for row in rows[:sample]]
        schema[name] = classify(model, values)
    return schema

/**
 * Guess what each column holds, instead of writing the schema by hand.
 *
 * Rung N1. Rung N0 needs a schema: somebody has to declare that `joined` holds
 * dates and `amount` holds numbers. On a file with three columns that takes a
 * minute. On the two hundred columns of an export nobody documented, it does
 * not happen, and the file gets loaded as text.
 *
 * The classifier never looks at a value on its own. It looks at eight traits
 * of a column taken as a whole — how often a value is all digits, how often it
 * carries a decimal mark, how many distinct values there are — and learns
 * which combination goes with which type. The training set is a few dozen
 * columns labelled by hand, which is an afternoon rather than a project.
 *
 * Written out rather than pulled from a library, because logistic regression
 * on eight features is thirty lines. That is the argument of this rung: the
 * classical tool is small enough to read.
 *
 * The output is exactly the schema `cleanCsv` of rung N0 takes as its second
 * argument. This rung replaces the typing, not the cleaning.
 */

const SPACES = /[\s\u00a0\u202f]/g;
const ALL_DIGITS = /^[+-]?\d+$/;
const DECIMAL = /^[+-]?\d*[.,]\d+$/;
const DATE_SHAPE = /^\d{2,4}[-/.]\d{1,2}[-/.]\d{2,4}$/;
const LETTER = /\p{L}/u;

const BOOLEAN_WORDS = new Set(
  ['true', 'false', 'yes', 'no', 'y', 'n', '0', '1', 'vrai', 'faux', 'oui', 'non'],
);
const LONG_VALUE = 20; // a length past which a column reads as free text

export const FEATURE_NAMES = [
  'all digits',
  'decimal mark',
  'date shape',
  'contains a letter',
  'boolean word',
  'distinct ratio',
  'mean length',
  'empty ratio',
];

/**
 * Describe one column with eight numbers, all between zero and one.
 *
 * Every trait is a proportion rather than a count, so a column sampled from
 * ten rows and one sampled from ten thousand are described on the same scale.
 * Blank values are set aside before the proportions are taken, and counted
 * separately: a column that is mostly empty is a fact about the file, not
 * about the type.
 */
export function columnFeatures(values) {
  const filled = values.map((value) => value.trim()).filter((value) => value !== '');
  if (filled.length === 0) return new Array(FEATURE_NAMES.length).fill(0);
  const bare = filled.map((value) => value.replace(SPACES, ''));
  const count = filled.length;
  const share = (list, test) => list.filter(test).length / count;
  return [
    share(bare, (v) => ALL_DIGITS.test(v)),
    share(bare, (v) => DECIMAL.test(v)),
    share(bare, (v) => DATE_SHAPE.test(v)),
    share(filled, (v) => LETTER.test(v)),
    share(filled, (v) => BOOLEAN_WORDS.has(v.toLowerCase())),
    new Set(filled).size / count,
    Math.min(filled.reduce((total, v) => total + v.length, 0) / count, LONG_VALUE) / LONG_VALUE,
    (values.length - count) / values.length,
  ];
}

/** One logistic regression, fitted by gradient descent on a tiny data set. */
function fitOne(rows, targets, epochs, rate) {
  const weights = new Float64Array(rows[0].length);
  let bias = 0;
  for (let epoch = 0; epoch < epochs; epoch += 1) {
    for (let i = 0; i < rows.length; i += 1) {
      let z = bias;
      for (let j = 0; j < weights.length; j += 1) z += weights[j] * rows[i][j];
      const error = 1 / (1 + Math.exp(-z)) - targets[i];
      for (let j = 0; j < weights.length; j += 1) weights[j] -= rate * error * rows[i][j];
      bias -= rate * error;
    }
  }
  return { weights, bias };
}

/**
 * `columns` is a list of column samples, `labels` the type name of each.
 *
 * The type names are the ones rung N0 coerces to: text, integer, number,
 * date, boolean. One regression is fitted per type, each answering "is it
 * this one?", and the strongest answer wins.
 */
export function train(columns, labels, { epochs = 600, rate = 0.5 } = {}) {
  const rows = columns.map(columnFeatures);
  const kinds = [...new Set(labels)].sort();
  const models = kinds.map((kind) =>
    fitOne(rows, labels.map((label) => (label === kind ? 1 : 0)), epochs, rate),
  );
  return { kinds, models };
}

/**
 * The type of one column.
 *
 * A column with nothing in it is text, and the model is not consulted: there
 * is no evidence to weigh, and text is the type that loses nothing.
 */
export function classify(model, values) {
  if (!values.some((value) => value.trim() !== '')) return 'text';
  const features = columnFeatures(values);
  let best = model.kinds[0];
  let bestScore = -Infinity;
  model.kinds.forEach((kind, k) => {
    const { weights, bias } = model.models[k];
    let z = bias;
    for (let j = 0; j < weights.length; j += 1) z += weights[j] * features[j];
    if (z > bestScore) {
      bestScore = z;
      best = kind;
    }
  });
  return best;
}

/**
 * Build the schema that `cleanCsv` of rung N0 asks for.
 *
 * The whole file is not needed. A couple of hundred rows say as much about the
 * shape of a column as a million do, and reading them costs nothing.
 */
export function inferSchema(model, header, rows, sample = 200) {
  const schema = {};
  header.forEach((name, index) => {
    const values = rows.slice(0, sample).map((row) => row[index] ?? '');
    schema[name] = classify(model, values);
  });
  return schema;
}

"""
Repair the rows rung N0 refused, and only those, by asking a model.

Rung N3. The word that makes this rung defensible on this entry is "only".
The file may hold a hundred thousand rows; the journal of rung N0 holds the
handful that did not fit. One call per refused row, and none at all for the
rest. Hand the whole file to a model instead and you have paid for the
ninety-nine per cent that a regular expression had already dealt with.

Note what this code has to do that rung N0 did not: build a prompt, retry a
failed call, parse an answer that is only probably valid JSON, and put the
answer back through the coercion of N0 before believing a word of it. That
plumbing is the real cost of this rung, and it is the part the tests have to
cover, because the model itself is not testable.
"""

from __future__ import annotations

import json

from n0 import Rejected, coerce_row

PROMPT = (
    "A row of a CSV file was refused by a type check. Repair it.\n"
    "Columns, in order, with the type each one expects:\n"
    "{schema}\n"
    "The row was refused because: {reason}\n"
    "Its fields, as they were read: {fields}\n\n"
    "Answer with JSON only: one object, one key per column, every value a\n"
    "string in the expected format. Dates are written YYYY-MM-DD. If the row\n"
    "cannot be repaired, answer with an empty object."
)


def repair_rejected_rows(header, rejects, schema, client=None, *, attempts: int = 3) -> dict:
    """
    Return the rows that were repaired, and the ones that were not.

    `rejects` is the journal returned by `clean_csv` of rung N0. Nothing else
    from the file is read, and the number of calls made is exactly the length
    of that journal.

    A row that cannot be repaired comes back in `unrepairable`, carrying its
    original line, column and fields, plus the reason the repair failed. It is
    never dropped: this rung exists because rung N0 refused to drop it either.

    `client` is injected so this function can be tested without a network
    call. In production it defaults to a real provider client.
    """
    if client is None:  # pragma: no cover - needs a key and a network
        from openai import OpenAI

        client = OpenAI()

    described = "\n".join(f"- {name}: {schema.get(name, 'text')}" for name in header)
    rows, unrepairable = [], []
    for reject in rejects:
        prompt = PROMPT.format(
            schema=described,
            reason=reject["reason"],
            fields=json.dumps(reject["fields"], ensure_ascii=False),
        )
        answer = _ask(client, prompt, attempts)
        if answer is None:
            unrepairable.append({**reject, "reason": "the model did not return a usable object"})
        elif not answer:
            unrepairable.append({**reject, "reason": "the model could not repair the row"})
        else:
            # The answer is only a proposal. It goes through the same coercion
            # every other row went through, and it is refused on the same terms.
            fields = [str(answer.get(name, "")) for name in header]
            try:
                rows.append(coerce_row(header, fields, schema))
            except Rejected as refusal:
                unrepairable.append(
                    {
                        **reject,
                        "column": refusal.column,
                        "reason": f"the repair was refused too: {refusal.reason}",
                    }
                )
    return {"rows": rows, "unrepairable": unrepairable}


def _ask(client, prompt: str, attempts: int):
    """
    Return the decoded object, or None when nothing usable came back.

    A failed call is retried; an unusable answer is not. At temperature zero
    the same prompt gives the same answer, so asking a second time buys
    nothing but a second bill.
    """
    for _ in range(attempts):
        try:
            answer = client.complete(prompt=prompt, temperature=0)
        except Exception:  # noqa: BLE001 - any provider failure is worth one more try
            continue
        try:
            parsed = json.loads(answer)
        except ValueError:
            return None
        return parsed if isinstance(parsed, dict) else None
    return None

/**
 * Repair the rows rung N0 refused, and only those, by asking a model.
 *
 * Rung N3. The word that makes this rung defensible on this entry is "only".
 * The file may hold a hundred thousand rows; the journal of rung N0 holds the
 * handful that did not fit. One call per refused row, and none at all for the
 * rest. Hand the whole file to a model instead and you have paid for the
 * ninety-nine per cent that a regular expression had already dealt with.
 *
 * Note what this code has to do that rung N0 did not: build a prompt, retry a
 * failed call, parse an answer that is only probably valid JSON, and put the
 * answer back through the coercion of N0 before believing a word of it. That
 * plumbing is the real cost of this rung, and it is the part the tests have to
 * cover, because the model itself is not testable.
 */

import { Rejected, coerceRow } from './n0.js';

const PROMPT = [
  'A row of a CSV file was refused by a type check. Repair it.',
  'Columns, in order, with the type each one expects:',
  '{schema}',
  'The row was refused because: {reason}',
  'Its fields, as they were read: {fields}',
  '',
  'Answer with JSON only: one object, one key per column, every value a',
  'string in the expected format. Dates are written YYYY-MM-DD. If the row',
  'cannot be repaired, answer with an empty object.',
].join('\n');

/**
 * Return the rows that were repaired, and the ones that were not.
 *
 * `rejects` is the journal returned by `cleanCsv` of rung N0. Nothing else
 * from the file is read, and the number of calls made is exactly the length of
 * that journal.
 *
 * A row that cannot be repaired comes back in `unrepairable`, carrying its
 * original line, column and fields, plus the reason the repair failed. It is
 * never dropped: this rung exists because rung N0 refused to drop it either.
 *
 * @param {string[]} header
 * @param {Array<object>} rejects the journal from rung N0
 * @param {Record<string,string>} schema
 * @param {object} options
 * @param {{complete: Function}} [options.client] injected so this can be
 *   tested without a network call; defaults to a real provider client
 * @param {number} [options.attempts]
 */
export async function repairRejectedRows(header, rejects, schema, { client, attempts = 3 } = {}) {
  if (!client) {
    // Needs a key and a network, so it is never reached in the tests.
    const { OpenAI } = await import('openai');
    client = new OpenAI();
  }

  const described = header.map((name) => `- ${name}: ${schema[name] ?? 'text'}`).join('\n');
  const rows = [];
  const unrepairable = [];

  for (const reject of rejects) {
    // The replacements are given as functions: a `$` in a field would
    // otherwise be read as a back-reference by the string form of `replace`.
    const prompt = PROMPT.replace('{schema}', () => described)
      .replace('{reason}', () => reject.reason)
      .replace('{fields}', () => JSON.stringify(reject.fields));
    const answer = await ask(client, prompt, attempts);

    if (answer === null) {
      unrepairable.push({ ...reject, reason: 'the model did not return a usable object' });
    } else if (Object.keys(answer).length === 0) {
      unrepairable.push({ ...reject, reason: 'the model could not repair the row' });
    } else {
      // The answer is only a proposal. It goes through the same coercion every
      // other row went through, and it is refused on the same terms.
      const fields = header.map((name) => String(answer[name] ?? ''));
      try {
        rows.push(coerceRow(header, fields, schema));
      } catch (refusal) {
        if (!(refusal instanceof Rejected)) throw refusal;
        unrepairable.push({
          ...reject,
          column: refusal.column,
          reason: `the repair was refused too: ${refusal.reason}`,
        });
      }
    }
  }
  return { rows, unrepairable };
}

/**
 * Return the decoded object, or null when nothing usable came back.
 *
 * A failed call is retried; an unusable answer is not. At temperature zero the
 * same prompt gives the same answer, so asking a second time buys nothing but
 * a second bill.
 */
async function ask(client, prompt, attempts) {
  for (let i = 0; i < attempts; i += 1) {
    let answer;
    try {
      answer = await client.complete({ prompt, temperature: 0 });
    } catch {
      continue; // any provider failure is worth one more try
    }
    try {
      const parsed = JSON.parse(answer);
      return parsed && typeof parsed === 'object' && !Array.isArray(parsed) ? parsed : null;
    } catch {
      return null;
    }
  }
  return null;
}