"""Generate a Collection Summary document.""" from os import path from typing import Any from typing import Generator import pandas import config from src.connectors import snowflake from src.connectors.ows_abacus_account import OwsAbacusAccount from src.connectors.ows_moneyhub import OwsMoneyhub from src.connectors.ows_royalties import OwsRoyalties 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 sanitize_filename from src.utils.constants import FeatureFlag from src.utils.features import is_feature_enabled def create_df(collection_summary: Generator[Any, None, None]) -> tuple[list, str]: """Create dataframe from sql data. Args: collection_summary (Generator): Collection summary rows from Snowflake. Returns: pandas.Dataframe: dataframe to create table from """ df = pandas.DataFrame(collection_summary) if df.empty: return [], '' currency_code = df['CURRENCY'][0] df.drop(columns=['CURRENCY'], inplace=True) df[['PRE_WHT_AMOUNT', 'WHT_AMOUNT', 'GROSS_AMOUNT', 'CLIENT_SHARE', 'NET_REVENUE']] = df[ ['PRE_WHT_AMOUNT', 'WHT_AMOUNT', 'GROSS_AMOUNT', 'CLIENT_SHARE', 'NET_REVENUE'] ].astype(float) sums = df[['PRE_WHT_AMOUNT', 'WHT_AMOUNT', 'GROSS_AMOUNT', 'NET_REVENUE']].agg(['sum']) sums['COLLECTION_SOCIETY'] = 'Total' sums['COUNTRY'] = '' sums['CLIENT_SHARE'] = '' sums = sums[ [ 'COLLECTION_SOCIETY', 'COUNTRY', 'PRE_WHT_AMOUNT', 'WHT_AMOUNT', 'GROSS_AMOUNT', 'CLIENT_SHARE', 'NET_REVENUE', ] ] df = pandas.concat([df, sums]).reset_index(drop=True) df = df.round(2) numeric_cols = df.select_dtypes(include=[float, int]).columns df[numeric_cols] = df[numeric_cols].map('{:,.2f}'.format) dict_list = df.to_dict('records') return dict_list, currency_code 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 Returns: str: Local path to the generated file. """ account = OwsAbacusAccount.get_account(statement_attachment['account_id']) contract = OwsRoyalties.get_contract(statement_attachment['contract_id']) signing_entity = OwsMoneyhub.get_reference_signing_entity(statement_attachment['contract_id']) statement_period = OwsRoyalties.get_statement_period( statement_attachment['statement_period_id'] ) address_list = signing_entity['address'].split(',', 1) account_name = account['account_name'] contract_id = contract['contract_id'] statement_period_name = statement_period['statement_period_name'] destination_path = path.join( config.FILE_OUTPUT_PATH, sanitize_filename( f'{account_name}_{contract_id}_{statement_period_name}_collection_summary' ) + '.pdf', ) collection_summary, total_rows = snowflake.get_collection_summary_data( statement_attachment['statement_period_id'], statement_attachment['account_id'], statement_attachment['contract_id'], statement_attachment['statement_attachment_type'], ) dict_list, currency_code = create_df(collection_summary) document_data = { 'account_id': account['account_id'], 'account_name': account_name, 'contract_id': contract_id, 'contract_name': contract['contract_name'], 'currency_code': currency_code, 'company_name': signing_entity['legal_name'], 'company_address': address_list, 'statement_period_name': statement_period_name, 'table_info': dict_list, 'logo_path': get_logo(signing_entity['company_code']), } page_options = { 'footer-html': 'src/documents/templates/includes/footer.html', 'javascript-delay': '1000', } if is_feature_enabled(FeatureFlag.NEW_PDF_LIBRARY, account['account_id']): generate_pdf_file_playwright( destination_path, 'collection_summary.html', 'includes/footer_playwright.html', document_data, ) else: generate_pdf_file( destination_path, 'collection_summary.html', document_data, page_options, ) return destination_path