"""Statement Attachment Logic.""" from datetime import datetime import re from fastapi import HTTPException from moneyhub import models from moneyhub.config import Config from moneyhub.connectors import sqs from moneyhub.connectors.mysql import db from moneyhub.connectors.ows_royalties import get_contracts_for_account from moneyhub.connectors.s3 import create_presigned_url from moneyhub.constants.constants import ContractType from moneyhub.constants.constants import KNR_SAP_IDS from moneyhub.constants.constants import NumberFormat from moneyhub.constants.constants import StatementAttachmentFileType from moneyhub.constants.constants import StatementAttachmentStatus from moneyhub.constants.constants import StatementAttachmentType from moneyhub.constants.constants import SYSTEM_TIMEZONE from moneyhub.constants.constants import VatCategory from moneyhub.constants.error import FORBIDDEN_INVOICE_URL_ACCESS from moneyhub.constants.error import FORBIDDEN_URL_ACCESS from moneyhub.constants.error import INVALID_FILE_TYPE from moneyhub.constants.error import INVALID_STATEMENT_ATTACHMENT from moneyhub.constants.error import NO_INVOICE_FILE_LOCATION from moneyhub.constants.error import SIGNING_ENTITY_NOT_SUPPORTED from moneyhub.constants.error import STATEMENT_PERIOD_NOT_VISIBLE from moneyhub.constants.features import FEATURE_INVOICE_ACCOUNT_SEQUENCE_NUMBERS from moneyhub.schemas.statement_attachment import StatementAttachmentDetailSchema from moneyhub.utils.aws import parse_s3_url from moneyhub.utils.features import is_feature_enabled from moneyhub.utils.request import profile_has_access_to_resource DISTRIBUTION_FEE_INVOICE = StatementAttachmentType.DISTRIBUTION_FEE_INVOICE SELF_BILLING_INVOICE = StatementAttachmentType.SELF_BILLING_INVOICE REVENUE_DETAIL = StatementAttachmentType.REVENUE_DETAIL NEIGHBOURING_RIGHTS_PERFORMER_REVENUE = StatementAttachmentType.NEIGHBOURING_RIGHTS_PERFORMER_REVENUE # noqa: E501 NEIGHBOURING_RIGHTS_LABEL_REVENUE = StatementAttachmentType.NEIGHBOURING_RIGHTS_LABEL_REVENUE COLLECTION_SUMMARY_LABEL = StatementAttachmentType.COLLECTION_SUMMARY_LABEL COLLECTION_SUMMARY_PERFORMER = StatementAttachmentType.COLLECTION_SUMMARY_PERFORMER CUSTOM_PAYMENT = VatCategory.CUSTOM_PAYMENT GROSS_REVENUE = VatCategory.GROSS_REVENUE DISTRIBUTION_FEE = VatCategory.DISTRIBUTION_FEE CLOSING_BALANCE = VatCategory.CLOSING_BALANCE COMMISSION = VatCategory.COMMISSION LEGACY_REVENUE_DETAIL_FULL = StatementAttachmentType.LEGACY_REVENUE_DETAIL_FULL LEGACY_REVENUE_DETAIL_PHYSICAL = StatementAttachmentType.LEGACY_REVENUE_DETAIL_PHYSICAL # Regular expression that matches a number at the end of a string # Ex: # abcd-fedc-1234 # 321_321_1234 # long text that ends in 1234 FINAL_NUMBER_REGEX = re.compile(r'(\d+)$') PUBLISHING_STATEMENT_FILENAME_PATTERN = '_publishing_statement_' ROYALTY_SHARE_STATEMENT_PATTERN = '_rps_001_' FILE_KEY_REGEXP = re.compile( r""" (?P[L]) (?P\d+) (?:\|(?P\d+))?_ (?P\d{1,4})_ (?P[a-zA-Z0-9-_.*'()]+\.(?P\w{3,4}))$""", re.VERBOSE, ) def get_statement_attachments_by_account_and_statement_periods( account_id: int, statement_period_ids: list | None, contract_id: int | None, subaccount_id: int | None = None, ) -> list[StatementAttachmentDetailSchema]: """Get statement attachment for an account and statement period. Args: account_id (int): The id of an account statement_period_ids (list): Optional list of statement periods attachments were applied to contract_id (int): Optional id of the contract subaccount_id (int): Optional id of the subaccount Returns: list: list of statement attachments """ result = models.StatementAttachment.get_by_account_id_and_statement_periods( account_id=account_id, contract_id=contract_id, statement_period_ids=statement_period_ids, subaccount_id=subaccount_id ) result = list(map(_add_display_name, result)) return result def _add_display_name(attachment: StatementAttachmentDetailSchema): """Add display file name to an attachment.""" if attachment.statement_attachment_type == 'royalty_share_statement': regex_match = re.search(FILE_KEY_REGEXP, attachment.file_location.split('/')[-1]) displayed_file_name = regex_match.group('filename') if regex_match else None attachment.displayed_file_name = displayed_file_name else: attachment.displayed_file_name = None return attachment def get_statement_attachment_by_id( statement_attachment_id: int, profile_type: str, profile_id: int ) -> models.StatementAttachment: """Get a single attachment by ID. Args: statement_attachment_id (int): ID of the attachment profile_type (str): Profile type of the user making the request profile_id (id): Profile ID of the user making the request Returns: StatementAttachment: The attachment """ statement_attachment = models.StatementAttachment.get_by_id_or_error(statement_attachment_id) has_access = profile_has_access_to_resource( profile_type, profile_id, statement_attachment.account_id, statement_attachment.subaccount_id, ) if not has_access: raise HTTPException(status_code=403, detail=FORBIDDEN_URL_ACCESS) return statement_attachment def get_latest_account_invoice_number_map(year: int, account_ids: list[int]) -> dict: """Get a mapping of account ID to invoice sequence number for a given year. Args: year (int): Year to get the invoice numbers for. account_ids (list): List of accounts to get the invoice numbers for. Returns: dict: Mapping of account ID to sequence number. """ invoice_numbers = models.StatementAttachment.get_latest_account_self_billing_invoice_number( year, account_ids) sequence_number_map = {} for number in invoice_numbers: value = FINAL_NUMBER_REGEX.search(number.invoice_number)[1] sequence_number_map[number.account_id] = int(value) return sequence_number_map def get_latest_entity_invoice_number_map(year: int) -> dict: """Get the latest invoice sequence number, mapped to the SAP ID and type. Args: year (int): Current year Returns dict: dict of invoice's latest sequence number """ latest_invoices = \ models.StatementAttachment.get_latest_statement_attachments_invoices(year) # Make a map of the {sap_id}_{type} to invoice sequence number sequence_number_map = {} for invoice in latest_invoices: if invoice.invoice_number is None: continue # skip attachments that don't have invoice numbers key = f'{invoice.sap_id}_{invoice.statement_attachment_type.value}' value = FINAL_NUMBER_REGEX.search(invoice.invoice_number)[1] sequence_number_map[key] = int(value) return sequence_number_map def generate_account_invoice_number( year: int, company_code: str, account_id: int, sequence_number_map: dict ) -> str: """Generate an invoice number for a year/account, taking the next number from the sequence map. NOTE: This modifies the sequence_number_map in-place. Args: year (int): Year of the invoice number. company_code (str): Company code. account_id (int): Account to generate invoice number for. sequence_number_map (dict): Mapping of account_id to current sequence number. Returns: str: Invoice number in YYYY-XXXX-NNN format. """ if account_id not in sequence_number_map: sequence_number_map[account_id] = 0 sequence_number_map[account_id] += 1 sequence_number = str(sequence_number_map[account_id]).zfill(3) return f'{company_code}_{year}_{account_id}_{sequence_number}' def generate_entity_invoice_number( year: int, company_code: str, statement_attachment_type: StatementAttachmentType, sequence_number_map: dict ) -> str: """Generate an invoice numbers for a statement attachment based on the entity company code. NOTE: This also modifies the sequence_number_map in-place. Args: year (int): Current year company_code (str): The SAP company code for the contract statement_attachment_type (StatementAttachmentType): Type of the statement attachment sequence_number_map (dict): Map of SAP ID and type to sequence number Returns str: Invoice number in XXXX_YYYY_NNNNNNNNNN format """ sequence_number_key = f'{company_code}_{statement_attachment_type.value}' if sequence_number_key not in sequence_number_map: sequence_number_map[sequence_number_key] = 1 else: sequence_number_map[sequence_number_key] += 1 sequence_number = sequence_number_map[sequence_number_key] if statement_attachment_type == DISTRIBUTION_FEE_INVOICE: return f'{company_code}_{year}_{str(sequence_number).zfill(10)}' if statement_attachment_type == SELF_BILLING_INVOICE: return f'{company_code}_{year}_SB{str(sequence_number).zfill(8)}' def create_revenue_detail_reports( account_id: int, statement_period_id: int, orchard_identity_id: str, correlation_id: str | None = None, contract_id: int | None = None, subaccount_id: int | None = None, ) -> list[models.StatementAttachment]: """Create revenue detail report attachments for a specific account and statement period. If one already exists then it will return that instead. Args: account_id (int): Account to create the report for statement_period_id (int): Period to create the report for orchard_identity_id (str): orchard identity id correlation_id (str): Optional request correlation ID contract_id (id): ID of contract subaccount_id (int): Optional id of the subaccount Returns: list: The created/existing statement attachments """ _check_statement_periods(account_id, [statement_period_id]) attachments = [] if subaccount_id: report_type = REVENUE_DETAIL if not models.StatementAttachment.exists( account_id, statement_period_id, attachment_type=report_type, subaccount_id=subaccount_id ): attachments.append(models.StatementAttachment.create( account_id=account_id, subaccount_id=subaccount_id, contract_id=None, statement_period_id=statement_period_id, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_attachment_type=report_type, file_type=StatementAttachmentFileType.CSV, number_format=NumberFormat.US, created_by=orchard_identity_id, )) elif not contract_id: # Enforce idempotency existing_attachments = models.StatementAttachment.get_by_statement_period( statement_period_id, account_id=account_id, types=[ REVENUE_DETAIL, NEIGHBOURING_RIGHTS_PERFORMER_REVENUE, NEIGHBOURING_RIGHTS_LABEL_REVENUE, ], subaccount_id=subaccount_id ) if len(existing_attachments): return existing_attachments contracts = get_contracts_for_account(account_id) for contract in contracts: report_type = _get_report_type(contract) attachments.append({ 'account_id': account_id, 'subaccount_id': subaccount_id, 'contract_id': contract['contract_id'], 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': report_type, 'file_type': StatementAttachmentFileType.CSV, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }) attachments = bulk_create_statement_attachments(attachments) else: contract = models.Contract.get_by_id(contract_id) report_type = _get_report_type(contract.__dict__) if not models.StatementAttachment.exists( account_id, statement_period_id, contract_id=contract_id, attachment_type=report_type, subaccount_id=subaccount_id ): attachments.append(models.StatementAttachment.create( account_id=account_id, subaccount_id=subaccount_id, contract_id=contract_id, statement_period_id=statement_period_id, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_attachment_type=report_type, file_type=StatementAttachmentFileType.CSV, number_format=NumberFormat.US, created_by=orchard_identity_id, )) if attachments: sqs.send_message( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, { 'account_id': account_id, 'subaccount_id': subaccount_id, 'statement_period_id': statement_period_id, } ) return attachments def _get_report_type(contract: dict) -> StatementAttachmentType: match contract['contract_type']: case ContractType.NEIGHBOURING_RIGHTS: return NEIGHBOURING_RIGHTS_PERFORMER_REVENUE case ContractType.DISTRIBUTION: signing_entity = models.ReferenceSigningEntity.get_by_contract_id( contract['contract_id']) if signing_entity.company_code in KNR_SAP_IDS: return NEIGHBOURING_RIGHTS_LABEL_REVENUE else: return REVENUE_DETAIL case _: return REVENUE_DETAIL def create_statement_attachments(statement_period_id: int, orchard_identity_id: str) -> list: """Create statement attachments for specified statement period. Args: statement_period_id (int): The id of statement period orchard_identity_id (str): orchard identity id Returns: list: list of statement attachments """ new_statement_attachments = list() current_year = (datetime.now(SYSTEM_TIMEZONE)).year invoices_sequence_numbers = get_latest_entity_invoice_number_map(current_year) contracts = models.LedgerAccountContract.get_contracts_by_statement_period( statement_period_id) account_invoice_sequence_numbers = get_latest_account_invoice_number_map( current_year, [contract.account_id for contract in contracts] ) vat_entries = models.LedgerAccountingRunVat.get_by_statement_period(statement_period_id) ledger_vat_summary_entries = models.LedgerVatSummary.get_by_activity_statement_period( statement_period_id) # Determine which contracts should get what attachments distribution_fee_contracts = [ entry.contract_id for entry in vat_entries if entry.distribution_fee is not None ] summary_distribution_contracts = [ entry.contract_id for entry in ledger_vat_summary_entries if entry.vat_category in [DISTRIBUTION_FEE, COMMISSION] ] distribution_fee_contracts = set(distribution_fee_contracts + summary_distribution_contracts) self_billing_contracts = set( entry.contract_id for entry in ledger_vat_summary_entries if entry.vat_category in [GROSS_REVENUE, CLOSING_BALANCE, CUSTOM_PAYMENT] ) existing_attachments = models.StatementAttachment.get_by_statement_period( statement_period_id) existing_attachments = [ f'{item.contract_id}-{item.statement_attachment_type}' for item in existing_attachments] for contract in contracts: statement_attachment_key = f'{contract.contract_id}-{DISTRIBUTION_FEE_INVOICE}' company_code = contract.tax_entity_company_code if contract.tax_entity_company_code \ else contract.company_code if (statement_attachment_key not in existing_attachments and contract.contract_id in distribution_fee_contracts): invoice_number = generate_entity_invoice_number( current_year, company_code, DISTRIBUTION_FEE_INVOICE, invoices_sequence_numbers) new_statement_attachments.append(dict( account_id=contract.account_id, contract_id=contract.contract_id, statement_period_id=statement_period_id, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_attachment_type=DISTRIBUTION_FEE_INVOICE, invoice_number=invoice_number, file_type=StatementAttachmentFileType.PDF, number_format=NumberFormat.US, created_by=orchard_identity_id )) statement_attachment_key = f'{contract.contract_id}-{SELF_BILLING_INVOICE}' if (statement_attachment_key not in existing_attachments and contract.contract_id in self_billing_contracts): if is_feature_enabled(FEATURE_INVOICE_ACCOUNT_SEQUENCE_NUMBERS): invoice_number = generate_account_invoice_number( current_year, company_code, contract.account_id, account_invoice_sequence_numbers) else: invoice_number = generate_entity_invoice_number( current_year, company_code, SELF_BILLING_INVOICE, invoices_sequence_numbers) new_statement_attachments.append(dict( account_id=contract.account_id, contract_id=contract.contract_id, statement_period_id=statement_period_id, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_attachment_type=SELF_BILLING_INVOICE, invoice_number=invoice_number, file_type=StatementAttachmentFileType.PDF, number_format=NumberFormat.US, created_by=orchard_identity_id )) return bulk_create_statement_attachments(new_statement_attachments) def add_new_collection_summary_attachments( filtered_contracts: list, account_id: int, statement_period_id: int, orchard_identity_id: str ) -> list: """Create collection summary docs for specified statement period. Args: filtered_contracts(list): List of filtered contracts account_id (int): Account to create the report for statement_period_id (int): Period to create the report for orchard_identity_id (str): orchard identity id Returns: list: list of collection summary attachments """ new_attachments = [] for contract in filtered_contracts: signing_entity = models.ReferenceSigningEntity.get_by_contract_id( contract['contract_id']) if signing_entity.company_code in KNR_SAP_IDS: if contract['contract_type'] == ContractType.NEIGHBOURING_RIGHTS: report_type = COLLECTION_SUMMARY_PERFORMER else: report_type = COLLECTION_SUMMARY_LABEL else: raise HTTPException(status_code=400, detail=SIGNING_ENTITY_NOT_SUPPORTED) new_attachments.append({ 'account_id': account_id, 'contract_id': contract['contract_id'], 'statement_period_id': statement_period_id, 'statement_attachment_status': StatementAttachmentStatus.IN_PROGRESS, 'statement_attachment_type': report_type, 'file_type': StatementAttachmentFileType.PDF, 'number_format': NumberFormat.US, 'created_by': orchard_identity_id }) return new_attachments def _check_statement_periods(account_id: int, statement_period_ids: list[int]) -> None: are_statement_periods_visible = models.AccountStatementPeriods.are_statement_period_ids_visible( account_id, statement_period_ids) if not are_statement_periods_visible: detail = STATEMENT_PERIOD_NOT_VISIBLE.format(statement_period_ids=statement_period_ids) raise HTTPException(status_code=400, detail=detail) def create_legacy_revenue_report( account_id: int, contract_id: int | None, statement_period_id: int, orchard_identity_id: str, report_type: StatementAttachmentType, file_type: StatementAttachmentFileType, number_format: NumberFormat, correlation_id: str | None, subaccount_id: int | None = None, filters: dict | None = None, statement_period_ids: list[int] | None = None ): """Create legacy revenue reports for an account, contract and statement period. Args: account_id (int): Account to make the report for contract_id (int): Contract id to make the report for statement_period_id (int): Period to make the report for orchard_identity_id (str): the orchard identity of the client making the request report_type (StatementAttachmentType): the type of legacy revenue report file_type (StatementAttachmentFileType): the requested file type number_format (str): Format for numbers ('us' or 'eu') correlation_id (str): Request correlation ID subaccount_id (int): Optional id of the subaccount statement_period_ids (str): Optional list of statement period ids Returns: StatementAttachmentDetailSchema: the created legacy revenue report """ if not statement_period_ids: statement_period_ids = [statement_period_id] statement_period_ids_str = str(statement_period_id) else: statement_period_ids_str = ', '.join(str(sid) for sid in statement_period_ids) _check_statement_periods(account_id, statement_period_ids) if report_type not in {LEGACY_REVENUE_DETAIL_FULL, LEGACY_REVENUE_DETAIL_PHYSICAL}: raise HTTPException(status_code=400, detail=INVALID_STATEMENT_ATTACHMENT) if file_type not in [StatementAttachmentFileType.XLS, StatementAttachmentFileType.TXT]: raise HTTPException(status_code=400, detail=INVALID_FILE_TYPE) report = models.StatementAttachment.create( account_id=account_id, subaccount_id=subaccount_id, contract_id=contract_id, statement_period_id=statement_period_id, statement_period_ids=statement_period_ids_str, file_type=file_type, number_format=number_format, created_by=orchard_identity_id, statement_attachment_status=StatementAttachmentStatus.IN_PROGRESS, statement_attachment_type=report_type, filters=filters ) sqs.send_message( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, { 'statement_attachment_id': report.statement_attachment_id } ) return [report] def create_collection_summary_documents( account_id: int, statement_period_id: int, orchard_identity_id: str, correlation_id: str | None) -> list: """Create collection summary docs for specified statement period. Args: account_id (int): Account to create the report for statement_period_id (int): Period to create the report for orchard_identity_id (str): orchard identity id correlation_id (str): Optional request correlation ID Returns: list: list of collection summary attachments """ contracts = get_contracts_for_account(account_id) visible_periods = models.StatementPeriodPaymentEntity.get_visible_statement_period_ids( account_id) if statement_period_id not in visible_periods: detail = STATEMENT_PERIOD_NOT_VISIBLE.format(statement_period_ids=[statement_period_id]) raise HTTPException(status_code=400, detail=detail) existing_attachments = models.StatementAttachment.get_by_statement_period( statement_period_id, account_id=account_id, types=[ COLLECTION_SUMMARY_PERFORMER, COLLECTION_SUMMARY_LABEL]) existing_contracts = [ attachment.contract_id for attachment in existing_attachments] filtered_contracts = [ contract for contract in contracts if contract['contract_id'] not in existing_contracts ] new_attachments = add_new_collection_summary_attachments( filtered_contracts, account_id, statement_period_id, orchard_identity_id) if new_attachments: new_attachments = bulk_create_statement_attachments(new_attachments) sqs.send_message( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, { 'account_id': account_id, 'statement_period_id': statement_period_id, } ) return existing_attachments + new_attachments def bulk_create_statement_attachments(statement_attachments: list) -> list: """Bulk create statement attachments. Args: statement_attachments (list): The statement attachments to create Returns: list: list of statement attachments """ statement_attachments_list = list() for param in statement_attachments: new_statement_attachment = models.StatementAttachment.build(**param) db.session.flush() statement_attachments_list.append(new_statement_attachment) if statement_attachments_list: models.StatementAttachment.commit_changes() return statement_attachments_list def update_statement_attachment( statement_attachment_id: int, **params: dict) -> StatementAttachmentDetailSchema: """Update statement attachment by statement_attachment_id. Args: statement_attachment_id (int): The id of a statement attachment params (dict): PUT parameters Returns: """ statement_attachment = models.StatementAttachment.get_by_id_or_error( statement_attachment_id) statement_attachment.update_attributes(**params) models.StatementAttachment.commit_changes() return statement_attachment def trigger_statement_attachment_generation( statement_period_id: int, account_id: int | None, correlation_id: str | None = None ) -> dict: """Trigger statement attachment generation for the specified period. Args: statement_period_id (int): The id of statement period account_id (int): ID of the account to generate attachments for correlation_id (str): Correlation Id from the header Returns: dict: status and result """ attachments = models.StatementAttachment.get_by_statement_period( statement_period_id, statuses=[ StatementAttachmentStatus.IN_PROGRESS, StatementAttachmentStatus.ERROR, ] ) account_ids = { (attachment.account_id, attachment.subaccount_id) for attachment in attachments } if account_id: account_ids = [item for item in account_ids if item[0] == account_id] if correlation_id is None: correlation_id = '' messages = [ { 'account_id': account_id, 'subaccount_id': subaccount_id, 'statement_period_id': statement_period_id, } for (account_id, subaccount_id) in account_ids ] sqs.send_messages( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, messages) return { 'status': 'OK', 'messages_sent': len(messages), } def get_invoice_presigned_url( statement_attachment_id: int, profile_type: str, profile_id: int ) -> dict: """Get presigned S3 URL for downloading invoice. Args: statement_attachment_id (int): The id of statement attachment profile_type (str): Profile type of the user making the request profile_id (id): Profile ID of the user making the request Returns: dict: presigned url of invoice file location """ statement_attachment = models.StatementAttachment.get_by_id_or_error( statement_attachment_id) has_access = profile_has_access_to_resource( profile_type, profile_id, statement_attachment.account_id, statement_attachment.subaccount_id, ) if not has_access: raise HTTPException(status_code=403, detail=FORBIDDEN_INVOICE_URL_ACCESS) file_location = statement_attachment.file_location if not file_location: raise Exception( NO_INVOICE_FILE_LOCATION.format( statement_attachment_id=statement_attachment_id ) ) bucket, key = parse_s3_url(file_location) presigned_url = create_presigned_url(bucket, key) return presigned_url def regenerate_statement_attachments( statement_period_id: int, account_id: int | None, statement_attachment_type: StatementAttachmentStatus | None, correlation_id: str | None = None ) -> list: """Regenerate statement attachments. Args: statement_period_id (int): Statement period to regenerate attachments for account_id (int): Optional account to filter by statement_attachment_type (str): Optional type to filter by correlation_id (str): Optional request correlation ID Returns: list: attachments which will be regenerated """ types = [statement_attachment_type] if statement_attachment_type else None attachments = models.StatementAttachment.get_by_statement_period( statement_period_id, account_id=account_id, types=types) if not attachments: return [] account_ids = set() for attachment in attachments: attachment.statement_attachment_status = StatementAttachmentStatus.IN_PROGRESS attachment.failure_reason = None account_ids.add(attachment.account_id) models.StatementAttachment.commit_changes() messages = [ {'account_id': acc_id, 'statement_period_id': statement_period_id} for acc_id in account_ids ] sqs.send_messages( Config.SQS_MH_GENERATE_ATTACHMENTS_QUEUE_NAME, correlation_id, messages) return attachments def delete_statement_attachment(statement_attachment_id: int): """Delete a statement attachment. Args: statement_attachment_id (int): Statement attachment to delete. """ models.StatementAttachment.delete_by_id(statement_attachment_id) def create_internal_attachment( account_id: int, statement_period_id: int, file_type: StatementAttachmentFileType, file_location: str, orchard_identity_id: str, upload_date: datetime, contract_id: int | None = None, ) -> object | None: """Create internal file statement attachments. Args: account_id(int): The id of the account. statement_period_id (int): The id of statement period. file_type(StatementAttachmentFileType): The file type of the file. file_location(str): The file location of the internal document. upload_date (datetime): the original uploaded timestamp orchard_identity_id(str): the orchard id of the user contract_id(int): the contract id of the user Returns: The created statement attachment """ if not models.Account.exists(account_id): return if not models.StatementAttachment.exists(account_id, statement_period_id, file_location): file_name_expression = file_location.lower() match file_name_expression: case _ if PUBLISHING_STATEMENT_FILENAME_PATTERN in file_name_expression: attachment_type = StatementAttachmentType.PUBLISHING_DETAIL case _ if ROYALTY_SHARE_STATEMENT_PATTERN in file_name_expression: attachment_type = StatementAttachmentType.ROYALTY_SHARE_STATEMENT case _: attachment_type = StatementAttachmentType.INTERNAL_UPLOAD attachment = models.StatementAttachment.create( account_id=account_id, contract_id=contract_id, statement_period_id=statement_period_id, file_type=file_type, file_location=file_location, created_by=orchard_identity_id, created_at=upload_date, statement_attachment_status=StatementAttachmentStatus.COMPLETE, statement_attachment_type=attachment_type ) attachment = _add_display_name(attachment) return attachment