ncf_manager/ncf_manager.py

import datetime

# This file is part of ncf_manager module for Tryton.  The COPYRIGHT file at
# the top level of this repository contains the full copyright notices and
# license terms.
from trytond.exceptions import UserError
from trytond.i18n import gettext

from trytond.model import ModelView, ModelSQL, Workflow, fields, Unique
from trytond.pool import Pool
from trytond.pyson import Eval
from trytond.transaction import Transaction


NCF_TYPE = [
    ('', ''),
    ('1', 'Factura de Crédito Fiscal'),
    ('2', 'Factura de Consumo'),
    ('3', 'Nota de Débito'),
    ('4', 'Nota de Crédito'),
    ('11', 'Comprobante de Compras'),
    ('12', 'Registro Único de Ingresos'),
    ('13', 'Comprobante para Gastos Menores'),
    ('14', 'Comprobante de Regímenes Especiales'),
    ('15', 'Comprobante Gubernamental'),
    ('16', 'Comprobante para exportaciones'),
    ('17', 'Comprobante para Pagos al Exterior'),
    ('31', 'Factura de Crédito Fiscal Electrónica'),
    ('32', 'Factura de Consumo Electrónica'),
    ('33', 'Nota de Débito Electrónica'),
    ('34', 'Nota de Crédito Electrónica'),
    ('41', 'Compras Electrónico'),
    ('43', 'Gastos Menores Electrónico'),
    ('44', 'Regímenes Especiales Electrónico'),
    ('45', 'Gubernamental Electrónico'),
    ('46', 'Exportación Electrónico'),
    ('47', 'Pagos al Exterior Electrónico'),
    ('99', 'Factura sin NCF'),
]

ELECTRONIC_NCF_TYPES = frozenset({
    '31', '32', '33', '34', '41', '43', '44', '45', '46', '47'})

NCF_SEQUENCE_TYPE_IDS = {
    '1': 'sequence_type_out_invoice_01',
    '2': 'sequence_type_out_invoice_02',
    '3': 'sequence_type_out_invoice_03',
    '4': 'sequence_type_out_credit_note_04',
    '11': 'sequence_type_out_invoice_11',
    '12': 'sequence_type_out_invoice_12',
    '13': 'sequence_type_out_invoice_13',
    '14': 'sequence_type_out_invoice_14',
    '15': 'sequence_type_out_invoice_15',
    '16': 'sequence_type_out_invoice_16',
    '17': 'sequence_type_out_invoice_17',
    '31': 'sequence_type_out_invoice_31',
    '32': 'sequence_type_out_invoice_32',
    '33': 'sequence_type_out_invoice_33',
    '34': 'sequence_type_out_credit_note_34',
    '41': 'sequence_type_out_invoice_41',
    '43': 'sequence_type_out_invoice_43',
    '44': 'sequence_type_out_invoice_44',
    '45': 'sequence_type_out_invoice_45',
    '46': 'sequence_type_out_invoice_46',
    '47': 'sequence_type_out_invoice_47',
}


def ncf_prefix(ncf_type):
    if not ncf_type or ncf_type == '99':
        return None
    series = 'E' if ncf_type in ELECTRONIC_NCF_TYPES else 'B'
    return f'{series}{ncf_type.zfill(2)}'


def ncf_padding(ncf_type):
    if not ncf_type or ncf_type == '99':
        return 0
    return 10 if ncf_type in ELECTRONIC_NCF_TYPES else 8


class NCFManager(Workflow, ModelSQL, ModelView):

    'NCF Manager'
    __name__ = 'ncf.manager'

    _states = {
        'readonly': Eval('state') != 'draft',
    }

    _depends = ['state']

    company = fields.Many2One('company.company', 'Company', required=True,
                              states=_states, depends=_depends)

    app_date = fields.Date('Application Date', required=True,
                           states=_states, depends=_depends)

    app_number = fields.Char('Application Number', required=True, size=10,
                             states=_states, depends=_depends)

    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)

    state = fields.Selection([
        ('draft', 'Draft'),
        ('active', 'Active'),
        ('finished', 'Finished'),
    ], "State", readonly=True, required=True, sort=False)

    state_string = state.translated('state')

    activated_by = fields.Many2One(
        'res.user', 'Activated By', readonly=True)
    activated_at = fields.DateTime('Activated At', readonly=True)
    finished_by = fields.Many2One(
        'res.user', 'Finished By', readonly=True)
    finished_at = fields.DateTime('Finished At', readonly=True)

    del _states, _depends

    @classmethod
    def __setup__(cls):
        super(NCFManager, cls).__setup__()
        cls._order = [
            ('app_date', 'DESC'),
            ('id', 'DESC'),
        ]
        cls._transitions |= {
            ('draft', 'active'),
            ('active', 'finished'),
        }
        cls._buttons.update({
            'activate': {
                'invisible': Eval('state') != 'draft',
                'depends': ['state'],
            },
            'finish': {
                'invisible': Eval('state') != 'active',
                'depends': ['state'],
            },
        })

    def get_rec_name(self, name):
        parts = []
        if self.company:
            parts.append(self.company.rec_name)
        if self.app_number:
            parts.append(self.app_number)
        return ' - '.join(parts) if parts else str(self.id)

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

    @staticmethod
    def default_state():
        return 'draft'

    @classmethod
    @ModelView.button
    @Workflow.transition('active')
    def activate(cls, managers):
        Line = Pool().get('ncf.manager.sequence')
        now = datetime.datetime.now()
        user = Transaction().user
        lines = []
        for manager in managers:
            if not manager.ncf_sequences:
                raise UserError(gettext(
                    'ncf_manager.msg_ncf_manager_without_sequences'))
            manager._validate_authorization()
            for line in manager.ncf_sequences:
                line.prepare_sequence()
                line._validate_fiscal_sequence()
                line._validate_no_overlap()
                lines.append(line)
        if lines:
            Line.save(lines)
            Line.activate(lines)
        cls.write(managers, {
            'activated_by': user,
            'activated_at': now,
        })

    def _validate_authorization(self):
        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
    @Workflow.transition('finished')
    def finish(cls, managers):
        Line = Pool().get('ncf.manager.sequence')
        lines = [line for manager in managers
            for line in manager.ncf_sequences if line.state == 'active']
        if lines:
            Line.complete(lines)
        cls.write(managers, {
            'finished_by': Transaction().user,
            'finished_at': datetime.datetime.now(),
        })

    @classmethod
    def delete(cls, managers):
        if any(manager.state != 'draft' for manager in managers):
            raise UserError(gettext(
                'ncf_manager.msg_ncf_authorization_delete_not_draft'))
        Invoice = Pool().get('account.invoice')
        if Invoice.search([
                    ('ncf_sequence.pos', 'in', [m.id for m in managers]),
                    ], limit=1):
            raise UserError(gettext(
                'ncf_manager.msg_ncf_authorization_used'))
        super().delete(managers)

    @classmethod
    def copy(cls, managers, default=None):
        default = dict(default or {})
        default.update({
            'state': 'draft',
            'ncf_sequences': None,
            'activated_by': None,
            '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)


class NCFManagerSequences(Workflow, ModelSQL, ModelView):

    'NCF Manager Sequences'
    __name__ = 'ncf.manager.sequence'

    _states = {
        'readonly': Eval('state') != 'draft',
    }

    _depends = ['state']

    pos = fields.Many2One('ncf.manager', 'Point of Sale',
                          ondelete='CASCADE', required=True)

    ncf_type = fields.Selection(NCF_TYPE, 'NCF Type',
                                required=True, sort=False,
                                states=_states, depends=_depends,
                                help='Choose a NCF Type')

    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')

    expected_prefix = fields.Function(
        fields.Char('Expected Prefix'), 'on_change_with_expected_prefix')
    expected_padding = fields.Function(
        fields.Integer('Expected Padding'),
        'on_change_with_expected_padding')

    sequence_mode = fields.Selection([
        ('existing', 'Select Existing Sequence'),
        ('create', 'Create Sequence on Activation'),
    ], 'Sequence Configuration', required=True, states=_states,
        depends=_depends, sort=False)

    ncf_sequences = fields.Many2One('ir.sequence.strict', 'Fiscal Sequence',
        domain=[
            ('company', '=', Eval('company', -1)),
            ('prefix', '=', Eval('expected_prefix')),
            ('active', '=', True),
        ],
        states={
            **_states,
            'required': Eval('sequence_mode') == 'existing',
            'invisible': Eval('sequence_mode') == 'create',
        },
        depends=_depends + [
            'company',
            'sequence_mode', 'expected_prefix', 'expected_padding'])

    from_number = fields.Integer(
        'Authorized From', states={
            **_states,
            'required': Eval('sequence_mode') == 'create',
            'invisible': Eval('sequence_mode') != 'create',
        }, depends=_depends + ['sequence_mode'])
    until_number = fields.Integer(
        'Authorized Until', states={
            **_states,
            'required': Eval('sequence_mode') == 'create',
            'invisible': Eval('sequence_mode') != 'create',
        }, depends=_depends + ['sequence_mode'])
    number_next = fields.Integer(
        'Next Number to Issue', states={
            **_states,
            'required': Eval('sequence_mode') == 'create',
            'invisible': Eval('sequence_mode') != 'create',
        }, depends=_depends + ['sequence_mode'])
    expiration_date = fields.Date(
        'Expiration Date', states={
            **_states,
            '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('Next 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')

    # Tree-specific values must remain visible for both assisted draft lines
    # and lines backed by an activated strict sequence.  The effective fields
    # above retain their form-only visibility rules.
    tree_from_number = fields.Function(
        fields.Integer('Authorized From'), 'get_effective_sequence_values')
    tree_until_number = fields.Function(
        fields.Integer('Authorized Until'), 'get_effective_sequence_values')
    tree_number_next = fields.Function(
        fields.Integer('Next Number to Issue'),
        'get_effective_sequence_values')
    tree_expiration_date = fields.Function(
        fields.Date('Expiration Date'), 'get_effective_sequence_values')

    invoice_type = fields.Selection([
        ('invoice_customer', 'Customer Invoice'),
        ('invoice_supplier', 'Supplier Invoice')
    ], 'Invoice Type', required=True,
        states=_states, depends=_depends)

    state = fields.Selection([
        ('draft', 'Draft'),
        ('active', 'Active'),
        ('finished', 'Finished'),
    ], "State", readonly=True, required=True, sort=False)

    state_string = state.translated('state')

    availability = fields.Function(fields.Selection([
        ('available', 'Available'),
        ('low', 'Running Low'),
        ('expired', 'Expired'),
        ('exhausted', 'Exhausted'),
    ], 'Availability', sort=False), 'get_availability')
    availability_string = availability.translated('availability')
    finished_by = fields.Many2One(
        'res.user', 'Finished By', readonly=True)
    finished_at = fields.DateTime('Finished At', readonly=True)

    del _states, _depends

    @staticmethod
    def default_state():
        return 'draft'

    @staticmethod
    def default_sequence_mode():
        return 'existing'

    @fields.depends('pos', '_parent_pos.company')
    def on_change_with_company(self, name=None):
        return self.pos.company.id if self.pos and self.pos.company else None

    @fields.depends('ncf_type')
    def on_change_with_expected_prefix(self, name=None):
        return ncf_prefix(self.ncf_type)

    @fields.depends('ncf_type')
    def on_change_with_expected_padding(self, name=None):
        return ncf_padding(self.ncf_type)

    @fields.depends(
        'sequence_mode', 'ncf_type', 'ncf_sequences',
        'from_number', 'until_number', 'number_next', 'expiration_date')
    def on_change_ncf_type(self):
        if (self.ncf_sequences
                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',
            'tree_from_number': 'from_number',
            'tree_until_number': 'until_number',
            'tree_number_next': 'number_next',
            'tree_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 ''
        return self.ncf_type_string.split('-')[0]

    @classmethod
    def validate_fields(cls, records, field_names):
        super().validate_fields(records, field_names)
        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
        if not sequence:
            return
        if self.ncf_type == '99':
            raise UserError(gettext(
                'ncf_manager.msg_ncf_non_fiscal_sequence'))
        if sequence.company != self.pos.company:
            raise UserError(gettext(
                'ncf_manager.msg_ncf_sequence_company_mismatch'))
        expected_prefix = ncf_prefix(self.ncf_type)
        if self.ncf_type != '99' and sequence.prefix != expected_prefix:
            raise UserError(gettext(
                'ncf_manager.msg_ncf_sequence_prefix_mismatch',
                prefix=expected_prefix))
        if sequence.padding != ncf_padding(self.ncf_type):
            raise UserError(gettext(
                'ncf_manager.msg_ncf_invalid_padding',
                padding=ncf_padding(self.ncf_type)))
        ModelData = Pool().get('ir.model.data')
        expected_type = ModelData.get_id(
            'ncf_manager', NCF_SEQUENCE_TYPE_IDS[self.ncf_type])
        if sequence.sequence_type.id != expected_type:
            raise UserError(gettext(
                'ncf_manager.msg_ncf_sequence_type_mismatch'))
        today = Transaction().context.get('date') or datetime.date.today()
        if sequence.expiration_date and sequence.expiration_date < today:
            raise UserError(gettext(
                'ncf_manager.msg_ncf_sequence_expired'))
        if (sequence.until_number is not None
                and sequence.number_next is not None
                and sequence.number_next > sequence.until_number):
            raise UserError(gettext(
                'ncf_manager.msg_ncf_sequence_finished'))

    def prepare_sequence(self):
        if self.sequence_mode != 'create':
            return
        if self.ncf_type == '99':
            raise UserError(gettext(
                'ncf_manager.msg_ncf_non_fiscal_sequence'))
        self._validate_assisted_values()
        pool = Pool()
        ModelData = pool.get('ir.model.data')
        Sequence = pool.get('ir.sequence.strict')
        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.auth_number}',
            'sequence_type': sequence_type,
            'company': self.pos.company.id,
            'prefix': ncf_prefix(self.ncf_type),
            'padding': ncf_padding(self.ncf_type),
            'number_increment': 1,
            'number_next': self.number_next,
            'from_number': self.from_number,
            'until_number': self.until_number,
            'expiration_date': self.expiration_date,
        }])
        self.ncf_sequences = sequence
        self.sequence_mode = 'existing'

    def _validate_assisted_values(self):
        if None in (self.from_number, self.until_number, self.number_next):
            raise UserError(gettext(
                'ncf_manager.msg_ncf_assisted_values_required'))
        if not (1 <= self.from_number <= self.number_next
                <= self.until_number):
            raise UserError(gettext('ncf_manager.msg_ncf_invalid_range'))

    def _validate_no_overlap(self):
        sequence = self.ncf_sequences
        if not sequence:
            return
        candidates = self.__class__.search([
            ('id', '!=', self.id),
            ('pos.company', '=', self.pos.company.id),
            ('ncf_type', '=', self.ncf_type),
            ('invoice_type', '=', self.invoice_type),
            ('state', '=', 'active'),
            ('ncf_sequences', '!=', None),
        ])
        for candidate in candidates:
            other = candidate.ncf_sequences
            if None in (sequence.from_number, sequence.until_number,
                    other.from_number, other.until_number):
                continue
            if (sequence.from_number <= other.until_number
                    and other.from_number <= sequence.until_number):
                raise UserError(gettext(
                    'ncf_manager.msg_ncf_sequence_overlap'))

    def get_availability(self, name):
        sequence = self.ncf_sequences
        if not sequence:
            return 'available'
        today = Transaction().context.get('date') or datetime.date.today()
        if sequence.expiration_date and sequence.expiration_date < today:
            return 'expired'
        if (sequence.until_number is not None
                and sequence.number_next is not None):
            remaining = sequence.until_number - sequence.number_next + 1
            if remaining <= 0:
                return 'exhausted'
            approved = ((sequence.until_number - sequence.from_number + 1)
                if sequence.from_number is not None else 0)
            if remaining <= max(10, int(approved * .1)):
                return 'low'
        return 'available'

    @classmethod
    def __setup__(cls):
        super().__setup__()
        table = cls.__table__()
        cls._sql_constraints += [
            ('fiscal_sequence_unique', Unique(table, table.ncf_sequences),
                'ncf_manager.msg_ncf_sequence_already_assigned'),
        ]
        cls._transitions |= {
            ('draft', 'active'),
            ('active', 'finished'),
        }
        cls._buttons.update({
            'finish': {
                'invisible': Eval('state') != 'active',
                'depends': ['state'],
            },
        })

    @classmethod
    @Workflow.transition('active')
    def activate(cls, lines):
        pass

    @classmethod
    @ModelView.button
    def finish(cls, lines):
        cls.complete(lines)

    @classmethod
    @Workflow.transition('finished')
    def complete(cls, lines):
        cls.write(lines, {
            'finished_by': Transaction().user,
            'finished_at': datetime.datetime.now(),
        })

    @classmethod
    def delete(cls, lines):
        if any(line.state != 'draft' for line in lines):
            raise UserError(gettext(
                'ncf_manager.msg_ncf_sequence_delete_not_draft'))
        Invoice = Pool().get('account.invoice')
        if Invoice.search([
                    ('ncf_sequence', 'in', [line.id for line in lines]),
                    ], limit=1):
            raise UserError(gettext(
                'ncf_manager.msg_ncf_sequence_used'))
        super().delete(lines)