This commit is contained in:
José Arturo García
+342
@@ -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']
|
||||
Reference in New Issue
Block a user