# Part of Odoo. See LICENSE file for full copyright and licensing details.
import contextlib
import json
import jwt
import logging
import re
from datetime import datetime, date
from markupsafe import Markup
from odoo import Command, _, api, fields, models, modules
from odoo.tools import split_every
from odoo.exceptions import RedirectWarning, UserError
from .irn_exception import IrnException
_logger = logging.getLogger(__name__)
UOM_REF_MAP = {
"CMS": "uom.product_uom_cm",
"CBM": "uom.product_uom_cubic_meter",
"DOZ": "uom.product_uom_dozen",
"GMS": "uom.product_uom_gram",
"KGS": "uom.product_uom_kgm",
"KME": "uom.product_uom_km",
"LTR": "uom.product_uom_litre",
"MTR": "uom.product_uom_meter",
"QTL": "uom.product_uom_yard",
"SQF": "uom.product_uom_square_foot",
"SQM": "uom.product_uom_square_meter",
"TON": "uom.product_uom_ton",
"UNT": "uom.product_uom_unit",
"YDS": "uom.product_uom_yard",
}
MOVE_TYPE_MAPPING = {
'INV': 'in_invoice',
'CRN': 'in_refund',
'DBN': 'in_invoice'
}
IRN_PATTERN = r'^[a-z0-9]{64}$'
class AccountMove(models.Model):
_inherit = 'account.move'
l10n_in_transaction_type = fields.Selection(
selection=[
("inter_state", "Inter State"),
("intra_state", "Intra State"),
],
string="Indian Transaction Type",
compute="_compute_l10n_in_transaction_type",
store=True,
)
l10n_in_fetch_vendor_edi_feature_enabled = fields.Boolean(related='company_id.l10n_in_fetch_vendor_edi_feature')
# gstr related fields
l10n_in_exception = fields.Html("Exception", copy=False)
l10n_in_account_return_id = fields.Many2one("account.return", "GST Tax Return", copy=False)
l10n_in_gstr2b_reconciliation_status = fields.Selection(selection=[
("pending", "Pending"),
("matched", "Fully Matched"),
("partially_matched", "Partially Matched"),
("exception", "Exception"),
("bills_not_in_gstr2", "Bills Not in GSTR-2"),
("manually_matched", "Manually Matched"),
("gstr2_bills_not_in_odoo", "GSTR-2 Bills not in Odoo")],
string="GSTR-2B Reconciliation",
readonly=True,
default="pending",
tracking=True,
copy=False
)
l10n_in_reversed_entry_warning = fields.Boolean(
string="Display reversed entry warning",
compute="_compute_l10n_in_reversed_entry_warning"
)
l10n_in_irn_number = fields.Char('IRN Number', readonly=True, copy=False)
l10n_in_gstr_activate_einvoice_fetch = fields.Selection(related="company_id.l10n_in_gstr_activate_einvoice_fetch")
l10n_in_gst_efiling_feature_enabled = fields.Boolean(related="company_id.l10n_in_gst_efiling_feature")
_uniq_l10n_in_irn_number = models.Constraint(
'unique(company_id, l10n_in_irn_number)',
'Irn number must be unique for bills per company.',
)
@api.depends("country_code", "l10n_in_state_id", "company_id", "commercial_partner_id", "journal_id")
def _compute_l10n_in_transaction_type(self):
self.fetch(['country_code', 'l10n_in_state_id', "company_id", "commercial_partner_id", "journal_id"])
for move in self:
if move.country_code == "IN":
# POS and other non-purchase journal moves' state should be compared against the company's state.
state = move.commercial_partner_id.state_id if move.journal_id.type == 'purchase' else move.company_id.state_id
if move.l10n_in_state_id and move.l10n_in_state_id == state:
move.l10n_in_transaction_type = 'intra_state'
else:
move.l10n_in_transaction_type = 'inter_state'
else:
move.l10n_in_transaction_type = False
@api.depends('move_type', 'reversed_entry_id', 'state', 'invoice_date', 'invoice_line_ids.tax_ids', 'checked')
def _compute_l10n_in_reversed_entry_warning(self):
for move in self:
if (
move.country_code == 'IN'
and move.l10n_in_gst_efiling_feature_enabled
and move.move_type == 'out_refund'
and move.invoice_date
and move.reversed_entry_id
and move.invoice_line_ids.tax_ids
and (move.state == 'draft' or (move.state == 'posted' and not move.checked))
):
move.l10n_in_reversed_entry_warning = move.reversed_entry_id.invoice_date < move._l10n_in_get_fiscal_year_start_date(
move.company_id, move.invoice_date)
else:
move.l10n_in_reversed_entry_warning = False
def _l10n_in_get_fiscal_year_start_date(self, company, reversal_move_date):
fiscal_year_start_month = (int(company.fiscalyear_last_month) % 12) + 1
fiscal_year_start_date = date(reversal_move_date.year, fiscal_year_start_month, 1)
if reversal_move_date.month <= 11:
fiscal_year_start_date = fiscal_year_start_date.replace(year=reversal_move_date.year - 1)
return fiscal_year_start_date
def _post(self, soft=True):
"""
The parent `_post` method sets the `checked` flag based on the journal
configuration. After posting, we explicitly reset `checked` to False
for invoices that have a credit note (reversed entry) warning.
"""
warning_invoices = self.filtered(lambda i: i.l10n_in_reversed_entry_warning)
to_post = super()._post(soft=soft)
for invoice in self:
if invoice.l10n_in_gstr2b_reconciliation_status == "gstr2_bills_not_in_odoo":
invoice.l10n_in_gstr2b_reconciliation_status = "pending"
warning_invoices.checked = False
return to_post
def button_draft(self):
res = super().button_draft()
self.write({
"l10n_in_exception": False,
"l10n_in_gstr2b_reconciliation_status": "pending",
"l10n_in_account_return_id": False,
})
return res
def l10n_in_update_move_using_irn(self):
""" Fetch the attachment from IRN and use it to update the invoice.
If an appropriate attachment already exists, update the invoice with that attachment instead,
reversing modifications to the invoice.
:returns: action to refresh the form view.
"""
for company in self.company_id:
self.env['account.return']._l10n_in_check_config(
company=company
)
JSON_MIMETYPE = 'application/json'
STATUS_CANCELLED = 'CNL'
for move in self:
# Filter valid JSON attachments
attachment = move.attachment_ids.filtered(
lambda a: a.mimetype == JSON_MIMETYPE and self._is_l10n_in_irn_json(a.raw)
)[:1]
if attachment:
move._extend_with_attachments(move._to_files_data(attachment))
continue
try:
# Retrieve IRN details if no valid attachment is found
gov_json_data = self._l10n_in_retrieve_details_from_irn(move.l10n_in_irn_number, move.company_id)
except IrnException as e:
if str(e) == 'no-credit':
message = self._l10n_in_edi_get_iap_buy_credits_message()
else:
message = str(e)
move.message_post(
body=Markup("%s
%s") % (_("Fetching IRN details failed with error(s):"), message)
)
continue
attachment = self.env['ir.attachment'].create({
# Limit the name to 45 characters to avoid exceeding the limit on e-invoice portal
'name': f'{move.l10n_in_irn_number[:45]}.json',
'mimetype': JSON_MIMETYPE,
'raw': json.dumps(gov_json_data),
'res_model': 'account.move',
'res_id': move.id,
})
move._extend_with_attachments(move._to_files_data(attachment), new=True)
if gov_json_data.get('Status') == STATUS_CANCELLED and move.state != 'cancel':
move.message_post(body=_("This bill has been marked as canceled based on the e-invoice status."))
move.button_cancel()
return {'type': 'ir.actions.act_window_close'} # Refresh the form to show the key
# ========================================
# Cron Method
# ========================================
def _l10n_in_cron_update_with_irn(self, job_count=10):
""" Update draft account moves with IRN details for Indian companies.
:param job_count: the number of moves to process in each batch.
"""
domain = [
('company_id.account_fiscal_country_id.code', '=', 'IN'),
('company_id.l10n_in_fetch_vendor_edi_feature', '=', True),
('company_id.l10n_in_gstr_activate_einvoice_fetch', '=', 'automatic'),
('company_id.l10n_in_gstr_gst_token_validity', '>', fields.Datetime.now()),
('state', '=', 'draft'),
('l10n_in_irn_number', '!=', False),
('posted_before', '=', False),
('line_ids', 'not any', [(1, '=', 1)]),
]
moves = self.search(domain)
for move_batch in split_every(job_count, moves):
try:
move_batch.l10n_in_update_move_using_irn()
except (RedirectWarning, UserError):
_logger.exception('Error when update Bill with IRN')
if not modules.module.current_test:
self.env.cr.commit()
# ========================================
# Import Vendor Bills and Credit Notes
# ========================================
@api.model
def _is_l10n_in_irn_json(self, content):
""" Determine whether the given file content is a vendor bill JSON retrieved from IRN. """
with contextlib.suppress(json.JSONDecodeError, UnicodeDecodeError):
content = json.loads(content)
return all(key in content for key in (
'Irn', 'AckNo', 'AckDt', 'SignedInvoice', 'Status',
))
def _get_import_file_type(self, file_data):
""" Identify IRN JSON files. """
# EXTENDS 'account'
if self._is_l10n_in_irn_json(file_data['raw']):
return 'l10n_in.irn'
return super()._get_import_file_type(file_data)
def _get_edi_decoder(self, file_data, new=False):
# EXTENDS 'account'
if file_data['import_file_type'] == 'l10n_in.irn':
return {
'priority': 20,
'decoder': self._l10n_in_irn_import_invoice,
}
return super()._get_edi_decoder(file_data, new)
def _l10n_in_irn_import_invoice(self, invoice, data, is_new):
""" Import invoice details from IRN data and update the corresponding invoice.
Signed invoice data is decoded before being passed to the import function.
:param invoice: the invoice record to be updated.
:param data: dict representing the signed invoice attachment.
:param is_new: indicates if the invoice is newly created.
:returns: True if the import was successful, False if it fails.
"""
if invoice.invoice_line_ids:
return invoice._reason_cannot_decode_has_invoice_lines()
try:
# Load content from the data
attachment_content = json.loads(data['raw'])
signed_invoice = attachment_content['SignedInvoice']
except (json.JSONDecodeError, KeyError):
return
# Decode the signed invoice using JWT
try:
decoded_data = jwt.decode(signed_invoice, options={'verify_signature': False})
decoded_invoice_data = json.loads(decoded_data.get('data', '{}'))
except (json.JSONDecodeError, jwt.exceptions.DecodeError):
# Post a message on the invoice regarding the failure
invoice.message_post(body=_("Failed to decode signed invoice."))
return
# Update the invoice with decoded data
with self._get_edi_creation() as self:
self._l10n_in_update_bill_with_irn_details(decoded_invoice_data)
def _l10n_in_update_bill_with_irn_details(self, content):
""" Update the invoice with details retrieved from IRN.
:param content: dict representing all invoice details.
:returns: True if the update was successful
"""
def _get_tax(rate, tag_ids, is_rcm_invoice=False):
domain = [
('type_tax_use', '=', 'purchase'),
('amount', '=', rate),
'|',
('repartition_line_ids.tag_ids', 'in', tag_ids),
('children_tax_ids.repartition_line_ids.tag_ids', 'in', tag_ids)
]
if is_rcm_invoice:
domain.append(('l10n_in_reverse_charge', '=', True))
return self.env['account.tax'].search(domain, limit=1)
bill_details = content['DocDtls']
seller_details = content['SellerDtls']
item_list = content['ItemList']
value_details = content['ValDtls']
trans_details = content['TranDtls']
buyer_details = content['BuyerDtls']
if self.company_id.sudo().l10n_in_edi_production_env and buyer_details.get('Gstin'):
if buyer_details['Gstin'] != self.company_id.vat:
self.message_post(
body=_(
"Mismatch in GSTIN: The buyer's GSTIN in the document (%(buyer)s) does not match the company's registered GSTIN (%(company)s).\n"
"Please verify that the document belongs to the current company.",
buyer=buyer_details['Gstin'],
company=self.company_id.vat,
)
)
return False
move_vals = {
'l10n_in_irn_number': content.get('Irn'),
'move_type': MOVE_TYPE_MAPPING.get(bill_details.get('Typ'), 'in_invoice'),
'invoice_line_ids': [Command.clear(),],
}
# Find a partner if one exists, else create one
if seller_details.get('Gstin'):
ResPartner = self.env['res.partner']
seller_partner = ResPartner.search([
('vat', '=', seller_details['Gstin']),
], limit=1)
if not seller_partner:
partner_vals = ResPartner._l10n_in_get_partner_vals_by_vat(seller_details['Gstin'])
if partner_vals:
seller_partner = ResPartner.create(partner_vals)
move_vals['partner_id'] = seller_partner.id
if (bill_date := bill_details.get('Dt')):
move_vals['invoice_date'] = datetime.strptime(bill_date, '%d/%m/%Y').strftime('%Y-%m-%d')
move_vals['ref'] = bill_details.get('No')
igst_tag_ids = self.env.ref('l10n_in.tax_tag_igst').ids
gst_tag_ids = (self.env.ref('l10n_in.tax_tag_cgst') + self.env.ref('l10n_in.tax_tag_sgst')).ids
uom_map = {
irn_uom: self.env['ir.model.data']._xmlid_to_res_id(xmlid)
for irn_uom, xmlid in UOM_REF_MAP.items()
}
other_charges = value_details.get('OthChrg', 0)
cess_charges = 0
not_mapped_tax_values = {}
is_rcm_invoice = 1 if trans_details.get('RegRev') == "Y" else 0
for item in item_list:
line_dict = {}
gst_rate = item.get('GstRt')
if gst_rate:
is_igst = bool(item.get('IgstAmt'))
tax_tag_ids = igst_tag_ids if is_igst else gst_tag_ids
applicable_tax = _get_tax(gst_rate, tax_tag_ids, is_rcm_invoice)
if applicable_tax:
line_dict['tax_ids'] = [Command.link(applicable_tax.id)]
else:
tax_key_base = (gst_rate, "IGST" if is_igst else "GST")
tax_amount = item.get('IgstAmt', 0) if is_igst else item.get('CgstAmt', 0) + item.get('SgstAmt', 0)
not_mapped_tax_values.setdefault(tax_key_base, 0)
not_mapped_tax_values[tax_key_base] += tax_amount
if is_rcm_invoice:
tax_key_rcm = tax_key_base + ("RC",)
not_mapped_tax_values.setdefault(tax_key_rcm, 0)
not_mapped_tax_values[tax_key_rcm] -= tax_amount
line_dict['product_uom_id'] = uom_map.get(item.get('Unit'))
quantity = item.get('Qty') or 1
price_unit = item.get('UnitPrice')
if price_unit:
line_dict['discount'] = (item.get('Discount', 0.0) / item.get('TotAmt', 1.0)) * 100.0 if item.get('TotAmt') else 0.0
else:
# For service-type products where UnitPrice might be 0,
# calculate and replace it with AssAmt divided by quantity to ensure accurate unit pricing and proper handling.
price_unit = item.get('AssAmt', 0) / quantity
move_vals['invoice_line_ids'].append(
Command.create({
**line_dict,
'name': item.get('PrdDesc'),
'l10n_in_hsn_code': item.get('HsnCd'),
'quantity': quantity,
'price_unit': price_unit,
})
)
other_charges += item.get('OthChrg', 0)
cess_charges += sum(
item.get(key, 0)
for key in ('CesAmt', 'CesNonAdvlAmt', 'StateCesAmt', 'StateCesNonAdvlAmt')
)
# Create other charges line
if other_charges:
move_vals['invoice_line_ids'].append(Command.create({
'name': _("Other Charges"),
'price_unit': other_charges,
}))
# Create cess charges line
if cess_charges:
move_vals['invoice_line_ids'].append(Command.create({
'name': _("CESS Charges"),
'price_unit': cess_charges,
}))
# Create invoice lines for the grouped tax values
for tax_group, tax_amount in not_mapped_tax_values.items():
line_name = f"{tax_group[0]}% ({tax_group[1]}){' - Reverse Charge' if is_rcm_invoice else ''}"
move_vals['invoice_line_ids'].append(Command.create({
'name': line_name,
'price_unit': tax_amount,
}))
# Create rounding value line
if (rounding_amount := value_details.get('RndOffAmt')):
move_vals['invoice_line_ids'].append(Command.create({
'name': _("Rounding Value"),
'price_unit': rounding_amount,
}))
# Create discount value line
if (discount_amount := value_details.get('Discount')):
move_vals['invoice_line_ids'].append(Command.create({
'name': _("Discount Value"),
'price_unit': discount_amount * -1,
}))
self.write(move_vals)
return True
def _l10n_in_retrieve_details_from_irn(self, irn_number, company_id):
""" Retrieve signed invoice details for a given IRN number via the IAP proxy service.
:param irn_number: IRN number for which signed details are to be retrieved.
:param company_id: company for which the request is made.
:returns: dict containing signed IRN details.
"""
response = self.env['account.return']._l10n_in_reports_request(
url="/iap/l10n_in_reports/1/einvoice/irndtl",
params={
"irn_number": irn_number,
"auth_token": company_id.sudo().l10n_in_gstr_gst_token,
},
company=company_id,
)
data = response.get('data', {}).get('data', {})
if data and 'SignedInvoice' in data:
return data
errors = response.get('error', {})
raise IrnException(errors)
def _l10n_in_get_bill_from_irn(self, irn):
bill_action = self.env['ir.actions.act_window']._for_xml_id('account.action_move_in_invoice_type')
irn_lower = irn and irn.lower() or irn
if not re.match(IRN_PATTERN, irn_lower):
return super()._l10n_in_get_bill_from_irn(irn)
if (
self.env.company.l10n_in_fetch_vendor_edi_feature
and self.env.company.l10n_in_gstr_activate_einvoice_fetch in ('manual', 'automatic')
):
match_bill = self.search([('l10n_in_irn_number', '=', irn_lower)], limit=1)
if not match_bill:
match_bill = self.create({
'l10n_in_irn_number': irn_lower,
'move_type': 'in_invoice',
})
if self.env.company.l10n_in_gstr_activate_einvoice_fetch == "automatic":
match_bill.l10n_in_update_move_using_irn()
bill_action['views'] = [(self.env.ref('account.view_move_form').id, 'form')]
bill_action['res_id'] = match_bill.id
return {'action': bill_action}
return {'warning': _("To Get Bill by IRN First activate Fetch Vendor E-Invoiced Document in setting.")}
# ==========================================
# Vendor Bills Manual Matching with GSTR-2B
# ==========================================
def _l10n_in_change_gstr2b_reconciliation_status(self, status, message=None):
""" Manually match the vendor bill with GSTR-2B bills. """
self.write({'l10n_in_gstr2b_reconciliation_status': status})
if message:
self.message_post(body=message)
def action_l10n_in_bill_set_gstr2b_manual_matching(self):
self.ensure_one()
if (
self.move_type in ['in_invoice', 'in_refund']
and self.l10n_in_gstr2b_reconciliation_status == 'partially_matched'
):
exception_message = Markup("%s
%s") % (_("Manually matched with GSTR-2B bills:"), self.l10n_in_exception)
self._l10n_in_change_gstr2b_reconciliation_status(status='manually_matched', message=exception_message)
def action_l10n_in_bill_reset_gstr2b_manual_matching(self):
""" Reset the manual matching of the vendor bill with GSTR-2B bills. """
self.ensure_one()
if (
self.move_type in ['in_invoice', 'in_refund']
and self.l10n_in_gstr2b_reconciliation_status == 'manually_matched'
):
self._l10n_in_change_gstr2b_reconciliation_status(status='partially_matched')
# ===============================
# QR Code Bill Scan Methods
# ===============================
@api.model
def l10n_in_get_bill_from_qr_raw(self, qr_raw):
try:
qr_json = jwt.decode(qr_raw, options={'verify_signature': False})
except jwt.exceptions.DecodeError:
res = self._l10n_in_get_bill_from_irn(qr_raw)
if res:
return res
return {'warning': _("Scanned QR/IRN it's not E-Invoice QR/IRN code. Please scan/enter E-invoice QR/IRN code")}
qr_json_data = json.loads(qr_json.get('data', '{}'))
is_valid = self._l10n_in_validate_qr_data(qr_json_data)
if not is_valid:
return {'warning': _("Scanned QR/IRN is not appropriate as per E-Invoice QR/IRN")}
default_journal = self.env['account.journal'].search([
('type', '=', 'purchase'),
('company_id', '=', self.env.company.id)], limit=1)
bill_action = self.env.ref('account.action_move_in_invoice_type')._get_action_dict()
bill_action.update({
'views': [[False, "form"]],
'context': {
'create': False, # If new button is clicked then below default values will be set again.
'default_ref': qr_json_data.get('DocNo'),
'default_l10n_in_irn_number': qr_json_data.get('Irn') and qr_json_data['Irn'].lower() or False,
'default_journal_id': default_journal.id,
'default_invoice_date': datetime.strptime(qr_json_data.get('DocDt'), "%d/%m/%Y"),
'default_move_type': MOVE_TYPE_MAPPING.get(qr_json_data.get('DocTyp')),
'default_partner_id': self._l10n_in_get_partner_from_gstin(qr_json_data.get('SellerGstin')),
'default_invoice_line_ids': self.env.company.extract_single_line_per_tax and [
Command.create(self._l10n_in_get_move_lines_vals_from_qr_data(default_journal, qr_json_data))] or [],
}
})
bill_action = {'action': bill_action}
return bill_action
@api.model
def _l10n_in_get_move_lines_vals_from_qr_data(self, journal, qr_data):
price_unit = qr_data.get('TotInvVal')
move_line_vals = {
'name': _('HSN: %s', qr_data.get('MainHsnCode')),
'quantity': 1,
'price_unit': price_unit,
'account_id': journal.default_account_id.id,
}
products = self.env['product.product'].search([('l10n_in_hsn_code', '=', qr_data.get('MainHsnCode'))], limit=2)
if products:
taxes = products[0].supplier_taxes_id # Same HSN code -> same taxes
if taxes:
taxes_on_line = igst_taxes = taxes
fiscal_position = self.env.ref(f'account.{self.env.company.id}_fiscal_position_in_inter_state', raise_if_not_found=False)
if fiscal_position:
# The tax computation of taxes i.e. GST 5%, GST 18%, etc.
# the `total_exluded` value is computed incorrectly because of
# two sub-taxes i.e. SGST and CGST because the tax amount of the first tax
# is added to the next tax base value, it is better to use IGST for `total_exluded` value computation
igst_taxes = fiscal_position.map_tax(igst_taxes)
# If the supplier and the buyer are from different states then IGST will be applied
if qr_data.get('SellerGstin')[0:2] != qr_data.get('BuyerGstin')[0:2]:
taxes_on_line = igst_taxes
computed_tax = igst_taxes.filtered(lambda tax: not any(tax.flatten_taxes_hierarchy().mapped('price_include'))
).with_context(force_price_include=True).compute_all(price_unit=price_unit)
move_line_vals.update({
'price_unit': computed_tax.get('total_excluded'),
'tax_ids': [Command.set(taxes_on_line.ids)],
})
if len(products) == 1:
move_line_vals.update({'product_id': products.id})
return move_line_vals
def _l10n_in_get_partner_from_gstin(self, gstin):
ResPartner = self.env['res.partner']
partner = ResPartner._retrieve_partner(vat=gstin)
if partner:
return partner.id
partner_vals = ResPartner._l10n_in_get_partner_vals_by_vat(gstin)
if partner_vals:
partner = ResPartner.create(partner_vals)
#read_by_vat method is not providing the state/country code
#by using the following methods the state and country will set from the partner vat
partner.onchange_vat()
partner._onchange_state()
return partner.id
return False
@api.model
def _l10n_in_validate_qr_data(self, qr_data):
return all(key in qr_data for key in ["DocNo", "SellerGstin", "TotInvVal", "DocDt", "DocTyp", "MainHsnCode"])