robopato/nb_account_etl.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "insert_query = \"INSERT IGNORE INTO default.TRANSACTION (ID, TDATE, ACCOUNT_ID, MEMO, COUNTRY, OUTFLOW, INFLOW, OWNER_ID, INSTALLMENT_NR, INSTALLMENT_TT, BILL, CREATED, UPDATED) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s )\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def build_account_input(file_name):\n",
    "\n",
    "    import csv\n",
    "    import hashlib\n",
    "    import re\n",
    "    from datetime import date, datetime\n",
    "    from decimal import Decimal\n",
    "\n",
    "    with open(file_name or \"./documents/extrato.csv\", newline=\"\", encoding=\"latin\") as csvfile:\n",
    "        spamreader = csv.reader(csvfile)\n",
    "        csv_list = list(spamreader)\n",
    "\n",
    "    data_list = csv_list[1:]\n",
    "\n",
    "    insert_bulk = []\n",
    "\n",
    "    install_pattern = r\"(?P<install>Transferência Periódica - \\d{2}\\/\\d{2} .*?(?P<p_nr>\\d{3})\\/(?P<p_tt>\\d{3}))\"\n",
    "\n",
    "    for item in data_list:\n",
    "        dateList = item[0].split(\"/\")\n",
    "\n",
    "        tTdate = str(date(int(dateList[2]), int(dateList[1]), int(dateList[0])))\n",
    "        tAccount = 1\n",
    "        tMemo = item[2]\n",
    "        tCountry = \"BR\"\n",
    "        tOutflow = None if Decimal(item[5]) > 0 else str(Decimal(item[5])*-1)\n",
    "        tInflow = None if Decimal(item[5]) < 0 else str(Decimal(item[5]))\n",
    "        tOwner = 1\n",
    "        # Check if it's a periodic PIX transaction\n",
    "        matches = re.match(install_pattern, item[2])\n",
    "        if matches:\n",
    "            tInstallmentNr = matches.group(\"p_nr\")\n",
    "            tInstallmentTt = matches.group(\"p_tt\")\n",
    "        else:\n",
    "            tInstallmentNr = None\n",
    "            tInstallmentTt = None\n",
    "        tBill = \"CC\"\n",
    "        tCreated = str(datetime.now(tz=None))\n",
    "        tUpdated = None\n",
    "\n",
    "        preHash = str(tTdate) + tMemo + str(tOutflow) + str(tInflow)\n",
    "        tId = hashlib.sha256(preHash.encode()).hexdigest()\n",
    "\n",
    "        # Remove balance totals if detected\n",
    "        if item[2].strip() != \"S A L D O\":\n",
    "            insert_bulk.append(\n",
    "                (\n",
    "                    tId,\n",
    "                    tTdate,\n",
    "                    tAccount,\n",
    "                    tMemo,\n",
    "                    tCountry,\n",
    "                    tOutflow,\n",
    "                    tInflow,\n",
    "                    tOwner,\n",
    "                    tInstallmentNr,\n",
    "                    tInstallmentTt,\n",
    "                    tBill,\n",
    "                    tCreated,\n",
    "                    tUpdated,\n",
    "                )\n",
    "            )\n",
    "    print(\"DONE\")\n",
    "    return insert_bulk"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Test\n",
    "# build_account_input(\"./documents/extrato-30d.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def db_insert(insert_bulk: list[tuple]):\n",
    "    from mysql.connector import connect, Error\n",
    "\n",
    "    try:\n",
    "        with connect(\n",
    "            host=\"127.0.0.1\",\n",
    "            user=\"root\",\n",
    "            password=\"pleasehashapasswordomg\",\n",
    "            database=\"default\",\n",
    "        ) as connection:\n",
    "            print(\"CONNECTED!\", connection)\n",
    "            with connection.cursor() as cursor:\n",
    "                cursor.executemany(insert_query, insert_bulk)\n",
    "            connection.commit()\n",
    "            print(\"DONE!\")\n",
    "    except Error as e:\n",
    "        print(e)\n",
    "    finally:\n",
    "        connection.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Test\n",
    "# db_insert(build_account_input(\"./documents/extrato-30d.csv\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "def executor():\n",
    "    import glob\n",
    "\n",
    "    matched_files = glob.glob(\"./documents/extrato*.csv\")\n",
    "\n",
    "    try:\n",
    "        for file_name in matched_files:\n",
    "            db_insert(build_account_input(file_name=file_name))\n",
    "    except:\n",
    "        print(\"executor(): Error\")\n",
    "\n",
    "    print(\"EXECUTOR COMPLETED.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC1ADDF0>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC1ADDF0>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC1D61E0>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC1D77D0>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BAD26C00>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC1D5820>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC1D5820>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC1708F0>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC1D5820>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC137B00>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC208200>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC23FFB0>\n",
      "DONE!\n",
      "DONE\n",
      "CONNECTED! <mysql.connector.connection_cext.CMySQLConnection object at 0x000001A7BC1D5880>\n",
      "DONE!\n",
      "EXECUTOR COMPLETED.\n"
     ]
    }
   ],
   "source": [
    "executor()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}