123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227 |
- '''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)
- self._calculate_payment_attributes(payments_df, mentor_names=mentor_names, day_slice=day_slice)
- 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']
- self._calculate_payment_attributes(income_df, mentor_names, day_slice=day_slice)
- 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)
- 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 = []
- 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
- 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': acc_symbol,
- 'std_client': std_client,
- 'obs': self.obs_dict
- }
- 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'])
|