agdar/finance/zatca_service.py
2025-11-02 14:35:35 +03:00

427 lines
22 KiB
Python

"""
ZATCA E-Invoice Integration Service
This module handles:
- XML generation (UBL 2.1 format)
- Cryptographic signing (ECDSA)
- QR code generation
- FATOORA API integration (Clearance & Reporting)
- CSID management
"""
import base64
import hashlib
import json
import logging
from decimal import Decimal
from typing import Dict, Optional, Tuple
from xml.etree import ElementTree as ET
import requests
from django.conf import settings
from django.utils import timezone as django_timezone
logger = logging.getLogger(__name__)
class ZATCAService:
"""Service for ZATCA e-invoice operations."""
# ZATCA API Endpoints
SANDBOX_BASE_URL = "https://gw-fatoora.zatca.gov.sa/e-invoicing/simulation"
PRODUCTION_BASE_URL = "https://gw-fatoora.zatca.gov.sa/e-invoicing/core"
def __init__(self, use_sandbox=True):
"""
Initialize ZATCA service.
Args:
use_sandbox: Whether to use sandbox or production environment
"""
self.base_url = self.SANDBOX_BASE_URL if use_sandbox else self.PRODUCTION_BASE_URL
self.use_sandbox = use_sandbox
def generate_xml_invoice(self, invoice) -> str:
"""
Generate UBL 2.1 XML for invoice.
Args:
invoice: Invoice model instance
Returns:
str: XML content
"""
# Create root element with namespaces
namespaces = {
'xmlns': 'urn:oasis:names:specification:ubl:schema:xsd:Invoice-2',
'xmlns:cac': 'urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2',
'xmlns:cbc': 'urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2',
'xmlns:ext': 'urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2',
}
root = ET.Element('Invoice', namespaces)
# UBL Extensions (for signatures and QR code)
self._add_ubl_extensions(root, invoice)
# Invoice Type Code
invoice_type_code = self._get_invoice_type_code(invoice.invoice_type)
ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}InvoiceTypeCode',
name=self._get_invoice_type_name(invoice.invoice_type)).text = invoice_type_code
# Document Currency Code
ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}DocumentCurrencyCode').text = 'SAR'
# Tax Currency Code
ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}TaxCurrencyCode').text = 'SAR'
# Invoice ID (IRN)
ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID').text = invoice.invoice_number
# UUID
ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}UUID').text = str(invoice.id)
# Issue Date
ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}IssueDate').text = invoice.issue_date.isoformat()
# Issue Time
if invoice.issue_time:
ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}IssueTime').text = invoice.issue_time.isoformat()
# Billing Reference (for credit/debit notes)
if invoice.is_credit_or_debit_note and invoice.billing_reference_id:
self._add_billing_reference(root, invoice)
# Supplier Party (Seller)
self._add_supplier_party(root, invoice)
# Customer Party (Buyer)
self._add_customer_party(root, invoice)
# Payment Means (if applicable)
self._add_payment_means(root, invoice)
# Tax Total
self._add_tax_total(root, invoice)
# Legal Monetary Total
self._add_legal_monetary_total(root, invoice)
# Invoice Lines
self._add_invoice_lines(root, invoice)
# Convert to string
xml_string = ET.tostring(root, encoding='unicode', method='xml')
return xml_string
def _add_ubl_extensions(self, root, invoice):
"""Add UBL extensions for signatures and QR code."""
ext_elem = ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2}UBLExtensions')
# Extension for signature
ext = ET.SubElement(ext_elem, '{urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2}UBLExtension')
ext_uri = ET.SubElement(ext, '{urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2}ExtensionURI')
ext_uri.text = 'urn:oasis:names:specification:ubl:dsig:enveloped:xades'
# Extension content (signature placeholder)
ext_content = ET.SubElement(ext, '{urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2}ExtensionContent')
# Signature will be added here during signing process
# Extension for QR code
qr_ext = ET.SubElement(ext_elem, '{urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2}UBLExtension')
qr_uri = ET.SubElement(qr_ext, '{urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2}ExtensionURI')
qr_uri.text = 'urn:zatca:names:specification:ubl:extension:qr'
qr_content = ET.SubElement(qr_ext, '{urn:oasis:names:specification:ubl:schema:xsd:CommonExtensionComponents-2}ExtensionContent')
qr_value = ET.SubElement(qr_content, 'QRCode')
qr_value.text = invoice.qr_code
def _get_invoice_type_code(self, invoice_type: str) -> str:
"""Get UBL invoice type code."""
mapping = {
'STANDARD': '388', # Tax invoice
'SIMPLIFIED': '388', # Tax invoice (simplified)
'STANDARD_DEBIT': '383', # Debit note
'STANDARD_CREDIT': '381', # Credit note
'SIMPLIFIED_DEBIT': '383', # Debit note (simplified)
'SIMPLIFIED_CREDIT': '381', # Credit note (simplified)
}
return mapping.get(invoice_type, '388')
def _get_invoice_type_name(self, invoice_type: str) -> str:
"""Get invoice type name for XML."""
if 'SIMPLIFIED' in invoice_type:
return '0200000' # Simplified tax invoice
return '0100000' # Standard tax invoice
def _add_billing_reference(self, root, invoice):
"""Add billing reference for credit/debit notes."""
billing_ref = ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}BillingReference')
invoice_doc_ref = ET.SubElement(billing_ref, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}InvoiceDocumentReference')
ET.SubElement(invoice_doc_ref, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID').text = invoice.billing_reference_id
if invoice.billing_reference_issue_date:
ET.SubElement(invoice_doc_ref, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}IssueDate').text = invoice.billing_reference_issue_date.isoformat()
def _add_supplier_party(self, root, invoice):
"""Add supplier (seller) party information."""
supplier = ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}AccountingSupplierParty')
party = ET.SubElement(supplier, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}Party')
# Party Identification (VAT Number)
party_id = ET.SubElement(party, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}PartyIdentification')
id_elem = ET.SubElement(party_id, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID', schemeID='TIN')
id_elem.text = getattr(invoice.tenant, 'vat_number', '300000000000003')
# Party Name
party_name = ET.SubElement(party, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}PartyLegalEntity')
name_elem = ET.SubElement(party_name, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}RegistrationName')
name_elem.text = invoice.tenant.name if invoice.tenant else "Agdar Centre"
# Postal Address
self._add_postal_address(party, invoice.tenant)
# Party Tax Scheme
tax_scheme = ET.SubElement(party, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}PartyTaxScheme')
ET.SubElement(tax_scheme, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}CompanyID').text = getattr(invoice.tenant, 'vat_number', '300000000000003')
tax_scheme_elem = ET.SubElement(tax_scheme, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}TaxScheme')
ET.SubElement(tax_scheme_elem, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID').text = 'VAT'
def _add_customer_party(self, root, invoice):
"""Add customer (buyer) party information."""
customer = ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}AccountingCustomerParty')
party = ET.SubElement(customer, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}Party')
# Party Name
party_name = ET.SubElement(party, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}PartyLegalEntity')
name_elem = ET.SubElement(party_name, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}RegistrationName')
name_elem.text = invoice.patient.full_name_en if hasattr(invoice.patient, 'full_name_en') else str(invoice.patient)
# For Standard invoices, add VAT number if available
if 'STANDARD' in invoice.invoice_type:
# Add party identification if available
if hasattr(invoice.patient, 'vat_number') and invoice.patient.vat_number:
party_id = ET.SubElement(party, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}PartyIdentification')
id_elem = ET.SubElement(party_id, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID', schemeID='TIN')
id_elem.text = invoice.patient.vat_number
def _add_postal_address(self, party, entity):
"""Add postal address to party."""
address = ET.SubElement(party, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}PostalAddress')
# Street name
street = ET.SubElement(address, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}StreetName')
street.text = getattr(entity, 'address', 'Riyadh')
# City
city = ET.SubElement(address, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}CityName')
city.text = getattr(entity, 'city', 'Riyadh')
# Postal Zone
postal = ET.SubElement(address, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}PostalZone')
postal.text = getattr(entity, 'postal_code', '12345')
# Country
country = ET.SubElement(address, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}Country')
country_id = ET.SubElement(country, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}IdentificationCode')
country_id.text = 'SA'
def _add_payment_means(self, root, invoice):
"""Add payment means information."""
payment_means = ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}PaymentMeans')
code = ET.SubElement(payment_means, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}PaymentMeansCode')
code.text = '10' # Cash (default)
def _add_tax_total(self, root, invoice):
"""Add tax total information."""
tax_total = ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}TaxTotal')
# Tax Amount
tax_amount = ET.SubElement(tax_total, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}TaxAmount', currencyID='SAR')
tax_amount.text = f"{invoice.tax:.2f}"
# Tax Subtotal
tax_subtotal = ET.SubElement(tax_total, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}TaxSubtotal')
taxable_amount = ET.SubElement(tax_subtotal, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}TaxableAmount', currencyID='SAR')
taxable_amount.text = f"{invoice.subtotal:.2f}"
tax_amt = ET.SubElement(tax_subtotal, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}TaxAmount', currencyID='SAR')
tax_amt.text = f"{invoice.tax:.2f}"
# Tax Category
tax_category = ET.SubElement(tax_subtotal, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}TaxCategory')
ET.SubElement(tax_category, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID').text = 'S' # Standard rate
ET.SubElement(tax_category, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}Percent').text = '15.00'
tax_scheme = ET.SubElement(tax_category, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}TaxScheme')
ET.SubElement(tax_scheme, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID').text = 'VAT'
def _add_legal_monetary_total(self, root, invoice):
"""Add legal monetary total."""
monetary_total = ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}LegalMonetaryTotal')
# Line Extension Amount
line_ext = ET.SubElement(monetary_total, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}LineExtensionAmount', currencyID='SAR')
line_ext.text = f"{invoice.subtotal:.2f}"
# Tax Exclusive Amount
tax_excl = ET.SubElement(monetary_total, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}TaxExclusiveAmount', currencyID='SAR')
tax_excl.text = f"{invoice.subtotal:.2f}"
# Tax Inclusive Amount
tax_incl = ET.SubElement(monetary_total, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}TaxInclusiveAmount', currencyID='SAR')
tax_incl.text = f"{invoice.total:.2f}"
# Allowance Total Amount (discount)
if invoice.discount > 0:
allowance = ET.SubElement(monetary_total, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}AllowanceTotalAmount', currencyID='SAR')
allowance.text = f"{invoice.discount:.2f}"
# Payable Amount
payable = ET.SubElement(monetary_total, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}PayableAmount', currencyID='SAR')
payable.text = f"{invoice.total:.2f}"
def _add_invoice_lines(self, root, invoice):
"""Add invoice line items."""
for idx, line_item in enumerate(invoice.line_items.all(), start=1):
line = ET.SubElement(root, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}InvoiceLine')
# Line ID
ET.SubElement(line, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID').text = str(idx)
# Invoiced Quantity
quantity = ET.SubElement(line, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}InvoicedQuantity', unitCode='PCE')
quantity.text = str(line_item.quantity)
# Line Extension Amount
line_ext = ET.SubElement(line, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}LineExtensionAmount', currencyID='SAR')
line_ext.text = f"{line_item.total:.2f}"
# Tax Total for line
line_tax_total = ET.SubElement(line, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}TaxTotal')
line_tax_amt = ET.SubElement(line_tax_total, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}TaxAmount', currencyID='SAR')
line_tax_value = line_item.total * Decimal('0.15') # 15% VAT
line_tax_amt.text = f"{line_tax_value:.2f}"
# Item
item = ET.SubElement(line, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}Item')
name = ET.SubElement(item, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}Name')
name.text = line_item.description or 'Service'
# Classified Tax Category
tax_category = ET.SubElement(item, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}ClassifiedTaxCategory')
ET.SubElement(tax_category, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID').text = 'S'
ET.SubElement(tax_category, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}Percent').text = '15.00'
tax_scheme = ET.SubElement(tax_category, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}TaxScheme')
ET.SubElement(tax_scheme, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID').text = 'VAT'
# Price
price = ET.SubElement(line, '{urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2}Price')
price_amt = ET.SubElement(price, '{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}PriceAmount', currencyID='SAR')
price_amt.text = f"{line_item.unit_price:.2f}"
def submit_for_clearance(self, invoice, csid: str, secret: str) -> Tuple[bool, Dict]:
"""
Submit standard invoice for clearance.
Args:
invoice: Invoice instance
csid: Cryptographic Stamp Identifier
secret: CSID secret
Returns:
Tuple[bool, Dict]: (success, response_data)
"""
try:
# Generate XML
xml_content = self.generate_xml_invoice(invoice)
# Encode XML in base64
xml_base64 = base64.b64encode(xml_content.encode('utf-8')).decode('utf-8')
# Prepare request
url = f"{self.base_url}/invoices/clearance/single"
headers = {
'Accept': 'application/json',
'Accept-Version': 'V2',
'Content-Type': 'application/json',
'Authorization': f'Basic {base64.b64encode(f"{csid}:{secret}".encode()).decode()}'
}
payload = {
'invoiceHash': invoice.invoice_hash,
'uuid': str(invoice.id),
'invoice': xml_base64
}
# Make request
response = requests.post(url, json=payload, headers=headers, timeout=30)
# Parse response
response_data = response.json() if response.content else {}
if response.status_code == 200:
logger.info(f"Invoice {invoice.invoice_number} cleared successfully")
return True, response_data
else:
logger.error(f"Clearance failed for invoice {invoice.invoice_number}: {response_data}")
return False, response_data
except Exception as e:
logger.error(f"Error submitting invoice for clearance: {e}")
return False, {'error': str(e)}
def submit_for_reporting(self, invoice, csid: str, secret: str) -> Tuple[bool, Dict]:
"""
Submit simplified invoice for reporting.
Args:
invoice: Invoice instance
csid: Cryptographic Stamp Identifier
secret: CSID secret
Returns:
Tuple[bool, Dict]: (success, response_data)
"""
try:
# Generate XML
xml_content = self.generate_xml_invoice(invoice)
# Encode XML in base64
xml_base64 = base64.b64encode(xml_content.encode('utf-8')).decode('utf-8')
# Prepare request
url = f"{self.base_url}/invoices/reporting/single"
headers = {
'Accept': 'application/json',
'Accept-Version': 'V2',
'Content-Type': 'application/json',
'Authorization': f'Basic {base64.b64encode(f"{csid}:{secret}".encode()).decode()}'
}
payload = {
'invoiceHash': invoice.invoice_hash,
'uuid': str(invoice.id),
'invoice': xml_base64
}
# Make request
response = requests.post(url, json=payload, headers=headers, timeout=30)
# Parse response
response_data = response.json() if response.content else {}
if response.status_code == 200:
logger.info(f"Invoice {invoice.invoice_number} reported successfully")
return True, response_data
else:
logger.error(f"Reporting failed for invoice {invoice.invoice_number}: {response_data}")
return False, response_data
except Exception as e:
logger.error(f"Error submitting invoice for reporting: {e}")
return False, {'error': str(e)}