Odoo18-Base/addons/l10n_hu_edi/models/account_move.py
2025-01-06 10:57:38 +07:00

1054 lines
50 KiB
Python

# Part of Odoo. See LICENSE file for full copyright and licensing details.
import math
import base64
import logging
import re
from lxml import etree
from psycopg2.errors import LockNotAvailable
from odoo import fields, models, api, _
from odoo.http import request
from odoo.exceptions import UserError, ValidationError
from odoo.tools import formatLang, float_round, float_repr, cleanup_xml_node, groupby
from odoo.tools.misc import split_every
from odoo.addons.base_iban.models.res_partner_bank import normalize_iban
from odoo.addons.l10n_hu_edi.models.l10n_hu_edi_connection import format_bool, L10nHuEdiConnection, L10nHuEdiConnectionError
_logger = logging.getLogger(__name__)
class AccountMove(models.Model):
_inherit = 'account.move'
l10n_hu_payment_mode = fields.Selection(
[
("TRANSFER", "Transfer"),
("CASH", "Cash"),
("CARD", "Credit/debit card"),
("VOUCHER", "Voucher"),
("OTHER", "Other"),
],
string="Payment mode",
help="NAV expected payment mode of the invoice.",
)
# === EDI Fields === #
l10n_hu_edi_state = fields.Selection(
######################################################################################################################
# STATE DIAGRAM
# * False, rejected, cancelled --[upload]--> False, sent, send_timeout
# * sent --[query_status]--> sent, confirmed, confirmed_warning, rejected
# * confirmed, confirmed_warning --[request_cancel]--> cancel_sent, cancel_timeout
# * cancel_sent, cancel_pending --[query_status]--> confirmed_warning, cancel_pending, cancelled
# * send_timeout --[recover_timeout]--> False, send_timeout, confirmed, confirmed_warning, rejected,
# * cancel_timeout --[recover_timeout]--> confirmed_warning, cancel_sent, cancel_timeout, cancel_pending, cancelled
######################################################################################################################
selection=[
('sent', 'Sent, waiting for response'),
('send_timeout', 'Timeout when sending'),
('confirmed', 'Confirmed'),
('confirmed_warning', 'Confirmed with warnings'),
('rejected', 'Rejected'),
('cancel_sent', 'Cancellation request sent'),
('cancel_timeout', 'Timeout when requesting cancellation'),
('cancel_pending', 'Cancellation request pending'),
('cancelled', 'Cancelled'),
],
string='NAV 3.0 status',
copy=False,
index='btree_not_null',
)
l10n_hu_edi_batch_upload_index = fields.Integer(
string='Index of invoice within a batch upload',
copy=False,
)
l10n_hu_edi_attachment = fields.Binary(
string='Invoice XML file',
attachment=True,
copy=False,
)
l10n_hu_edi_send_time = fields.Datetime(
string='Invoice upload time',
copy=False,
)
l10n_hu_edi_transaction_code = fields.Char(
string='Transaction Code',
index='trigram',
copy=False,
tracking=True,
)
# A dict with the following structure:
# {
# 'error_title': the main heading of the message
# 'errors': a list of message items
# 'blocking_level': {'error' | 'warning' | None}
# directs which blocking behaviour to adopt in the Send and Print:
# * error: blocks PDF generation and sending by e-mail
# * warning: PDF is generated and sent by e-mail, but a warning appears in the banner
# * None: PDF is generated and sent by e-mail, no warning appears
# }
l10n_hu_edi_messages = fields.Json(
string='Transaction messages (JSON)',
copy=False,
)
l10n_hu_invoice_chain_index = fields.Integer(
string='Invoice Chain Index',
help="""
Index in the chain of modification invoices:
-1 for a base invoice;
1, 2, 3, ... for modification invoices;
0 for rejected/cancelled invoices or if it has not yet been set.
""",
copy=False,
)
l10n_hu_edi_attachment_filename = fields.Char(
string='Invoice XML filename',
compute='_compute_l10n_hu_edi_attachment_filename',
)
l10n_hu_edi_message_html = fields.Html(
string='Transaction messages',
compute='_compute_message_html',
)
# === Constraints === #
@api.constrains('l10n_hu_edi_state', 'state')
def _check_posted_if_active(self):
""" Enforce the constraint that you cannot reset to draft / cancel a posted invoice if it was already sent to NAV. """
for move in self:
if move.state in ['draft', 'cancel'] and move.l10n_hu_edi_state not in [False, 'rejected', 'cancelled']:
raise ValidationError(_('Cannot reset to draft or cancel invoice %s because an electronic document was already sent to NAV!', move.name))
# === Computes === #
@api.depends('l10n_hu_edi_messages')
def _compute_message_html(self):
for move in self:
if move.l10n_hu_edi_messages:
move.l10n_hu_edi_message_html = self.env['account.move.send']._format_error_html(move.l10n_hu_edi_messages)
else:
move.l10n_hu_edi_message_html = False
@api.depends('l10n_hu_edi_state', 'state')
def _compute_show_reset_to_draft_button(self):
super()._compute_show_reset_to_draft_button()
self.filtered(lambda m: m.l10n_hu_edi_state not in [False, 'rejected', 'cancelled']).show_reset_to_draft_button = False
@api.depends('l10n_hu_edi_state')
def _compute_need_cancel_request(self):
# EXTEND 'account' to add dependencies
return super()._compute_need_cancel_request()
@api.depends('name')
def _compute_l10n_hu_edi_attachment_filename(self):
for move in self:
move.l10n_hu_edi_attachment_filename = f'{move.name.replace("/", "_")}.xml' if move.name else 'nav30.xml'
# === Overrides === #
def _need_cancel_request(self):
# EXTEND account
# Technical annulment should be available only in debug mode
return super()._need_cancel_request() or (self.l10n_hu_edi_state in ['confirmed', 'confirmed_warning'] and request and request.session.debug)
def button_request_cancel(self):
# EXTEND 'account'
if self._need_cancel_request() and self.l10n_hu_edi_state in ['confirmed', 'confirmed_warning']:
return {
"name": _("Technical Annulment"),
"type": "ir.actions.act_window",
"view_type": "form",
"view_mode": "form",
"res_model": "l10n_hu_edi.cancellation",
"target": "new",
"context": {"default_invoice_id": self.id},
}
return super().button_request_cancel()
# === Actions === #
def l10n_hu_edi_button_update_status(self, from_cron=False):
""" Attempt to update the status of the invoices in `self` """
invoices_to_query = self.filtered(lambda m: 'query_status' in m._l10n_hu_edi_get_valid_actions())
with L10nHuEdiConnection(self.env) as connection:
# Call `query_status` on the invoices.
invoices_to_query._l10n_hu_edi_query_status(connection)
# Attempt to recover missing transactions, if any invoice is missing a transaction code
# or has a duplicate error.
recover_transactions_error = False
if any(
not m.l10n_hu_edi_transaction_code
or any(
'INVOICE_NUMBER_NOT_UNIQUE' in error or 'ANNULMENT_IN_PROGRESS' in error
for error in m.l10n_hu_edi_messages['errors']
)
for m in self
):
recover_transactions_error = self.company_id._l10n_hu_edi_recover_transactions(connection)
# Error handling.
for invoice in invoices_to_query:
# Log invoice status in chatter.
formatted_message = self.env['account.move.send']._format_error_html(invoice.l10n_hu_edi_messages)
invoice.with_context(no_new_invoice=True).message_post(body=formatted_message)
if self.env['account.move.send']._can_commit():
self.env.cr.commit()
# If blocking errors, raise UserError, or log if we are in a cron.
for invoice in invoices_to_query:
if invoice.l10n_hu_edi_messages.get('blocking_level') == 'error' or recover_transactions_error:
if invoice.l10n_hu_edi_messages.get('blocking_level') == 'error':
error_text = self.env['account.move.send']._format_error_text(invoice.l10n_hu_edi_messages)
else:
error_text = self.env['account.move.send']._format_error_text(recover_transactions_error)
if not from_cron:
raise UserError(error_text)
else:
_logger.error(error_text)
def l10n_hu_edi_button_hide_banner(self):
messages = self.l10n_hu_edi_messages
if messages:
messages['hide_banner'] = True
self.l10n_hu_edi_messages = messages
# === Helpers === #
def _l10n_hu_edi_get_valid_actions(self):
""" If any NAV 3.0 flows are applicable to the given invoice, return them, else None. """
self.ensure_one()
valid_actions = []
if (
self.country_code == 'HU'
and self.is_sale_document()
and self.state == 'posted'
):
if self.l10n_hu_edi_state in [False, 'rejected', 'cancelled']:
valid_actions.append('upload')
if self.l10n_hu_edi_transaction_code:
valid_actions.append('query_status')
if self.l10n_hu_edi_state in ['confirmed', 'confirmed_warning']:
valid_actions.append('request_cancel')
if not valid_actions:
# Placeholder to denote that the invoice was already processed with a NAV flow
valid_actions.append(True)
return valid_actions
def _l10n_hu_get_chain_base(self):
""" Get the base invoice of the invoice chain. """
modification_invoices = self
base_invoices = self.env['account.move']
while modification_invoices:
base_invoices |= modification_invoices.filtered(lambda m: not m.reversed_entry_id and not m.debit_origin_id)
modification_invoices = modification_invoices.reversed_entry_id | modification_invoices.debit_origin_id
return base_invoices
def _l10n_hu_get_chain_invoices(self):
""" Given base invoices, get all invoices in the chain. """
chain_invoices = self
next_invoices = self
while (next_invoices := next_invoices.reversal_move_ids | next_invoices.debit_note_ids):
chain_invoices |= next_invoices
return chain_invoices
def _l10n_hu_get_currency_rate(self):
""" Get the invoice currency / HUF rate.
If the company currency is HUF, we estimate this based on the invoice lines
(or if this is not an invoice, based on the AMLs), using a MMSE estimator.
If the company currency is not HUF (e.g. Hungarian companies that do their accounting in euro),
we get the rate from the currency rates.
"""
if self.currency_id.name == 'HUF':
return 1
if self.company_id.currency_id.name == 'HUF':
squared_amount_currency = sum(line.amount_currency ** 2 for line in (self.invoice_line_ids or self.line_ids))
squared_balance = sum(line.balance ** 2 for line in self.invoice_line_ids)
return math.sqrt(squared_balance / squared_amount_currency)
return self.env['res.currency']._get_conversion_rate(
from_currency=self.currency_id,
to_currency=self.env.ref('base.HUF'),
company=self.company_id,
date=self.invoice_date,
)
def _l10n_hu_edi_set_chain_index(self):
""" Set the l10n_hu_invoice_chain_index field. """
self.ensure_one()
base_invoice = self._l10n_hu_get_chain_base()
if base_invoice == self:
self.l10n_hu_invoice_chain_index = -1 # -1 indicates a base invoice (0 indicates the chain index was not set).
else:
# Lock base invoice to prevent concurrent updates, ensuring sequence integrity.
base_invoice._l10n_hu_edi_acquire_lock()
chain_indexes_already_sent = base_invoice._l10n_hu_get_chain_invoices().filtered(
lambda m: m.l10n_hu_edi_state not in ['rejected', 'cancelled']
).mapped('l10n_hu_invoice_chain_index')
for i in range(1, len(chain_indexes_already_sent) + 1):
if i not in chain_indexes_already_sent:
self.l10n_hu_invoice_chain_index = i
break
def _l10n_hu_edi_acquire_lock(self):
""" Acquire a write lock on the invoices in self. """
if not self:
return
try:
with self.env.cr.savepoint(flush=False):
self.env.cr.execute('SELECT * FROM account_move WHERE id = ANY(%s) FOR UPDATE NOWAIT', [self.ids])
except LockNotAvailable:
raise UserError(_('Could not acquire lock on invoices - is another user performing operations on them?')) from None
# === EDI: Flow === #
def _l10n_hu_edi_check_invoices(self):
hu_vat_regex = re.compile(r'\d{8}-[1-5]-\d{2}')
hu_bank_account_regex = re.compile(r'\d{8}-\d{8}-\d{8}|\d{8}-\d{8}|[A-Z]{2}\d{2}[0-9A-Za-z]{11,30}')
# This contains all the advance invoices that correspond to final invoices in `self`.
advance_invoices = self.filtered(lambda m: not m._is_downpayment()).invoice_line_ids._get_downpayment_lines().mapped('move_id')
checks = {
'company_vat_missing': {
'records': self.company_id.filtered(lambda c: not c.vat),
'message': _('Please set company VAT number!'),
'action_text': _('View Company/ies'),
},
'company_vat_invalid': {
'records': self.company_id.filtered(
lambda c: (
(c.vat and not hu_vat_regex.fullmatch(c.vat))
or (c.l10n_hu_group_vat and not hu_vat_regex.fullmatch(c.l10n_hu_group_vat))
)
),
'message': _('Please enter the Hungarian VAT (and/or Group VAT) number in 12345678-1-12 format!'),
'action_text': _('View Company/ies'),
},
'company_address_missing': {
'records': self.company_id.filtered(lambda c: not c.country_id or not c.zip or not c.city or not c.street),
'message': _('Please set company Country, Zip, City and Street!'),
'action_text': _('View Company/ies'),
},
'company_not_huf': {
'records': self.company_id.filtered(lambda c: c.currency_id.name != 'HUF'),
'message': _('Please use HUF as company currency!'),
'action_text': _('View Company/ies'),
},
'partner_bank_account_invalid': {
'records': self.partner_bank_id.filtered(lambda p: not hu_bank_account_regex.fullmatch(p.acc_number)),
'message': _('Please set a valid recipient bank account number!'),
'action_text': _('View partner(s)'),
},
'partner_vat_missing': {
'records': self.partner_id.commercial_partner_id.filtered(
lambda p: p.is_company and not p.vat
),
'message': _('Please set partner Tax ID on company partners!'),
'action_text': _('View partner(s)'),
},
'partner_vat_invalid': {
'records': self.partner_id.commercial_partner_id.filtered(
lambda p: (
p.is_company and p.country_code == 'HU'
and (
(p.vat and not hu_vat_regex.fullmatch(p.vat))
or (p.l10n_hu_group_vat and not hu_vat_regex.fullmatch(p.l10n_hu_group_vat))
)
)
),
'message': _('Please enter the Hungarian VAT (and/or Group VAT) number in 12345678-1-12 format!'),
'action_text': _('View partner(s)'),
},
'partner_address_missing': {
'records': self.partner_id.commercial_partner_id.filtered(
lambda p: p.is_company and (not p.country_id or not p.zip or not p.city or not p.street),
),
'message': _('Please set partner Country, Zip, City and Street!'),
'action_text': _('View partner(s)'),
},
'invoice_date_not_today': {
'records': self.filtered(lambda m: m.invoice_date != fields.Date.context_today(m)),
'message': _('Please set invoice date to today!'),
'action_text': _('View invoice(s)'),
},
'invoice_chain_not_confirmed': {
'records': self.env['account.move'].union(*[
move._l10n_hu_get_chain_base()._l10n_hu_get_chain_invoices().filtered(
lambda m: (
m.id < move.id
and m.l10n_hu_edi_state in [False, 'rejected', 'cancelled']
and m not in self
)
)
for move in self
]),
'message': _('The following invoices appear to be earlier in the chain, but have not yet been sent. Please send them first.'),
'action_text': _('View invoice(s)'),
},
'invoice_advance_not_paid': {
'records': advance_invoices.filtered(
lambda m: (
m.payment_state not in ['in_payment', 'paid', 'partial']
or m.l10n_hu_edi_state in [False, 'rejected', 'cancelled']
and m not in self # It's okay to send an advance and a final invoice together, as we sort by id before sending.
)
),
'message': _('All advance invoices must be paid and sent to NAV before the final invoice is issued.'),
'action_text': _('View advance invoice(s)'),
},
'invoice_line_not_one_vat_tax': {
'records': self.filtered(
lambda m: any(
len(l.tax_ids.filtered(lambda t: t.l10n_hu_tax_type)) != 1
for l in m.invoice_line_ids.filtered(lambda l: l.display_type == 'product')
)
),
'message': _('Please set exactly one VAT tax on each invoice line!'),
'action_text': _('View invoice(s)'),
},
'invoice_line_non_vat_taxes_misconfigured': {
'records': self.invoice_line_ids.tax_ids.filtered(
lambda t: not t.l10n_hu_tax_type and (not t.price_include or not t.include_base_amount)
),
'message': _("Please set any non-VAT (excise) taxes to be 'Included in Price' and 'Affects subsequent taxes'!"),
'action_text': _('View tax(es)'),
},
'invoice_line_vat_taxes_misconfigured': {
'records': self.invoice_line_ids.tax_ids.filtered(
lambda t: t.l10n_hu_tax_type and not t.is_base_affected
),
'message': _("Please set any VAT taxes to be 'Affected by previous taxes'!"),
'action_text': _('View tax(es)'),
},
}
errors = {
f"l10n_hu_edi_{check}": {
'message': values['message'],
'action_text': values['action_text'],
'action': values['records']._get_records_action(name=values['action_text']),
}
for check, values in checks.items()
if values['records']
}
if companies_missing_credentials := self.company_id.filtered(lambda c: not c.l10n_hu_edi_server_mode):
errors['l10n_hu_edi_company_credentials_missing'] = {
'message': _('Please set NAV credentials in the Accounting Settings!'),
'action_text': _('Open Accounting Settings'),
'action': self.env.ref('account.action_account_config').with_company(companies_missing_credentials[0])._get_action_dict(),
}
return errors
def _l10n_hu_edi_upload(self, connection):
""" Generate invoice XMLs and send to NAV. """
invoices_sorted = self.sorted(lambda m: m.id)
for invoice in invoices_sorted:
# If we come from the 'cancelled' state, this means the previous XML had been confirmed
# before it was cancelled.
# In that case, we want to keep it as a regular invoice attachment, for future reference.
if invoice.l10n_hu_edi_state == 'cancelled':
self.env['ir.attachment'].search([
('res_model', '=', self._name),
('res_id', '=', invoice.id),
('res_field', '=', 'l10n_hu_edi_attachment'),
]).write({
'res_field': False,
'name': f'{invoice.name.replace("/", "_")}_cancelled_{invoice.l10n_hu_edi_transaction_code}.xml',
})
# Set chain index
invoice._l10n_hu_edi_set_chain_index()
# Generate XML
invoice.l10n_hu_edi_attachment = base64.b64encode(invoice._l10n_hu_edi_generate_xml())
# Set name & mimetype on newly-created attachment.
attachment = self.env['ir.attachment'].search([
('res_model', '=', self._name),
('res_id', '=', invoice.id),
('res_field', '=', 'l10n_hu_edi_attachment'),
])
attachment.write({
'name': invoice.l10n_hu_edi_attachment_filename,
'mimetype': 'application/xml',
})
# Batch by company, with max 100 invoices per batch.
for __, batch_company in groupby(invoices_sorted, lambda m: m.company_id):
for batch in split_every(100, batch_company):
self.env['account.move'].union(*batch)._l10n_hu_edi_upload_single_batch(connection)
def _l10n_hu_edi_upload_single_batch(self, connection):
try:
token_result = connection.do_token_exchange(self.company_id.sudo()._l10n_hu_edi_get_credentials_dict())
except L10nHuEdiConnectionError as e:
return self.write({
'l10n_hu_edi_state': 'rejected',
'l10n_hu_edi_transaction_code': False,
'l10n_hu_edi_messages': {
'error_title': _('Could not authenticate with NAV. Check your credentials and try again.'),
'errors': e.errors,
'blocking_level': 'error',
},
})
for i, invoice in enumerate(self, start=1):
invoice.l10n_hu_edi_batch_upload_index = i
invoice_operations = [
{
'index': invoice.l10n_hu_edi_batch_upload_index,
'operation': 'CREATE' if invoice._l10n_hu_get_chain_base() == invoice else 'MODIFY',
'invoice_data': base64.b64decode(invoice.l10n_hu_edi_attachment),
}
for invoice in self
]
self.write({'l10n_hu_edi_send_time': fields.Datetime.now()})
try:
transaction_code = connection.do_manage_invoice(
self.company_id.sudo()._l10n_hu_edi_get_credentials_dict(),
token_result['token'],
invoice_operations,
)
except L10nHuEdiConnectionError as e:
if e.code == 'timeout':
return self.write({
'l10n_hu_edi_state': 'send_timeout',
'l10n_hu_edi_transaction_code': False,
'l10n_hu_edi_messages': {
'error_title': _('Invoice submission timed out. Please wait at least 6 minutes, then update the status.'),
'errors': e.errors,
'blocking_level': 'warning',
},
})
return self.write({
'l10n_hu_edi_state': 'rejected',
'l10n_hu_edi_transaction_code': False,
'l10n_hu_invoice_chain_index': 0,
'l10n_hu_edi_messages': {
'error_title': _('Invoice submission failed.'),
'errors': e.errors,
'blocking_level': 'error',
},
})
self.write({
'l10n_hu_edi_state': 'sent',
'l10n_hu_edi_transaction_code': transaction_code,
'l10n_hu_edi_messages': {
'error_title': _('Invoice submitted, waiting for response.'),
'errors': [],
}
})
def _l10n_hu_edi_query_status(self, connection):
""" Check the NAV invoice status. """
# We should update all invoices with the same company and transaction code at once.
invoices = self | self.search([
('company_id', 'in', self.company_id.ids),
('l10n_hu_edi_transaction_code', 'in', self.mapped('l10n_hu_edi_transaction_code')),
('l10n_hu_edi_state', 'in', ['sent', 'cancel_sent']),
])
# Querying status should be grouped by company and transaction code
for __, invoices_grouped in groupby(invoices, lambda m: (m.company_id, m.l10n_hu_edi_transaction_code)):
self.env['account.move'].union(*invoices_grouped)._l10n_hu_edi_query_status_single_batch(connection)
def _l10n_hu_edi_query_status_single_batch(self, connection):
""" Check the NAV status for invoices that share the same transaction code (uploaded in a single batch). """
try:
results = connection.do_query_transaction_status(
self.company_id.sudo()._l10n_hu_edi_get_credentials_dict(),
self[0].l10n_hu_edi_transaction_code,
)
except L10nHuEdiConnectionError as e:
if self.l10n_hu_edi_state == 'sent':
return self.write({
'l10n_hu_edi_messages': {
'error_title': _('The invoice was sent to the NAV, but there was an error querying its status.'),
'errors': e.errors,
'blocking_level': 'warning',
},
})
else:
return self.write({
'l10n_hu_edi_messages': {
'error_title': _('The annulment was sent to the NAV, but there was an error querying its status.'),
'errors': e.errors,
'blocking_level': 'warning',
},
})
for processing_result in results['processing_results']:
invoice = self.filtered(lambda m: str(m.l10n_hu_edi_batch_upload_index) == processing_result['index'])
if not invoice:
_logger.error(_('Could not match NAV transaction_code %(code)s, index %(index)s to an invoice in Odoo',
code=self[0].l10n_hu_edi_transaction_code,
index=processing_result['index']))
continue
invoice._l10n_hu_edi_process_query_transaction_result(processing_result, results['annulment_status'])
def _l10n_hu_edi_process_query_transaction_result(self, processing_result, annulment_status):
def get_errors_from_processing_result(processing_result):
return [
f'({message["validation_result_code"]}) {message["validation_error_code"]}: {message["message"]}'
for message in processing_result.get('business_validation_messages', []) + processing_result.get('technical_validation_messages', [])
]
self.ensure_one()
if processing_result['invoice_status'] in ['RECEIVED', 'PROCESSING', 'SAVED']:
# The invoice/annulment has not been processed yet.
if self.l10n_hu_edi_state in ['sent', 'send_timeout']:
self.write({
'l10n_hu_edi_state': 'sent',
'l10n_hu_edi_messages': {
'error_title': _('The invoice was received by the NAV, but has not been confirmed yet.'),
'errors': get_errors_from_processing_result(processing_result),
'blocking_level': 'warning',
},
})
elif self.l10n_hu_edi_state in ['cancel_sent', 'cancel_timeout']:
self.write({
'l10n_hu_edi_state': 'cancel_sent',
'l10n_hu_edi_messages': {
'error_title': _('The annulment request was received by the NAV, but has not been confirmed yet.'),
'errors': get_errors_from_processing_result(processing_result),
'blocking_level': 'warning',
},
})
elif processing_result['invoice_status'] == 'DONE':
if self.l10n_hu_edi_state in ['sent', 'send_timeout']:
if not processing_result['business_validation_messages'] and not processing_result['technical_validation_messages']:
self.write({
'l10n_hu_edi_state': 'confirmed',
'l10n_hu_edi_messages': {
'error_title': _('The invoice was successfully accepted by the NAV.'),
'errors': get_errors_from_processing_result(processing_result),
},
})
else:
self.write({
'l10n_hu_edi_state': 'confirmed_warning',
'l10n_hu_edi_messages': {
'error_title': _(
'The invoice was accepted by the NAV, but warnings were reported. '
'To reverse, create a credit note / debit note.'
),
'errors': get_errors_from_processing_result(processing_result),
'blocking_level': 'warning',
},
})
elif self.l10n_hu_edi_state in ['cancel_sent', 'cancel_timeout', 'cancel_pending']:
if annulment_status == 'NOT_VERIFIABLE':
self.write({
'l10n_hu_edi_state': 'confirmed_warning',
'l10n_hu_edi_messages': {
'error_title': _('The annulment request was rejected by NAV.'),
'errors': get_errors_from_processing_result(processing_result),
'blocking_level': 'error',
},
})
elif annulment_status == 'VERIFICATION_PENDING':
self.write({
'l10n_hu_edi_state': 'cancel_pending',
'l10n_hu_edi_messages': {
'error_title': _('The annulment request is pending, please confirm it on the OnlineSzámla portal.'),
'errors': get_errors_from_processing_result(processing_result),
'blocking_level': 'warning',
}
})
elif annulment_status == 'VERIFICATION_DONE':
# Annulling a base invoice will also annul all its modification invoices on NAV.
to_cancel = self if self.reversed_entry_id or self.debit_origin_id else self._l10n_hu_get_chain_invoices().filtered(lambda m: m.l10n_hu_edi_state)
to_cancel.write({
'l10n_hu_edi_state': 'cancelled',
'l10n_hu_invoice_chain_index': 0,
'l10n_hu_edi_messages': {
'error_title': _('The annulment request has been approved by the user on the OnlineSzámla portal.'),
'errors': get_errors_from_processing_result(processing_result),
}
})
to_cancel.button_cancel()
elif annulment_status == 'VERIFICATION_REJECTED':
self.write({
'l10n_hu_edi_state': 'confirmed_warning',
'l10n_hu_edi_messages': {
'error_title': _('The annulment request was rejected by the user on the OnlineSzámla portal.'),
'errors': get_errors_from_processing_result(processing_result),
'blocking_level': 'error',
}
})
elif processing_result['invoice_status'] == 'ABORTED':
if self.l10n_hu_edi_state in ['sent', 'send_timeout']:
self.write({
'l10n_hu_edi_state': 'rejected',
'l10n_hu_invoice_chain_index': 0,
'l10n_hu_edi_messages': {
'error_title': _('The invoice was rejected by the NAV.'),
'errors': get_errors_from_processing_result(processing_result),
'blocking_level': 'error',
},
})
elif self.l10n_hu_edi_state in ['cancel_sent', 'cancel_timeout', 'cancel_pending']:
self.write({
'l10n_hu_edi_state': 'confirmed_warning',
'l10n_hu_edi_messages': {
'error_title': _('The cancellation request could not be performed.'),
'errors': get_errors_from_processing_result(processing_result),
'blocking_level': 'error',
},
})
def _l10n_hu_edi_request_cancel(self, connection, code, reason):
""" Send a cancellation request for all invoices in `self`. """
# Batch by company, with max 100 annulment requests per batch.
for __, batch_company in groupby(self, lambda m: m.company_id):
for batch in split_every(100, batch_company):
self.env['account.move'].union(*batch)._l10n_hu_edi_request_cancel_single_batch(connection, code, reason)
def _l10n_hu_edi_request_cancel_single_batch(self, connection, code, reason):
for i, invoice in enumerate(self, start=1):
invoice.l10n_hu_edi_batch_upload_index = i
annulment_operations = [
{
'index': invoice.l10n_hu_edi_batch_upload_index,
'annulmentReference': invoice.name,
'annulmentCode': code,
'annulmentReason': reason,
}
for invoice in self
]
try:
token_result = connection.do_token_exchange(self.company_id.sudo()._l10n_hu_edi_get_credentials_dict())
except L10nHuEdiConnectionError as e:
return self.write({
'l10n_hu_edi_messages': {
'error_title': _('Could not authenticate with NAV. Check your credentials and try again.'),
'errors': e.errors,
'blocking_level': 'error',
},
})
self.write({'l10n_hu_edi_send_time': fields.Datetime.now()})
try:
transaction_code = connection.do_manage_annulment(
self.company_id.sudo()._l10n_hu_edi_get_credentials_dict(),
token_result['token'],
annulment_operations,
)
except L10nHuEdiConnectionError as e:
if e.code == 'timeout':
return self.write({
'l10n_hu_edi_state': 'cancel_timeout',
'l10n_hu_edi_messages': {
'error_title': _('Cancellation request timed out. Please wait at least 6 minutes, then update the status.'),
'errors': e.errors,
'blocking_level': 'warning',
},
})
return self.write({
'l10n_hu_edi_messages': {
'error_title': _('Cancellation request failed.'),
'errors': e.errors,
'blocking_level': 'error',
},
})
self.write({
'l10n_hu_edi_state': 'cancel_sent',
'l10n_hu_edi_transaction_code': transaction_code,
'l10n_hu_edi_messages': {
'error_title': _('Cancellation request submitted, waiting for response.'),
'errors': [],
}
})
# === EDI: XML generation === #
def _l10n_hu_edi_generate_xml(self):
invoice_data = self.env['ir.qweb']._render(
self._l10n_hu_edi_get_electronic_invoice_template(),
self._l10n_hu_edi_get_invoice_values(),
)
return etree.tostring(cleanup_xml_node(invoice_data, remove_blank_nodes=False), xml_declaration=True, encoding='UTF-8')
def _l10n_hu_edi_get_electronic_invoice_template(self):
""" For feature extensibility. """
return 'l10n_hu_edi.nav_online_invoice_xml_3_0'
def _l10n_hu_edi_get_invoice_values(self):
eu_country_codes = set(self.env.ref('base.europe').country_ids.mapped('code'))
def get_vat_data(partner, force_vat=None):
if partner.country_code == 'HU' or force_vat:
return {
'tax_number': partner.l10n_hu_group_vat or (force_vat or partner.vat),
'group_member_tax_number': partner.l10n_hu_group_vat and (force_vat or partner.vat),
}
elif partner.country_code in eu_country_codes:
return {'community_vat_number': partner.vat}
else:
return {'third_state_tax_id': partner.vat}
def format_bank_account_number(bank_account):
# Normalize IBANs (no spaces!)
if bank_account.acc_type == 'iban':
return normalize_iban(bank_account.acc_number)
else:
return bank_account.acc_number
supplier = self.company_id.partner_id
customer = self.partner_id.commercial_partner_id
currency_huf = self.env.ref('base.HUF')
currency_rate = self._l10n_hu_get_currency_rate()
base_invoice = self._l10n_hu_get_chain_base()
invoice_values = {
'invoice': self,
'invoiceIssueDate': self.invoice_date,
'completenessIndicator': False,
'modifyWithoutMaster': False,
'base_invoice': base_invoice if base_invoice != self else None,
'supplier': supplier,
'supplier_vat_data': get_vat_data(supplier, self.fiscal_position_id.foreign_vat),
'supplierBankAccountNumber': format_bank_account_number(self.partner_bank_id or supplier.bank_ids[:1]),
'individualExemption': self.company_id.l10n_hu_tax_regime == 'ie',
'customer': customer,
'customerVatStatus': (not customer.is_company and 'PRIVATE_PERSON') or (customer.country_code == 'HU' and 'DOMESTIC') or 'OTHER',
'customer_vat_data': get_vat_data(customer) if customer.is_company else None,
'customerBankAccountNumber': format_bank_account_number(customer.bank_ids[:1]),
'smallBusinessIndicator': self.company_id.l10n_hu_tax_regime == 'sb',
'exchangeRate': currency_rate,
'cashAccountingIndicator': self.company_id.l10n_hu_tax_regime == 'ca',
'shipping_partner': self.partner_shipping_id,
'sales_partner': self.user_id,
'mergedItemIndicator': False,
'format_bool': format_bool,
'float_repr': float_repr,
'lines_values': [],
}
sign = 1.0 if self.is_inbound() else -1.0
prev_chain_invoices = base_invoice._l10n_hu_get_chain_invoices().filtered(
lambda m: m.l10n_hu_invoice_chain_index and m.l10n_hu_invoice_chain_index < self.l10n_hu_invoice_chain_index
)
first_line_number = sum(
len(move.line_ids.filtered(lambda l: l.display_type in ['product', 'rounding']))
for move in prev_chain_invoices
) + 1
for (line_number, line) in enumerate(
self.line_ids.filtered(lambda l: l.display_type in ['product', 'rounding']).sorted(lambda l: l.display_type),
start=first_line_number,
):
line_values = {
'line': line,
'lineNumber': line_number - first_line_number + 1,
'lineNumberReference': base_invoice != self and line_number,
'lineExpressionIndicator': line.product_id and line.product_uom_id,
'lineNatureIndicator': {False: 'OTHER', 'service': 'SERVICE'}.get(line.product_id.type, 'PRODUCT'),
'lineDescription': line.name.replace('\n', ' '),
}
if 'is_downpayment' in line and line.is_downpayment:
# Advance and final invoices.
line_values['advanceIndicator'] = True
if not self._is_downpayment():
# This is a final invoice that deducts one or more advance invoices.
# In this case, we add a reference to the *last-paid* advance invoice (NAV only allows us to report one) if one exists,
# otherwise we don't add anything.
advance_invoices = line._get_downpayment_lines().mapped('move_id').filtered(lambda m: m.state == 'posted')
reconciled_moves = advance_invoices._get_reconciled_amls().move_id
last_reconciled_payment = reconciled_moves.filtered(lambda m: m.origin_payment_id or m.statement_line_id).sorted('date', reverse=True)[:1]
if last_reconciled_payment:
line_values.update({
'advanceOriginalInvoice': advance_invoices.filtered(lambda m: last_reconciled_payment in m._get_reconciled_amls().move_id)[0].name,
'advancePaymentDate': last_reconciled_payment.date,
'advanceExchangeRate': last_reconciled_payment._l10n_hu_get_currency_rate(),
})
if line.display_type == 'product':
vat_tax = line.tax_ids.filtered(lambda t: t.l10n_hu_tax_type)
price_unit_signed = sign * line.price_unit
price_net_signed = self.currency_id.round(price_unit_signed * line.quantity * (1 - line.discount / 100.0))
discount_value_signed = self.currency_id.round(price_unit_signed * line.quantity - price_net_signed)
price_total_signed = sign * line.price_total
vat_amount_signed = self.currency_id.round(price_total_signed - price_net_signed)
line_values.update({
'vat_tax': vat_tax,
'vatPercentage': float_round(vat_tax.amount / 100.0, 4),
'quantity': line.quantity,
'unitPrice': price_unit_signed,
'unitPriceHUF': currency_huf.round(price_unit_signed * currency_rate),
'discountValue': discount_value_signed,
'discountRate': line.discount / 100.0,
'lineNetAmount': price_net_signed,
'lineNetAmountHUF': currency_huf.round(price_net_signed * currency_rate),
'lineVatData': not self.currency_id.is_zero(vat_amount_signed),
'lineVatAmount': vat_amount_signed,
'lineVatAmountHUF': currency_huf.round(vat_amount_signed * currency_rate),
'lineGrossAmountNormal': price_total_signed,
'lineGrossAmountNormalHUF': currency_huf.round(price_total_signed * currency_rate),
})
elif line.display_type == 'rounding':
atk_tax = self.env['account.tax'].search(
[
('type_tax_use', '=', 'sale'),
('l10n_hu_tax_type', '=', 'ATK'),
('company_id', '=', self.company_id.id),
],
limit=1,
)
if not atk_tax:
raise UserError(_('Please create a sales tax with type ATK (outside the scope of the VAT Act).'))
amount_huf = line.balance if self.company_id.currency_id == currency_huf else currency_huf.round(line.amount_currency * currency_rate)
line_values.update({
'vat_tax': atk_tax,
'vatPercentage': float_round(atk_tax.amount / 100.0, 4),
'quantity': 1.0,
'unitPrice': -line.amount_currency,
'unitPriceHUF': -amount_huf,
'lineNetAmount': -line.amount_currency,
'lineNetAmountHUF': -amount_huf,
'lineVatData': False,
'lineGrossAmountNormal': -line.amount_currency,
'lineGrossAmountNormalHUF': -amount_huf,
})
line_values['lineDescription'] = line_values['lineDescription'] or line.product_id.display_name
invoice_values['lines_values'].append(line_values)
is_company_huf = self.company_id.currency_id == currency_huf
tax_amounts_by_tax = {
line.tax_line_id: {
'vatRateVatAmount': -line.amount_currency,
'vatRateVatAmountHUF': -line.balance if is_company_huf else currency_huf.round(-line.amount_currency * currency_rate),
}
for line in self.line_ids.filtered(lambda l: l.tax_line_id.l10n_hu_tax_type)
}
invoice_values['tax_summary'] = [
{
'vat_tax': vat_tax,
'vatPercentage': float_round(vat_tax.amount / 100.0, 4),
'vatRateNetAmount': self.currency_id.round(sum(l['lineNetAmount'] for l in lines_values_by_tax)),
'vatRateNetAmountHUF': currency_huf.round(sum(l['lineNetAmountHUF'] for l in lines_values_by_tax)),
'vatRateVatAmount': tax_amounts_by_tax.get(vat_tax, {}).get('vatRateVatAmount', 0.0),
'vatRateVatAmountHUF': tax_amounts_by_tax.get(vat_tax, {}).get('vatRateVatAmountHUF', 0.0),
}
for vat_tax, lines_values_by_tax in groupby(invoice_values['lines_values'], lambda l: l['vat_tax'])
]
total_vat = self.currency_id.round(sum(tax_vals['vatRateVatAmount'] for tax_vals in invoice_values['tax_summary']))
total_vat_huf = currency_huf.round(sum(tax_vals['vatRateVatAmountHUF'] for tax_vals in invoice_values['tax_summary']))
total_gross = self.amount_total_in_currency_signed
total_gross_huf = self.amount_total_signed if is_company_huf else currency_huf.round(self.amount_total_in_currency_signed * currency_rate)
total_net = self.currency_id.round(total_gross - total_vat)
total_net_huf = currency_huf.round(total_gross_huf - total_vat_huf)
invoice_values.update({
'invoiceNetAmount': total_net,
'invoiceNetAmountHUF': total_net_huf,
'invoiceVatAmount': total_vat,
'invoiceVatAmountHUF': total_vat_huf,
'invoiceGrossAmount': total_gross,
'invoiceGrossAmountHUF': total_gross_huf,
})
return invoice_values
# === PDF generation === #
def _get_name_invoice_report(self):
self.ensure_one()
return 'l10n_hu_edi.report_invoice_document' if self.country_code == 'HU' else super()._get_name_invoice_report()
def _l10n_hu_get_invoice_totals_for_report(self):
""" In Hungary, tax amounts should appear negative on credit notes.
We therefore apply a post-processing to the tax totals to make them negative. """
def invert_dict(dictionary, keys_to_invert):
""" Replace the values of keys_to_invert by their negative. """
dictionary.update({
key: -value
for key, value in dictionary.items()
if key in keys_to_invert
})
self.ensure_one()
tax_totals = self.tax_totals
if not tax_totals:
return tax_totals
fields_to_reverse = (
'base_amount_currency', 'base_amount',
'display_base_amount_currency', 'display_base_amount',
'tax_amount_currency', 'tax_amount',
'total_amount_currency', 'total_amount',
'cash_rounding_base_amount_currency', 'cash_rounding_base_amount',
)
if self.move_type in ('out_refund', 'in_refund'):
invert_dict(tax_totals, fields_to_reverse)
for subtotal in tax_totals['subtotals']:
invert_dict(subtotal, fields_to_reverse)
for tax_group in subtotal['tax_groups']:
invert_dict(tax_group, fields_to_reverse)
currency_huf = self.env.ref('base.HUF')
tax_totals['total_vat_amount_in_huf'] = sum(
line.balance for line in self.line_ids.filtered(lambda l: l.tax_line_id.l10n_hu_tax_type)
) * (1 if self.is_purchase_document() else -1)
tax_totals['formatted_total_vat_amount_in_huf'] = formatLang(
self.env, tax_totals['total_vat_amount_in_huf'], currency_obj=currency_huf
)
return tax_totals
class AccountInvoiceLine(models.Model):
_inherit = 'account.move.line'
@api.depends('move_id.delivery_date')
def _compute_currency_rate(self):
super()._compute_currency_rate()
# In Hungary, the currency rate should be based on the delivery date.
for line in self.filtered(lambda l: l.move_id.country_code == 'HU' and l.currency_id):
line.currency_rate = self.env['res.currency']._get_conversion_rate(
from_currency=line.company_currency_id,
to_currency=line.currency_id,
company=line.company_id,
date=line.move_id.delivery_date or line.move_id.invoice_date or line.move_id.date or fields.Date.context_today(line),
)