"""Generate a distribution fee VAT invoice document.""" from datetime import datetime from decimal import Decimal from os import path import tempfile from typing import Any import config from config import ALTAFONTE_SAP_IDS from src.connectors.ows_abacus_account import OwsAbacusAccount from src.connectors.ows_moneyhub import OwsMoneyhub from src.connectors.ows_payee import OwsPayee from src.connectors.ows_royalties import OwsRoyalties from src.documents.utils import extract_vat_info from src.documents.utils import extract_wht_info from src.documents.utils import format_address from src.documents.utils import generate_pdf_file from src.documents.utils import generate_pdf_file_playwright from src.documents.utils import get_logo from src.documents.utils import process_transactions from src.documents.utils import render_template from src.utils.constants import fake_tax_details from src.utils.constants import FeatureFlag from src.utils.constants import TaxDetails from src.utils.constants import VatCategoryType from src.utils.error_handling import LambdaException from src.utils.features import is_feature_enabled def build_old_document( destination_path: str, invoice_number: str, account: dict, contract: dict, signing_entity: dict, tax_details: TaxDetails | dict, statement_attachment: dict, ) -> None: """Create the document, saving the resulting file in the destination path. Args: destination_path (str): Path to store the document to invoice_number (str): Invoice number account (dict): Account info contract (dict): Contract info signing_entity (dict): Orchard specific info tax_details (dict): Tax details statement_attachment (dict): Statement attachment to generate """ vat_data = OwsMoneyhub.get_account_statement_period_vat( statement_attachment['account_id'], statement_attachment['contract_id'], statement_attachment['statement_period_id'], ) statement_period = OwsRoyalties.get_statement_period( statement_attachment['statement_period_id'] ) se_data = { 'se_name': signing_entity['legal_name'], 'se_address': (signing_entity['address'] or '').replace('\n', ' '), 'se_vat_number': signing_entity['vat_number'], } customer_data = { 'customer_name': account['account_name'], 'customer_id': account['account_id'], 'contract_name': contract['contract_name'], 'contract_id': contract['contract_id'], 'customer_address': format_address(tax_details['address']), } invoice_data = { 'invoice_number': invoice_number, 'invoice_date': datetime.fromisoformat(statement_attachment['created_at']).date(), 'invoice_currency': vat_data['currency_code'], } transactions: list[dict[str, Any]] = [ { 'description': 'Commission {}'.format(statement_period['statement_period_name']), 'quantity': 1, 'amount': Decimal(vat_data['distribution_fee'] or 0) * -1, 'amount_ex_vat': Decimal(vat_data['distribution_fee'] or 0) * -1, 'vat_rate': Decimal(vat_data['distribution_vat_rate'] or 0), } ] subtotal_amount = total_amount = Decimal('0') for tr in transactions: tr['amount'] = tr['amount'] or Decimal('0') amount_ex_vat = Decimal(tr['amount_ex_vat'] or 0) tr['amount_ex_vat'] = amount_ex_vat subtotal_amount += amount_ex_vat total_amount += subtotal_amount total_vat = Decimal(vat_data['distribution_vat'] or 0) * -1 vat_info = { 'vat_items': [ { 'vat_rate': Decimal(vat_data['distribution_vat_rate'] or 0), 'vat_amount': Decimal(vat_data['distribution_vat'] or 0) * -1, 'amount_ex_vat': Decimal(vat_data['distribution_fee'] or 0) * -1, } ], 'amount_ex_vat': Decimal(vat_data['distribution_fee'] or 0) * -1, 'total_vat': total_vat, } total_amount += total_vat document_data = { **se_data, **customer_data, **invoice_data, **vat_info, 'transactions': transactions, 'subtotal_amount': subtotal_amount, 'total_amount': total_amount, 'logo_path': get_logo(signing_entity['company_code']), } if is_feature_enabled(FeatureFlag.NEW_PDF_LIBRARY, account['account_id']): generate_pdf_file_playwright( destination_path, 'distribution_fee_invoice_old.html', 'includes/footer_playwright.html', document_data, '2.75in', ) else: generate_pdf_file(destination_path, 'distribution_fee_invoice_old.html', document_data) def build_document(statement_attachment: dict) -> str: """Create the document, saving the resulting file in the destination path. Args: statement_attachment (dict): Statement attachment to generate """ invoice_number = statement_attachment['invoice_number'] account_id = statement_attachment['account_id'] destination_path = path.join(config.FILE_OUTPUT_PATH, f'{invoice_number}.pdf') ledger_vat_data = OwsMoneyhub.get_ledger_vat_summary( statement_attachment['account_id'], statement_attachment['contract_id'], statement_attachment['statement_period_id'], [VatCategoryType.DISTRIBUTION_FEE, VatCategoryType.COMMISSION], ) account = OwsAbacusAccount.get_account(statement_attachment['account_id']) contract = OwsRoyalties.get_contract(statement_attachment['contract_id']) payment_entity_id = OwsAbacusAccount.get_account_payment_term(account_id).get( 'payment_entity_id' ) # noqa:E501 assert payment_entity_id is not None payment_entity = OwsRoyalties.get_reference_payment_entity(payment_entity_id) is_spanish_entity = payment_entity.get('country_of_tax_reporting') == config.Countries.SPAIN signing_entity = OwsMoneyhub.get_reference_signing_entity(statement_attachment['contract_id']) if is_spanish_entity: tax_details = ( fake_tax_details if config.ENVIRONMENT == config.DEV_ENVIRONMENT else (OwsPayee.get_tax_details(account['account_payee_id'], ['local_tax_id'])) ) tax_number = tax_details.get('local_tax_id') else: tax_details = ( fake_tax_details if config.ENVIRONMENT == config.DEV_ENVIRONMENT else (OwsPayee.get_tax_details(account['account_payee_id'], ['address'])) ) tax_number = signing_entity['vat_number'] statement_period = OwsRoyalties.get_statement_period( statement_attachment['statement_period_id'] ) if not ledger_vat_data: build_old_document( destination_path, invoice_number, account, contract, signing_entity, tax_details, statement_attachment, ) return destination_path is_altafonte_networks = signing_entity['company_code'] in ALTAFONTE_SAP_IDS payee_currencies = [item['payee_currency_code'] for item in ledger_vat_data] if not all(currency == payee_currencies[0] for currency in payee_currencies): raise LambdaException('VAT entries contain more than one payee currency') vat_currencies = [item['vat_currency_code'] for item in ledger_vat_data] if not all(currency == vat_currencies[0] for currency in vat_currencies): raise LambdaException('VAT entries contain more than one VAT currency') se_data = { 'se_name': signing_entity['legal_name'], 'se_address': (signing_entity['address'] or '').replace('\n', ' '), 'se_vat_number': signing_entity['vat_number'], } customer_data = { 'customer_tax_number': tax_number, 'account_name': account['account_name'], 'account_id': account['account_id'], 'contract_name': contract['contract_name'], 'contract_id': contract['contract_id'], 'customer_name': tax_details['business_name'], 'customer_address': format_address(tax_details['address']), } invoice_data = { 'invoice_number': invoice_number, 'invoice_date': datetime.fromisoformat(statement_attachment['created_at']).date(), 'invoice_currency': ledger_vat_data[0]['payee_currency_code'], } transactions: list[dict] = [] process_transactions(ledger_vat_data, statement_period, transactions, flip_signs=True) subtotal_amount = total_amount = Decimal('0') for tr in transactions: tr['amount'] = tr['amount'] or Decimal('0') tr['amount_ex_tax'] = tr['amount_ex_tax'] or Decimal('0') subtotal_amount += tr['amount_ex_tax'] total_amount += subtotal_amount vat_info = extract_vat_info(ledger_vat_data, True) wht_info = extract_wht_info(ledger_vat_data, True) has_wht = is_altafonte_networks and wht_info['has_wht'] total_amount += vat_info['payee_total_vat'] total_amount += wht_info['total_wht_amount_payee_currency'] if has_wht else 0 document_data = { **se_data, **customer_data, **invoice_data, **vat_info, **wht_info, 'transactions': transactions, 'subtotal_amount': subtotal_amount, 'total_amount': total_amount, 'logo_path': get_logo(signing_entity['company_code']), 'has_vat': vat_info['has_vat'], 'has_wht': has_wht, } if is_feature_enabled(FeatureFlag.NEW_PDF_LIBRARY, account['account_id']): generate_pdf_file_playwright( destination_path, 'distribution_fee_invoice/body.html', 'distribution_fee_invoice/footer_playwright.html', document_data, '2.15in', ) else: page_options = { 'margin-top': '15', 'margin-right': '20', 'margin-bottom': '55', 'margin-left': '20', 'footer-spacing': '5', 'javascript-delay': '1000', } with tempfile.NamedTemporaryFile(suffix='.html', delete=False) as footer: page_options['--footer-html'] = footer.name footer.write( render_template('distribution_fee_invoice/footer.html', document_data).encode( 'utf-8' ) ) generate_pdf_file( destination_path, 'distribution_fee_invoice/body.html', document_data, page_options ) return destination_path