ncf_manager/importer.py

import csv
import datetime
import hashlib
from html.parser import HTMLParser
import json
import re
import unicodedata
import xml.etree.ElementTree as ET

from trytond.exceptions import UserError
from trytond.i18n import gettext
from trytond.model import ModelView, fields
from trytond.pool import Pool
from trytond.transaction import Transaction
from trytond.wizard import Button, StateTransition, StateView, Wizard

from .ncf_manager import NCF_SEQUENCE_TYPE_IDS, ncf_padding, ncf_prefix


REQUIRED_COLUMNS = {
    'NUM_SOL', 'NUM_AUTOR', 'FECHA_AUTOR', 'OBSERVACIONES',
    'NUMERO_DESDE', 'NUMERO_HASTA', 'CODTIPO', 'FECHA_SOLICITUD',
    'ESTATUS', 'CANTIDAD_APROBADA',
}

SUPPLIER_NCF_TYPES = frozenset({'11', '13', '17', '41', '43', '47'})
MAX_IMPORT_SIZE = 5 * 1024 * 1024


def _text(value):
    return (value or '').replace('\xa0', ' ').strip()


def _normalized_name(value):
    value = unicodedata.normalize('NFKD', _text(value).upper())
    return ''.join(character for character in value
        if character.isalnum())


def _parse_datetime(value):
    value = _text(value)
    if not value:
        return None
    try:
        return datetime.datetime.fromisoformat(value).replace(tzinfo=None)
    except ValueError:
        pass
    for pattern in ('%d/%m/%Y %H:%M:%S', '%d/%m/%Y %H:%M'):
        try:
            return datetime.datetime.strptime(value, pattern)
        except ValueError:
            continue
    raise ValueError(f'Invalid date: {value}')


class _HTMLTableParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.rows = []
        self._row = None
        self._cell = None

    def handle_starttag(self, tag, attrs):
        if tag == 'tr':
            self._row = []
        elif tag in {'td', 'th'} and self._row is not None:
            self._cell = []

    def handle_data(self, data):
        if self._cell is not None:
            self._cell.append(data)

    def handle_endtag(self, tag):
        if tag in {'td', 'th'} and self._cell is not None:
            self._row.append(_text(''.join(self._cell)))
            self._cell = None
        elif tag == 'tr' and self._row:
            self.rows.append(self._row)
            self._row = None


def _parse_xml(data):
    root = ET.fromstring(data)
    rows = []
    for element in list(root):
        row = {
            child.tag.rsplit('}', 1)[-1]: _text(child.text)
            for child in list(element)
        }
        if row:
            rows.append(row)
    return rows


def _parse_html(data):
    parser = _HTMLTableParser()
    parser.feed(data.decode('utf-8-sig', errors='replace'))
    if len(parser.rows) < 2:
        return []
    header = parser.rows[0]
    return [dict(zip(header, row)) for row in parser.rows[1:]]


def _parse_csv(data):
    # DGII does not always quote DETALLE_LOG even when it contains commas.
    # Splitting at most len(header)-1 times preserves that final field.
    text = data.decode('utf-8-sig', errors='strict')
    lines = [line for line in text.splitlines() if line.strip()]
    if len(lines) < 2:
        return []
    header = next(csv.reader([lines[0]]))
    rows = []
    for line in lines[1:]:
        values = line.split(',', len(header) - 1)
        rows.append(dict(zip(header, values)))
    return rows


def parse_dgii_authorizations(data):
    data = bytes(data)
    if not data or len(data) > MAX_IMPORT_SIZE:
        raise UserError(gettext('ncf_manager.msg_dgii_import_invalid_size'))
    stripped = data.lstrip()
    try:
        if stripped.startswith(b'<'):
            if re.search(br'<\s*(?:\w+:)?SolicitudNCF\b', stripped[:500]):
                rows = _parse_xml(data)
                source_format = 'xml'
            elif re.search(br'<\s*(?:html|div|table)\b', stripped[:500], re.I):
                rows = _parse_html(data)
                source_format = 'xls_html'
            else:
                raise ValueError('Unsupported XML/HTML document')
        else:
            rows = _parse_csv(data)
            source_format = 'csv'
    except (UnicodeDecodeError, ET.ParseError, ValueError) as exception:
        raise UserError(gettext(
            'ncf_manager.msg_dgii_import_read_error', error=str(exception)))
    if not rows or not REQUIRED_COLUMNS.issubset(rows[0]):
        raise UserError(gettext('ncf_manager.msg_dgii_import_invalid_file'))
    return source_format, [{key: _text(value) for key, value in row.items()}
        for row in rows]


def normalize_dgii_row(row):
    if row['ESTATUS'].upper() != 'APROBADA':
        raise UserError(gettext(
            'ncf_manager.msg_dgii_import_not_approved',
            authorization=row['NUM_AUTOR']))
    ncf_type = row['CODTIPO'].lstrip('0')
    if ncf_type not in NCF_SEQUENCE_TYPE_IDS:
        raise UserError(gettext(
            'ncf_manager.msg_dgii_import_unknown_type', type=ncf_type))
    start = row['NUMERO_DESDE'].upper()
    end = row['NUMERO_HASTA'].upper()
    prefix = ncf_prefix(ncf_type)
    padding = ncf_padding(ncf_type)
    pattern = rf'{re.escape(prefix)}(\d{{{padding}}})'
    start_match = re.fullmatch(pattern, start)
    end_match = re.fullmatch(pattern, end)
    if not start_match or not end_match:
        raise UserError(gettext(
            'ncf_manager.msg_dgii_import_invalid_range',
            authorization=row['NUM_AUTOR']))
    from_number = int(start_match.group(1))
    until_number = int(end_match.group(1))
    approved = int(row['CANTIDAD_APROBADA'])
    if from_number < 1 or until_number < from_number:
        raise UserError(gettext('ncf_manager.msg_ncf_invalid_range'))
    if approved != until_number - from_number + 1:
        raise UserError(gettext(
            'ncf_manager.msg_dgii_import_quantity_mismatch',
            authorization=row['NUM_AUTOR']))
    try:
        requested = int(row.get('CANTIDAD_SOLICITADA') or 0) or None
        authorization_date = _parse_datetime(row['FECHA_AUTOR'])
        application_date = _parse_datetime(row['FECHA_SOLICITUD'])
        expiration = _parse_datetime(row.get('FECHA_VIGENCIA'))
    except ValueError as exception:
        raise UserError(gettext(
            'ncf_manager.msg_dgii_import_read_error', error=str(exception)))
    if not application_date or not authorization_date:
        raise UserError(gettext(
            'ncf_manager.msg_dgii_import_missing_dates',
            authorization=row['NUM_AUTOR']))
    normalized = {
        'application_number': row['NUM_SOL'],
        'authorization_number': row['NUM_AUTOR'],
        'application_date': application_date,
        'authorization_date': authorization_date,
        'company_name': row.get('OBSERVACIONES', ''),
        'ncf_type': ncf_type,
        'invoice_type': ('invoice_supplier'
            if ncf_type in SUPPLIER_NCF_TYPES else 'invoice_customer'),
        'from_number': from_number,
        'until_number': until_number,
        'approved_quantity': approved,
        'requested_quantity': requested,
        'expiration_date': expiration.date() if expiration else None,
        'usage_type': row.get('TIPO_USO', ''),
        'status': row['ESTATUS'].upper(),
        'detail': row.get('DETALLE_LOG', ''),
    }
    fingerprint_values = {key: value.isoformat()
        if isinstance(value, (datetime.date, datetime.datetime)) else value
        for key, value in normalized.items()
        if key not in {'detail', 'company_name'}}
    normalized['fingerprint'] = hashlib.sha256(json.dumps(
        fingerprint_values, ensure_ascii=False, sort_keys=True,
        separators=(',', ':')).encode()).hexdigest()
    return normalized


def import_dgii_authorizations(company, data, filename, activate=False):
    pool = Pool()
    Manager = pool.get('ncf.manager')
    Line = pool.get('ncf.manager.sequence')
    data = bytes(data)
    _source_format, rows = parse_dgii_authorizations(data)
    normalized_rows = [normalize_dgii_row(row) for row in rows]

    company_name = _normalized_name(company.party.name)
    for row in normalized_rows:
        supplied_name = _normalized_name(row['company_name'])
        if supplied_name and supplied_name != company_name:
            raise UserError(gettext(
                'ncf_manager.msg_dgii_import_company_mismatch',
                company=row['company_name']))

    file_hash = hashlib.sha256(data).hexdigest()
    managers_to_activate = []
    imported_lines = []
    for row in normalized_rows:
        managers = Manager.search([
            ('company', '=', company.id),
            ('app_number', '=', row['application_number']),
        ], limit=1)
        if managers:
            manager = managers[0]
        else:
            manager, = Manager.create([{
                'company': company.id,
                'app_number': row['application_number'],
                'app_date': row['application_date'].date(),
                'source_filename': filename,
                'source_hash': file_hash,
                'source_file': data,
                'imported_by': Transaction().user,
                'imported_at': datetime.datetime.now(),
            }])

        duplicates = Line.search([
            ('pos.company', '=', company.id),
            ('auth_number', '=', row['authorization_number']),
        ], limit=1)
        if duplicates:
            if duplicates[0].import_fingerprint != row['fingerprint']:
                raise UserError(gettext(
                    'ncf_manager.msg_dgii_import_conflict',
                    authorization=row['authorization_number']))
            continue
        if manager.state != 'draft':
            raise UserError(gettext(
                'ncf_manager.msg_dgii_import_closed_request',
                request=manager.app_number))
        line, = Line.create([{
            'pos': manager.id,
            'auth_number': row['authorization_number'],
            'authorization_date': row['authorization_date'],
            'ncf_type': row['ncf_type'],
            'invoice_type': row['invoice_type'],
            'sequence_mode': 'create',
            'from_number': row['from_number'],
            'number_next': row['from_number'],
            'until_number': row['until_number'],
            'expiration_date': row['expiration_date'],
            'dgii_status': row['status'],
            'requested_quantity': row['requested_quantity'],
            'approved_quantity': row['approved_quantity'],
            'usage_type': row['usage_type'],
            'dgii_detail': row['detail'],
            'import_fingerprint': row['fingerprint'],
            'source_filename': filename,
            'source_hash': file_hash,
            'source_file': data,
            'imported_by': Transaction().user,
            'imported_at': datetime.datetime.now(),
        }])
        imported_lines.append(line)
        if manager not in managers_to_activate:
            managers_to_activate.append(manager)

    if activate and managers_to_activate:
        Manager.activate(managers_to_activate)
    return managers_to_activate, imported_lines


class ImportNCFStart(ModelView):
    'Import DGII NCF Authorization'
    __name__ = 'ncf.import.start'

    company = fields.Many2One(
        'company.company', 'Company', required=True)
    file_ = fields.Binary('DGII File', required=True, filename='filename')
    filename = fields.Char('Filename', required=True)
    activate = fields.Boolean(
        'Activate Imported Authorization',
        help='Creates the strict sequence and activates it immediately. '
        'Leave disabled when numbers may already have been issued outside '
        'Tryton so the first number can be reviewed.')

    @staticmethod
    def default_company():
        return Transaction().context.get('company')

    @staticmethod
    def default_activate():
        return False


class ImportNCF(Wizard):
    'Import DGII NCF Authorization'
    __name__ = 'ncf.import'

    start = StateView('ncf.import.start',
        'ncf_manager.ncf_import_start_view_form', [
            Button('Cancel', 'end', 'tryton-cancel'),
            Button('Import', 'import_', 'tryton-ok', default=True),
        ])
    import_ = StateTransition()

    def transition_import_(self):
        import_dgii_authorizations(
            self.start.company, self.start.file_, self.start.filename,
            activate=self.start.activate)
        return 'end'


__all__ = [
    'ImportNCF', 'ImportNCFStart', 'import_dgii_authorizations',
    'normalize_dgii_row', 'parse_dgii_authorizations']