ncf_manager/invoice.py

# -*- coding: utf-8 -*-
# This file is part of Tryton.  The COPYRIGHT file at the top level of
# this repository contains the full copyright notices and license terms.
from decimal import Decimal
from datetime import date
import re
from sql.aggregate import Sum
from sql.conditionals import Coalesce
from sql.functions import Round
from trytond import backend
from trytond.model import ModelView, Workflow, fields
from trytond.pool import Pool, PoolMeta
from trytond.pyson import Eval, If, And, Bool
from trytond.transaction import Transaction
from trytond.exceptions import UserError
from trytond.i18n import gettext
from trytond.modules.ncf_manager.ncf_manager import NCF_TYPE
from trytond.modules.ncf_manager.fiscal import (
    EXPENSE_TYPE, PAYMENT_METHOD, REVENUE_TYPE)
from trytond.modules.currency.fields import Monetary
import logging
logger = logging.getLogger(__name__)


def _backend_name():
    name = backend.name
    return name() if callable(name) else name


_ISR_WITHHOLDING_TYPE = [
    ('', ''),
    ('01', "01 - Alquileres"),
    ('02', "02 - Honorarios por Servicios"),
    ('03', "03 - Otras Rentas"),
    ('04', "04 - Otras Rentas (Rentas Presuntas)"),
    ('05', "05 - Intereses Pagados a Personas Jurídicas Residentes"),
    ('06', "06 - Intereses Pagados a Personas Físicas Residentes"),
    ('07', "07 - Retención por Proveedores del Estado"),
    ('08', "08 - Juegos Telefónicos"),
    ('09', "09 - Retenciones Subsector Ganadería de Carne Bovina"),
]

_CANCELLATION_TYPE = [
    ('', ''),
    ('01', "01 Deterioro de Factura Pre-Impresa"),
    ('02', "02 Errores de Impresión (Factura Pre-Impresa)"),
    ('03', "03 Impresión Defectuosa"),
    ('04', "04 Corrección de la Información"),
    ('05', "05 Cambio de Productos"),
    ('06', "06 Devolución de Productos"),
    ('07', "07 Omisión de Productos"),
    ('08', "08 Errores en Secuencias de NCF"),
    ('09', "09 Por Cese de Operaciones"),
    ('10', "10 Pérdida O Hurto De Talonario(S)"),
]

_ZERO = Decimal('0.00')

_DEBIT_NOTE_CODES = ('3', '33')
_CREDIT_NOTE_CODES = ('4', '34')
_MODIFIED_NOTE_CODES = _DEBIT_NOTE_CODES + _CREDIT_NOTE_CODES

_ECF_ORIGIN_CODES = (
    '31', '32', '33', '34',
    '41', '43', '44', '45', '46', '47',
)

_ECF_CODES = [
    '31', '32', '33', '34',
    '41', '43', '44', '45', '46', '47',
]

_ELECTRONIC_INVOICE_MODULES = (
    'account_invoice_electronic',
    'account_invoice_electronic_do',
    'dgii_electronic_invoice',
    'electronic_invoice_do',
    'ncf_electronic',
)

class Invoice(metaclass=PoolMeta):
    __name__ = 'account.invoice'

    ncf_type = fields.Selection(
        NCF_TYPE,
        'NCF Type',
        sort=False,
        states={
            'required': And(Eval('type') == 'out', Eval('state') != 'draft'),
            'readonly': True,
        },
        depends=['state', 'type', 'company'],
        help='Fiscal receipt type resolved from the selected NCF sequence.'
    )

    ncf_sequence = fields.Many2One('ncf.manager.sequence', 'Tipo NCF',
        domain=[
            If(
                Eval('state') == 'draft',
                [
                    ('state', '=', 'active'),
                    If(
                        Eval('type') == 'out',
                        ('invoice_type', '=', 'invoice_customer'),
                        ('invoice_type', '=', 'invoice_supplier')
                    )
                ],
                []
            )
        ],
        states={
           'required': And(Eval('type') == 'out', Eval('state') != 'draft'),
           'readonly': Eval('state') != 'draft',
        },
        depends=['state', 'type'])

    ncf_type_code = fields.Function(fields.Char('NCF Type Code'),
        'on_change_with_ncf_type_code')

    ncf_number = fields.Char('NCF Number', size=19,
        states={
             'readonly': (Eval('type') == 'out') | (Eval('state') != 'draft'),
        },
        depends=['state', 'type'])

    ncf_number_modified = fields.Char('NCF Modified', size=19,
        states={
          'readonly': Eval('state') != 'draft',
          'invisible': ~Eval('ncf_type_code').in_(['3', '33', '4', '34']),
        }, depends=['state'])

    security_code = fields.Char(
        'Security Code',
        states={
            'readonly': Eval('state') != 'draft',
            'invisible': ~Eval('ncf_type').in_(_ECF_CODES),
        },
        depends=['state', 'ncf_type'],
    )

    itbis_invoiced = fields.Function(Monetary(
        "ITBIS Invoiced", currency='currency', digits='currency',
        states={
            'invisible': ~Eval('itbis_invoiced', 0),
        }), 'on_change_with_itbis_invoiced', searcher='search_itbis_invoiced')

    itbis_withholding = fields.Function(Monetary(
        "ITBIS Withholding", currency='currency', digits='currency',
        states={
            'invisible': ~Eval('itbis_withholding', 0),
        }), 'on_change_with_itbis_withholding', searcher='search_itbis_withholding')

    itbis_withholding_acquirer = fields.Function(Monetary(
        "ITBIS Withholding Acquirer", currency='currency', digits='currency',
        states={
            'invisible': ~Eval('itbis_withholding_acquirer', 0),
        }), 'on_change_with_itbis_withholding_acquirer',
        searcher='search_itbis_withholding_acquirer')

    isc_invoiced = fields.Function(Monetary(
        "ISC Invoiced", currency='currency', digits='currency',
        states={
            'invisible': ~Eval('isc_invoiced', 0),
        }), 'on_change_with_isc_invoiced', searcher='search_isc_invoiced')

    cdt_invoiced = fields.Function(Monetary(
        "CDT Invoiced", currency='currency', digits='currency',
        states={
            'invisible': ~Eval('cdt_invoiced', 0),
        }), 'on_change_with_cdt_invoiced', searcher='search_cdt_invoiced')

    isr_withholding = fields.Function(Monetary(
        "ISR Withholding", currency='currency', digits='currency',
        states={
            'invisible': ~Eval('isr_withholding', 0),
        }), 'on_change_with_isr_withholding', searcher='search_isr_withholding')

    tip = fields.Function(Monetary(
        "Tip", currency='currency', digits='currency',
        states={
            'invisible': ~Eval('tip', 0),
        }), 'on_change_with_tip', searcher='search_tip')

    id_type = fields.Function(fields.Integer('ID Type',
                                             help='The identifier type of party in the invoice.'),
                              'get_id_type')

    expiration_date = fields.Date('Expiration Date')

    amount_invoiced_services = fields.Function(Monetary(
        "Amount Invoiced in Services", currency='currency', digits='currency'),
        'get_amount_invoiced_services', searcher='search_amount_invoiced_services')

    amount_invoiced_goods = fields.Function(Monetary(
        "Amount Invoiced in Goods", currency='currency', digits='currency'),
        'get_amount_invoiced_goods', searcher='search_amount_invoiced_goods')

    revenue_type = fields.Selection(REVENUE_TYPE, 'Revenue Type',
                                    states={
                                        'readonly': Eval('state').in_(['posted', 'paid', 'cancelled']),
                                        'invisible': Eval('type') == 'in',
                                    },
                                    depends=['state'])

    revenue_type_string = revenue_type.translated('revenue_type')

    expense_type = fields.Selection(EXPENSE_TYPE, 'Expense Type',
                                    states={
                                        'readonly': Eval('state').in_(['posted', 'paid', 'cancelled']),
                                        'invisible': Eval('type') == 'out',
                                    },
                                    depends=['state'])

    expense_type_string = expense_type.translated('expense_type')

    payment_method = fields.Selection(PAYMENT_METHOD, 'Method of Payment',
                                      states={
                                          'readonly': Eval('state').in_(['posted', 'paid', 'cancelled']),
                                      },
                                      depends=['state'])

    payment_method_string = payment_method.translated('payment_method')

    payment_method_cash = fields.Function(Monetary(
            "Cash", currency='currency', digits='currency'),
        'on_change_with_payment_method_cash')

    payment_method_check = fields.Function(Monetary(
            "Checks/Transfers/Deposit", currency='currency', digits='currency'),
        'on_change_with_payment_method_check')

    payment_method_card = fields.Function(Monetary(
        "Credit/Debit Card", currency='currency', digits='currency'),
        'on_change_with_payment_method_card')

    payment_method_purchase_on_credit = fields.Function(Monetary(
        "Purchase on Credit", currency='currency', digits='currency'),
        'on_change_with_payment_method_purchase_on_credit')

    payment_method_barter = fields.Function(Monetary(
        "Barter", currency='currency', digits='currency'),
        'on_change_with_payment_method_barter')

    payment_method_credit_note = fields.Function(Monetary(
        "Credit Note", currency='currency', digits='currency'),
        'on_change_with_payment_method_credit_note')

    payment_method_mixed_payment = fields.Function(Monetary(
        'Mixed Payment', currency='currency', digits='currency'),
        'on_change_with_payment_method_mixed_payment')

    isr_withholding_type = fields.Selection(_ISR_WITHHOLDING_TYPE, 'ISR Withholding Type',
                                            states={
                                                'readonly': Eval('state').in_(['posted', 'paid', 'cancelled']),
                                                'invisible': Eval('type') == 'out',
                                            },
                                            depends=['state'])

    isr_withholding_type_string = isr_withholding_type.translated(
        'isr_withholding_type')

    ncf_cancellation_type = fields.Selection(
        _CANCELLATION_TYPE, 'NCF Cancellation Type',
        states={
            'required': (
                (Eval('state') == 'cancelled')
                & (Eval('type') == 'out')
                & Bool(Eval('ncf_number'))
                & ~Eval('context', {}).get('invoice_force_cancel', False)),
            'readonly': Eval('state') == 'cancelled',
            'invisible': Eval('type') == 'in',
            },
        depends=['state', 'type', 'ncf_number'])

    ncf_cancellation_type_string = ncf_cancellation_type.translated(
        'ncf_cancellation_type')

    ncf_cancellation_reason = fields.Text(
        'NCF Cancellation Reason',
        states={
            'required': (
                (Eval('state') == 'cancelled')
                & (Eval('type') == 'out')
                & Bool(Eval('ncf_number'))
                & ~Eval('context', {}).get('invoice_force_cancel', False)),
            'readonly': Eval('state') == 'cancelled',
            'invisible': Eval('type') == 'in',
            },
        depends=['state', 'type', 'ncf_number'])

    @classmethod
    def _ncf_get_configuration_default(cls, field_name, **pattern):
        "Return a default value from account configuration."
        Configuration = Pool().get('account.configuration')
        config = Configuration(1)
        return config.get_multivalue(field_name, **pattern) or None

    @classmethod
    def default_revenue_type(cls, **pattern):
        if Transaction().context.get('type') == 'in':
            return None
        return cls._ncf_get_configuration_default(
            'default_customer_revenue_type', **pattern)

    @classmethod
    def default_expense_type(cls, **pattern):
        if Transaction().context.get('type') == 'out':
            return None
        return cls._ncf_get_configuration_default(
            'default_supplier_expense_type', **pattern)

    @classmethod
    def default_payment_method(cls, **pattern):
        return cls._ncf_get_configuration_default(
            'default_payment_method', **pattern)

    @classmethod
    def __setup__(cls):
        super().__setup__()
        cls._check_modify_exclude.update({
                'ncf_cancellation_type',
                'ncf_cancellation_reason',
                })

    @classmethod
    def __register__(cls, module_name):
        """
        Migrate legacy account_invoice.ncf_type safely.

        Old structure:
            account_invoice.ncf_type = ncf.manager.sequence.id as integer

        New structure:
            account_invoice.ncf_type = fiscal NCF type code as varchar
            account_invoice.ncf_sequence = ncf.manager.sequence.id
        """
        cls._ncf_prepare_legacy_invoice_ncf_type_column()
        super().__register__(module_name)
        cls._ncf_migrate_invoice_ncf_type_from_legacy_column()
        cls._ncf_migrate_legacy_cancellation_fields()

    @classmethod
    def _ncf_column_exists(cls, cursor, table_name, column_name):
        if _backend_name() == 'sqlite':
            cursor.execute(f'PRAGMA table_info("{table_name}")')
            return any(row[1] == column_name for row in cursor.fetchall())
        cursor.execute("""
            SELECT 1
              FROM information_schema.columns
             WHERE table_name = %s
               AND column_name = %s
        """, (table_name, column_name))
        return cursor.fetchone() is not None


    @classmethod
    def _ncf_column_udt_name(cls, cursor, table_name, column_name):
        if _backend_name() == 'sqlite':
            cursor.execute(f'PRAGMA table_info("{table_name}")')
            for row in cursor.fetchall():
                if row[1] == column_name:
                    return (row[2] or '').lower()
            return None
        cursor.execute("""
            SELECT udt_name
              FROM information_schema.columns
             WHERE table_name = %s
               AND column_name = %s
        """, (table_name, column_name))
        row = cursor.fetchone()
        return row[0] if row else None


    @classmethod
    def _ncf_prepare_legacy_invoice_ncf_type_column(cls):
        """
        Rename old integer ncf_type column before Tryton tries to migrate it.

        This avoids:
            Unable to migrate column ncf_type on table account_invoice
            from int4 to varchar.
        """
        cursor = Transaction().connection.cursor()

        table_name = 'account_invoice'
        old_column = 'ncf_type'
        legacy_column = 'ncf_type_legacy_sequence_id'

        if not cls._ncf_column_exists(cursor, table_name, old_column):
            return

        if cls._ncf_column_exists(cursor, table_name, legacy_column):
            return

        udt_name = cls._ncf_column_udt_name(cursor, table_name, old_column)

        if udt_name in {'int2', 'int4', 'int8', 'integer'}:
            logger.info(
                "Renaming legacy %s.%s integer column to %s before Tryton "
                "creates the new Selection column",
                table_name,
                old_column,
                legacy_column,
            )

            cursor.execute(f"""
                ALTER TABLE {table_name}
                RENAME COLUMN {old_column} TO {legacy_column}
            """)


    @classmethod
    def _ncf_migrate_invoice_ncf_type_from_legacy_column(cls):
        """
        Convert legacy sequence IDs into real NCF type values.

        Example:
            account_invoice.ncf_type_legacy_sequence_id = 2
            ncf_manager_sequence.id = 2
            ncf_manager_sequence.ncf_type = '34'

        Result:
            account_invoice.ncf_type = '34'
            account_invoice.ncf_sequence = 2
        """
        cursor = Transaction().connection.cursor()

        table_name = 'account_invoice'
        legacy_column = 'ncf_type_legacy_sequence_id'

        if not cls._ncf_column_exists(cursor, table_name, legacy_column):
            logger.info(
                "No legacy invoice NCF column found. Skipping invoice NCF migration."
            )
            return

        logger.info(
            "Starting migration of account_invoice legacy ncf_type sequence IDs "
            "to fiscal NCF type codes"
        )

        cursor.execute("""
            UPDATE account_invoice AS invoice
               SET ncf_sequence = seq.id,
                   ncf_type = seq.ncf_type
              FROM ncf_manager_sequence AS seq
             WHERE invoice.ncf_type_legacy_sequence_id IS NOT NULL
               AND invoice.ncf_type_legacy_sequence_id = seq.id
               AND seq.ncf_type IS NOT NULL
               AND seq.ncf_type <> ''
               AND (
                    invoice.ncf_type IS NULL
                    OR invoice.ncf_type = ''
                    OR invoice.ncf_type <> seq.ncf_type
               )
        """)

        logger.info(
            "Migrated %s account_invoice rows from legacy ncf_type sequence IDs",
            cursor.rowcount,
        )

        cursor.execute("""
            SELECT COUNT(*)
              FROM account_invoice AS invoice
              LEFT JOIN ncf_manager_sequence AS seq
                ON invoice.ncf_type_legacy_sequence_id = seq.id
             WHERE invoice.ncf_type_legacy_sequence_id IS NOT NULL
               AND seq.id IS NULL
        """)
        unmatched = cursor.fetchone()[0]

        if unmatched:
            logger.warning(
                "%s account_invoice rows have legacy ncf_type values that do "
                "not match any ncf_manager_sequence.id",
                unmatched,
            )

    @classmethod
    def _ncf_migrate_legacy_cancellation_fields(cls):
        """
        Copy data from the former account_invoice_cancel_reason module.

        The old module used generic cancellation_type/cancellation_reason
        columns.  ncf_manager stores those values as fiscal DGII cancellation
        data so they can later feed the NCF annulment report.
        """
        cursor = Transaction().connection.cursor()
        table_name = 'account_invoice'
        field_pairs = (
            ('cancellation_type', 'ncf_cancellation_type'),
            ('cancellation_reason', 'ncf_cancellation_reason'),
            )

        for old_column, new_column in field_pairs:
            if not cls._ncf_column_exists(cursor, table_name, old_column):
                continue
            if not cls._ncf_column_exists(cursor, table_name, new_column):
                continue
            cursor.execute(f"""
                UPDATE {table_name}
                   SET {new_column} = {old_column}
                 WHERE {old_column} IS NOT NULL
                   AND {old_column} <> ''
                   AND ({new_column} IS NULL OR {new_column} = '')
            """)

    @classmethod
    def search_rec_name(cls, name, clause):
        domain = super().search_rec_name(name, clause)
        return domain + [
            ('ncf_number', *clause[1:]),
            ('ncf_number_modified', *clause[1:]),
        ]

    @staticmethod
    def _ncf_type_label_map():
        """
        Return NCF/e-CF labels indexed by fiscal code.

        This keeps invoice.ncf_type as a simple fiscal code while allowing
        the selection list to be generated dynamically from active and valid
        ncf.manager.sequence records.
        """
        return dict(NCF_TYPE)

    @classmethod
    def _ncf_invoice_type_from_type(cls, invoice_type):
        """
        Convert account.invoice.type into ncf.manager.sequence.invoice_type.

        Mapping:
        - account.invoice.type = 'out' -> ncf.manager.sequence.invoice_type = 'invoice_customer'
        - account.invoice.type = 'in'  -> ncf.manager.sequence.invoice_type = 'invoice_supplier'

        Important:
        Do not return invoice_supplier by default. If invoice_type is empty,
        unknown or not loaded yet, return None to avoid showing wrong NCF types.
        """
        if invoice_type == 'out':
            return 'invoice_customer'

        if invoice_type == 'in':
            return 'invoice_supplier'

        return None

    @classmethod
    def _get_available_ncf_type_domain(
            cls, company_id=None, sequence_invoice_type=None):
        """
        Build the domain used to find NCF types available for invoices.

        sequence_invoice_type must be:
        - invoice_customer
        - invoice_supplier
        """
        today = Transaction().context.get('date') or date.today()

        domain = [
            ('state', '=', 'active'),
            ('pos.state', '=', 'active'),
            ('ncf_type', '!=', ''),
            ('ncf_sequences', '!=', None),
            [
                'OR',
                ('ncf_sequences.expiration_date', '=', None),
                ('ncf_sequences.expiration_date', '>=', today),
            ],
        ]

        if company_id:
            domain.append(('pos.company', '=', company_id))

        if sequence_invoice_type:
            domain.append(('invoice_type', '=', sequence_invoice_type))

        return domain

    @fields.depends('company', 'type')
    def get_available_ncf_types(self):
        """
        Dynamic values for invoice.ncf_type.

        This method must use the current invoice values, not only the transaction
        context, because the invoice type that controls the fiscal sequence is the
        value from the current account.invoice record.

        Expected behavior:
        - type = 'out' -> show only ncf.manager.sequence records with
          invoice_type = 'invoice_customer'
        - type = 'in' -> show only ncf.manager.sequence records with
          invoice_type = 'invoice_supplier'
        """
        pool = Pool()
        NCFSequence = pool.get('ncf.manager.sequence')

        label_map = self.__class__._ncf_type_label_map()
        values = [('', '')]

        company_id = self.company.id if self.company else None
        sequence_invoice_type = self.__class__._ncf_invoice_type_from_type(
            self.type)

        if not company_id or not sequence_invoice_type:
            return values

        sequences = NCFSequence.search(
            self.__class__._get_available_ncf_type_domain(
                company_id=company_id,
                sequence_invoice_type=sequence_invoice_type,
            ),
            order=[
                ('ncf_type', 'ASC'),
                ('id', 'ASC'),
            ],
        )

        seen = set()

        for sequence in sequences:
            strict_sequence = sequence.ncf_sequences

            if not strict_sequence:
                continue

            if (
                strict_sequence.until_number is not None
                and strict_sequence.number_next is not None
                and strict_sequence.number_next > strict_sequence.until_number
            ):
                continue

            ncf_type = sequence.ncf_type

            if not ncf_type:
                continue

            if ncf_type in seen:
                continue

            label = label_map.get(ncf_type)

            if not label:
                continue

            values.append((ncf_type, label))
            seen.add(ncf_type)

        return values

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

    @fields.depends('company', 'type', 'ncf_type', 'ncf_sequence', 'expiration_date')
    def on_change_ncf_type(self):
        """
        Keep compatibility when ncf_type is assigned automatically from party
        defaults or from migration logic.
        """
        if not self.ncf_type:
            self.ncf_sequence = None
            self.expiration_date = None
            return

        sequence = self._ncf_get_active_sequence()

        if not sequence:
            self.ncf_sequence = None
            self.expiration_date = None
            return

        self.ncf_sequence = sequence

        if sequence.ncf_sequences:
            self.expiration_date = sequence.ncf_sequences.expiration_date
        else:
            self.expiration_date = None

    @fields.depends('company', 'type', 'ncf_type', 'ncf_sequence', 'expiration_date')
    def on_change_ncf_sequence(self):
        """
        When the fiscal sequence changes, resolve the fiscal NCF type from it.

        The user selects the available sequence.
        The system fills invoice.ncf_type automatically.
        """
        if not self.ncf_sequence:
            self.ncf_type = None
            self.expiration_date = None
            return

        expected_invoice_type = self.__class__._ncf_invoice_type_from_type(
            self.type)

        if (
            expected_invoice_type
            and self.ncf_sequence.invoice_type != expected_invoice_type
        ):
            self.ncf_sequence = None
            self.ncf_type = None
            self.expiration_date = None
            return

        self.ncf_type = self.ncf_sequence.ncf_type

        if self.ncf_sequence.ncf_sequences:
            self.expiration_date = self.ncf_sequence.ncf_sequences.expiration_date
        else:
            self.expiration_date = None

    def _ncf_get_active_sequence(self):
        """
        Return the active ncf.manager.sequence related to the selected ncf_type.
        """
        if not self.ncf_type or not self.company:
            return None

        return self.__class__._ncf_find_active_sequence(self, self.ncf_type)

    def _ncf_clear_fiscal_sequence_fields(self):
        """
        Clear NCF/e-CF fields related to the selected fiscal sequence.
        """
        self.ncf_type = None
        self.ncf_sequence = None
        self.expiration_date = None

    def _ncf_get_party_default_ncf_type(self):
        """
        Return the default NCF type configured on the selected party according
        to the current invoice type.

        Mapping:
        - invoice.type = 'out' -> party.customer_ncf_type
        - invoice.type = 'in'  -> party.supplier_ncf_type

        The party fields are MultiValue fields, so the company context must be
        considered when reading them.
        """
        if not self.party or not self.type:
            return None

        if self.type == 'out':
            field_name = 'customer_ncf_type'
        elif self.type == 'in':
            field_name = 'supplier_ncf_type'
        else:
            return None

        company_id = self.company.id if self.company else Transaction().context.get('company')

        if getattr(self.party, 'id', None) and company_id:
            Party = Pool().get('party.party')

            with Transaction().set_context(company=company_id):
                party = Party(self.party.id)
                return getattr(party, field_name, None) or None

        return getattr(self.party, field_name, None) or None

    def _ncf_apply_party_default_ncf_type(self):
        """
        Apply the party default NCF type to the invoice.

        The selected NCF type is accepted only if there is an active and valid
        ncf.manager.sequence for:
        - invoice company
        - invoice type
        - selected party default NCF type

        If the party default does not have a valid active sequence, the invoice
        fiscal fields are cleared to avoid assigning an invalid NCF type.
        """
        party_ncf_type = self._ncf_get_party_default_ncf_type()

        if not party_ncf_type:
            self._ncf_clear_fiscal_sequence_fields()
            return

        self.ncf_type = party_ncf_type

        sequence = self._ncf_get_active_sequence()

        if not sequence:
            self._ncf_clear_fiscal_sequence_fields()
            return

        self.ncf_sequence = sequence

        if sequence.ncf_sequences:
            self.expiration_date = sequence.ncf_sequences.expiration_date
        else:
            self.expiration_date = None

    def _ncf_get_configuration_default_invoice_field(self, field_name):
        "Return an invoice default configured on account configuration."
        Configuration = Pool().get('account.configuration')
        config = Configuration(1)
        company_id = (
            self.company.id if self.company
            else Transaction().context.get('company'))

        if company_id:
            with Transaction().set_context(company=company_id):
                return config.get_multivalue(field_name) or None

        return config.get_multivalue(field_name) or None

    def _ncf_apply_configuration_invoice_defaults(self):
        """
        Apply configured fiscal reporting defaults without replacing values
        already selected by the user or copied from another invoice.
        """
        if self.type == 'out' and not self.revenue_type:
            self.revenue_type = (
                self._ncf_get_configuration_default_invoice_field(
                    'default_customer_revenue_type'))
        elif self.type == 'in' and not self.expense_type:
            self.expense_type = (
                self._ncf_get_configuration_default_invoice_field(
                    'default_supplier_expense_type'))

        if not self.payment_method:
            self.payment_method = (
                self._ncf_get_configuration_default_invoice_field(
                    'default_payment_method'))

    @fields.depends(
        'party',
        'company',
        'type',
        'ncf_type',
        'ncf_sequence',
        'expiration_date',
        'revenue_type',
        'expense_type',
        'payment_method',
        '_parent_party.customer_ncf_type',
        '_parent_party.supplier_ncf_type',
    )
    def on_change_party(self):
        """
        When the party changes, assign the default NCF type configured on the
        party according to the invoice type.

        Customer invoice:
            type = 'out' -> party.customer_ncf_type

        Supplier invoice:
            type = 'in' -> party.supplier_ncf_type
        """
        super().on_change_party()
        self._ncf_apply_party_default_ncf_type()
        self._ncf_apply_configuration_invoice_defaults()


    @fields.depends(
        'company',
        'type',
        'ncf_type',
        'ncf_sequence',
        'expiration_date',
        'revenue_type',
        'expense_type',
        'payment_method',
    )
    def on_change_company(self):
        if self.ncf_type:
            sequence = self._ncf_get_active_sequence()
            if not sequence:
                self.ncf_type = None
                self.ncf_sequence = None
                self.expiration_date = None
            else:
                self.ncf_sequence = sequence
                if sequence.ncf_sequences:
                    self.expiration_date = sequence.ncf_sequences.expiration_date
        self._ncf_apply_configuration_invoice_defaults()

    @fields.depends(
        'company',
        'type',
        'party',
        'journal',
        'account',
        'payment_term',
        'invoice_address',
        'party_tax_identifier',
        'ncf_type',
        'ncf_sequence',
        'expiration_date',
        'revenue_type',
        'expense_type',
        'payment_method',
        methods=['set_journal', 'on_change_party'],
    )
    def on_change_type(self):
        """
        Preserve the core account_invoice behavior and then apply NCF logic.

        Core behavior:
        - type = 'out' assigns the revenue journal.
        - type = 'in' assigns the expense journal.
        - party/account/payment term values are refreshed.

        NCF behavior:
        - clears fiscal sequence values when the selected sequence does not match
          the new invoice type.
        """
        super().on_change_type()
        self._ncf_apply_configuration_invoice_defaults()

        if not self.ncf_sequence:
            self.ncf_type = None
            self.expiration_date = None
            return

        expected_invoice_type = self.__class__._ncf_invoice_type_from_type(
            self.type)

        if (
            expected_invoice_type
            and self.ncf_sequence.invoice_type != expected_invoice_type
        ):
            self.ncf_sequence = None
            self.ncf_type = None
            self.expiration_date = None
            return

        self.ncf_type = self.ncf_sequence.ncf_type

        if self.ncf_sequence.ncf_sequences:
            self.expiration_date = self.ncf_sequence.ncf_sequences.expiration_date
        else:
            self.expiration_date = None

    def _ncf_validate_selected_sequence(self):
        """
        Validate that the selected NCF sequence matches the invoice company,
        invoice type, active state and selected NCF type.
        """
        if not self.ncf_type:
            return

        if not self.ncf_sequence:
            raise UserError(gettext(
                'ncf_manager.msg_missing_ncf_sequence',
                invoice=self.rec_name,
            ))

        expected_invoice_type = self.__class__._ncf_invoice_type_from_type(
            self.type)

        if (
            expected_invoice_type
            and self.ncf_sequence.invoice_type != expected_invoice_type
        ):
            raise UserError(gettext(
                'ncf_manager.msg_invalid_ncf_sequence_invoice_type',
                invoice=self.rec_name,
            ))

        if self.ncf_sequence.ncf_type != self.ncf_type:
            raise UserError(gettext(
                'ncf_manager.msg_invalid_ncf_sequence_type',
                invoice=self.rec_name,
            ))

        if self.ncf_sequence.state != 'active':
            raise UserError(gettext(
                'ncf_manager.msg_inactive_ncf_sequence',
                invoice=self.rec_name,
            ))

        if self.ncf_sequence.pos.state != 'active':
            raise UserError(gettext(
                'ncf_manager.msg_inactive_ncf_manager',
                invoice=self.rec_name,
            ))

        strict_sequence = self.ncf_sequence.ncf_sequences

        if not strict_sequence:
            raise UserError(gettext(
                'ncf_manager.msg_missing_ncf_sequence',
                invoice=self.rec_name,
            ))

        today = Transaction().context.get('date') or date.today()

        if (
            strict_sequence.expiration_date
            and strict_sequence.expiration_date < today
        ):
            raise UserError(gettext(
                'ncf_manager.msg_ncf_sequence_expired',
                invoice=self.rec_name,
            ))

        if (
            strict_sequence.until_number is not None
            and strict_sequence.number_next is not None
            and strict_sequence.number_next > strict_sequence.until_number
        ):
            raise UserError(gettext(
                'ncf_manager.msg_ncf_sequence_finished',
                invoice=self.rec_name,
            ))

    @fields.depends('payment_method', 'total_amount')
    def on_change_with_payment_method_cash(self, name=None):
        if self.payment_method and self.payment_method == '01':
            return self.total_amount

    @fields.depends('payment_method', 'total_amount')
    def on_change_with_payment_method_check(self, name=None):
        if self.payment_method and self.payment_method == '02':
            return self.total_amount

    @fields.depends('payment_method', 'total_amount')
    def on_change_with_payment_method_card(self, name=None):
        if self.payment_method and self.payment_method == '03':
            return self.total_amount

    @fields.depends('payment_method', 'total_amount')
    def on_change_with_payment_method_purchase_on_credit(self, name=None):
        if self.payment_method and self.payment_method == '04':
            return self.total_amount

    @fields.depends('payment_method', 'total_amount')
    def on_change_with_payment_method_barter(self, name=None):
        if self.payment_method and self.payment_method == '05':
            return self.total_amount

    @fields.depends('payment_method', 'total_amount')
    def on_change_with_payment_method_credit_note(self, name=None):
        if self.payment_method and self.payment_method == '06':
            return self.total_amount

    @fields.depends('payment_method', 'total_amount')
    def on_change_with_payment_method_mixed_payment(self, name=None):
        if self.payment_method and self.payment_method == '07':
            return self.total_amount

    # def get_untaxed_amount_invoiced(self, name=None):
    #     amount = _ZERO
    #     amount = self.untaxed_amount
    #     return abs(amount)

    # def get_itbis_amount_invoiced(self, name=None):
    #     amount = _ZERO
    #     amount = self.itbis_invoiced
    #     return abs(amount)

    # def get_total_amount_invoiced(self, name=None):
    #     amount = _ZERO
    #     amount = self.total_amount
    #     return abs(amount)

    @classmethod
    def get_amount_invoiced_services(cls, invoices, names):
        amount_invoiced_services = {i.id: _ZERO for i in invoices}
        for invoice in invoices:
            amount_invoiced_services[invoice.id] = sum(
                (line.amount for line in invoice.lines
                    if line.type == 'line' and line.product_type == 'service'),
                _ZERO)
        return {'amount_invoiced_services': amount_invoiced_services}

    @classmethod
    def search_amount_invoiced_services(cls, name, clause):
        return cls._search_amount_invoiced_by_product_type(
            name, clause, 'service')

    @classmethod
    def search_amount_invoiced_goods(cls, name, clause):
        return cls._search_amount_invoiced_by_product_type(
            name, clause, 'goods')

    @classmethod
    def _search_amount_invoiced_by_product_type(
            cls, name, clause, product_type):
        pool = Pool()
        Rule = pool.get('ir.rule')
        Line = pool.get('account.invoice.line')
        Invoice = pool.get('account.invoice')
        Currency = pool.get('currency.currency')
        Product = pool.get('product.product')
        Template = pool.get('product.template')
        type_name = getattr(cls, name)._field.sql_type().base
        line = Line.__table__()
        invoice = Invoice.__table__()
        currency = Currency.__table__()
        product = Product.__table__()
        template = Template.__table__()

        _, operator, value = clause
        if value is None:
            return []
        invoice_query = Rule.query_get('account.invoice')
        Operator = fields.SQL_OPERATORS[operator]
        # SQLite uses float for sum
        if _backend_name() == 'sqlite':
            value = float(value)

        query = (line.join(invoice, condition=(invoice.id == line.invoice))
            .join(currency, condition=(currency.id == invoice.currency))
            .join(product, condition=(product.id == line.product))
            .join(template, condition=(template.id == product.template))
            .select(
                line.invoice,
                where=(line.invoice.in_(invoice_query)
                    & (line.type == 'line')
                    & (template.type == product_type)),
                group_by=line.invoice,
                having=Operator(Coalesce(Sum(
                    Round((line.quantity * line.unit_price).cast(type_name),
                        currency.digits)), 0).cast(type_name), value)))
        return [('id', 'in', query)]

    @classmethod
    def get_amount_invoiced_goods(cls, invoices, names):
        amount_invoiced_goods = {i.id: _ZERO for i in invoices}
        for invoice in invoices:
            amount_invoiced_goods[invoice.id] = sum(
                (line.amount for line in invoice.lines
                    if line.type == 'line' and line.product_type == 'goods'),
                _ZERO)
        return {'amount_invoiced_goods': amount_invoiced_goods}

    def get_id_type(self, name):
        if not self.party or not self.party_tax_identifier:
            return None

        identifier = self.party_tax_identifier
        identifier_type = getattr(identifier, 'type', None)
        if identifier_type in {'do_rnc', 'do_vat'}:
            return 1
        if identifier_type == 'do_cedula':
            return 2

        # Tryton 8.0 stores a formatted display code (for example
        # ``1-01-01063-2`` or ``001-1391820-5``). Use code_compact when
        # available and still normalize it for migrated records whose compact
        # value may retain the old formatting.
        code = (getattr(identifier, 'code_compact', None)
            or getattr(identifier, 'code', None) or '')
        digits = re.sub(r'\D', '', code)
        if len(digits) == 9:
            return 1
        if len(digits) == 11:
            return 2
        return 3

    @fields.depends('taxes', 'lines')
    def on_change_with_itbis_invoiced(self, name=None):
        itbis_invoiced = _ZERO
        for tax_line in self.taxes:
            if tax_line.tax and tax_line.tax.tax_kind == 'itbis':
                itbis_invoiced += tax_line.amount

        return abs(itbis_invoiced)

    @fields.depends('taxes', 'lines')
    def on_change_with_isc_invoiced(self, name=None):
        isc_invoiced = _ZERO
        for tax_line in self.taxes:
            if tax_line.tax and tax_line.tax.tax_kind == 'isc':
                isc_invoiced += tax_line.amount

        return abs(isc_invoiced)

    @fields.depends('taxes', 'lines')
    def on_change_with_cdt_invoiced(self, name=None):
        cdt_invoiced = _ZERO
        for tax_line in self.taxes:
            if tax_line.tax and tax_line.tax.tax_kind == 'cdt':
                cdt_invoiced += tax_line.amount

        return abs(cdt_invoiced)

    @fields.depends('taxes', 'lines')
    def on_change_with_itbis_withholding(self, name=None):
        itbis_withholding = _ZERO
        for tax_line in self.taxes:
            if tax_line.tax and tax_line.tax.tax_kind == 'itbis_withholding':
                itbis_withholding += tax_line.amount
        return abs(itbis_withholding)

    @fields.depends('taxes', 'lines')
    def on_change_with_itbis_withholding_acquirer(self, name=None):
        total = _ZERO
        for tax_line in self.taxes:
            if (tax_line.tax
                    and tax_line.tax.tax_kind == 'itbis_withholding_acquirer'):
                total += tax_line.amount
        return abs(total)

    @fields.depends('taxes', 'lines')
    def on_change_with_isr_withholding(self, name=None):
        isr_withholding = _ZERO
        for tax_line in self.taxes:
            if tax_line.tax and tax_line.tax.tax_kind == 'isr_withholding':
                isr_withholding += tax_line.amount

        return abs(isr_withholding)

    @fields.depends('taxes', 'lines')
    def on_change_with_tip(self, name=None):
        tip = _ZERO
        for tax_line in self.taxes:
            if tax_line.tax and tax_line.tax.tax_kind == 'tip':
                tip += tax_line.amount

        return abs(tip)

    @classmethod
    def _search_tax_amount_by_kind(cls, field, tax_kind, clause):
        pool = Pool()
        Rule = pool.get('ir.rule')
        InvoiceTax = pool.get('account.invoice.tax')
        AccountTax = pool.get('account.tax')
        type_name = field.sql_type().base
        invoice_tax = InvoiceTax.__table__()
        account_tax = AccountTax.__table__()

        _, operator, value = clause
        invoice_query = Rule.query_get('account.invoice')
        Operator = fields.SQL_OPERATORS[operator]

        query = invoice_tax.join(
            account_tax,
            condition=(account_tax.id == invoice_tax.tax),
        ).select(
            invoice_tax.invoice,
            where=(
                invoice_tax.invoice.in_(invoice_query)
                & (account_tax.tax_kind == tax_kind)
            ),
            group_by=invoice_tax.invoice,
            having=Operator(
                Coalesce(Sum(invoice_tax.amount), 0).cast(type_name),
                value,
            ),
        )
        return [('id', 'in', query)]

    @classmethod
    def search_itbis_invoiced(cls, name, clause):
        return cls._search_tax_amount_by_kind(
            cls.itbis_invoiced._field, 'itbis', clause)

    @classmethod
    def search_itbis_withholding(cls, name, clause):
        return cls._search_tax_amount_by_kind(
            cls.itbis_withholding._field, 'itbis_withholding', clause)

    @classmethod
    def search_itbis_withholding_acquirer(cls, name, clause):
        return cls._search_tax_amount_by_kind(
            cls.itbis_withholding_acquirer._field,
            'itbis_withholding_acquirer', clause)

    @classmethod
    def search_isc_invoiced(cls, name, clause):
        return cls._search_tax_amount_by_kind(
            cls.isc_invoiced._field, 'isc', clause)

    @classmethod
    def search_cdt_invoiced(cls, name, clause):
        return cls._search_tax_amount_by_kind(
            cls.cdt_invoiced._field, 'cdt', clause)

    @classmethod
    def search_isr_withholding(cls, name, clause):
        return cls._search_tax_amount_by_kind(
            cls.isr_withholding._field, 'isr_withholding', clause)

    @classmethod
    def search_tip(cls, name, clause):
        return cls._search_tax_amount_by_kind(
            cls.tip._field, 'tip', clause)

    @classmethod
    def _ncf_invoice_type(cls, invoice):
        """
        Return ncf.manager.sequence invoice_type according to account.invoice.type.
        """
        return cls._ncf_invoice_type_from_type(invoice.type)

    @classmethod
    def _ncf_code(cls, invoice):
        """
        Return the fiscal/e-CF code from invoice.ncf_type.
        """
        return invoice.ncf_type or None


    @classmethod
    def _ncf_is_credit_note(cls, invoice):
        """
        Return True when the invoice is fiscally a credit note.
        """
        return cls._ncf_code(invoice) in _CREDIT_NOTE_CODES


    @classmethod
    def _ncf_is_debit_note(cls, invoice):
        """
        Return True when the invoice is fiscally a debit note.
        """
        return cls._ncf_code(invoice) in _DEBIT_NOTE_CODES


    @classmethod
    def _ncf_resolve_credit_note_code(cls, origin):
        """
        Resolve the preferred NCF/e-NCF type for a credit note.

        Electronic credit notes are only allowed when the electronic invoicing
        module is active.
        """
        if cls._ncf_electronic_invoicing_enabled():
            return '34'
        return '4'


    @classmethod
    def _ncf_resolve_debit_note_code(cls, origin):
        """
        Resolve the preferred NCF/e-NCF type for a debit note.

        Electronic debit notes are only allowed when the electronic invoicing
        module is active.
        """
        if cls._ncf_electronic_invoicing_enabled():
            return '33'
        return '3'


    @classmethod
    def _ncf_find_active_sequence(cls, invoice, ncf_code):
        """
        Find active and valid ncf.manager.sequence for the invoice company,
        invoice type and selected fiscal NCF/e-CF type.
        """
        if not ncf_code or not invoice.company:
            return None

        NCFSequence = Pool().get('ncf.manager.sequence')
        today = Transaction().context.get('date') or date.today()

        sequence_invoice_type = cls._ncf_invoice_type(invoice)

        if not sequence_invoice_type:
            return None

        sequences = NCFSequence.search([
            ('ncf_type', '=', ncf_code),
            ('invoice_type', '=', sequence_invoice_type),
            ('pos.company', '=', invoice.company.id),
            ('pos.state', '=', 'active'),
            ('state', '=', 'active'),
            ('ncf_sequences', '!=', None),
            [
                'OR',
                ('ncf_sequences.expiration_date', '=', None),
                ('ncf_sequences.expiration_date', '>=', today),
            ],
        ], order=[
            ('id', 'ASC'),
        ])

        for sequence in sequences:
            strict_sequence = sequence.ncf_sequences

            if not strict_sequence:
                continue

            if (
                strict_sequence.until_number is not None
                and strict_sequence.number_next is not None
                and strict_sequence.number_next > strict_sequence.until_number
            ):
                continue

            return sequence

        return None


    @classmethod
    def _ncf_electronic_invoicing_enabled(cls):
        """
        Return whether electronic invoicing is enabled for NCF selection.

        ncf_manager alone must only issue physical B NCFs. Electronic E NCFs
        are enabled by an electronic invoicing module. A concrete electronic
        invoicing module may also override this hook if its technical name is
        not in _ELECTRONIC_INVOICE_MODULES.
        """
        Module = Pool().get('ir.module')
        return bool(Module.search([
            ('name', 'in', list(_ELECTRONIC_INVOICE_MODULES)),
            ('state', '=', 'activated'),
            ], limit=1))


    @classmethod
    def _ncf_note_code_candidates(cls, origin, electronic_code, physical_code):
        """
        Return note NCF codes in preference order.

        ncf_manager alone issues B notes. When electronic invoicing is active,
        E notes are preferred and B notes remain a fallback.
        """
        if not cls._ncf_electronic_invoicing_enabled():
            return [physical_code]

        preferred = (
            electronic_code
            if cls._ncf_code(origin) in _ECF_ORIGIN_CODES
            else physical_code)
        candidates = [electronic_code, physical_code]
        if preferred in candidates:
            candidates.remove(preferred)
            candidates.insert(0, preferred)
        return candidates


    @classmethod
    def _ncf_find_note_sequence(
            cls, origin, note, electronic_code, physical_code):
        """
        Return the selected NCF code and active sequence for a debit/credit note.
        """
        for ncf_code in cls._ncf_note_code_candidates(
                origin, electronic_code, physical_code):
            sequence = cls._ncf_find_active_sequence(note, ncf_code)
            if sequence:
                return ncf_code, sequence
        return None, None


    @classmethod
    def _ncf_prepare_credit_note_from_origin(cls, origin, credit_note):
        """
        Fiscal preparation for credit notes generated from an existing invoice.

        This is the specific fiscal responsibility of ncf_manager:
        - Set ncf_number_modified from the original invoice ncf_number.
        - Assign the correct credit-note NCF type: 34 or 4.
        - Set fiscal payment method to credit note when available.
        """
        if not origin.ncf_number:
            return

        credit_note.ncf_number_modified = origin.ncf_number

        ncf_code, sequence = cls._ncf_find_note_sequence(
            origin, credit_note, '34', '4')

        if not sequence:
            raise UserError(gettext(
                'ncf_manager.msg_missing_credit_note_sequence',
                ncf_type='/'.join(cls._ncf_note_code_candidates(
                    origin, '34', '4')),
                company=credit_note.company.rec_name))

        credit_note.ncf_type = ncf_code
        credit_note.ncf_sequence = sequence

        if hasattr(credit_note, 'payment_method'):
            credit_note.payment_method = '06'

        if hasattr(credit_note, 'revenue_type') and credit_note.type == 'out':
            credit_note.revenue_type = credit_note.revenue_type or '01'

    @classmethod
    def _ncf_prepare_debit_note_from_origin(cls, origin, debit_note):
        """
        Fiscal preparation for debit notes generated from an existing invoice.

        This is the specific fiscal responsibility of ncf_manager:
        - Set ncf_number_modified from the original invoice ncf_number.
        - Assign the correct debit-note NCF type: 33 or 3.
        - Preserve fiscal report fields from the original invoice when present.
        """
        if not origin.ncf_number:
            return

        debit_note.ncf_number_modified = origin.ncf_number

        ncf_code, sequence = cls._ncf_find_note_sequence(
            origin, debit_note, '33', '3')

        if not sequence:
            raise UserError(gettext(
                'ncf_manager.msg_missing_debit_note_sequence',
                ncf_type='/'.join(cls._ncf_note_code_candidates(
                    origin, '33', '3')),
                company=debit_note.company.rec_name))

        debit_note.ncf_type = ncf_code
        debit_note.ncf_sequence = sequence

        if hasattr(debit_note, 'payment_method'):
            debit_note.payment_method = getattr(origin, 'payment_method', None)

        if hasattr(debit_note, 'revenue_type') and debit_note.type == 'out':
            debit_note.revenue_type = (
                getattr(origin, 'revenue_type', None)
                or debit_note.revenue_type
                or '01')

        if hasattr(debit_note, 'expense_type') and debit_note.type == 'in':
            debit_note.expense_type = (
                getattr(origin, 'expense_type', None)
                or debit_note.expense_type)

        if hasattr(debit_note, 'isr_withholding_type'):
            debit_note.isr_withholding_type = getattr(
                origin, 'isr_withholding_type', None)

    @classmethod
    def _debit_prepare_note_from_origin(cls, origin, debit_note, values):
        super_method = getattr(super(), '_debit_prepare_note_from_origin', None)
        if super_method:
            super_method(origin, debit_note, values)
        cls._ncf_prepare_debit_note_from_origin(origin, debit_note)

    @classmethod
    def credit(cls, invoices, refund=False, **values):
        """
        Extend credit-note creation with fiscal NCF behavior.

        This method must live in ncf_manager, not in account_credit_note.

        It works for:
        - Credit note from form view.
        - Credit note from tree view.
        - Full credit note.
        - Partial credit note created by account_credit_note.

        Important:
        We call super(..., refund=False) first because ncf_number_modified and
        ncf_type must be assigned before posting the generated credit note.
        """
        credit_mode = values.get('credit_mode') or 'full'

        if refund and credit_mode in ('percent', 'amount'):
            raise UserError(gettext(
                'ncf_manager.msg_partial_credit_note_refund_not_allowed'))

        new_invoices = super().credit(invoices, refund=False, **values)

        to_save = []
        for origin, credit_note in zip(invoices, new_invoices):
            cls._ncf_prepare_credit_note_from_origin(origin, credit_note)
            to_save.append(credit_note)

        if to_save:
            cls.save(to_save)

        if refund:
            cls.post(new_invoices)

            for origin, credit_note in zip(invoices, new_invoices):
                if origin.state != 'posted':
                    raise UserError(gettext(
                        'account_invoice.msg_invoice_credit_refund_not_posted',
                        invoice=origin.rec_name))
                origin.cancel_move = credit_note.move

            cls.save(invoices)
            cls.cancel(invoices)

        return new_invoices


    @classmethod
    def set_ncf_number(cls, invoices):
        """
        Assign NCF/e-NCF number to invoices before posting.

        The user selects invoice.ncf_type as a simple Selection value.
        The system resolves invoice.ncf_sequence automatically from the active
        ncf.manager.sequence configuration.
        """
        for invoice in invoices:
            if invoice.ncf_number:
                continue

            if not invoice.ncf_type:
                raise UserError(gettext('ncf_manager.msg_no_ncf_type'))

            ncf_code = invoice.ncf_type

            if ncf_code == '99':
                continue

            ncf_sequence = invoice.ncf_sequence or cls._ncf_find_active_sequence(
                invoice, ncf_code)

            if not ncf_sequence:
                raise UserError(gettext(
                    'ncf_manager.msg_missing_ncf_sequence',
                    invoice=invoice.rec_name,
                ))

            strict_sequence = ncf_sequence.ncf_sequences

            if not strict_sequence:
                raise UserError(gettext(
                    'ncf_manager.msg_missing_strict_sequence'))

            invoice.ncf_sequence = ncf_sequence

            if not invoice.expiration_date:
                invoice.expiration_date = strict_sequence.expiration_date

            invoice.ncf_number, invoice.sequence = invoice.get_next_ncf_number(
                ncf_sequence=ncf_sequence)

        cls.save(invoices)

    def get_next_ncf_number(self, pattern=None, ncf_sequence=None):
        """
        Return next fiscal number and sequence id used.
        """
        if not self.ncf_type:
            raise UserError(gettext('ncf_manager.msg_no_ncf_type'))

        ncf_sequence = ncf_sequence or self.ncf_sequence or self._ncf_get_active_sequence()

        if not ncf_sequence:
            raise UserError(gettext(
                'ncf_manager.msg_missing_ncf_sequence',
                invoice=self.rec_name,
            ))

        strict_sequence = ncf_sequence.ncf_sequences

        if not strict_sequence:
            raise UserError(gettext(
                'ncf_manager.msg_missing_strict_sequence'))

        today = Transaction().context.get('date') or date.today()

        with Transaction().set_context(company=self.company.id):
            if (
                strict_sequence.until_number is not None
                and strict_sequence.number_next is not None
                and strict_sequence.number_next > strict_sequence.until_number
            ):
                raise UserError(gettext(
                    'ncf_manager.msg_ncf_sequence_finished'))

            if strict_sequence.expiration_date and today > strict_sequence.expiration_date:
                raise UserError(gettext(
                    'ncf_manager.msg_ncf_sequence_expired'))

            number = strict_sequence.get()
            StrictSequence = Pool().get('ir.sequence.strict')
            ManagerSequence = Pool().get('ncf.manager.sequence')
            strict_sequence = StrictSequence(strict_sequence.id)
            if (strict_sequence.until_number is not None
                    and strict_sequence.number_next is not None
                    and strict_sequence.number_next
                    > strict_sequence.until_number
                    and ncf_sequence.state == 'active'):
                ManagerSequence.complete([ncf_sequence])
            return number, strict_sequence.id


    def check_credit_note(self):
        if self.total_amount > 0:
            raise UserError(gettext(
                'ncf_manager.msg_amount_credit_note_negative'))

    def check_invoice(self):
        if self.total_amount < 0:
            raise UserError(gettext(
                'ncf_manager.msg_amount_invoice_no_negative'))

    def check_payment_term(self):
        if not self.payment_term:
            raise UserError(gettext(
                'ncf_manager.msg_no_payment_term'))

    def check_revenue_type(self):
        if not self.revenue_type and self.type == 'out':
            raise UserError(gettext(
                'ncf_manager.msg_no_revenue_type'))

    def check_expense_type(self):
        if not self.expense_type and self.type == 'in':
            raise UserError(gettext(
                'ncf_manager.msg_no_expense_type'))

    def check_payment_method(self):
        if not self.payment_method:
            raise UserError(gettext(
                'ncf_manager.msg_no_method_payment'))

    def check_isr_withholding_type(self):
        if self.isr_withholding > _ZERO and not self.isr_withholding_type:
            raise UserError(gettext(
                'ncf_manager.msg_no_isr_withholding_type'))

    def check_no_expiration_date(self):
        if self.type == 'in':
            if not self.expiration_date:
                raise UserError(gettext(
                    'ncf_manager.msg_no_expiration_date'))

    @classmethod
    def _ncf_check_fiscal_classification(cls, invoice):
        """
        Validate fiscal classification fields in DGII-friendly order.
        """
        if invoice.type == 'out':
            invoice.check_revenue_type()
        else:
            invoice.check_expense_type()
        invoice.check_isr_withholding_type()
        invoice.check_payment_method()

    def check_ncf_number_modified(self):
        """
        Validate modified NCF only for fiscal/electronic debit/credit notes.
        """
        if not self.ncf_type:
            return

        if self.ncf_type not in _MODIFIED_NOTE_CODES:
            return

        if not self.ncf_number_modified:
            raise UserError(gettext(
                'ncf_manager.msg_no_ncf_number_modified'))

        if self.type == 'in':
            length = len(self.ncf_number_modified or '')

            if length < 11:
                raise UserError(gettext(
                    'ncf_manager.msg_incorrect_lenght_ncf_number_modified'))

            if 11 < length < 13:
                raise UserError(gettext(
                    'ncf_manager.msg_incorrect_lenght_ncf_number_modified'))

            if 13 < length < 19:
                raise UserError(gettext(
                    'ncf_manager.msg_incorrect_lenght_ncf_number_modified'))

    @staticmethod
    def _ncf_normalize_number(number):
        return (number or '').strip().upper()

    @classmethod
    def _ncf_validate_number(cls, invoice):
        number = cls._ncf_normalize_number(invoice.ncf_number)
        if not number:
            return
        legacy_supplier_ncf = (
            getattr(invoice, 'type', None) == 'in'
            and re.fullmatch(r'[A-Z]\d{18}', number))
        if not legacy_supplier_ncf and not re.fullmatch(
                r'(?:B\d{10}|E\d{12})', number):
            raise UserError(gettext(
                'ncf_manager.msg_invalid_ncf_structure', number=number))
        if legacy_supplier_ncf:
            invoice.ncf_number = number
            return
        expected = ('E' if invoice.ncf_type in _ECF_CODES else 'B') + \
            invoice.ncf_type.zfill(2)
        if invoice.ncf_type != '99' and not number.startswith(expected):
            raise UserError(gettext(
                'ncf_manager.msg_invalid_ncf_type_prefix',
                number=number, prefix=expected))
        invoice.ncf_number = number

    @classmethod
    def _ncf_check_duplicate(cls, invoice):
        if not invoice.ncf_number:
            return
        domain = [
            ('id', '!=', invoice.id),
            ('company', '=', invoice.company.id),
            ('ncf_number', '=', invoice.ncf_number),
            ]
        # Supplier NCF sequences are unique per issuer, not globally.
        if invoice.type == 'in':
            domain.extend([
                ('type', '=', 'in'),
                ('party', '=', invoice.party.id),
                ])
        else:
            domain.append(('type', '=', 'out'))
        if cls.search(domain, limit=1):
            raise UserError(gettext(
                'ncf_manager.msg_duplicated_ncf_number_invoice'))

    @staticmethod
    def _ncf_is_fiscal_cancellation_required(invoice):
        return (
            invoice.type == 'out'
            and bool(getattr(invoice, 'ncf_number', None)))

    @classmethod
    def _ncf_check_cancellation_information(cls, invoice):
        if not cls._ncf_is_fiscal_cancellation_required(invoice):
            return
        if not invoice.ncf_cancellation_type:
            raise UserError(gettext(
                    'ncf_manager.msg_missing_ncf_cancellation_type',
                    invoice=invoice.rec_name))
        if not invoice.ncf_cancellation_reason:
            raise UserError(gettext(
                    'ncf_manager.msg_missing_ncf_cancellation_reason',
                    invoice=invoice.rec_name))

    @classmethod
    def copy(cls, invoices, default=None):
        if default is None:
            default = {}
        else:
            default = default.copy()

        default.setdefault('ncf_type', None)
        default.setdefault('ncf_sequence', None)
        default.setdefault('ncf_number', None)
        default.setdefault('ncf_number_modified', None)
        default.setdefault('expiration_date', None)
        default.setdefault('security_code', None)
        default.setdefault('ncf_cancellation_type', None)
        default.setdefault('ncf_cancellation_reason', None)
        
        return super(Invoice, cls).copy(invoices, default=default)

    @classmethod
    def view_attributes(cls):
        return super().view_attributes() + [
            ('/tree/field[@name="expense_type"]', 'tree_invisible',
             Eval('type') == 'out'),
            ('/tree/field[@name="revenue_type"]', 'tree_invisible',
             Eval('type') == 'in'),
            ('/form/notebook/page[@id="tax_information"]/separator[@id="revenue"]', 'states', {
                'invisible': Eval('type') == 'in',
            }),
            ('/form/notebook/page[@id="tax_information"]/separator[@id="expense"]', 'states', {
                'invisible': Eval('type') == 'out',
            }),
            ('/form/notebook/page[@id="ncf_cancellation"]', 'states', {
                'invisible': (
                    (Eval('type') == 'in') | ~Bool(Eval('ncf_number'))),
            }),
        ]

    @classmethod
    @ModelView.button
    @Workflow.transition('cancelled')
    def cancel(cls, invoices):
        if not Transaction().context.get('invoice_force_cancel'):
            for invoice in invoices:
                cls._ncf_check_cancellation_information(invoice)
        super().cancel(invoices)

    @classmethod
    def _post(cls, invoices):
        cls.set_ncf_number(invoices)

        for invoice in invoices:
            cls._ncf_validate_number(invoice)
            cls._ncf_check_duplicate(invoice)
            ncf_code = invoice.ncf_type
            is_debit_note = cls._ncf_is_debit_note(invoice)
            is_credit_note = cls._ncf_is_credit_note(invoice)
            is_modified_note = is_debit_note or is_credit_note
            is_without_ncf = ncf_code == '99'

            if invoice.type == 'out':
                if not is_without_ncf:
                    invoice._ncf_validate_selected_sequence()
                if is_credit_note:
                    invoice.check_credit_note()
                    invoice.check_ncf_number_modified()
                    cls._ncf_check_fiscal_classification(invoice)
                    invoice.check_payment_term()
                else:
                    invoice.check_invoice()
                    if is_modified_note:
                        invoice.check_ncf_number_modified()
                    cls._ncf_check_fiscal_classification(invoice)
                    invoice.check_payment_term()

            else:
                if is_without_ncf:
                    continue

                if is_credit_note:
                    invoice.check_credit_note()
                    invoice.check_ncf_number_modified()
                    invoice.check_no_expiration_date()
                    cls._ncf_check_fiscal_classification(invoice)
                    invoice.check_payment_term()
                else:
                    if is_modified_note:
                        invoice.check_ncf_number_modified()
                    invoice.check_no_expiration_date()
                    cls._ncf_check_fiscal_classification(invoice)
                    invoice.check_payment_term()

        super()._post(invoices)


class InvoiceReport(metaclass=PoolMeta):
    __name__ = 'account.invoice'

    @classmethod
    def get_context(cls, records, header, data):
        context = super().get_context(records, header, data)
        invoice = context.get('invoice')
        ncf_type_map = dict(NCF_TYPE)
        ncf_type_name = ''
        if invoice and invoice.ncf_type:
            ncf_type_name = ncf_type_map.get(invoice.ncf_type, '')
        context['ncf_type_name'] = ncf_type_name
        return context