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.

import re
from typing import Dict, List, NoReturn, Optional, Set, Tuple

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}")


def _fail(row: int, message: str) -> NoReturn:
    raise ValueError(f"row {row}: {message}")


def _read_record(text: str, pos: int, row: int) -> Tuple[List[str], int]:
    """One RFC 4180 record from pos: its fields and where the next record starts."""
    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:
                    _fail(row, "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":
                _fail(row, "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] == '"':
                    _fail(row, "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:
            _fail(row, "a CR outside quotes must be followed by LF")


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


def _whole(row: int, column: str, text: str) -> int:
    if not _WHOLE.fullmatch(text):
        _fail(row, f'{column} must be a whole number, found "{text}"')
    return int(text)


def _read_todo(fields: List[str], at: List[int], row: int) -> Todo:
    if len(fields) != len(_COLUMNS):
        _fail(row, f"expected {len(_COLUMNS)} fields, found {len(fields)}")
    (id_, title, notes, done, priority, due, tags, frequency, interval, anchor, created_at, completed_at,
     order) = [fields[i] for i in at]
    if done not in ("true", "false"):
        _fail(row, f'done must be true or false, found "{done}"')
    filled = sum(1 for text in (frequency, interval, anchor) if text != "")
    recurrence: Optional[Recurrence] = None
    if filled == 3:
        recurrence = Recurrence(frequency=frequency, interval=_whole(row, "recurrenceInterval", interval), anchor=anchor)
    elif filled != 0:
        _fail(row, "fill in all three recurrence columns or leave them all empty")
    return 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=_whole(row, "order", order),
    )


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)."""
    text = csv[1:] if csv.startswith("") else csv
    todos: List[Todo] = []
    ids: Set[str] = set()
    at: Optional[List[int]] = None
    row = 0
    pos = 0
    size = len(text)
    while pos < size:
        if text[pos] == "\n":
            pos += 1
            continue
        if text.startswith("\r\n", pos):
            pos += 2
            continue
        row += 1
        fields, pos = _read_record(text, pos, row)
        if at is None:
            at = _read_header(fields)
            continue
        todo = _read_todo(fields, at, row)
        errors = validate_todo(todo).errors
        for field, message in errors.items():
            _fail(row, f"{field}: {message}")
        if todo.id in ids:
            _fail(row, f'duplicate id "{todo.id}"')
        ids.add(todo.id)
        todos.append(todo)
    if at is None:
        raise ValueError("the CSV has no header row")
    return todos