'''Created 15 Jun 2023 Levi''' import zipfile import os from datetime import date import pandas as pd from pymt940.utils import (get_partners, dig_it, get_last_month) class PaymentsProcessorBrd: def __init__(self, csv_path, partner_data, account_name=None, account_number=None, currency=None, account_location=None, account_agency=None, account_symbol=None): self.csv_path = csv_path self.partner_data = partner_data self._account_name = account_name self._account_number = account_number self._currency = currency self._account_location = account_location self._account_agency = account_agency self._account_symbol = account_symbol self.trans_dict = None self.names_dict = None self.obs_dict = None self.counts = None self.doc_list = None self.unique_count = None self.fiscal_code_dict = None self.days = None self.columns = ['Banca', 'Cont', 'ID1', 'Data extras', 'Moneda', 'Total debit', 'Total credit', 'Data inregistrarii', 'Data valutei', 'Suma', 'tip', 'Referinta client', 'BLANK1', 'Detalii', 'BLANK2', 'BLANK3', 'Banca parteber', 'Cod parteber', 'Partener', 'ID6', 'Referinta interna', 'Detalii2', 'Referinta banca', 'Blank4', 'Blank5', 'Blank6', 'Blank7', 'Iban partener'] @property def account_name(self): return self._account_name @account_name.setter def account_name(self, value): self._account_name = value @property def account_number(self): return self._account_number @account_number.setter def account_number(self, value): self._account_number = value @property def currency(self): return self._currency @currency.setter def currency(self, value): self._currency = value @property def account_location(self): return self._account_location @account_location.setter def account_location(self, value): self._account_location = value @property def account_agency(self): return self._account_agency @account_agency.setter def account_agency(self, value): self._account_agency = value @property def account_symbol(self): return self._account_symbol @account_symbol.setter def account_symbol(self, value): self._account_symbol = value def _concat_dfs(self): with zipfile.ZipFile(self.csv_path, 'r') as zip_ref: csv_files = [file for file in zip_ref.namelist() if file.endswith('.csv'.upper())] print(csv_files) dir_ = self.csv_path.split(sep='\\')[:-1] # print(dir_) dir_ = os.path.join(*dir_) _extracted_files = zip_ref.extractall(path='C:/' + dir_[2:]) dfs = [] for csv_file in csv_files: df = pd.read_csv(os.path.join('C:/' + dir_[2:], csv_file), sep=';', header=None, names=self.columns, keep_default_na=False) dfs.append(df) concatenated_df = pd.concat(dfs, ignore_index=True) return concatenated_df def process_payment_data(self, day_slice=(0, 2)): df_dict = get_partners(self.partner_data) fiscal_codes = df_dict['CodFiscal'] mentor_names = df_dict['Denumire'] self.fiscal_code_dict = {k: v for k, v in zip(mentor_names, fiscal_codes)} # payments_df = pd.read_csv(self.csv_path, sep=';', header=None, names=self.columns, keep_default_na=False) payments_df = self._concat_dfs() payments_df['Suma'] = payments_df['Suma'].apply(dig_it) payments_df = payments_df.loc[(payments_df['Suma'] < 0)] # print(payments_df['Debit (suma)']) payments_df['Amount'] = payments_df['Suma'].map(abs) payments_df['symbols'] = '' payments_df['types'] = '' payments_df.loc[payments_df['symbols'] == '', 'symbols'] = 401 payments_df.loc[payments_df['types'] == '', 'types'] = 'ALIMENTARE CREDIT' self._calculate_payment_attributes(payments_df, mentor_names=mentor_names, day_slice=day_slice) return payments_df def process_income_data(self, day_slice=(0, 2)): df_dict = get_partners(self.partner_data) fiscal_codes = df_dict['CodFiscal'] mentor_names = df_dict['Denumire'] self.fiscal_code_dict = {k: v for k, v in zip(mentor_names, fiscal_codes)} # income_df = pd.read_csv(self.csv_path, sep=';', header=None, names=self.columns, keep_default_na=False) income_df = self._concat_dfs() income_df['Suma'] = income_df['Suma'].apply(dig_it) # income_df = income_df.iloc[1:] income_df = income_df.loc[(income_df['Suma'] > 0)] # print(income_df['Credit (suma)']) income_df['Amount'] = income_df['Suma'] income_df['symbols'] = '' income_df['types'] = '' income_df.loc[income_df['symbols'] == '', 'symbols'] = 411.01 income_df.loc[income_df['types'] == '', 'types'] = 'ALIMENTARE CREDIT' self._calculate_payment_attributes(income_df, mentor_names, day_slice=day_slice) return income_df def _calculate_payment_attributes(self, df, mentor_names, day_slice=(0, 2)): # df['Amount'] = df['Credit'].map(str).apply(dig_it).map(abs) df['Name'] = df['Partener'] names = [name for name in df['Name'] if name != ''] for name2 in names: index_ = name2.count(' ') # nname = name2 # ' '.join(name2.split()[:index_]).upper() # print(name2) for name in mentor_names: # index_ = name.count(' ') if len(' '.join(name.split()[:index_]).upper()) < 3: index_ = 2 elif len(' '.join(name.split()[:index_]).upper()) < 2: index_ = 1 if ' '.join(name2.split()[:index_]).upper() in name.upper(): # print(name) for index, row in df.iterrows(): if row['Name'] == name2: df.loc[index, 'Name'] = name df['Transaction details'] = df['Detalii'] df['day'] = df['Data inregistrarii'].str.slice(*day_slice) df.loc[df['Transaction details'].str.lower().str.startswith('comision'), 'symbols'] = 627 df.loc[df['Transaction details'].str.lower().str.startswith('abonament'), 'symbols'] = 627 # df.loc[df['Transaction details'].str.lower().str.startswith('inchidere depozit'), 'symbols'] = 508.01 df.loc[df['symbols'] == 627, 'types'] = 'DIRECT PE CHELTUIELI' df.loc[df['Transaction details'].str.lower().str.startswith('dobanda'), 'symbols'] = 766 # df.loc[df['Transaction details'].str.lower().str.startswith('alimentare cont depozit'), 'symbols'] = 508.01 df.loc[df['symbols'] == 766, 'types'] = 'DIRECT PE VENITURI' # df.loc[df['symbols'] == 508.01, 'types'] = direction unique_count = df['day'].nunique() self.doc_list = list(range(1, unique_count + 1)) days = df['day'].unique().tolist() self.days = days trans = [] cl_names = [] obs = [] counts = [] symbols = [] types = [] for day in days: values = df.loc[df['day'] == day, 'Amount'].tolist() trans.append(values) self.trans_dict = {k: v for k, v in zip(days, trans)} for day in days: values = df.loc[df['day'] == day, 'Name'].tolist() for i in range(len(values)): if values[i] not in mentor_names: values[i] = '' cl_names.append(values) self.names_dict = {k: v for k, v in zip(days, cl_names)} # print(self.names_dict) for day in days: values = df.loc[df['day'] == day, 'Transaction details'].tolist() obs.append(values) self.obs_dict = {k: v for k, v in zip(days, obs)} for day in days: value_counts = df['day'].value_counts() count_of_value = value_counts[day] counts.append(count_of_value) self.counts = counts for day in days: acc_symbol = df.loc[df['day'] == day, 'symbols'].tolist() symbols.append(acc_symbol) self.symbol_list = {k: v for k, v in zip(days, symbols)} for day in days: type_ = df.loc[df['day'] == day, 'types'].tolist() types.append(type_) self.type_list = {k: v for k, v in zip(days, types)} self.unique_count = unique_count def to_dict(self, kind: str, month_back: int, std_client: str, acc_symbol: int): kinds = ['TotalIncasari', 'TotalPlati'] if kind not in kinds: raise TypeError(f'type not in {kinds}') else: data = {'AnLucru': date.today().year, 'LunaLucru': get_last_month(month_back), 'TotalDocumente': self.unique_count, 'documents': self.doc_list, 'AccountName': self.account_name, 'AccountSymbol': self.account_symbol, 'AccountNumber': self.account_number, 'AccountLocation': self.account_location, 'AccountAgencyLocation': self.account_agency, 'Currency': self.currency, 'days': self.days, kind: self.counts, 'incasari': self.counts, 'clients': self.names_dict, 'cods': self.fiscal_code_dict, 'amount': self.trans_dict, 'cont': self.symbol_list, 'std_client': std_client, 'obs': self.obs_dict, 'represents': self.type_list } return data if __name__ == '__main__': columns = ['Banca', 'Cont', 'ID1', 'Data extras', 'Moneda', 'Total debit', 'Total credit', 'Data inregistrarii', 'Data valutei', 'Suma', 'tip', 'Referinta client', 'BLANK1', 'Detalii', 'BLANK2', 'BLANK3', 'Banca parteber', 'Cod parteber', 'Partener', 'ID6', 'Referinta interna', 'Detalii2', 'Referinta banca', 'Blank4', 'Blank5', 'Blank6', 'Blank7', 'Iban partener'] df = pd.read_csv('../csv/230502_05130927.CSV', sep=';', header=None, names=columns, keep_default_na=False) df['Suma'] = df['Suma'].apply(dig_it) # df = df.loc[(df['Suma'] < 0)] # df.columns = columns # print(len(columns)) print(df['Partener'])