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_normalise_tags import normalise_tags _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])