""" 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)}