122 lines
5.1 KiB
Python
122 lines
5.1 KiB
Python
# Importing libraries
|
|
from numpy import genfromtxt, savetxt
|
|
from time import time
|
|
from datetime import datetime
|
|
from models import models
|
|
from models.database import engine, SessionLocal
|
|
#import sustApp.models.models as models
|
|
#from sustApp.models.database import engine, SessionLocal
|
|
|
|
import os, re
|
|
from pytz import timezone
|
|
import pandas as pd
|
|
import glob
|
|
#import csv
|
|
|
|
def extract_year_from_reference(reference):
|
|
"""Extract year from payment reference (e.g., '2025-4732-163-1' -> '2025')"""
|
|
if reference:
|
|
year = reference.split('-')[0]
|
|
return year
|
|
return None
|
|
|
|
def Load_Data(file_name):
|
|
df = pd.read_csv(file_name, delimiter=';', encoding='utf-8-sig', dtype=str, header=None)
|
|
print(df.head())
|
|
return df.values.tolist()
|
|
|
|
# def Load_Data(file_name):
|
|
# #data = genfromtxt(file_name, delimiter=';', skip_header=1, converters={0: lambda s: str(s)})
|
|
# #str2date = lambda x: datetime.strptime(x, '%d-%m-%y')
|
|
# data = genfromtxt(file_name, delimiter=';', dtype=None, skip_header=0, encoding='utf-8-sig')
|
|
|
|
# return data.tolist()
|
|
|
|
# def Load_Data(file_name):
|
|
# data = []
|
|
# with open(file_name, encoding='utf-8-sig') as f:
|
|
# reader = csv.reader(f, delimiter=';')
|
|
# for row in reader:
|
|
# data.append(row)
|
|
# return data
|
|
|
|
def extract_values(text):
|
|
# Regular expressions to match the required fields
|
|
regex_patterns = {
|
|
'Auftraggeber': r'Auftraggeber: (.*?) (Verwendungszweck|Zahlungsreferenz|IBAN|BIC)',
|
|
'Verwendungszweck': r'Verwendungszweck: (.*?) (Zahlungsreferenz|IBAN|BIC)',
|
|
'Zahlungsreferenz': r'Zahlungsreferenz: (.*?) (Verwendungszweck|IBAN|BIC)',
|
|
'IBAN': r'IBAN Auftraggeber: (.*?) ',
|
|
'BIC': r'BIC Auftraggeber: (.*?)$'
|
|
}
|
|
|
|
# Initialize an empty list to store extracted information
|
|
extracted_info = []
|
|
|
|
# Initialize an empty dictionary to store information from this row
|
|
row_info = {}
|
|
|
|
# Extract information using regular expressions
|
|
for key, pattern in regex_patterns.items():
|
|
match = re.search(pattern, text)
|
|
if match:
|
|
row_info[key] = match.group(1)
|
|
else:
|
|
row_info[key] = None
|
|
|
|
# Append the extracted information from this row to the list
|
|
extracted_info.append(row_info)
|
|
|
|
return extracted_info
|
|
|
|
if __name__ == "__main__":
|
|
t = time()
|
|
|
|
db = SessionLocal()
|
|
models.Base.metadata.create_all(bind=engine)
|
|
|
|
try:
|
|
home_directory = os.path.expanduser('~')
|
|
source_path = os.path.join(home_directory, "OneDrive", "Dokumente", "SUST", "import_csv", "*.csv")
|
|
for source_csv in glob.glob(source_path):
|
|
if '_out.csv' not in source_csv:
|
|
#source_csv = os.path.join(home_directory, "OneDrive", "Dokumente", "SUST", "import_csv", "UmsatzListe_AT493443700000113407_20250529_091945_mod.csv")
|
|
target_csv = source_csv.replace(".csv", "_out.csv")
|
|
data = Load_Data(source_csv)
|
|
|
|
tz = timezone('UTC') # Replace with your desired timezone
|
|
outData = []
|
|
for i in data:
|
|
if ('Auftraggeber' in i[1]):
|
|
# extract key-value pairs Auftraggeber, Verwendungszweck, Zahlungsreferenz, IBAN
|
|
new_list = extract_values(i[1])
|
|
|
|
if len(new_list) > 0:
|
|
zahlungsreferenz = new_list[0].get('Zahlungsreferenz')
|
|
year = extract_year_from_reference(zahlungsreferenz)
|
|
|
|
#outData.append([i[0], new_list[0].get('Auftraggeber'), new_list[0].get('Zahlungsreferenz'), new_list[0].get('Verwendungszweck'), new_list[0].get('IBAN'), i[2], i[3].replace(',','.'), i[4], i[5], new_list[0].get('Zahlungsreferenz')])
|
|
outData.append([
|
|
i[0], # valuta_date
|
|
new_list[0].get('Auftraggeber'), # client
|
|
zahlungsreferenz, # reference
|
|
new_list[0].get('Verwendungszweck'), # intended
|
|
new_list[0].get('IBAN'), # IBAN
|
|
i[2], # entry_date
|
|
i[3].replace(',','.'), # amount
|
|
i[4], # currency
|
|
i[5], # timestamp
|
|
year # year (extracted from reference)
|
|
])
|
|
|
|
# Replace None with empty string before saving
|
|
outData = [['' if x is None else x for x in row] for row in outData]
|
|
savetxt(target_csv, outData, delimiter=";", encoding='utf-8', header='valuta_date;client;reference;intended;IBAN;entry_date;amount;currency;timestamp;year', fmt='%s')
|
|
bak_dir = os.path.join(os.path.dirname(source_csv), "bak")
|
|
os.makedirs(bak_dir, exist_ok=True)
|
|
os.rename(source_csv, os.path.join(bak_dir, os.path.basename(source_csv)))
|
|
|
|
except Exception as e:
|
|
print (e)
|
|
finally:
|
|
print("Time elapsed: " + str(time() - t) + " s.") #0.091s |