From 3d038bd9fe94e71a43ecb1dec919396a5fde3c81 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jose=CC=81=20Arturo=20Garci=CC=81a?= Date: Sun, 16 Aug 2026 14:09:42 -0400 Subject: [PATCH] Import DGII fiscal authorizations safely --- README.rst | 17 ++ __init__.py | 2 + importer.py | 342 ++++++++++++++++++++++++++++ importer.xml | 26 +++ locale/es.po | 72 ++++++ locale/es_419.po | 72 ++++++ message.xml | 39 ++++ ncf_manager.py | 105 ++++++++- tests/test_module.py | 57 +++++ tests/test_sequence_helpers.py | 34 +++ tryton.cfg | 4 + view/ncf_import_start_form.xml | 16 ++ view/ncf_manager_form.xml | 13 +- view/ncf_manager_sequences_form.xml | 32 +++ view/ncf_manager_sequences_tree.xml | 9 +- 15 files changed, 830 insertions(+), 10 deletions(-) create mode 100644 importer.py create mode 100644 importer.xml create mode 100644 view/ncf_import_start_form.xml diff --git a/README.rst b/README.rst index d02fa86..8cd3137 100644 --- a/README.rst +++ b/README.rst @@ -42,3 +42,20 @@ an authorization closes every remaining active line without reusing numbers. Electronic fiscal numbers are only reserved by this module. Activating an E-series authorization never signs or transmits an e-CF to DGII. + +Importing DGII authorizations +***************************** + +Use **NCF Manager → Import DGII Authorization** to load an authorization +downloaded from the DGII Virtual Office. XML, CSV and the HTML table exported +with an ``.xls`` extension are supported and detected from their content. +The importer groups lines by ``NUM_SOL`` and identifies each authorization by +``NUM_AUTOR``; file names are not used as fiscal identifiers. It validates +approval status, company name, NCF type, prefix, range and approved quantity, +and stores the original file, SHA-256, user and import date for audit. + +Imports are idempotent. Reimporting an equivalent CSV, XML or XLS record does +not create another line, while different fiscal data for an existing +authorization is rejected. The safe default leaves the result in draft so +the first number can be reviewed. Immediate activation is explicit and still +does not transmit electronic documents. diff --git a/__init__.py b/__init__.py index 00e440f..8d94fed 100644 --- a/__init__.py +++ b/__init__.py @@ -1,4 +1,5 @@ from . import invoice +from . import importer from . import ncf_manager from . import sequence from . import configuration @@ -8,6 +9,7 @@ from . import user __all__ = [ 'configuration', 'invoice', + 'importer', 'ncf_manager', 'party', 'sequence', diff --git a/importer.py b/importer.py new file mode 100644 index 0000000..551b8c9 --- /dev/null +++ b/importer.py @@ -0,0 +1,342 @@ +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'] diff --git a/importer.xml b/importer.xml new file mode 100644 index 0000000..a860fe0 --- /dev/null +++ b/importer.xml @@ -0,0 +1,26 @@ + + + + + + ncf.import.start + form + ncf_import_start_form + + + + Import DGII Authorization + ncf.import + + + + + + + + + + diff --git a/locale/es.po b/locale/es.po index 1970946..18b2cd1 100644 --- a/locale/es.po +++ b/locale/es.po @@ -605,3 +605,75 @@ msgstr "Finalizar secuencia" msgctxt "view:ncf.manager:" msgid "Authorized Fiscal Sequences" msgstr "Secuencias fiscales autorizadas" + +msgctxt "field:ncf.import.start,company:" +msgid "Company" +msgstr "Empresa" + +msgctxt "field:ncf.import.start,file_:" +msgid "DGII File" +msgstr "Archivo de la DGII" + +msgctxt "field:ncf.import.start,filename:" +msgid "Filename" +msgstr "Nombre del archivo" + +msgctxt "field:ncf.import.start,activate:" +msgid "Activate Imported Authorization" +msgstr "Activar autorización importada" + +msgctxt "field:ncf.manager,source_filename:" +msgid "Source Filename" +msgstr "Nombre del archivo fuente" + +msgctxt "field:ncf.manager,source_hash:" +msgid "Source SHA-256" +msgstr "SHA-256 del archivo fuente" + +msgctxt "field:ncf.manager,source_file:" +msgid "DGII Source File" +msgstr "Archivo fuente de la DGII" + +msgctxt "field:ncf.manager,imported_by:" +msgid "Imported By" +msgstr "Importado por" + +msgctxt "field:ncf.manager,imported_at:" +msgid "Imported At" +msgstr "Fecha de importación" + +msgctxt "field:ncf.manager.sequence,auth_number:" +msgid "Authorization Number" +msgstr "Número de autorización" + +msgctxt "field:ncf.manager.sequence,authorization_date:" +msgid "Authorization Date" +msgstr "Fecha de autorización" + +msgctxt "field:ncf.manager.sequence,dgii_status:" +msgid "DGII Status" +msgstr "Estado en DGII" + +msgctxt "field:ncf.manager.sequence,requested_quantity:" +msgid "Requested Quantity" +msgstr "Cantidad solicitada" + +msgctxt "field:ncf.manager.sequence,approved_quantity:" +msgid "Approved Quantity" +msgstr "Cantidad aprobada" + +msgctxt "field:ncf.manager.sequence,usage_type:" +msgid "Usage Type" +msgstr "Tipo de uso" + +msgctxt "field:ncf.manager.sequence,dgii_detail:" +msgid "DGII Detail" +msgstr "Detalle de la DGII" + +msgctxt "model:ir.action,name:act_ncf_import" +msgid "Import DGII Authorization" +msgstr "Importar autorización de la DGII" + +msgctxt "model:ir.ui.menu,name:menu_ncf_import" +msgid "Import DGII Authorization" +msgstr "Importar autorización de la DGII" diff --git a/locale/es_419.po b/locale/es_419.po index 7d72390..fcb3cc1 100644 --- a/locale/es_419.po +++ b/locale/es_419.po @@ -1120,3 +1120,75 @@ msgstr "Finalizar secuencia" msgctxt "view:ncf.manager:" msgid "Authorized Fiscal Sequences" msgstr "Secuencias fiscales autorizadas" + +msgctxt "field:ncf.import.start,company:" +msgid "Company" +msgstr "Empresa" + +msgctxt "field:ncf.import.start,file_:" +msgid "DGII File" +msgstr "Archivo de la DGII" + +msgctxt "field:ncf.import.start,filename:" +msgid "Filename" +msgstr "Nombre del archivo" + +msgctxt "field:ncf.import.start,activate:" +msgid "Activate Imported Authorization" +msgstr "Activar autorización importada" + +msgctxt "field:ncf.manager,source_filename:" +msgid "Source Filename" +msgstr "Nombre del archivo fuente" + +msgctxt "field:ncf.manager,source_hash:" +msgid "Source SHA-256" +msgstr "SHA-256 del archivo fuente" + +msgctxt "field:ncf.manager,source_file:" +msgid "DGII Source File" +msgstr "Archivo fuente de la DGII" + +msgctxt "field:ncf.manager,imported_by:" +msgid "Imported By" +msgstr "Importado por" + +msgctxt "field:ncf.manager,imported_at:" +msgid "Imported At" +msgstr "Fecha de importación" + +msgctxt "field:ncf.manager.sequence,auth_number:" +msgid "Authorization Number" +msgstr "Número de autorización" + +msgctxt "field:ncf.manager.sequence,authorization_date:" +msgid "Authorization Date" +msgstr "Fecha de autorización" + +msgctxt "field:ncf.manager.sequence,dgii_status:" +msgid "DGII Status" +msgstr "Estado en DGII" + +msgctxt "field:ncf.manager.sequence,requested_quantity:" +msgid "Requested Quantity" +msgstr "Cantidad solicitada" + +msgctxt "field:ncf.manager.sequence,approved_quantity:" +msgid "Approved Quantity" +msgstr "Cantidad aprobada" + +msgctxt "field:ncf.manager.sequence,usage_type:" +msgid "Usage Type" +msgstr "Tipo de uso" + +msgctxt "field:ncf.manager.sequence,dgii_detail:" +msgid "DGII Detail" +msgstr "Detalle de la DGII" + +msgctxt "model:ir.action,name:act_ncf_import" +msgid "Import DGII Authorization" +msgstr "Importar autorización de la DGII" + +msgctxt "model:ir.ui.menu,name:menu_ncf_import" +msgid "Import DGII Authorization" +msgstr "Importar autorización de la DGII" diff --git a/message.xml b/message.xml index 415b728..9b625b6 100644 --- a/message.xml +++ b/message.xml @@ -159,5 +159,44 @@ this repository contains the full copyright notices and license terms. --> Only draft fiscal authorizations can be deleted. + + Every fiscal sequence line requires its DGII authorization number. + + + The DGII file could not be read: %(error)s. + + + The file does not contain the required DGII authorization columns. + + + DGII authorization "%(authorization)s" is not approved and cannot be imported. + + + DGII NCF type "%(type)s" is not supported. + + + DGII authorization "%(authorization)s" has an invalid fiscal range or prefix. + + + The approved quantity for DGII authorization "%(authorization)s" does not match its range. + + + The DGII file belongs to "%(company)s", which does not match the selected company. + + + DGII authorization "%(authorization)s" already exists with different fiscal data. + + + DGII request "%(request)s" is no longer in draft and cannot receive another imported authorization. + + + DGII authorization number "%(authorization)s" is already registered for this company. + + + The DGII import file is empty or exceeds the 5 MB safety limit. + + + DGII authorization "%(authorization)s" does not contain the required request and authorization dates. + diff --git a/ncf_manager.py b/ncf_manager.py index b262d13..f928257 100644 --- a/ncf_manager.py +++ b/ncf_manager.py @@ -99,9 +99,16 @@ class NCFManager(Workflow, ModelSQL, ModelView): app_number = fields.Char('Application Number', required=True, size=10, states=_states, depends=_depends) - auth_number = fields.Char('Authorization Number', required=True, size=10, + auth_number = fields.Char('Legacy Authorization Number', size=10, states=_states, depends=_depends) + source_filename = fields.Char('Source Filename', readonly=True) + source_hash = fields.Char('Source SHA-256', readonly=True) + source_file = fields.Binary( + 'DGII Source File', filename='source_filename', readonly=True) + imported_by = fields.Many2One('res.user', 'Imported By', readonly=True) + imported_at = fields.DateTime('Imported At', readonly=True) + ncf_sequences = fields.One2Many('ncf.manager.sequence', 'pos', 'NCF Sequences', context={'company': Eval('company', -1)}, states=_states, depends=_depends) @@ -188,9 +195,16 @@ class NCFManager(Workflow, ModelSQL, ModelView): }) def _validate_authorization(self): - if not self.app_number.strip() or not self.auth_number.strip(): + if not self.app_number.strip(): raise UserError(gettext( 'ncf_manager.msg_ncf_missing_authorization')) + for line in self.ncf_sequences: + if not line.auth_number: + if self.auth_number: + line.auth_number = self.auth_number + else: + raise UserError(gettext( + 'ncf_manager.msg_ncf_line_missing_authorization')) @classmethod @ModelView.button @@ -229,6 +243,11 @@ class NCFManager(Workflow, ModelSQL, ModelView): 'activated_at': None, 'finished_by': None, 'finished_at': None, + 'source_filename': None, + 'source_hash': None, + 'source_file': None, + 'imported_by': None, + 'imported_at': None, }) return super().copy(managers, default=default) @@ -254,6 +273,23 @@ class NCFManagerSequences(Workflow, ModelSQL, ModelView): ncf_type_string = ncf_type.translated('ncf_type') + auth_number = fields.Char( + 'Authorization Number', size=10, states=_states, depends=_depends) + authorization_date = fields.DateTime( + 'Authorization Date', states=_states, depends=_depends) + dgii_status = fields.Char('DGII Status', readonly=True) + requested_quantity = fields.Integer('Requested Quantity', readonly=True) + approved_quantity = fields.Integer('Approved Quantity', readonly=True) + usage_type = fields.Char('Usage Type', readonly=True) + dgii_detail = fields.Text('DGII Detail', readonly=True) + import_fingerprint = fields.Char('Import Fingerprint', readonly=True) + source_filename = fields.Char('Source Filename', readonly=True) + source_hash = fields.Char('Source SHA-256', readonly=True) + source_file = fields.Binary( + 'DGII Source File', filename='source_filename', readonly=True) + imported_by = fields.Many2One('res.user', 'Imported By', readonly=True) + imported_at = fields.DateTime('Imported At', readonly=True) + company = fields.Function( fields.Many2One('company.company', 'Company'), 'on_change_with_company') @@ -309,6 +345,24 @@ class NCFManagerSequences(Workflow, ModelSQL, ModelView): 'invisible': Eval('sequence_mode') != 'create', }, depends=_depends + ['sequence_mode']) + effective_from_number = fields.Function( + fields.Integer('Authorized From', states={ + 'invisible': Eval('sequence_mode') == 'create', + }), 'get_effective_sequence_values') + effective_until_number = fields.Function( + fields.Integer('Authorized Until', states={ + 'invisible': Eval('sequence_mode') == 'create', + }), 'get_effective_sequence_values') + effective_number_next = fields.Function( + fields.Integer('First Number to Issue', states={ + 'invisible': Eval('sequence_mode') == 'create', + }), + 'get_effective_sequence_values') + effective_expiration_date = fields.Function( + fields.Date('Expiration Date', states={ + 'invisible': Eval('sequence_mode') == 'create', + }), 'get_effective_sequence_values') + invoice_type = fields.Selection([ ('invoice_customer', 'Customer Invoice'), ('invoice_supplier', 'Supplier Invoice') @@ -364,6 +418,35 @@ class NCFManagerSequences(Workflow, ModelSQL, ModelView): and self.ncf_sequences.prefix != ncf_prefix(self.ncf_type)): self.ncf_sequences = None + @fields.depends( + 'ncf_sequences', 'from_number', 'until_number', 'number_next', + 'expiration_date') + def on_change_ncf_sequences(self): + if not self.ncf_sequences: + return + self.from_number = self.ncf_sequences.from_number + self.until_number = self.ncf_sequences.until_number + self.number_next = self.ncf_sequences.number_next + self.expiration_date = self.ncf_sequences.expiration_date + + @classmethod + def get_effective_sequence_values(cls, records, names): + values = {name: {} for name in names} + mapping = { + 'effective_from_number': 'from_number', + 'effective_until_number': 'until_number', + 'effective_number_next': 'number_next', + 'effective_expiration_date': 'expiration_date', + } + for record in records: + sequence = record.ncf_sequences + for name in names: + source = mapping[name] + value = (getattr(sequence, source, None) if sequence + else getattr(record, source, None)) + values[name][record.id] = value + return values + def get_rec_name(self, name): if not self.ncf_type_string: return '' @@ -372,9 +455,23 @@ class NCFManagerSequences(Workflow, ModelSQL, ModelView): @classmethod def validate_fields(cls, records, field_names): super().validate_fields(records, field_names) - if field_names & {'pos', 'ncf_sequences', 'ncf_type'}: + if field_names & { + 'pos', 'ncf_sequences', 'ncf_type', 'auth_number'}: for record in records: record._validate_fiscal_sequence() + record._validate_unique_authorization() + + def _validate_unique_authorization(self): + if not self.auth_number or not self.pos or not self.pos.company: + return + if self.__class__.search([ + ('id', '!=', self.id), + ('pos.company', '=', self.pos.company.id), + ('auth_number', '=', self.auth_number), + ], limit=1): + raise UserError(gettext( + 'ncf_manager.msg_ncf_authorization_number_duplicate', + authorization=self.auth_number)) def _validate_fiscal_sequence(self): sequence = self.ncf_sequences @@ -424,7 +521,7 @@ class NCFManagerSequences(Workflow, ModelSQL, ModelView): sequence_type = ModelData.get_id( 'ncf_manager', NCF_SEQUENCE_TYPE_IDS[self.ncf_type]) sequence, = Sequence.create([{ - 'name': f'{ncf_prefix(self.ncf_type)} - {self.pos.auth_number}', + 'name': f'{ncf_prefix(self.ncf_type)} - {self.auth_number}', 'sequence_type': sequence_type, 'company': self.pos.company.id, 'prefix': ncf_prefix(self.ncf_type), diff --git a/tests/test_module.py b/tests/test_module.py index c3e146f..af17586 100644 --- a/tests/test_module.py +++ b/tests/test_module.py @@ -7,6 +7,7 @@ import unittest from trytond.exceptions import UserError from trytond.modules.ncf_manager.sequence import Sequence from trytond.modules.ncf_manager.invoice import Invoice +from trytond.modules.ncf_manager.importer import import_dgii_authorizations from trytond.pool import Pool from trytond.tests.test_tryton import ModuleTestCase, with_transaction from trytond.transaction import Transaction @@ -164,6 +165,29 @@ class NcfManagerTestCase(ModuleTestCase): }]) return sequence + @staticmethod + def _dgii_xml(ncf_type='31', authorization='6005423851', + start='E310000000001', end='E310000000100', quantity='100'): + return f''' + + 6009912073 + {authorization} + 2026-08-07T13:45:02-04:00 + FACTURA ELECTRÓNICA + Empresa NCF + {start} + {end} + {ncf_type} + 2026-08-07T13:36:23-04:00 + APROBADA + {quantity} + ELECTRÓNICA + {quantity} + 2027-12-31T00:00:00-04:00 + Autorizada por DGII + + '''.encode() + @with_transaction() def test_customer_invoice_post_assigns_b01_ncf(self): 'Test full NCF assignment workflow on posted customer invoice' @@ -248,6 +272,12 @@ class NcfManagerTestCase(ModuleTestCase): self.assertEqual(line.ncf_sequences.prefix, 'B01') self.assertEqual(line.ncf_sequences.padding, 8) self.assertEqual(line.ncf_sequences.number_next, 100) + self.assertEqual(line.effective_from_number, 100) + self.assertEqual(line.effective_until_number, 109) + self.assertEqual(line.effective_number_next, 100) + self.assertEqual( + line.effective_expiration_date, + datetime.date(today.year, 12, 31)) self.assertEqual(line.availability, 'low') Manager.finish([manager]) @@ -256,6 +286,33 @@ class NcfManagerTestCase(ModuleTestCase): self.assertEqual(manager.ncf_sequences[0].state, 'finished') self.assertIsNotNone(manager.finished_at) + @with_transaction() + def test_import_dgii_authorization_is_idempotent(self): + pool = Pool() + Manager = pool.get('ncf.manager') + company = self._create_company() + data = self._dgii_xml() + + managers, lines = import_dgii_authorizations( + company, data, 'authorization.xml') + self.assertEqual(len(managers), 1) + self.assertEqual(len(lines), 1) + manager = managers[0] + line = lines[0] + self.assertEqual(manager.app_number, '6009912073') + self.assertEqual(line.auth_number, '6005423851') + self.assertEqual(line.ncf_type, '31') + self.assertEqual(line.from_number, 1) + self.assertEqual(line.until_number, 100) + self.assertEqual(line.expiration_date, datetime.date(2027, 12, 31)) + + managers, lines = import_dgii_authorizations( + company, data, 'authorization-again.xml') + self.assertEqual(managers, []) + self.assertEqual(lines, []) + self.assertEqual(Manager.search_count([ + ('app_number', '=', '6009912073')]), 1) + del ModuleTestCase diff --git a/tests/test_sequence_helpers.py b/tests/test_sequence_helpers.py index 776eeda..d6f4f7b 100644 --- a/tests/test_sequence_helpers.py +++ b/tests/test_sequence_helpers.py @@ -4,10 +4,24 @@ import unittest from trytond.modules.ncf_manager.ncf_manager import ( NCFManager, NCFManagerSequences) from trytond.modules.ncf_manager.sequence import Sequence +from trytond.modules.ncf_manager.importer import ( + normalize_dgii_row, parse_dgii_authorizations) class NcfManagerUnitTestCase(unittest.TestCase): + HEADERS = ( + 'NUM_SOL,NUM_AUTOR,FECHA_AUTOR,COMP,OBSERVACIONES,NUMERO_DESDE,' + 'NUMERO_HASTA,CANTIDAD,CODTIPO,FECHA_SOLICITUD,ESTATUS,' + 'CANTIDAD_SOLICITADA,TIPO_USO,RNC_IMPRENTA,IMPRENTA,' + 'CANTIDAD_APROBADA,FECHA_VIGENCIA,DETALLE_LOG') + + ROW = ( + '6009912073,6005423851,07/08/2026 13:45:02,CRÉDITO ELECTRÓNICO,' + 'SOLUTEMA SRL,E310000000001,E310000000100,,31,' + '07/08/2026 13:36:23,APROBADA,1000,ELECTRÓNICA,,,100,' + '31/12/2027 0:00:00,Aviso, con comas') + def test_manager_defaults_and_rec_name(self): self.assertEqual(NCFManager.default_state(), 'draft') manager = SimpleNamespace() @@ -44,3 +58,23 @@ class NcfManagerUnitTestCase(unittest.TestCase): sequence.sequence_type = SimpleNamespace(name='Unknown') self.assertIsNone(Sequence.on_change_with_prefix(sequence)) self.assertEqual(Sequence.on_change_with_padding(sequence), 0) + + def test_csv_parser_preserves_unquoted_detail(self): + source_format, rows = parse_dgii_authorizations( + f'{self.HEADERS}\r\n{self.ROW}\r\n'.encode()) + self.assertEqual(source_format, 'csv') + self.assertEqual(rows[0]['NUM_AUTOR'], '6005423851') + self.assertEqual(rows[0]['DETALLE_LOG'], 'Aviso, con comas') + normalized = normalize_dgii_row(rows[0]) + self.assertEqual(normalized['from_number'], 1) + self.assertEqual(normalized['until_number'], 100) + + def test_html_xls_parser(self): + cells = ''.join(f'{header}' + for header in self.HEADERS.split(',')) + values = self.ROW.split(',', len(self.HEADERS.split(',')) - 1) + row = ''.join(f'{value}' for value in values) + source_format, rows = parse_dgii_authorizations( + f'
{cells}{row}
'.encode()) + self.assertEqual(source_format, 'xls_html') + self.assertEqual(rows[0]['CODTIPO'], '31') diff --git a/tryton.cfg b/tryton.cfg index 80412ad..b8fcb89 100644 --- a/tryton.cfg +++ b/tryton.cfg @@ -9,6 +9,7 @@ xml: sequence.xml configuration.xml ncf_manager.xml + importer.xml invoice.xml party.xml user.xml @@ -26,5 +27,8 @@ model: party.PartyNcfType party.Party user.User + importer.ImportNCFStart +wizard: + importer.ImportNCF report: invoice.InvoiceReport diff --git a/view/ncf_import_start_form.xml b/view/ncf_import_start_form.xml new file mode 100644 index 0000000..34cd798 --- /dev/null +++ b/view/ncf_import_start_form.xml @@ -0,0 +1,16 @@ + + +
+