todo.import-csv

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from typing import List

from .todo_import_csv_validate_todo_csv import read_todo_csv  ← validateTodoCsv, another function of this group · built into the same file, even by a slim install
from .todo_item import Todo  ← from todo.item ^1.0.0 · built alongside by fune


def import_csv(csv: str) -> List[Todo]:
    """Todos from CSV: 13 columns in any order, CRLF or LF, optional BOM; errors name the row (header = row 1)."""
    reading = read_todo_csv(csv)
    if reading.problems:
        raise ValueError(reading.problems[0].thrown)
    return reading.todos

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

import re
from dataclasses import dataclass
from typing import Dict, List, Optional, Set, Tuple, Union

from .todo_import_csv_types import CsvRowError, CsvValidation
from .todo_item import Recurrence, Todo, validate_todo  ← from todo.item ^1.0.0 · built alongside by fune
from .todo_normalise_tags import normalise_tags  ← from todo.normalise-tags ^1.0.0 · built alongside by fune

_COLUMNS = (
    "id", "title", "notes", "done", "priority", "due", "tags",
    "recurrenceFrequency", "recurrenceInterval", "recurrenceAnchor", "createdAt", "completedAt", "order",
)
_HEADER_RULE = "the header must have the columns " + ", ".join(_COLUMNS)
_WHOLE = re.compile(r"-?[0-9]{1,15}")


@dataclass(frozen=True)
class TodoCsvProblem:
    """One problem, and the message import_csv raises when it is the first."""

    error: CsvRowError
    thrown: str


@dataclass(frozen=True)
class TodoCsvReading:
    """What reading a whole file found: the good rows' todos, and every problem."""

    todos: List[Todo]
    rows: int
    problems: List[TodoCsvProblem]


def _record_at(text: str, pos: int) -> Union[Tuple[List[str], int], str]:
    """One RFC 4180 record from pos: its fields and where the next starts, or why it cannot be read."""
    fields: List[str] = []
    size = len(text)
    while True:
        if pos < size and text[pos] == '"':
            pos += 1
            parts: List[str] = []
            while True:
                quote = text.find('"', pos)
                if quote < 0:
                    return "a quoted field is not closed"
                parts.append(text[pos:quote])
                if quote + 1 < size and text[quote + 1] == '"':
                    parts.append('"')
                    pos = quote + 2
                else:
                    pos = quote + 1
                    break
            if pos < size and text[pos] not in ",\r\n":
                return "text after the closing quote of a field"
            value = "".join(parts)
        else:
            end = pos
            while end < size and text[end] not in ",\r\n":
                if text[end] == '"':
                    return "a field with a double quote in it must be quoted"
                end += 1
            value = text[pos:end]
            pos = end
        fields.append(value)
        if pos >= size:
            return fields, pos
        if text[pos] == ",":
            pos += 1
        elif text[pos] == "\n":
            return fields, pos + 1
        elif pos + 1 < size and text[pos + 1] == "\n":
            return fields, pos + 2
        else:
            return "a CR outside quotes must be followed by LF"


def _columns_at(names: List[str]) -> Union[List[int], str]:
    at: Dict[str, int] = {}
    for i, name in enumerate(names):
        if name not in _COLUMNS:
            return f'{_HEADER_RULE}: unknown column "{name}"'
        if name in at:
            return f'{_HEADER_RULE}: column "{name}" appears twice'
        at[name] = i
    for name in _COLUMNS:
        if name not in at:
            return f'{_HEADER_RULE}: missing column "{name}"'
    return [at[name] for name in _COLUMNS]


def _whole_number(text: str) -> Optional[int]:
    return int(text) if _WHOLE.fullmatch(text) else None


def read_todo_csv(csv: str) -> TodoCsvReading:
    """Reads the whole file, collecting every problem; the first is the one import_csv 1.0.0 raised."""
    text = csv[1:] if csv.startswith("") else csv
    todos: List[Todo] = []
    problems: List[TodoCsvProblem] = []
    ids: Set[str] = set()
    at: Optional[List[int]] = None
    row = 0
    rows = 0
    pos = 0
    size = len(text)

    def add(field: Optional[str], message: str, thrown: str) -> None:
        problems.append(TodoCsvProblem(error=CsvRowError(row=row, field=field, message=message), thrown=thrown))

    def plain(field: str, message: str) -> None:
        add(field, message, f"row {row}: {message}")

    while pos < size:
        if text[pos] == "\n":
            pos += 1
            continue
        if text.startswith("\r\n", pos):
            pos += 2
            continue
        row += 1
        record = _record_at(text, pos)
        if isinstance(record, str):
            add(None, record, f"row {row}: {record}")
            break
        fields, pos = record
        if at is None:
            header = _columns_at(fields)
            if isinstance(header, str):
                add(None, header, header)
                break
            at = header
            continue
        rows += 1
        if len(fields) != len(_COLUMNS):
            message = f"expected {len(_COLUMNS)} fields, found {len(fields)}"
            add(None, message, f"row {row}: {message}")
            continue
        before = len(problems)
        (id_, title, notes, done, priority, due, tags, frequency, interval, anchor, created_at, completed_at,
         order) = [fields[i] for i in at]

        done_read = done in ("true", "false")
        if not done_read:
            plain("done", f'done must be true or false, found "{done}"')
        filled = sum(1 for t in (frequency, interval, anchor) if t != "")
        recurrence: Optional[Recurrence] = None
        recurrence_read = True
        if filled == 3:
            every = _whole_number(interval)
            if every is None:
                plain("recurrenceInterval", f'recurrenceInterval must be a whole number, found "{interval}"')
                recurrence_read = False
            else:
                recurrence = Recurrence(frequency=frequency, interval=every, anchor=anchor)
        elif filled != 0:
            plain("recurrence", "fill in all three recurrence columns or leave them all empty")
            recurrence_read = False
        position = _whole_number(order)
        if position is None:
            plain("order", f'order must be a whole number, found "{order}"')

        todo = Todo(
            id=id_,
            title=title,
            notes=None if notes == "" else notes,
            done=done == "true",
            priority=priority,
            due=None if due == "" else due,
            tags=normalise_tags(tags.split(" ")),
            recurrence=recurrence,
            created_at=created_at,
            completed_at=None if completed_at == "" else completed_at,
            order=0 if position is None else position,
        )
        # A column that could not be read has already been reported; the
        # stand-in value must not raise a second, misleading message.
        errors = validate_todo(todo).errors
        for field, message in errors.items():
            if field == "completedAt" and not done_read:
                continue
            if field == "recurrence" and not recurrence_read:
                continue
            if field == "order" and position is None:
                continue
            add(field, message, f"row {row}: {field}: {message}")
        if "id" not in errors:
            if id_ in ids:
                plain("id", f'duplicate id "{id_}"')
            ids.add(id_)
        if len(problems) == before:
            todos.append(todo)
    if at is None and not problems:
        row = 1
        add(None, "the CSV has no header row", "the CSV has no header row")
    return TodoCsvReading(todos=todos, rows=rows, problems=problems)


def validate_todo_csv(csv: str) -> CsvValidation:
    """Every problem in a todo CSV file, row by row: the row (header = 1), the field and the message."""
    reading = read_todo_csv(csv)
    return CsvValidation(valid=not reading.problems, rows=reading.rows, errors=[p.error for p in reading.problems])