"""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 datetime import date from calendar import monthrange 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.*?)\s+(\S+)\s+(\S+)\s*$') section, section_count = '1', 0 for number, line in enumerate(text.splitlines(), 1): locator = f'line:{number}' if partial and number == 21: try: _creation_date(line) except ParseError: pass else: skipped['metadata'] += 1 continue 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]) elif partial and not re.match(r'^\s*(?:\d{1,2}[./]|DATA\s+DE\s+EMISS)', line, re.I): skipped['metadata'] += 1 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 _decode_source(data: bytes) -> tuple[str, str]: 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') try: encoding = 'utf-8-sig' if data.startswith(b'\xef\xbb\xbf') else 'utf-8' return encoding, data.decode(encoding) except UnicodeDecodeError: return 'latin-1', data.decode('latin-1') def _creation_date(line: str) -> date: line = line.strip() match = re.fullmatch(r'DATA\s+DE\s+EMISS(?:A|Ã)O\s*:\s*(\d{2}/\d{2}/\d{4})', line, re.I) if match is None: match = re.fullmatch( r'(\d{2}/\d{2}/\d{4})\s{2,}[A-ZÀ-ÖØ-öø-ÿ -]+\s{2,}\d{2}:\d{2}:\d{2}', line, re.I, ) if match is None: raise ParseError('INVALID_CREATION_DATE') try: return date.fromisoformat('-'.join(reversed(match[1].split('/')))) except ValueError: raise ParseError('INVALID_CREATION_DATE') from None def partial_context_from_creation_line(data: bytes) -> dict[str, str]: """Build the bounded partial-statement interval from its line-21 issue date.""" _, text = _decode_source(data) lines = text.splitlines() if len(lines) < 21: raise ParseError('MISSING_CREATION_DATE') creation = _creation_date(lines[20]) start_year = creation.year if creation.month == 12 else creation.year - 1 start_month = creation.month % 12 + 1 try: start = date(start_year, start_month, 1) end = date(creation.year, creation.month, monthrange(creation.year, creation.month)[1]) except ValueError: raise ParseError('INVALID_CREATION_DATE') from None if (end - start).days > 366: raise ParseError('INVALID_CREATION_DATE') return {'start_date': start.isoformat(), 'end_date': end.isoformat()} def parse_document(data: bytes, profile: str, context: dict | None = None) -> ParseResult: if profile not in PROFILES: raise ParseError('UNSUPPORTED_PROFILE') 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') encoding, text = _decode_source(data) 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 {})))