text extraction.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# for the LLM\n",
    "\n",
    "!pip3 install openai"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# only for OCR \n",
    "\n",
    "#!pip3 install langdetect\n",
    "#!pip3 install pytesseract\n",
    "#!pip3 install pdf2image Pillow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import re\n",
    "from openai import OpenAI"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Initial Text Extraction + cleaning\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Using PyMuPDF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "import fitz  # the PyMuPDF library\n",
    "\n",
    "def extract_text_pymupdf(pdf_path):\n",
    "    doc = fitz.open(pdf_path)  # Open the PDF file\n",
    "    text = \"\"  # Initialize an empty string to store text\n",
    "\n",
    "    for page in doc:  # Iterate through each page\n",
    "        text += page.get_text()  # Extract text from the page and append it\n",
    "\n",
    "    doc.close()  # Close the document\n",
    "    return text\n",
    "\n",
    "# here change path to pdf\n",
    "#pdf_path = '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/Jaarrekening-2022-DG-Press-HoldinG-B.V.pdf'\n",
    "#pdf_path = '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/Jaarrekening-Momo-Medical-Holding-B.V.-2022.pdf'\n",
    "pdf_path = '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/Jaarrekening-2022-Informed-IT-Holding-B.V.pdf'\n",
    "#pdf_path = 'KVK Sample Files 2 - Julia/01016572-docType-sd_jaarrek_art394_lid1-docJaar-2022-docCreatie-2023-12-21-docId-090299cc61e0f819.pdf'\n",
    "#pdf_path = '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/Jaarrekening-2022-RR-Mechatronics-International-B.V.pdf'\n",
    "#pdf_path= '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/Jaarverslag-2022-TABS-Holland.pdf'\n",
    "\n",
    "text = extract_text_pymupdf(pdf_path)\n",
    "raw_text = text # save raw text for later use\n",
    "text_data = text\n",
    "\n",
    "#print(text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# getting descriptives\n",
    "doc = fitz.open(pdf_path)\n",
    "num_pages = doc.page_count\n",
    "doc.close()\n",
    "\n",
    "# Count the number of words\n",
    "num_words = len(text.split())\n",
    "\n",
    "print(f\"Number of pages: {num_pages}\")\n",
    "print(f\"Number of words: {num_words}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# defining functions for handling the raw text and cleaning it\n",
    "\n",
    "def is_number(s: str) -> [str, bool]:\n",
    "    if s.startswith(\"(\") and s.endswith(\")\"):\n",
    "        s = s[1:-1]\n",
    "    elif s == \"-\":\n",
    "        return \"0\", True\n",
    "    \n",
    "    # Normalize number format\n",
    "    s = s.replace(\",\", \".\")\n",
    "    num = s.replace(\".\", \"\")\n",
    "\n",
    "    if s.startswith(\"-\"):\n",
    "        return s, num[1:].isdigit()\n",
    "\n",
    "    # Check for positive numbers\n",
    "    if num.isdigit():\n",
    "        # Check if it's a year and ignore if it is\n",
    "        if 2000 <= int(num) <= 2025:\n",
    "            return s, False\n",
    "        return s, True\n",
    "\n",
    "    return s, False\n",
    "\n",
    "def is_year(s: str) -> bool:\n",
    "    try:\n",
    "        num = int(s)\n",
    "        return 2000 <= num <= 2025\n",
    "    except ValueError:\n",
    "        return False\n",
    "    \n",
    "def ends_with_percent(s: str) -> bool: return s.endswith(\"%\")\n",
    "def is_page_number(s: str) -> bool:   return s.strip().startswith(\"Page\") or s.strip().startswith(\"Pagina\") \n",
    "\n",
    "def remove_small_numbers(text, threshold=100):\n",
    "    # Remove numbers smaller than 100 from the first 1200 characters\n",
    "    first_1200_chars = text[:1200]\n",
    "    modified_first_1200_chars = re.sub(r'\\b([1-9][0-9]?|100)\\b', '', first_1200_chars)\n",
    "    \n",
    "    return modified_first_1200_chars + text[1000:]\n",
    "\n",
    "def remove_empty_lines(text):\n",
    "    # Split the text into lines\n",
    "    lines = text.split('\\n')\n",
    "    \n",
    "    # Remove empty lines\n",
    "    non_empty_lines = [line for line in lines if line.strip() != '']\n",
    "    \n",
    "    # Join the non-empty lines back together\n",
    "    cleaned_text = '\\n'.join(non_empty_lines)\n",
    "    \n",
    "    return cleaned_text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# logic for extracting currency and handling the values that need to be multiplied: if x € mln. or x € mld. or  € x 1.000 , ...,\n",
    "# multiply the numbers on the page by the corresponding factor, separately page by page\n",
    "\n",
    "def multiply_numbers_in_text(text, multiplier):\n",
    "    # Function to replace and multiply numbers\n",
    "    def replace_numbers(match):\n",
    "        number_str = match.group(0).replace(',', '')\n",
    "        number = float(number_str)\n",
    "        # return actual value if number looks like year\n",
    "        # CAREFUL - this check is performed a second time in `parse_adjusted_financial_text`, relevant for other notations\n",
    "        if is_year(number):\n",
    "            return number_str\n",
    "        multiplied_number = number * multiplier\n",
    "        return f\"{multiplied_number:.2f}\"\n",
    "\n",
    "    # Regular expression to find all numbers in the text\n",
    "    number_pattern = re.compile(r'\\b\\d+[.,]?\\d*\\b')\n",
    "    # Replace all found numbers with their multiplied values\n",
    "    return number_pattern.sub(replace_numbers, text)\n",
    "\n",
    "def process_text_with_multipliers(text):\n",
    "    if re.search(r'\\d+ € mln\\.', text) or re.search(r'€ mln\\.', text) or re.search(r'€ mln', text) or re.search(r'Resultaten mln\\.', text) or re.search(r'mln\\.', text):\n",
    "        text = multiply_numbers_in_text(text, 1_000_000)\n",
    "    elif re.search(r'\\d+ € mld\\.', text) or re.search(r'€ mld\\.', text) or re.search(r'€ mld', text) or re.search(r'Resultaten mld\\.', text) or re.search(r'mld\\.', text):\n",
    "        text = multiply_numbers_in_text(text, 1_000_000_000)\n",
    "    elif re.search(r'€ \\d+ 1\\.000', text) or re.search(r'\\(x € 1\\.000\\)', text) or re.search(r'€ 1000', text) or re.search(r'€1000', text) or re.search(r'€ x 1.000', text) or re.search(r'x 1.000',text):\n",
    "        text = multiply_numbers_in_text(text, 1_000)\n",
    "    return text\n",
    "\n",
    "def remove_percentages(text):\n",
    "    def replace_numbers(match):\n",
    "        number_str = match.group(0).replace(',', '')\n",
    "        number = float(number_str)\n",
    "        if number < 100:\n",
    "            return ''\n",
    "        return number_str\n",
    "\n",
    "    number_pattern = re.compile(r'\\b\\d+[.,]?\\d*\\b')\n",
    "    return number_pattern.sub(replace_numbers, text)\n",
    "\n",
    "def process_pdf_with_multipliers(pdf_path):\n",
    "    doc = fitz.open(pdf_path)  # Open the PDF file\n",
    "    final_text = \"\"  # Initialize an empty string to store the processed text\n",
    "\n",
    "    for i, page in enumerate(doc):  # Iterate through each page with index\n",
    "        page_text = page.get_text()  # Extract text from the page\n",
    "\n",
    "        # Apply process_text_with_multipliers to all pages\n",
    "        page_text = process_text_with_multipliers(page_text)\n",
    "        \n",
    "        if '%' in page_text:\n",
    "            # If '%' sign is present, remove numbers below 100\n",
    "            page_text = remove_percentages(page_text)\n",
    "\n",
    "        final_text += page_text  # Append the processed text from the page\n",
    "\n",
    "    doc.close()  # Close the document\n",
    "    return final_text\n",
    "\n",
    "processed_text = process_pdf_with_multipliers(pdf_path)\n",
    "\n",
    "#print(processed_text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# function for cleaning the text and extract the relevant information usning the functions defined above, handling long descriptions\n",
    "MAX_DESCRIPTION_LENGTH = 220\n",
    "LLM_length=100\n",
    "\n",
    "def parse_adjusted_financial_text(text):\n",
    "\n",
    "    text = remove_empty_lines(text)\n",
    "    text= remove_small_numbers(text)\n",
    "    \n",
    "    # Split the text into lines\n",
    "    lines = text.strip().split(\"\\n\")\n",
    "    non_empty_lines = [line for line in lines if line.strip() != '']\n",
    "    \n",
    "    # Join the non-empty lines back together\n",
    "    cleaned_text = '\\n'.join(non_empty_lines)\n",
    "    \n",
    "    # Initialize an empty list to store our parsed data\n",
    "    parsed_data = []\n",
    "    \n",
    "    # this variable tracks if we found any numeric data on the current page\n",
    "    no_numeric_data_on_page = True\n",
    "    misc_text_data = []\n",
    "    \n",
    "    # Initialize empty strings for current description and values\n",
    "    current_description = \"\"\n",
    "    value_year1 = \"\"\n",
    "     # Iterate over each line in the text\n",
    "    for line in lines:\n",
    "        # Check if line is a description or a value\n",
    "        maybe_num, is_num = is_number(line)\n",
    "        # skipping over % values to not mistake them for financial records\n",
    "        # question czy to dziala wtedy dla wszystkich is num czy tylko dla tych konkretnych??\n",
    "        if ends_with_percent(line):\n",
    "            continue\n",
    "        elif is_num:\n",
    "            if value_year1 == \"\":\n",
    "                value_year1 = maybe_num\n",
    "            else:\n",
    "                # Remove thousand separators before converting to float\n",
    "                cleaned_value_year1 = value_year1.replace('.', '').replace(',', '.')\n",
    "                cleaned_maybe_num = maybe_num.replace('.', '').replace(',', '.')\n",
    "                parsed_data.append([\n",
    "                    current_description.strip().lower(),\n",
    "                    int(float(cleaned_value_year1)),  # Convert to float, then to int\n",
    "                    int(float(cleaned_maybe_num))  # Convert to float, then to int\n",
    "                ])\n",
    "                current_description = \"\"\n",
    "                value_year1 = \"\"\n",
    "                no_numeric_data_on_page = False\n",
    "        else:\n",
    "            # we might find long streches of text unrelated to data labeling\n",
    "            # in that case we have the following solutions:\n",
    "            #  * cap descripiton length (x chars)\n",
    "            #  * try to extract category name using llms\n",
    "            # If it's not a digit, it's a description\n",
    "            # Accumulate descriptions until we reach a digit\n",
    "            if current_description:\n",
    "                if is_page_number(line):\n",
    "                    if no_numeric_data_on_page:\n",
    "                        misc_text_data.append(current_description)\n",
    "                        current_description = \"\"\n",
    "                    # we want to reset the tracker at every new page\n",
    "                    no_numeric_data_on_page = True\n",
    "                else: current_description += \" \" + line.strip()\n",
    "               \n",
    "                #skip if the length is more than MAX_DESCRIPTION_LENGTH\n",
    "                if len(current_description) > MAX_DESCRIPTION_LENGTH:\n",
    "                    current_description = \"\"        \n",
    "            else:\n",
    "                current_description = line\n",
    "              \n",
    "    # Convert the parsed data into a DataFrame\n",
    "    return pd.DataFrame(parsed_data), misc_text_data\n",
    "\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Parse the adjusted text data\n",
    "df_parsed, misc_text_data = parse_adjusted_financial_text(processed_text)\n",
    "\n",
    "# Display the parsed DataFrame\n",
    "# display(df_parsed)\n",
    "# df_parsed.to_csv('parsed_data.csv', index=False)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def is_description_valid(description):\n",
    "    # Ensure the description is a string\n",
    "    if isinstance(description, bytes):\n",
    "        description = description.decode('utf-8') \n",
    "    # Remove all non-alphanumeric characters except spaces, including currency symbols, then strip excess whitespace\n",
    "    clean_description = re.sub(r'[^a-zA-Z0-9\\s]', '', description).strip()\n",
    "    # Check if there's any alphanumeric content left\n",
    "    return bool(re.search(r'\\w', clean_description))\n",
    "\n",
    "def remove_currency_symbols(text):\n",
    "    return re.sub(r'[$€£¥₩₹]', '', text)\n",
    "\n",
    "def refine_parsed_data(parsed_data):\n",
    "    refined_data = []\n",
    "    for i in range(len(parsed_data)):\n",
    "        # Handle cases where entries may have variable number of columns\n",
    "        row = parsed_data.iloc[i] if hasattr(parsed_data, 'iloc') else parsed_data[i]\n",
    "        description = row[0]\n",
    "        values = row[1:]\n",
    "\n",
    "        # Remove currency symbols from description\n",
    "        description = remove_currency_symbols(description)\n",
    "\n",
    "        # Check for valid description\n",
    "        if is_description_valid(description):\n",
    "            if refined_data and all(isinstance(v, int) and isinstance(prev_v, int) for v, prev_v in zip(values, refined_data[-1][1:])):\n",
    "                # Add current line's numbers to the previous line's numbers\n",
    "                refined_data[-1][1:] = [v + prev_v for v, prev_v in zip(values, refined_data[-1][1:])]\n",
    "            else:\n",
    "                refined_data.append([description] + list(values))\n",
    "\n",
    "    return refined_data\n",
    "\n",
    "refined_data = refine_parsed_data(df_parsed)\n",
    "display(refined_data)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# set up in a table, to have description, year1, year2 values, in df_parsed\n",
    "\n",
    "# df parsed= 1st column: 1st element of the line, called \"Description\", 2nd column: 2nd element of the line, called \"Year 1\", 3rd column: 3rd element of the line, called \"Year 2\"\n",
    "df_parsed = pd.DataFrame(refined_data)\n",
    "df_parsed.columns = [\"Description\", \"Year 1\", \"Year 2\"]\n",
    "display(df_parsed)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using OCR"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "# need to install poppler and tesseract for OCR to work (brew install poppler tesseract)\n",
    "\n",
    "import pytesseract\n",
    "from PIL import Image\n",
    "from pdf2image import convert_from_path\n",
    "import re\n",
    "\n",
    "# deinfe the classes for OCR\n",
    "class Page:\n",
    "    def __init__(self, text):\n",
    "        self.lines = []\n",
    "        for line in text.strip().split('\\n'):\n",
    "            # TODO: additonal cleaning steps here\n",
    "            cl = line.strip()\n",
    "            if cl != \"\":\n",
    "                self.lines.append(cl)\n",
    "            \n",
    "    def __len__(self):\n",
    "        acc = 0\n",
    "        for line in self.lines:\n",
    "            acc += len(line)\n",
    "        return acc\n",
    "    \n",
    "    def num_lines(self): return len(self.lines)\n",
    "        \n",
    "    def apply_to_lines(self, func):\n",
    "        self.lines = [func(line) for line in self.lines]\n",
    "        \n",
    "    def apply_to_page(self, transform):\n",
    "        # first create a single string from the lines\n",
    "        page_text = '\\n'.join(self.lines)\n",
    "        page_text = transform(page_text)\n",
    "        # then split the text back into lines\n",
    "        self.lines = page_text.strip().split('\\n')\n",
    "\n",
    "class Document:\n",
    "    def __init__(self, pdf_path):\n",
    "        self.pdf_path = pdf_path\n",
    "        self.pages = self._extract_text_from_pdf()\n",
    "\n",
    "    def _extract_text_from_pdf(self):\n",
    "        lang = 'eng+nld'  # both English and Dutch\n",
    "        pages = []\n",
    "        for page in convert_from_path(self.pdf_path, 500):  # here change DPI\n",
    "            text = pytesseract.image_to_string(page, lang=lang)\n",
    "            pages.append(Page(text))\n",
    "        return pages\n",
    "\n",
    "    def get_page(self, page_number) -> Page:\n",
    "        return self.pages[page_number]\n",
    "\n",
    "    def get_min_chars(self, min_chars):\n",
    "        acc = 0\n",
    "        ret_pages = []\n",
    "        for page in self.pages:\n",
    "            acc += len(page)\n",
    "            ret_pages.append(page)\n",
    "            if acc >= min_chars:\n",
    "                return ret_pages\n",
    "\n",
    "    def insert_pages(self, pages, overwrite=False):\n",
    "        if overwrite:  # start overwriting pages from the beginning\n",
    "            self.pages = pages + self.pages[len(pages):]\n",
    "        else:\n",
    "            self.pages += pages\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pytesseract\n",
    "from PIL import Image\n",
    "from pdf2image import convert_from_path\n",
    "import re\n",
    "\n",
    "# defining functions for cleaning the output\n",
    "\n",
    "class DocumentCleaner:\n",
    "    def __init__(self):\n",
    "        self.dupa = True\n",
    "\n",
    "    @staticmethod\n",
    "    def try_or_false(func, error_type):\n",
    "        try:\n",
    "            return func()\n",
    "        except error_type:\n",
    "            return False\n",
    "\n",
    "    @staticmethod\n",
    "    def is_year(s):\n",
    "        try:\n",
    "            num = int(s.replace(',', '').replace('.', ''))\n",
    "            return 2000 <= num <= 2025\n",
    "        except ValueError:\n",
    "            return False\n",
    "\n",
    "    @staticmethod\n",
    "    def delete_year_numbers(line):\n",
    "        words = line.split()\n",
    "        return ' '.join(word for word in words if not DocumentCleaner.is_year(word))\n",
    "\n",
    "    @staticmethod\n",
    "    def remove_small_numbers(line):\n",
    "        def should_remove(word):\n",
    "            try:\n",
    "                # Check if word is a number (integer or float) and less than 100\n",
    "                num = float(word)\n",
    "                return num < 100\n",
    "            except ValueError:\n",
    "                return False\n",
    "        words = line.split()\n",
    "        return ' '.join(word for word in words if not (len(word) <= 2 or should_remove(word)))\n",
    "\n",
    "    @staticmethod\n",
    "    def remove_percentages(line):\n",
    "        return re.sub(r'\\b\\d{1,2}\\b%', '', line)\n",
    "\n",
    "    @staticmethod\n",
    "    def remove_brackets(line):\n",
    "        return re.sub(r'\\((\\d+)\\)', r'-\\1', line) \n",
    "\n",
    "    @staticmethod\n",
    "    def is_page_number(line):\n",
    "        return line.startswith(\"Pagina\")\n",
    "\n",
    "    @staticmethod\n",
    "    def remove_underscores(line):\n",
    "        return line.replace('_', ' ')\n",
    "\n",
    "    @staticmethod\n",
    "    def replace_dash_with_zero(line):\n",
    "        return line.replace('-', '0').replace(':', '0')\n",
    "\n",
    "    @staticmethod\n",
    "    def remove_currency_symbols(line):\n",
    "        return line.replace('€', '').replace('$', '')\n",
    "    \n",
    "    @staticmethod\n",
    "    def delete_long_words(line):\n",
    "        words = line.split()\n",
    "        return ' '.join(word for word in words if len(word) <= 100)\n",
    "\n",
    "    @staticmethod\n",
    "    def clean_line(line):\n",
    "        if not DocumentCleaner.is_page_number(line):\n",
    "            line = DocumentCleaner.delete_year_numbers(line)\n",
    "            line = DocumentCleaner.remove_small_numbers(line)\n",
    "            line = DocumentCleaner.remove_percentages(line)\n",
    "            line = DocumentCleaner.remove_currency_symbols(line)\n",
    "            line = DocumentCleaner.remove_underscores(line)\n",
    "            line = DocumentCleaner.replace_dash_with_zero(line)\n",
    "            line = DocumentCleaner.remove_brackets(line)\n",
    "            line = DocumentCleaner.delete_long_words(line)\n",
    "        return line\n",
    "    \n",
    "    @staticmethod\n",
    "    def multiply_numbers_in_text(text, multiplier):\n",
    "    # Function to replace and multiply numbers\n",
    "        def replace_numbers(match):\n",
    "            number_str = match.group(0).replace(',', '')\n",
    "            number = float(number_str)\n",
    "            # return actual value if number looks like year to avoid multiplying it\n",
    "            if DocumentCleaner.is_year(number):\n",
    "                return number_str\n",
    "            multiplied_number = number * multiplier\n",
    "            return f\"{multiplied_number:.2f}\"\n",
    "\n",
    "        # Regular expression to find all numbers in the text\n",
    "        number_pattern = re.compile(r'\\b\\d+[.,]?\\d*\\b')\n",
    "        # Replace all found numbers with their multiplied values\n",
    "        return number_pattern.sub(replace_numbers, text)\n",
    "\n",
    "    @staticmethod\n",
    "    def process_text_with_multipliers(text):\n",
    "        if re.search(r'\\d+ € mln\\.', text) or re.search(r'€ mln\\.', text) or re.search(r'€ mln', text) or re.search(r'Resultaten mln.', text) or re.search(r'mln.', text):\n",
    "            text = DocumentCleaner.multiply_numbers_in_text(text, 1_000_000)\n",
    "        elif re.search(r'\\d+ € mld\\.', text) or re.search(r'€ mld\\.', text) or re.search(r'€ mld', text) or re.search(r'Resultaten mld.', text) or re.search(r'mld.', text):\n",
    "            text =DocumentCleaner.multiply_numbers_in_text(text, 1_000_000_000)\n",
    "        elif re.search(r'€ \\d+ 1\\.000', text) or re.search(r'\\(x € 1\\.000\\)', text) or re.search(r'€ 1000', text) or re.search(r'€1000', text) or re.search(r'€ x 1.000', text) or re.search(r'x 1.000', text):\n",
    "            text = DocumentCleaner.multiply_numbers_in_text(text, 1_000)\n",
    "        return text\n",
    "\n",
    "    def remove_contents(self, document: Document, table_len=1500):\n",
    "        clean_table_of_contents = []\n",
    "        table_of_contents = document.get_min_chars(table_len)\n",
    "        for page in table_of_contents:\n",
    "            clean_table_of_contents.append(page.apply_to_lines(DocumentCleaner.remove_small_numbers))\n",
    "        document.insert_pages(clean_table_of_contents, overwrite=True)\n",
    "\n",
    "    def clean_page(self, page: Page):\n",
    "        page.apply_to_page(DocumentCleaner.process_text_with_multipliers)\n",
    "        page.apply_to_lines(DocumentCleaner.clean_line)\n",
    "        \n",
    "    def clean_document(self, document: Document):\n",
    "        for page in document.pages:\n",
    "            self.clean_page(page)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# here change path to pdf\n",
    "# with OCR, this takes a while, but most of the time the output is easier to work with/more accurate\n",
    "\n",
    "# pdf_path = '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/Jaarrekening-2022-DG-Press-HoldinG-B.V.pdf'\n",
    "pdf_path = '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/Jaarrekening-Momo-Medical-Holding-B.V.-2022.pdf'\n",
    "#pdf_path = '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/Jaarrekening-2022-Informed-IT-Holding-B.V.pdf'\n",
    "# pdf_path = 'KVK Sample Files 2 - Julia/01042818-docType-sd_jaarrek_art394_lid1-docJaar-2022-docCreatie-2023-12-28-docId-090299cc626872c5.pdf'\n",
    "# pdf_path = 'KVK Sample Files 2 - Julia/01016572-docType-sd_jaarrek_art394_lid1-docJaar-2022-docCreatie-2023-12-21-docId-090299cc61e0f819.pdf'\n",
    "#pdf_path = '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/Jaarrekening-2022-RR-Mechatronics-International-B.V.pdf'\n",
    "#pdf_path= '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/Jaarverslag-2022-TABS-Holland.pdf'\n",
    "#text = extract_text_from_pdf(pdf_path)\n",
    "#text_data = text\n",
    "#print(text)\n",
    "doc = Document(pdf_path)\n",
    "text=doc\n",
    "text_description = doc\n",
    "\n",
    "# for text, join all the lines in the text\n",
    "text = \"\\n\".join([line for page in doc.pages for line in page.lines])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# getting descriptives\n",
    "\n",
    "# Calculate the number of pages\n",
    "num_pages = len(doc.pages)\n",
    "\n",
    "# Count the number of words\n",
    "num_words = len(text.split())\n",
    "\n",
    "\n",
    "print(f\"Number of pages: {num_pages}\")\n",
    "print(f\"Number of words: {num_words}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# printing the text\n",
    "for page in doc.pages:\n",
    "    print(page.lines)\n",
    "    print(page.num_lines())\n",
    "    print(\"---------\")\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cleaner = DocumentCleaner()\n",
    "cleaner.clean_document(doc)\n",
    "num_pages = len(doc.pages)\n",
    "\n",
    "for i in range(num_pages):\n",
    "    print(doc.get_page(i).lines)\n",
    "    print(\"---------\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# creating output from cleaned text\n",
    "\n",
    "def try_parse_float(s):\n",
    "    try:\n",
    "        float(s.replace(',', '').replace('(', '').replace(')', ''))\n",
    "        return True\n",
    "    except ValueError:\n",
    "        return False\n",
    "\n",
    "def extract_lines_with_descriptions_and_numbers(document: Document):\n",
    "    extracted_lines = []\n",
    "    buffer_line = \"\"\n",
    "\n",
    "    for page in document.pages:\n",
    "        for line in page.lines:\n",
    "            words = line.split()\n",
    "            numeric_words = [word for word in words if try_parse_float(word)]\n",
    "            non_numeric_words = [word for word in words if not try_parse_float(word)]\n",
    "\n",
    "            # Check if the line has both text and numbers\n",
    "            if non_numeric_words and numeric_words:\n",
    "                # Add any buffered line before processing this line\n",
    "                if buffer_line:\n",
    "                    extracted_lines.append(buffer_line)\n",
    "                    buffer_line = \"\"\n",
    "                extracted_lines.append(line)\n",
    "            # If the line contains only numbers, append to the buffer\n",
    "            elif numeric_words and not non_numeric_words:\n",
    "                if buffer_line:\n",
    "                    buffer_line += \" \" + line\n",
    "                else:\n",
    "                    buffer_line = line\n",
    "            # Reset buffer if the line contains only text\n",
    "            elif non_numeric_words:\n",
    "                if buffer_line:\n",
    "                    buffer_line = \"\"\n",
    "                buffer_line = \"\"\n",
    "\n",
    "    # Add any remaining buffer line that contains numbers and text\n",
    "    if buffer_line and any(try_parse_float(word) for word in buffer_line.split()) and any(not try_parse_float(word) for word in buffer_line.split()):\n",
    "        extracted_lines.append(buffer_line)\n",
    "\n",
    "    return extracted_lines\n",
    "\n",
    "extracted_lines = extract_lines_with_descriptions_and_numbers(doc)\n",
    "\n",
    "#for line in extracted_lines:\n",
    "#    print(line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# extracting tables from the text, adjusting the output\n",
    "\n",
    "def try_parse_float(s):\n",
    "    try:\n",
    "        float(s.replace(',', '').replace('(', '').replace(')', ''))\n",
    "        return True\n",
    "    except ValueError:\n",
    "        return False\n",
    "\n",
    "def refine_lines_with_descriptions_and_numbers(document: Document):\n",
    "    extracted_lines = []\n",
    "    for page in document.pages:\n",
    "        for line in page.lines:\n",
    "            words = line.split()\n",
    "            if len(words) > 1 and all(try_parse_float(word) for word in words[-2:]):\n",
    "                extracted_lines.append(line)\n",
    "    return extracted_lines\n",
    "\n",
    "extracted_lines = refine_lines_with_descriptions_and_numbers(doc)\n",
    "for line in extracted_lines:\n",
    "    print(line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# saving output in a table\n",
    "def save_extracted_lines_to_table(extracted_lines):\n",
    "    data = []\n",
    "    for line in extracted_lines:\n",
    "        words = line.split()\n",
    "        numeric_words = [word for word in words if try_parse_float(word)]\n",
    "        non_numeric_words = [word for word in words if not try_parse_float(word)]\n",
    "        description = ' '.join(non_numeric_words)\n",
    "        row = [description] + numeric_words\n",
    "        data.append(row)\n",
    "\n",
    "    # Determine the maximum number of columns needed\n",
    "    max_columns = max(len(row) for row in data)\n",
    "\n",
    "    # Pad rows with '-' to ensure all rows have the same number of columns\n",
    "    for row in data:\n",
    "        row.extend('-' * (max_columns - len(row)))\n",
    "\n",
    "    # Create DataFrame\n",
    "    columns = ['Description'] + [f'Year {i}' for i in range(1, max_columns)]\n",
    "    df = pd.DataFrame(data, columns=columns)\n",
    "    return df\n",
    "\n",
    "df = save_extracted_lines_to_table(extracted_lines)\n",
    "df_parsed = df\n",
    "\n",
    "print(df)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. extracting text (descriptions) from pages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# when using PyMuPDF\n",
    "\n",
    "def extract_descriptions_pymupdf(pdf_path):\n",
    "    doc = fitz.open(pdf_path)  # Open the PDF file\n",
    "    pages_text = []  # Initialize a list to store text for each page\n",
    "\n",
    "    for page in doc:  # Iterate through each page\n",
    "        page_text = page.get_text()  # Extract text from the page\n",
    "        filtered_text = \"\"  # Initialize a string to store filtered text\n",
    "\n",
    "        # Split the extracted text into lines and filter out lines containing numeric values\n",
    "        lines = page_text.split('\\n')\n",
    "        for line in lines:\n",
    "            if not re.search(r'\\d', line):  # Check if the line contains digits\n",
    "                filtered_text += line + '\\n'  # Add the line to filtered text if no digits found\n",
    "\n",
    "        pages_text.append(filtered_text)  # Append the filtered text of the current page to the list\n",
    "\n",
    "    doc.close()  # Close the document\n",
    "    return pages_text  # Return a list of filtered text for each page\n",
    "\n",
    "\n",
    "\n",
    "pages_text = extract_descriptions_pymupdf(pdf_path)  # Call the function to extract and filter text\n",
    "for page_number, text in enumerate(pages_text, start=1):\n",
    "    print(f\"Page {page_number}:\\n{text}\\n---\\n\")  # Print each page's text\n",
    "    \n",
    "# TODO: if we can work with text in the future, add proper processing of the contents to create cohesive paragraphs    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## for OCR \n",
    "\n",
    "page_number = 1  # Initialize page number counter\n",
    "\n",
    "for page in text_description.pages:\n",
    "    filtered_lines = []  # Prepare a list to hold filtered lines\n",
    "\n",
    "    # Iterate through each line in the current page\n",
    "    for line in page.lines:\n",
    "        # Check if the line contains any digits\n",
    "        if not re.search(r'\\d', line):\n",
    "            filtered_lines.append(line)\n",
    "\n",
    "    # Join the filtered lines into a single string for output\n",
    "    filtered_text = '\\n'.join(filtered_lines)\n",
    "\n",
    "    # Output the page number and the filtered text\n",
    "    print(f\"Page {page_number}:\\n{filtered_text}\\n---\\n\")\n",
    "    \n",
    "    # Increment the page number for the next iteration\n",
    "    page_number += 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. getting company info"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# getting company info from file name\n",
    "\n",
    "#TODO: for final files (depending on the format), add also extraction of company numerical identifier\n",
    "\n",
    "#get the file name from path\n",
    "file_name = pdf_path.split(\"/\")[-1]\n",
    "#transform the file name to lower case\n",
    "file_name = file_name.lower()\n",
    "#remove the words pdf, Jaarverslag and Jaarrekening,... from the file name\n",
    "file_name = file_name.replace(\".pdf\", \"\").replace(\"jaarverslag\", \"\").replace(\"jaarrekening\", \"\").replace(\"geconsolideerd\", \"\").replace(\"geconsolideerde\", \"\").replace(\"annual\", \"\").replace(\"report\", \"\").replace(\"consolidated\", \"\").replace(\"financial\", \"\").replace(\"statement\", \"\").replace(\"statements\", \"\") .replace(\"jaarbericht\", \"\") \n",
    "\n",
    "# if there is a number in the file name, it is the year\n",
    "year = re.findall(r'\\d{4}', file_name)\n",
    "# company name is the file name without the year\n",
    "company_name = file_name.replace(year[0], \"\").replace(\"-\", \" \").strip()\n",
    "\n",
    "#from the first 1000 characters of the text, extract the currency\n",
    "\n",
    "# List of currencies and their symbols/abbreviations\n",
    "currency_list = [\n",
    "    '€', 'usd', 'dollar', 'eur', 'gbp', '£', 'jpy', '¥', 'cny', 'rmb', 'cad', '$', \n",
    "    'aud', 'nzd', 'chf', 'sek', 'nok', 'kr', 'rub', '₽', 'inr', '₹', 'brl', 'real', \n",
    "    'zar', 'rand', 'mxn', 'peso', 'sgd', 'hkd', 'twd', 'won', '₩', 'idr', 'rupiah',\n",
    "    'try', 'lira', 'pln', 'zł', 'ron', 'lei'\n",
    "]\n",
    "\n",
    "# Join the list to create the regex pattern\n",
    "pattern = '|'.join(re.escape(currency) for currency in currency_list)\n",
    "\n",
    "# Extract the currency from the first 1000 characters of the text\n",
    "currency = re.findall(pattern, text[:1000].lower())\n",
    "\n",
    "if currency:\n",
    "    currency = currency[0]\n",
    "else:\n",
    "    currency = \"EUR\"\n",
    "    \n",
    "#year is Year_1, Year_2 is the year -1, display the extracted company name and year 1 and year 2\n",
    "company_information_df = pd.DataFrame([[company_name, year[0], str(int(year[0])-1), currency ]], columns=['Company Name', 'Year 1', 'Year 2', 'Currency'])\n",
    "\n",
    "display(company_information_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Extracting, Categorising, and Processing data "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "# setup for replacing all outputs longer than XXX characters with a category name extracted by gemma\n",
    "def prepare_messages(text: str) -> list:\n",
    "    return [\n",
    "        # below change prompt if needed, this one worked best in testing\n",
    "        {\"role\": \"system\", \"content\": \"This is a fragment of description from a financial statement. Extract a financial category name from the description. Output only the category and no other text or explanations. If there is no category or not applicable, output only: -.\"},\n",
    "        {\"role\": \"user\", \"content\": text},\n",
    "        {\"role\": \"assistant\", \"content\": \"category name:\"}\n",
    "    ]\n",
    "\n",
    "def output_category_name(text: str) -> str:\n",
    "    messages = prepare_messages(text)\n",
    "    response = client.chat.completions.create(\n",
    "        # TODO: try on smaller models\n",
    "        model=\"gemma:7b\",\n",
    "        messages = messages,\n",
    "        temperature=0.8\n",
    "    )\n",
    "    category_name = response.choices[0].message.content\n",
    "     # convert to lowercase\n",
    "    category_name = category_name.lower()\n",
    "    # remove whitespaces\n",
    "    category_name = category_name.strip()\n",
    "    return category_name\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "MAX_DESCRIPTION_LENGTH = 220\n",
    "LLM_length=100\n",
    "\n",
    "#change the 'Year 1 EUR' and 'Year 2 EUR' columns in df_parsed to the values under Year 1 and Year 2 in company_information_df\n",
    "client = OpenAI(base_url = 'http://localhost:11434/v1',api_key='ollama')\n",
    "\n",
    "df_parsed = df_parsed.rename(columns={'Year 1': company_information_df['Year 1'][0], 'Year 2': company_information_df['Year 2'][0]})\n",
    "df_parsed['Category'] = df_parsed['Description']\n",
    "\n",
    "# extracring category names for descriptions longer than LLM_length, for very long documens might take a few minutes to run\n",
    "\n",
    "for i, row in enumerate(df_parsed['Description']):\n",
    "    if len(row) > LLM_length and len(row) < MAX_DESCRIPTION_LENGTH:\n",
    "        category_name = output_category_name(row)\n",
    "        category_name = category_name.split(\"\\n\")[0]\n",
    "        df_parsed.at[i, 'Category'] = category_name\n",
    "        print(row, \"->\", category_name)\n",
    "  \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Mapping data to ledger"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "from financial_ledger import financial_ledger as ledger\n",
    "import jellyfish as jf\n",
    "from jellyfish import jaro_winkler_similarity as jws\n",
    "from tqdm import tqdm"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "# the data is mapped to ledger (financial_ledger.py) in the following way: \n",
    "# first, the jarowinkler similarity is calculated between the description and each term in the ledger\n",
    "# next, the cosine similarity is calculated between the description and each term in the ledger\n",
    "# if both best match of jarowinkler and best match cosine  = \"not found\" or \"not applicable\", we delete the row since we cannot match it to any category\n",
    "# if there is no match in cosine similarity, replace it with the match from jarowinkler, so the default match is obtained from the cosine similarity calculation\n",
    "\n",
    "# functions for calculating the Jaro-Winkler similarity for word matching\n",
    "# it is a string metric measuring an edit distance between two sequences.\n",
    "# edit distance is measured by counting the minimum number of operations required to transform one string into the other\n",
    "\n",
    "def add_matching_info_to_df(df, ledger):\n",
    "    # Initialize lists to hold match results\n",
    "    best_matches = []\n",
    "    match_scores = []\n",
    "    \n",
    "    for index, row in df.iterrows():\n",
    "        description = row['Description'].lower()\n",
    "        max_score = 0\n",
    "        best_match = \"\"\n",
    "        for key, ledger_entry in ledger.items():\n",
    "            # Iterate through both 'English' and 'Dutch' lists if they exist\n",
    "            for lang in ['English', 'Dutch']:\n",
    "                if lang in ledger_entry:  # Check if the language key exists\n",
    "                    for term in ledger_entry[lang]:\n",
    "                        score = jf.jaro_winkler(description, term.lower())\n",
    "                        if score > max_score:\n",
    "                            max_score = score\n",
    "                            best_match = key\n",
    "        # Append match result or indicate no match found\n",
    "        if max_score > 0.7:\n",
    "            best_matches.append(best_match)\n",
    "            match_scores.append(max_score)\n",
    "        else:\n",
    "            best_matches.append(\"No match found\")\n",
    "            match_scores.append(max_score)\n",
    "    \n",
    "    # Add the match results to the DataFrame\n",
    "    df['Best Match_JW'] = best_matches\n",
    "    df['Match Score_JW'] = match_scores\n",
    "\n",
    "add_matching_info_to_df(df_parsed, ledger)\n",
    "\n",
    "# Now df_parsed contains two new columns: 'Best Match' and 'Match Score'\n",
    "#display(df_parsed)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "# NOTE: do not run unless you want to display all rows\n",
    "# this is used to display all rows in the dataframe when checking output, not necessary to run as it makes the output very long, only for testing/checking purposes\n",
    "\n",
    "pd.set_option('display.max_rows', None)  # Show all rows\n",
    "pd.set_option('display.max_columns', None)  # Show all columns\n",
    "pd.set_option('display.width', None)  # Auto-detect the display width\n",
    "pd.set_option('display.max_colwidth', -1)  # Display full width of columns\n",
    "\n",
    "#display(df_parsed)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "# cosine similarity\n",
    "\n",
    "from sklearn.feature_extraction.text import TfidfVectorizer\n",
    "from sklearn.metrics.pairwise import cosine_similarity\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "\n",
    "# Combine all texts to build the vocabulary for vectorization\n",
    "all_texts = list(df_parsed['Category']) + [item for sublist in ledger.values() for lang in sublist if lang in ['English', 'Dutch'] for item in sublist[lang]]\n",
    "\n",
    "vectorizer = TfidfVectorizer().fit(all_texts)\n",
    "description_vectors = vectorizer.transform(df_parsed['Category'])\n",
    "\n",
    "# Prepare and vectorize ledger entries\n",
    "ledger_entries = [item for sublist in ledger.values() for lang in sublist if lang in ['English', 'Dutch'] for item in sublist[lang]]\n",
    "ledger_vectors = vectorizer.transform(ledger_entries)\n",
    "\n",
    "# Calculate cosine similarity between description vectors and ledger vectors\n",
    "similarity_scores = cosine_similarity(description_vectors, ledger_vectors)\n",
    "\n",
    "# Determine the best match for each description based on the highest cosine similarity score\n",
    "best_matches = [ledger_entries[np.argmax(row)] if max(row) > 0.7 else \"No match found\" for row in similarity_scores]\n",
    "df_parsed['Best Match_Cosine'] = best_matches\n",
    "df_parsed['Highest Match Score_Cosine'] = [max(row) for row in similarity_scores]\n",
    "\n",
    "# for each value in `Best Match_Cosine` assign the corresponding higher level key from the ledger\n",
    "for i, row in enumerate(df_parsed['Best Match_Cosine']):\n",
    "    for key, value in ledger.items():\n",
    "        if row in value['English'] or row in value['Dutch']:\n",
    "            df_parsed.at[i, 'Best Match_Cosine'] = key\n",
    "            break\n",
    "\n",
    "#display(df_parsed)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Creating output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "# if in df_parsed, best match JR and best match cosine  = \"not found\" or \"not applicable\", delete the row\n",
    "#if there is no match in 'Best Match_Cosine', replace it with the match from 'Best Match_JW'\n",
    "\n",
    "df_parsed['Best Match_Cosine'] = np.where(df_parsed['Best Match_Cosine'] == \"No match found\", df_parsed['Best Match_JW'], df_parsed['Best Match_Cosine'])\n",
    "\n",
    "df_parsed = df_parsed[df_parsed['Best Match_Cosine'] != \"No match found\"]\n",
    "df_parsed = df_parsed[df_parsed['Best Match_Cosine'] != \"not applicable\"]\n",
    "\n",
    "# if categories are found, make a new df with catgegory name, year 1 value,  year 2 value, output info from the ledger like id and other categories based on best match\n",
    "df_output = df_parsed[['Category', company_information_df['Year 1'][0], company_information_df['Year 2'][0], 'Best Match_JW', 'Match Score_JW', 'Best Match_Cosine', 'Highest Match Score_Cosine']]\n",
    "\n",
    "# based on the best match in the ledger, make new columns with balance, id, category, and statement type from the ledger for each row in df_output\n",
    "def add_ledger_info_to_df(df, ledger):\n",
    "    # Initialize lists to hold match results\n",
    "    balance = []\n",
    "    id = []\n",
    "    category = []\n",
    "    statement_type = []\n",
    "    postencode= []\n",
    "     \n",
    "    for index, row in df.iterrows():\n",
    "        best_match = row['Best Match_Cosine']\n",
    "        for key, ledger_entry in ledger.items():\n",
    "            if key == best_match:\n",
    "                balance.append(ledger_entry['balance'])\n",
    "                id.append(ledger_entry['id'])\n",
    "                category.append(ledger_entry['category'])\n",
    "                statement_type.append(ledger_entry['statement_type'])\n",
    "                postencode.append(ledger_entry['postencode'])\n",
    "    \n",
    "    # Add the match results to the DataFrame\n",
    "    df_output['Balance'] = balance\n",
    "    df_output['ID'] = id\n",
    "    df_output['Category'] = category\n",
    "    df_output['Statement Type'] = statement_type\n",
    "    df_output['Postencode'] = postencode\n",
    "\n",
    "    \n",
    "add_ledger_info_to_df(df_output, ledger)\n",
    "# display(df_output)\n",
    "df_output['Postencode'] = df_output['Postencode'].astype('int64')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "# joining the output with kvk postencodes and information\n",
    "\n",
    "# import excel file from path: /Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/20231212_kvk_finposten_dictionary - Antwoorden.xlsx\n",
    "kvk_file_path = '/Users/juliamarkusiewicz/Documents/research assistant work code/PDF_Mining_Julia/20231212_kvk_finposten_dictionary - Antwoorden.xlsx'\n",
    "kvk_df = pd.read_excel(kvk_file_path, sheet_name='Postencodes')\n",
    "# leave only the first 5 columns\n",
    "kvk_df = kvk_df.iloc[:, :5]\n",
    "\n",
    "#add a row with postencode equal to 1 and all other columns equal to \"totals\"\n",
    "kvk_df.loc[-1] = [1, 'Total', 'Total', 'Total', 'Total']\n",
    "kvk_df.index = kvk_df.index + 1\n",
    "kvk_df = kvk_df.sort_index()\n",
    "\n",
    "#join output df with kvk_df on the postencode column\n",
    "df_output = df_output.merge(kvk_df, how='left', left_on='Postencode', right_on='Postencode')\n",
    "\n",
    "#display(df_output)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "#delete the Match Score_JW\tBest Match_Cosine, Highest Match Score_Cosine, Category, Balance, and ID columns \n",
    "df_output = df_output.drop(columns=['Match Score_JW', 'Best Match_JW', 'Highest Match Score_Cosine', 'Balance', 'ID', \"Statement Type\"])\n",
    "\n",
    "# display NaN as '-'\n",
    "df_output = df_output.fillna('-')\n",
    "\n",
    "# display the dataframe\n",
    "# display(df_output)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "metadata": {}
   },
   "outputs": [],
   "source": [
    "# Melting the DataFrame to have 'year' and 'value' columns\n",
    "\n",
    "df_melted = df_output.melt(id_vars=[\"Category\", \"Postencode\", \"Hoofd posten code:\", \"Balans/res.rek. indicatie:\", \"Debet/credit indicatie: \", \"Omschr. postencode: \"],\n",
    "                           value_vars=[\"2022\", \"2021\"],\n",
    "                           var_name=\"Year\", value_name=\"Value\")\n",
    "\n",
    "    \n",
    "# add company name and currency to the df\n",
    "df_melted['Company Name'] = company_information_df['Company Name'][0]\n",
    "df_melted['Currency'] = company_information_df['Currency'][0]\n",
    "display(df_melted)\n",
    "\n",
    "# save the dataframe to a csv file\n",
    "df_melted.to_csv('output.csv', index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Data Validation\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# checking if assets=liabilities+equity for each year\n",
    "\n",
    "# Clean the 'Value' column\n",
    "def clean_value(value):\n",
    "    if isinstance(value, str):\n",
    "        value = value.replace('(', '-').replace(')', '')  # Replace parentheses\n",
    "        value = value.replace(',', '')  # Remove commas\n",
    "    return value\n",
    "\n",
    "df_melted['Value'] = df_melted['Value'].apply(clean_value)  # Apply cleaning function\n",
    "df_melted['Value'] = pd.to_numeric(df_melted['Value'], errors='coerce')  # Convert to numeric\n",
    "\n",
    "df_checks = df_melted[df_melted['Postencode'] != 1]    \n",
    "\n",
    "# From Category of df_melted, sum all values for assets, liabilities, and equity for each year\n",
    "assets_y1 = df_checks[(df_checks['Category'] == 'assets') & (df_checks['Year'] == company_information_df['Year 1'][0])]['Value'].sum()\n",
    "liabilities_y1 = df_checks[(df_checks['Category'] == 'liabilities') & (df_checks['Year'] == company_information_df['Year 1'][0])]['Value'].sum()\n",
    "equity_y1 = df_checks[(df_checks['Category'] == 'equity') & (df_checks['Year'] == company_information_df['Year 1'][0])]['Value'].sum()\n",
    "\n",
    "# Check if the sum of assets is equal to the sum of liabilities and equity\n",
    "if assets_y1 == liabilities_y1 + equity_y1:\n",
    "    print(\"The balance sheet balances for\", company_information_df['Year 1'][0])\n",
    "else:\n",
    "    print(\"The balance sheet does not balance for\", company_information_df['Year 1'][0])\n",
    "\n",
    "# Print the values for assets, liabilities, and equity for year 1\n",
    "print(\"Assets:\", assets_y1)\n",
    "print(\"Liabilities:\", liabilities_y1)\n",
    "print(\"Equity:\", equity_y1)\n",
    "\n",
    "# Print sum of equity and liabilities\n",
    "print(\"Sum of liabilities and equity:\", liabilities_y1 + equity_y1)\n",
    "\n",
    "# Perform the same check for year 2\n",
    "assets_y2 = df_checks[(df_checks['Category'] == 'assets') & (df_checks['Year'] == company_information_df['Year 2'][0])]['Value'].sum()\n",
    "liabilities_y2 = df_checks[(df_checks['Category'] == 'liabilities') & (df_checks['Year'] == company_information_df['Year 2'][0])]['Value'].sum()\n",
    "equity_y2 = df_checks[(df_checks['Category'] == 'equity') & (df_checks['Year'] == company_information_df['Year 2'][0])]['Value'].sum()\n",
    "\n",
    "# Check if the sum of assets is equal to the sum of liabilities and equity\n",
    "if assets_y2 == liabilities_y2 + equity_y2:\n",
    "    print(\"The balance sheet balances for\", company_information_df['Year 2'][0])\n",
    "else:\n",
    "    print(\"The balance sheet does not balance for\", company_information_df['Year 2'][0])\n",
    "\n",
    "# Print the values for assets, liabilities, and equity for year 2\n",
    "print(\"Assets:\", assets_y2)\n",
    "print(\"Liabilities:\", liabilities_y2)\n",
    "print(\"Equity:\", equity_y2)\n",
    "print(\"Sum of liabilities and equity:\", liabilities_y2 + equity_y2)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# checking if totals for each category and year match the reported totals (not correct for most documents due to different notations of totals)\n",
    "\n",
    "# Strip any whitespace and convert 'Year' column to integer type\n",
    "df_melted['Year'] = df_melted['Year'].astype(str).str.strip().astype(int)\n",
    "\n",
    "# Extract years from company_information_df\n",
    "year_1 = int(company_information_df['Year 1'].iloc[0])\n",
    "year_2 = int(company_information_df['Year 2'].iloc[0])\n",
    "\n",
    "# Function to check totals for a specific category and year\n",
    "def check_totals(category, year):\n",
    "    # Calculate the sum of values for the given category and year excluding Postencode == 1\n",
    "    category_sum = df_melted[(df_melted['Category'] == category) & (df_melted['Year'] == year) & (df_melted['Postencode'] != 1)]['Value'].sum()\n",
    "    \n",
    "    # Calculate the sum of values for the given category and year where Postencode == 1\n",
    "    reported_total = df_melted[(df_melted['Category'] == category) & (df_melted['Year'] == year) & (df_melted['Postencode'] == 1)]['Value'].sum()\n",
    "    \n",
    "    # Display the values for comparison\n",
    "    print(f\"\\nChecking totals for {category} in {year}:\")\n",
    "    print(f\"Sum of all entries (excluding Postencode == 1): {category_sum}\")\n",
    "    print(f\"Total with Postencode == 1: {reported_total}\")\n",
    "    \n",
    "    if category_sum == reported_total:\n",
    "        print(f\"Result: The calculated sum matches the reported total for {category} in {year}.\")\n",
    "    else:\n",
    "        print(f\"Result: Discrepancy found for {category} in {year}.\")\n",
    "        print(f\"Difference: {reported_total - category_sum}\")\n",
    "\n",
    "# Example usage for the years in company_information_df\n",
    "check_totals('assets', year_1)\n",
    "check_totals('liabilities', year_1)\n",
    "check_totals('equity', year_1)\n",
    "\n",
    "check_totals('assets', year_2)\n",
    "check_totals('liabilities', year_2)\n",
    "check_totals('equity', year_2)\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}