Odoo18-Base/addons/l10n_eg_edi_eta/models/account_move.py

130 lines
6.6 KiB
Python
Raw Permalink Normal View History

2025-01-06 10:57:38 +07:00
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import logging
import json
from odoo import api, models, fields, _
from odoo.exceptions import ValidationError, UserError
from odoo.tools import float_is_zero
from odoo.tools.sql import column_exists, create_column
from datetime import datetime
_logger = logging.getLogger(__name__)
class AccountMove(models.Model):
_inherit = 'account.move'
l10n_eg_long_id = fields.Char(string='ETA Long ID', compute='_compute_eta_long_id')
l10n_eg_qr_code = fields.Char(string='ETA QR Code', compute='_compute_eta_qr_code_str')
l10n_eg_submission_number = fields.Char(string='Submission ID', compute='_compute_eta_response_data', store=True, copy=False)
l10n_eg_uuid = fields.Char(string='Document UUID', compute='_compute_eta_response_data', store=True, copy=False)
l10n_eg_eta_json_doc_id = fields.Many2one('ir.attachment', copy=False)
l10n_eg_signing_time = fields.Datetime('Signing Time', copy=False)
l10n_eg_is_signed = fields.Boolean(copy=False)
def _auto_init(self):
if not column_exists(self.env.cr, "account_move", "l10n_eg_uuid"):
create_column(self.env.cr, "account_move", "l10n_eg_uuid", "VARCHAR")
# Since l10n_eg_uuid columns does not exist we can assume l10n_eg_submission_number doesn't exist either
create_column(self.env.cr, "account_move", "l10n_eg_submission_number", "VARCHAR")
return super()._auto_init()
@api.depends('l10n_eg_eta_json_doc_id.raw')
def _compute_eta_long_id(self):
for rec in self:
response_data = rec.l10n_eg_eta_json_doc_id and json.loads(rec.l10n_eg_eta_json_doc_id.raw).get('response')
if response_data:
rec.l10n_eg_long_id = response_data.get('l10n_eg_long_id')
else:
rec.l10n_eg_long_id = False
@api.depends('invoice_date', 'l10n_eg_uuid', 'l10n_eg_long_id')
def _compute_eta_qr_code_str(self):
for move in self:
if move.invoice_date and move.l10n_eg_uuid and move.l10n_eg_long_id:
is_prod = move.company_id.l10n_eg_production_env
base_url = self.env['account.edi.format']._l10n_eg_get_eta_qr_domain(production_enviroment=is_prod)
qr_code_str = '%s/documents/%s/share/%s' % (base_url, move.l10n_eg_uuid, move.l10n_eg_long_id)
move.l10n_eg_qr_code = qr_code_str
else:
move.l10n_eg_qr_code = ''
@api.depends('l10n_eg_eta_json_doc_id.raw')
def _compute_eta_response_data(self):
for rec in self:
response_data = rec.l10n_eg_eta_json_doc_id and json.loads(rec.l10n_eg_eta_json_doc_id.raw).get('response')
if response_data:
rec.l10n_eg_uuid = response_data.get('l10n_eg_uuid')
rec.l10n_eg_submission_number = response_data.get('l10n_eg_submission_number')
rec.l10n_eg_long_id = response_data.get('l10n_eg_long_id')
else:
rec.l10n_eg_uuid = False
rec.l10n_eg_submission_number = False
rec.l10n_eg_long_id = False
def button_draft(self):
self.l10n_eg_eta_json_doc_id = False
self.l10n_eg_is_signed = False
return super().button_draft()
def action_post_sign_invoices(self):
# only sign invoices that are confirmed and not yet sent to the ETA.
invoices = self.filtered(lambda r: r.country_code == 'EG' and r.state == 'posted' and not r.l10n_eg_submission_number and r.edi_document_ids.filtered(lambda e: e.edi_format_id.code == 'eg_eta'))
if not invoices:
return
company_ids = invoices.mapped('company_id')
# since the middleware accepts only one drive at a time, we have to limit signing to one company at a time
if len(company_ids) > 1:
raise UserError(_('Please only sign invoices from one company at a time'))
company_id = company_ids[0]
drive_id = self.env['l10n_eg_edi.thumb.drive'].search([('user_id', '=', self.env.user.id),
('company_id', '=', company_id.id)])
if not drive_id:
raise ValidationError(_('Please setup a personal drive for company %s', company_id.name))
if not drive_id.certificate:
raise ValidationError(_('Please setup the certificate on the thumb drive menu'))
self.write({'l10n_eg_signing_time': datetime.utcnow()})
for invoice in invoices:
eta_invoice = self.env['account.edi.format']._l10n_eg_eta_prepare_eta_invoice(invoice)
attachment = self.env['ir.attachment'].create({
'name': _('ETA_INVOICE_DOC_%s', invoice.name),
'res_id': invoice.id,
'res_model': invoice._name,
'type': 'binary',
'raw': json.dumps(dict(request=eta_invoice)),
'mimetype': 'application/json',
'description': _('Egyptian Tax authority JSON invoice generated for %s.', invoice.name),
})
invoice.l10n_eg_eta_json_doc_id = attachment.id
return drive_id.action_sign_invoices(self)
def action_get_eta_invoice_pdf(self):
""" This is a pdf with the structure from the government. While we can use our own format,
some clients appreciate this to verify that all the data is there in case of confusion."""
self.ensure_one()
eta_invoice_pdf = self.env['account.edi.format']._l10n_eg_get_eta_invoice_pdf(self)
if eta_invoice_pdf.get('error', False):
_logger.warning('PDF Content Error: %s.', eta_invoice_pdf.get('error'))
return
self.with_context(no_new_invoice=True).message_post(body=_('ETA invoice has been received'),
attachments=[('ETA invoice of %s.pdf' % self.name,
eta_invoice_pdf.get('data'))])
def _l10n_eg_edi_exchange_currency_rate(self):
""" Calculate the rate based on the balance and amount_currency, so we recuperate the one used at the time"""
self.ensure_one()
from_currency = self.currency_id
to_currency = self.company_id.currency_id
if from_currency != to_currency and self.invoice_line_ids:
amount_currency = self.invoice_line_ids[0].amount_currency
if not float_is_zero(amount_currency, precision_rounding=from_currency.rounding):
return abs(self.invoice_line_ids[0].balance / amount_currency)
return 1.0