import arrow import calendar from time import perf_counter from decimal import Decimal from datetime import datetime, timedelta from Schemas import BuildDecisionBookPayments, AccountRecords, AccountDelayInterest, ApiEnumDropdown, Build, BuildDecisionBook from sqlalchemy import select,text, Integer, Float, func, distinct, cast, Date, String, literal, desc, and_, or_, case from sqlalchemy.sql.expression import literal from Controllers.Postgres.engine import get_session_factory def find_last_day_of_month(date_value): today = date_value.date() _, last_day = calendar.monthrange(today.year, today.month) return datetime(today.year, today.month, last_day, 23, 59, 59) def find_first_day_of_month(date_value): today = date_value.date() return datetime(today.year, today.month, 1) def add_month_to_date(date_value): month_to_process = date_value.month year_to_process = date_value.year if date_value.month == 12: month_to_process = 1 year_to_process += 1 else: month_to_process += 1 _, last_day = calendar.monthrange(year_to_process, month_to_process) return datetime(year_to_process, month_to_process, 1), datetime(year_to_process, month_to_process, last_day) class BuildDuesTypes: def __init__(self): self.debit: ApiEnumDropdownShallowCopy = None self.add_debit: ApiEnumDropdownShallowCopy = None self.renovation: ApiEnumDropdownShallowCopy = None self.lawyer_expence: ApiEnumDropdownShallowCopy = None self.service_fee: ApiEnumDropdownShallowCopy = None self.information: ApiEnumDropdownShallowCopy = None class ApiEnumDropdownShallowCopy: id: int uuid: str enum_class: str key: str value: str def __init__(self, id: int, uuid: str, enum_class: str, key: str, value: str): self.id = id self.uuid = uuid self.enum_class = enum_class self.key = key self.value = value class ExactMatch: def __init__(self, account_record_id, build_decision_book_payment_id, living, diff, day_df, process_comment, bank_date, process_date, fund, debt, payment_correct): self.account_record_id = account_record_id self.build_decision_book_payment_id = build_decision_book_payment_id self.living = living self.diff = diff self.day_df = day_df self.process_comment = process_comment self.bank_date = str(bank_date) self.process_date = str(process_date) self.fund = fund self.debt = debt self.payment_correct = payment_correct def to_dict(self): return { 'account_record_id': self.account_record_id, 'build_decision_book_payment_id': self.build_decision_book_payment_id, # 'process_comment': self.process_comment, 'living': self.living, 'diff': self.diff, 'day_df': self.day_df, 'bank_date': self.bank_date, 'process_date': self.process_date, 'fund': self.fund, 'debt': self.debt, 'payment_correct': self.payment_correct } def close_payment_book(payment_row_book, account_record, value, session, is_commission_applicable: bool = False): """Create a credit entry in BuildDecisionBookPayments to close a debt. Args: payment_row_book: The debit entry to be paid account_record: The account record containing the funds value: The amount to pay session: Database session Returns: The newly created payment record """ BuildDecisionBookPayments.set_session(session) # account_record_remainder_balance = retrieve_remainder_balance_set_if_needed(account_record_id=account_record.id, session=session) # print(f'NOT Updated remainder balance: {account_record_remainder_balance} | Account record id: {account_record.id}') # Create a new credit entry (payment) new_row = BuildDecisionBookPayments.create( ref_id=str(payment_row_book.uu_id), payment_plan_time_periods=payment_row_book.payment_plan_time_periods, period_time=payment_row_book.period_time, currency=payment_row_book.currency, account_records_id=account_record.id, account_records_uu_id=str(account_record.uu_id), build_parts_id=payment_row_book.build_parts_id, build_parts_uu_id=str(payment_row_book.build_parts_uu_id), payment_amount=abs(value), # Negative for credit entries payment_types_id=payment_row_book.payment_types_id, payment_types_uu_id=str(payment_row_book.payment_types_uu_id), process_date_m=payment_row_book.process_date.month, process_date_y=payment_row_book.process_date.year, process_date=payment_row_book.process_date, build_decision_book_item_id=payment_row_book.build_decision_book_item_id if payment_row_book.build_decision_book_item_id else None, build_decision_book_item_uu_id=str(payment_row_book.build_decision_book_item_uu_id) if payment_row_book.build_decision_book_item_uu_id else None, decision_book_project_id=payment_row_book.decision_book_project_id if payment_row_book.decision_book_project_id else None, decision_book_project_uu_id=str(payment_row_book.decision_book_project_uu_id) if payment_row_book.decision_book_project_uu_id else None, build_decision_book_id=payment_row_book.build_decision_book_id if payment_row_book.build_decision_book_id else None, build_decision_book_uu_id=str(payment_row_book.build_decision_book_uu_id) if payment_row_book.build_decision_book_uu_id else None, is_confirmed=True, account_is_debit=False, ) # Save the new payment record saved_row = new_row.save() session.commit() session.refresh(saved_row) if is_commission_applicable: AccountDelayInterest.set_session(session) new_row_interest = AccountDelayInterest.create( account_record_bank_date=account_record.bank_date, book_payment_process_date=payment_row_book.process_date, debt=abs(value), account_records_id=account_record.id, account_records_uu_id=str(account_record.uu_id), build_decision_book_payment_id=saved_row.id, build_decision_book_payment_uu_id=str(saved_row.uu_id), replication_id=99, # Triggers for trigger_account_delay_interest calculation [active trigger] bank_resp_company_id=1, # Evyos interest rates will be applied when id is given bank_resp_company_uu_id="CMP", ) new_row_interest.save() session.commit() session.refresh(new_row_interest) print(f'Commission applicable: amount : {value} bank date: {account_record.bank_date} process date: {payment_row_book.process_date}') # account_record_remainder_balance = retrieve_remainder_balance_set_if_needed(account_record_id=account_record.id, session=session) # print(f'Updated remainder balance: {account_record_remainder_balance} | Account record id: {account_record.id}') # check_current_debt_to_pay_from_database_is_closed(ref_id=new_row.ref_id, session=session) return None def get_enums_from_database(): build_dues_types = BuildDuesTypes() with ApiEnumDropdown.new_session() as session: ApiEnumDropdown.set_session(session) debit_enum_shallow = ApiEnumDropdown.query.filter_by(enum_class="BuildDuesTypes", key="BDT-D").first() # Debit add_debit_enum_shallow = ApiEnumDropdown.query.filter_by(enum_class="BuildDuesTypes", key="BDT-A").first() # Add Debit renovation_enum_shallow = ApiEnumDropdown.query.filter_by(enum_class="BuildDuesTypes", key="BDT-R").first() # Renovation late_payment_enum_shallow = ApiEnumDropdown.query.filter_by(enum_class="BuildDuesTypes", key="BDT-L").first() # Lawyer expence service_fee_enum_shallow = ApiEnumDropdown.query.filter_by(enum_class="BuildDuesTypes", key="BDT-S").first() # Service fee information_enum_shallow = ApiEnumDropdown.query.filter_by(enum_class="BuildDuesTypes", key="BDT-I").first() # Information build_dues_types.debit = ApiEnumDropdownShallowCopy( debit_enum_shallow.id, str(debit_enum_shallow.uu_id), debit_enum_shallow.enum_class, debit_enum_shallow.key, debit_enum_shallow.value ) build_dues_types.add_debit = ApiEnumDropdownShallowCopy( add_debit_enum_shallow.id, str(add_debit_enum_shallow.uu_id), add_debit_enum_shallow.enum_class, add_debit_enum_shallow.key, add_debit_enum_shallow.value ) build_dues_types.renovation = ApiEnumDropdownShallowCopy( renovation_enum_shallow.id, str(renovation_enum_shallow.uu_id), renovation_enum_shallow.enum_class, renovation_enum_shallow.key, renovation_enum_shallow.value ) build_dues_types.lawyer_expence = ApiEnumDropdownShallowCopy( late_payment_enum_shallow.id, str(late_payment_enum_shallow.uu_id), late_payment_enum_shallow.enum_class, late_payment_enum_shallow.key, late_payment_enum_shallow.value ) build_dues_types.service_fee = ApiEnumDropdownShallowCopy( service_fee_enum_shallow.id, str(service_fee_enum_shallow.uu_id), service_fee_enum_shallow.enum_class, service_fee_enum_shallow.key, service_fee_enum_shallow.value ) build_dues_types.information = ApiEnumDropdownShallowCopy( information_enum_shallow.id, str(information_enum_shallow.uu_id), information_enum_shallow.enum_class, information_enum_shallow.key, information_enum_shallow.value ) return [build_dues_types.debit, build_dues_types.lawyer_expence, build_dues_types.add_debit, build_dues_types.renovation, build_dues_types.service_fee, build_dues_types.information] def get_read_payment_type(payment_result_type_uu_id): payment_type_list = get_enums_from_database() for payment_type in payment_type_list: if payment_type.uuid == payment_result_type_uu_id: return payment_type.key return None if __name__ == "__main__": # import sys # Default values # id_given = 1 count_row = 0 session_factory = get_session_factory() session = session_factory() payment_type_list = get_enums_from_database() date_from = '2022-01-01' results = session.query(distinct(AccountRecords.build_parts_id)).filter( AccountRecords.active == True, func.abs(AccountRecords.currency_value) - func.abs(AccountRecords.remainder_balance) > 0, AccountRecords.bank_date >= date_from, AccountRecords.bank_date <= datetime.now(), AccountRecords.currency_value > 0 ).order_by(AccountRecords.build_parts_id.asc()).all() build_parts_ids = [result[0] for result in results] build_parts_ids.remove(1) for build_parts_id in build_parts_ids: AccountRecords.set_session(session) BuildDecisionBookPayments.set_session(session) all_account_records_of_selected_build_parts = session.query(AccountRecords).filter( AccountRecords.active == True, func.abs(AccountRecords.currency_value) - func.abs(AccountRecords.remainder_balance) > 0, AccountRecords.bank_date >= date_from, AccountRecords.bank_date <= datetime.now(), AccountRecords.currency_value > 0, AccountRecords.build_parts_id == build_parts_id ).order_by(AccountRecords.bank_date.desc()).limit(10).offset(0).all() for account_record in all_account_records_of_selected_build_parts: AccountRecords.set_session(session) BuildDecisionBookPayments.set_session(session) selected_bank_date = account_record.bank_date selected_bank_date_month = selected_bank_date.month selected_bank_date_year = selected_bank_date.year year_month = f"{selected_bank_date_year}-{selected_bank_date_month:02d}" print(f"Build decision book id: {account_record.build_decision_book_id}") total_amount_collected_this_month = session.query(func.sum(AccountRecords.currency_value)).filter( AccountRecords.currency_value > 0, AccountRecords.build_decision_book_id == account_record.build_decision_book_id, func.abs(AccountRecords.currency_value) - func.abs(AccountRecords.remainder_balance) > 0, func.extract('month', AccountRecords.bank_date) == selected_bank_date_month, func.extract('year', AccountRecords.bank_date) == selected_bank_date_year, AccountRecords.build_parts_id == build_parts_id ).scalar() or 0 total_amount_of_debt_of_this_month = session.query(func.sum(BuildDecisionBookPayments.debt_to_pay)).filter( BuildDecisionBookPayments.account_is_debit == True, BuildDecisionBookPayments.build_decision_book_id == account_record.build_decision_book_id, func.extract('month', BuildDecisionBookPayments.process_date) == selected_bank_date_month, func.extract('year', BuildDecisionBookPayments.process_date) == selected_bank_date_year, BuildDecisionBookPayments.process_date <= datetime.now() ).scalar() or 0 print(f"{year_month} total amount of debt in this month : ", f"{total_amount_of_debt_of_this_month:,.2f}") print(f"{year_month} total amount collected this month : ", f"{total_amount_collected_this_month:,.2f}") avaliable_fund = abs(account_record.currency_value) - abs(account_record.remainder_balance) total_debt_to_pay_of_this_month = session.query(func.sum(BuildDecisionBookPayments.debt_to_pay)).filter( BuildDecisionBookPayments.account_is_debit == True, BuildDecisionBookPayments.build_decision_book_id == account_record.build_decision_book_id, func.extract('month', BuildDecisionBookPayments.process_date) == selected_bank_date_month, func.extract('year', BuildDecisionBookPayments.process_date) == selected_bank_date_year, BuildDecisionBookPayments.process_date <= datetime.now() ).scalar() or 0 exact_match_with_account_fund_and_debt_to_pay = BuildDecisionBookPayments.query.filter( BuildDecisionBookPayments.account_is_debit == True, BuildDecisionBookPayments.build_decision_book_id == account_record.build_decision_book_id, func.extract('month', BuildDecisionBookPayments.process_date) == selected_bank_date_month, func.extract('year', BuildDecisionBookPayments.process_date) == selected_bank_date_year, BuildDecisionBookPayments.process_date <= datetime.now(), BuildDecisionBookPayments.debt_to_pay == avaliable_fund ).first() is_fund_available_is_completely_closes_all_debts = avaliable_fund >= total_debt_to_pay_of_this_month is_one_of_the_debt_to_pay_exactly_matches = avaliable_fund == exact_match_with_account_fund_and_debt_to_pay.debt_to_pay if exact_match_with_account_fund_and_debt_to_pay else False print(f"account record id: {account_record.id} | bank date: {account_record.bank_date.date()} | total debt to pay: {total_debt_to_pay_of_this_month:,.2f}") print(f"avaliable fund: {avaliable_fund:,.2f} | exact match: {is_one_of_the_debt_to_pay_exactly_matches}") print(f"Build part processed: {build_parts_id}") session.close() print(f"Processed {count_row} rows") # Parse command line arguments # if len(sys.argv) > 1: # try: # id_given = int(sys.argv[1]) # except ValueError: # print(f"Error: Invalid limit value '{sys.argv[1]}'. Using default limit={limit}")