You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
92 lines
3.1 KiB
92 lines
3.1 KiB
insert_query ="INSERT INTO default.TRANSACTION (TDATE, ACCOUNTID, MEMO, CITY, COUNTRY, OUTFLOW, INFLOW, OWNERID, INSTALLMENT_NR, INSTALLMENT_TT, CREATED, UPDATED) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s )" |
|
|
|
def create_lists(): |
|
import re |
|
|
|
# Open the text file |
|
with open('OUROCARD_VISA_INFINITE-Ago_24.txt', 'r', encoding='latin') as file: |
|
# Read the contents of the file |
|
contents = file.readlines() |
|
|
|
# Define the regex patterns |
|
dan_pattern = r'1 - DANIEL.*' |
|
iza_pattern = r'4 - IZABELY.*' |
|
line_pattern = r'\d{2}\.\d{2}\.\d{4}.{23}.{14}.{2}\s*\d+,\d{2}\s*\d+,\d{2}' |
|
|
|
# Lists |
|
list_dan = [] |
|
list_iza = [] |
|
current_list = None |
|
|
|
# Iterate all lines |
|
for line in contents: |
|
line = line.strip() |
|
if re.match(dan_pattern, line): |
|
current_list = 'list_dan' |
|
elif re.match(iza_pattern, line): |
|
current_list = 'list_iza' |
|
else: |
|
if re.match(line_pattern, line): |
|
if current_list == 'list_dan': |
|
list_dan.append(line) |
|
if current_list == 'list_iza': |
|
list_iza.append(line) |
|
|
|
return [list_dan, list_iza] |
|
|
|
def build_insert(lists: list[list, list], account: int, owner: int): |
|
from datetime import date, datetime |
|
|
|
insert_bulk = [] |
|
line_group_pattern = r'(\d{2})\.(\d{2})\.(\d{4})((.+PARC (\d+)\/(\d+))(\s.{12})|(.{23})(.{14}))(.{2})(\s*\d+,\d{2})(\s*\d+,\d{2})' |
|
|
|
for batch in lists: |
|
for item in batch: |
|
match = re.search(line_group_pattern, item) |
|
tTdate = str(date(int(match.group(3)), int(match.group(2)), int(match.group(1)))) |
|
tAccount = account |
|
|
|
#* check for Installments |
|
if match.group(5): |
|
tMemo = match.group(5) |
|
tCity = match.group(8) |
|
tInstallmentNr = int(match.group(6)) |
|
tInstallmentTt = int(match.group(7)) |
|
else: |
|
tMemo = match.group(9) |
|
tCity = match.group(10) |
|
tInstallmentNr = 1 |
|
tInstallmentTt = None |
|
|
|
tCountry = match.group(11) |
|
tOutflow = match.group(12).strip().replace(',', '.') |
|
tInflow = match.group(13).strip().replace(',', '.') |
|
tOwner = owner |
|
|
|
tCreated = str(datetime.now(tz=None)) |
|
tUpdated = None |
|
insert_bulk.append(( tTdate, tAccount, tMemo, tCity, tCountry, tOutflow, tInflow, tOwner, tInstallmentNr, tInstallmentTt, tCreated, tUpdated )) |
|
|
|
return insert_bulk |
|
|
|
def db_insert(insert_bulk: list[tuple]): |
|
from mysql.connector import connect, Error |
|
|
|
try: |
|
with connect( |
|
host='localhost', |
|
user='root', |
|
password='pleasehashapasswordomg', |
|
database='default' |
|
) as connection: |
|
print("CONNECTED!", connection) |
|
with connection.cursor() as cursor: |
|
cursor.executemany(insert_query, insert_bulk) |
|
connection.commit() |
|
print("DONE!") |
|
except Error as e: |
|
print(e) |
|
finally: |
|
connection.close() |
|
|
|
db_insert(build_insert(create_lists(), 1, 1)) |