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: 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 # A DGII request number may appear in more than one downloaded # authorization. Keep every NUM_AUTOR in its own workflow record so # importing a later authorization never mutates an active one. 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(), }]) 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) 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']