"""Account model.""" import re from abacus_common_logic.connectors.database import db from abacus_common_logic.models.base import BaseModel from sqlalchemy import bindparam, func, or_, text from abacus_account.constants import templates from abacus_account.constants.constants import BULK_PAYOUTS_PAYONEER_PAYMENT_TYPES from abacus_account.models.account_payment_term import AccountPaymentTerm class Account(BaseModel): """Account model.""" __tablename__ = 'account' account_id = db.Column(db.Integer, primary_key=True) # TODO: Temporary added this field. We will remove it later on account_name = db.Column(db.String(180), nullable=False) # TODO Defaulting created_by to '' was a short-term fix done during the TSA cutover created_by = db.Column(db.String(255), nullable=False, default='') sap_created_at = db.Column(db.DateTime, nullable=False) account_tax_info = db.relationship( 'AccountTaxInfo', backref='account', uselist=False, cascade='all, delete-orphan' ) payment_hold = db.relationship( 'PaymentHold', backref='account', uselist=False, cascade='all, delete-orphan', order_by='desc(PaymentHold.payment_hold_id)' ) payment_hold_history = db.relationship( 'PaymentHoldHistory', backref='account', cascade='all, delete-orphan' ) account_payee = db.relationship( 'AccountPayee', backref='account', cascade='all, delete-orphan', uselist=False ) account_payee_history = db.relationship( 'AccountPayeeHistory', backref='account', cascade='all, delete-orphan' ) account_payment_term = db.relationship( 'AccountPaymentTerm', backref='account', cascade='all, delete-orphan', uselist=False ) account_tax_info_history = db.relationship( 'AccountTaxInfoHistory', backref='account', cascade='all, delete-orphan' ) @classmethod def default_order(cls): """Override to customize default ordering.""" return func.lower(cls.account_name) @classmethod def search_order(cls): """Override to customize search ordering with fuzzy search.""" inf_param = bindparam('inf', type_=db.Integer, value=999999999) search_term_param = bindparam('search_term', type_=db.String) # Match position: lower index = better match_position = text("""LEAST( IFNULL(NULLIF(POSITION( :search_term IN LOWER(account.account_name) ), 0), :inf), IFNULL(NULLIF(POSITION( :search_term IN CAST(account.account_id AS CHAR) ), 0), :inf) )""").bindparams(inf_param, search_term_param) # Match quality: shorter difference = better match_quality = text("""LEAST( IF( POSITION(:search_term IN LOWER(account.account_name)) > 0, LENGTH(account.account_name) - LENGTH(:search_term), :inf ), IF( POSITION(:search_term IN CAST(account.account_id AS CHAR)) > 0, LENGTH(CAST(account.account_id AS CHAR)) - LENGTH(:search_term), :inf ) )""").bindparams(inf_param, search_term_param) return (match_quality, match_position) @classmethod def _filter_by_search_term(cls, query, search_term): """Filter accounts by search term.""" search_term_text = re \ .sub(r'([\\%_])', r'\\\1', str(search_term)) \ .lower() return query \ .filter( or_( cls.account_name.ilike(f'%{search_term_text}%'), cls.account_id.ilike(f'%{search_term_text}%') ) ) \ .order_by(*cls.search_order()) \ .params(search_term=search_term_text) @classmethod def get_filtered_query( cls, account_name=None, account_ids=None, search_term=None, payment_entity_id=None, reference_payment_type_id=None, agreement_type_ids=None ): """Get accounts by field values. To prevent filters from being injected, they are applied to the account search in an ad-hoc fashion. """ query = cls.query # Account Name Filter -- Deprecated if account_name is not None: search_account_name_text = \ account_name.replace('\\', '\\\\').replace('%', '\\%') query = query.filter( cls.account_name.ilike(f'%{search_account_name_text}%') ) # Account IDs Filter if account_ids is not None: query = query.filter(cls.account_id.in_(account_ids)) # Filter by account_name or account_id if search_term is not None: query = cls._filter_by_search_term(query, search_term) if payment_entity_id is not None: query = query.join(cls.account_payment_term).filter_by( payment_entity_id=payment_entity_id ) if reference_payment_type_id is not None: query = query.join(cls.account_payee).filter_by( reference_payment_type_id=reference_payment_type_id ) if agreement_type_ids is not None: query = query.join(cls.account_payment_term).filter( AccountPaymentTerm.agreement_type_id.in_(agreement_type_ids) ) return query @staticmethod def get_eligible_for_payment( payment_group_id: int ) -> list: """Get eligible Accounts meeting the specified payment group's criteria. An account's payment eligibility is based on having: - a payment_minimum and payment_schedule - complete payment information in payoneer - complete tax information in the secure database - no current payment_holds - no "pending" payments with payoneer - a current balance greater than both the account's payment minimum and the payment method's payment minimum Args: payment_group_id (int): payment group identifier Returns: a list of accounts meeting the payment eligibility criteria """ return db.engine.execute( # TODO: SQL formatting is a bad practice from security perspective # as may lead to SQL injections and should be avoided in favour of # using query params templates.GET_ACCOUNTS_ELIGIBLE_FOR_PAYMENT.format( payment_group_id=payment_group_id ) ).fetchall() @staticmethod def get_eligible_for_payment_via_closing_balance( payment_group_id: int ) -> list: """Get eligible Accounts meeting the specified payment group's criteria. An account's payment eligibility is based on having: - a payment_schedule - complete payment information in payoneer - complete tax information in the secure database - no current payment_holds - no "pending" payments with payoneer Args: payment_group_id (int): payment group identifier Returns: a list of accounts meeting the payment eligibility criteria """ return db.engine.execute( text(templates.GET_ACCOUNTS_ELIGIBLE_FOR_PAYMENT_CLOSING_BALANCE), { 'payment_group_id': payment_group_id, 'bulk_payouts_payoneer_payment_types': list(BULK_PAYOUTS_PAYONEER_PAYMENT_TYPES.values()) } ).fetchall()