gizmoduck/ingestion/parsers.py

"""Pure, versioned source grammars. No filesystem or database access."""

import csv
import io
import re
import unicodedata
from collections import Counter
from dataclasses import dataclass
from types import MappingProxyType
from typing import Mapping


PARSER_VERSION = '1'
PROFILES = frozenset({'BANK_CSV_V1', 'OUROCARD_FULL_V1', 'OUROCARD_PARTIAL_V1'})
MAX_BYTES = 10 * 1024 * 1024
MAX_ITEMS = 20000


class ParseError(ValueError):
    def __init__(self, code):
        self.code = code
        super().__init__(code)


@dataclass(frozen=True)
class ParsedItem:
    locator: str
    section: str
    date_text: str = ''
    description: str = ''
    debit: str = ''
    credit: str = ''
    payment: bool = False
    errors: tuple[str, ...] = ()


@dataclass(frozen=True)
class ParseResult:
    encoding: str
    profile: str
    items: tuple[ParsedItem, ...]
    skipped: Mapping[str, int]
    context: Mapping[str, str]


def _header_key(value):
    return ''.join(c for c in unicodedata.normalize('NFD', value.strip().casefold()) if not unicodedata.combining(c))


def _balance(value):
    return bool(re.fullmatch(r'(?:S\s*A\s*L\s*D\s*O|SALDO ANTERIOR)(?:\s+[-\d.,]+)?', value.strip(), re.I))


def _csv_items(text, skipped):
    reader = csv.reader(io.StringIO(text, newline=''), strict=True)
    try:
        header = next(reader)
        aliases = ({'data', 'data movimento', 'data do movimento', 'data lancamento'}, {'historico', 'descricao', 'descricao do lancamento'}, {'valor', 'valor lancamento', 'valor do lancamento'})
        if len(header) < 6 or any(_header_key(header[i]) not in allowed for i, allowed in zip((0, 2, 5), aliases)):
            raise ParseError('INVALID_HEADER')
        while True:
            line = reader.line_num + 1
            try:
                row = next(reader)
            except StopIteration:
                return
            if not row or all(not cell.strip() for cell in row):
                skipped['blank'] += 1
                continue
            if len(row) != len(header):
                description = row[2] if len(row) > 2 else ''
                yield ParsedItem(f'line:{line}', '1', description=description,
                                 errors=('INVALID_ROW',))
            elif _balance(row[2]):
                skipped['balance'] += 1
            else:
                yield ParsedItem(f'line:{line}', '1', row[0].strip(), row[2], row[5].strip())
    except StopIteration:
        raise ParseError('INVALID_HEADER') from None
    except csv.Error:
        raise ParseError('INVALID_CSV') from None


def _txt_items(text, profile, skipped):
    partial = profile == 'OUROCARD_PARTIAL_V1'
    date_size, body_size = (5, 43) if partial else (10, 37)
    date_re = r'\d{2}/\d{2}' if partial else r'\d{2}\.\d{2}\.\d{4}'
    regular = [re.compile(rf'^({date_re})(.{{{body_size}}})([A-Z ]{{2}})\s+(\S+)\s+(\S+)\s*$')]
    if profile == 'OUROCARD_FULL_V1':
        regular.append(re.compile(rf'^({date_re})(.{{33}})([A-Za-z ]{{7}})\s+(\S+)\s+(\S+)\s*$'))
    payment = re.compile(rf'^({date_re})\s*(PGTO DEBITO CONTA.*?)200211\s+(\S+)\s+(\S+)\s*$')
    section, section_count = '1', 0
    for number, line in enumerate(text.splitlines(), 1):
        locator = f'line:{number}'
        if not line.strip():
            skipped['blank'] += 1
        elif re.match(r'^\s*\d\s*-\s*[A-Z]', line):
            section_count += 1
            section = str(section_count)
            skipped['section'] += 1
        elif _balance(line):
            skipped['balance'] += 1
        elif match := payment.fullmatch(line):
            yield ParsedItem(locator, section, match[1], match[2].strip(), match[3], match[4], True)
        elif match := next((candidate for pattern in regular if (candidate := pattern.fullmatch(line))), None):
            if _balance(match[2]):
                skipped['balance'] += 1
            else:
                yield ParsedItem(locator, section, match[1], match[2].rstrip(), match[4], match[5])
        else:
            code = 'INVALID_ROW' if re.match(r'^\s*\d{1,2}[./]', line) else 'UNRECOGNIZED_LINE'
            yield ParsedItem(locator, section, description=line, errors=(code,))


def parse_document(data: bytes, profile: str, context: dict | None = None) -> ParseResult:
    if profile not in PROFILES:
        raise ParseError('UNSUPPORTED_PROFILE')
    if not isinstance(data, bytes):
        raise ParseError('INVALID_INPUT')
    if len(data) > MAX_BYTES:
        raise ParseError('INPUT_TOO_LARGE')
    if b'\x00' in data:
        raise ParseError('NUL_BYTE')
    if context is not None and (not isinstance(context, dict) or any(not isinstance(v, str) for v in context.values())):
        raise ParseError('INVALID_DATE_CONTEXT')
    try:
        encoding = 'utf-8-sig' if data.startswith(b'\xef\xbb\xbf') else 'utf-8'
        text = data.decode(encoding)
    except UnicodeDecodeError:
        encoding, text = 'latin-1', data.decode('latin-1')
    skipped = Counter()
    source = _csv_items(text, skipped) if profile == 'BANK_CSV_V1' else _txt_items(text, profile, skipped)
    items = []
    for item in source:
        items.append(item)
        if len(items) > MAX_ITEMS:
            raise ParseError('TOO_MANY_ITEMS')
    return ParseResult(encoding, profile, tuple(items), MappingProxyType(dict(skipped)), MappingProxyType(dict(context or {})))