Table_extracter_cv2_mistral_final.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "source": [
        "!pip install opencv-python-headless\n",
        "!pip install pytesseract\n",
        "!pip install Pillow\n",
        "!pip install pdf2image\n",
        "!pip install langchain"
      ],
      "metadata": {
        "id": "8SpFUpYjeGbS"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!sudo apt-get install poppler-utils\n",
        "!sudo apt-get install tesseract-ocr"
      ],
      "metadata": {
        "id": "Qo6omoFzel98"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import cv2\n",
        "import numpy as np\n",
        "import pytesseract\n",
        "from pdf2image import convert_from_path\n",
        "from PIL import Image\n",
        "import os\n",
        "import logging\n",
        "import re\n",
        "from langchain import HuggingFaceHub, PromptTemplate, LLMChain\n",
        "# from langchain.document_loaders import UnstructuredPDFLoader\n",
        "import re\n",
        "import json\n",
        "from typing import List,Tuple,Any,Union,Dict\n",
        "logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')\n",
        "\n",
        "class TableExtractor:\n",
        "\n",
        "    def __init__(self, pdf_path):\n",
        "        self.huggingfacehub_api_token = \"YOUR_HUGGINGFACE_API_KEY\"  # Replace with your Hugging Face token\n",
        "        self.repo_id = \"mistralai/Mistral-7B-Instruct-v0.1\"\n",
        "        self.llm = HuggingFaceHub(\n",
        "            huggingfacehub_api_token=self.huggingfacehub_api_token,\n",
        "            repo_id=self.repo_id,\n",
        "            model_kwargs={\"temperature\": 0.1, \"max_new_tokens\":3000}\n",
        "        )\n",
        "        self.pdf_path = pdf_path\n",
        "\n",
        "    def _image_list_(self, pdf_path: str) -> List[str]:\n",
        "        \"\"\"\n",
        "        Converts all pages in a PDF file to images, saving them locally and returning a list of image filenames.\n",
        "\n",
        "        Parameters:\n",
        "        - pdf_path (str): The file path of the PDF document to be converted.\n",
        "\n",
        "        Returns:\n",
        "        - List[str]: A list of filenames for the images created, one per page of the PDF.\n",
        "\n",
        "        Raises:\n",
        "        - Exception: Propagates any exception that occurs during the PDF to image conversion process,\n",
        "                    after logging the error.\n",
        "        \"\"\"\n",
        "\n",
        "        try:\n",
        "            images = convert_from_path(self.pdf_path)\n",
        "            img_list = []\n",
        "            for i, image in enumerate(images):\n",
        "                image_name = f'page_{i}.jpg'\n",
        "                image.save(image_name, 'JPEG')\n",
        "                img_list.append(image_name)\n",
        "            return img_list\n",
        "        except Exception as e:\n",
        "            logging.error(f\"Error converting PDF to images: {e}\")\n",
        "            raise\n",
        "\n",
        "    def _preprocess_image_(self, image_path: str) -> Any:\n",
        "      \"\"\"\n",
        "      Preprocesses an image to enhance table detection and OCR accuracy by converting it to grayscale,\n",
        "      applying noise reduction, and performing thresholding to obtain a binary image.\n",
        "\n",
        "      Parameters:\n",
        "      - image_path (str): The file path of the image to preprocess.\n",
        "\n",
        "      Returns:\n",
        "      - Any: The preprocessed image in a binary format suitable for further processing. The actual type\n",
        "            is dependent on the OpenCV version used, but it generally corresponds to a numpy array.\n",
        "\n",
        "      Raises:\n",
        "      - FileNotFoundError: If the specified image file does not exist.\n",
        "      - Exception: For issues related to reading the image or preprocessing steps.\n",
        "      \"\"\"\n",
        "      try:\n",
        "        img = cv2.imread(image_path)\n",
        "        # Convert to grayscale\n",
        "        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)\n",
        "        # Noise removal\n",
        "        denoised = cv2.fastNlMeansDenoising(gray, None, 30, 7, 21)\n",
        "        # Thresholding to get a binary image\n",
        "        _, thresh = cv2.threshold(denoised, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)\n",
        "        return thresh\n",
        "\n",
        "      except Exception as e:\n",
        "          logging.error(\"Error during the preprocessing of the image\", exc_info=True)\n",
        "          raise\n",
        "\n",
        "\n",
        "    def _detect_tables_(self, image: Any) -> List[Tuple[int, int, int, int]]:\n",
        "      \"\"\"\n",
        "      Detects tables in an image using morphological transformations for line detection\n",
        "      and contour detection to identify table boundaries.\n",
        "\n",
        "      Parameters:\n",
        "      - image (Any): The preprocessed binary image where tables are to be detected. The type is\n",
        "                    typically a NumPy array, though it is annotated as `Any` to accommodate for\n",
        "                    flexibility in input image types.\n",
        "\n",
        "      Returns:\n",
        "      - List[Tuple[int, int, int, int]]: A list of tuples, each representing the bounding box of a detected\n",
        "                                        table in the format (x, y, width, height).\n",
        "\n",
        "      Note:\n",
        "      This method assumes the input image is preprocessed, ideally binary, to highlight table structures.\n",
        "      \"\"\"\n",
        "      try:\n",
        "        # Use morphological transformations to detect lines\n",
        "        vertical_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (1, int(image.shape[0] / 30)))\n",
        "        horiz_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (int(image.shape[1] / 30), 1))\n",
        "        vertical_lines = cv2.morphologyEx(image, cv2.MORPH_OPEN, vertical_kernel, iterations=2)\n",
        "        horizontal_lines = cv2.morphologyEx(image, cv2.MORPH_OPEN, horiz_kernel, iterations=2)\n",
        "\n",
        "        # Combine lines\n",
        "        table_grid = cv2.add(horizontal_lines, vertical_lines)\n",
        "        # Find contours\n",
        "        contours, _ = cv2.findContours(table_grid, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)\n",
        "\n",
        "        tables = []\n",
        "        for contour in contours:\n",
        "            x, y, w, h = cv2.boundingRect(contour)\n",
        "            if w * h > image.size * 0.001:  # Filter out small contours\n",
        "                tables.append((x, y, w, h))\n",
        "\n",
        "        logging.info(f\"Detected {len(tables)} tables in the image.\")\n",
        "        return tables\n",
        "\n",
        "      except Exception as e:\n",
        "        logging.error(\"Error during table detection\", exc_info=True)\n",
        "        raise\n",
        "\n",
        "\n",
        "    def _extract_text_from_tables_(self, image: Any, tables: List[Tuple[int, int, int, int]]) -> List[str]:\n",
        "      \"\"\"\n",
        "      Extracts text from specified table regions in an image using OCR.\n",
        "\n",
        "      Parameters:\n",
        "      - image (Any): The image from which text is to be extracted. The type is typically a NumPy array,\n",
        "                    though it is annotated as `Any` to accommodate for flexibility in input image types.\n",
        "      - tables (List[Tuple[int, int, int, int]]): A list of tuples, each representing the bounding box\n",
        "                                                  of a table to extract text from, in the format\n",
        "                                                  (x, y, width, height).\n",
        "\n",
        "      Returns:\n",
        "      - List[str]: A list of strings, where each string contains the text extracted from the corresponding\n",
        "                  table region defined in the `tables` parameter.\n",
        "\n",
        "      Raises:\n",
        "      - Exception: For issues during the image cropping or OCR process.\n",
        "      \"\"\"\n",
        "      try:\n",
        "        texts = []\n",
        "        for (x, y, w, h) in tables:\n",
        "            table_image = image[y:y+h, x:x+w]\n",
        "            text = pytesseract.image_to_string(table_image, lang='eng')\n",
        "            texts.append(text)\n",
        "        logging.info(f\"Extracted text from {len(tables)} tables.\")\n",
        "        return texts\n",
        "      except Exception as e:\n",
        "        logging.error(\"Error extracting text from tables\", exc_info=True)\n",
        "        raise\n",
        "\n",
        "\n",
        "    def extract_tables_and_text(self) -> List[str]:\n",
        "      \"\"\"\n",
        "      Extracts tables and their respective text from the document specified by `self.pdf_path`.\n",
        "\n",
        "      This method integrates the workflow of converting PDF pages to images, preprocessing images for table\n",
        "      detection, detecting table boundaries, and extracting text from these tables.\n",
        "\n",
        "      Returns:\n",
        "      - List[str]: A list of strings, each string contains the text extracted from a table detected in the\n",
        "                  document. The list is compiled from all tables detected across all pages of the document.\n",
        "\n",
        "      Raises:\n",
        "      - Exception: For any issues encountered during the processes of image conversion, preprocessing,\n",
        "                  table detection, or text extraction.\n",
        "      \"\"\"\n",
        "      try:\n",
        "        logging.info(\"Starting table and text extraction process.\")\n",
        "        # Convert all pages of the PDF to images and store the paths in `images`.\n",
        "        images = self._image_list_(self.pdf_path)\n",
        "\n",
        "        # Initialize an empty list to hold all extracted texts from tables.\n",
        "        all_tables_text = []\n",
        "\n",
        "        # Iterate through each image path in the list of images.\n",
        "        for image_path in images:\n",
        "            preprocessed_image = self._preprocess_image_(image_path)\n",
        "            tables = self._detect_tables_(preprocessed_image)\n",
        "            texts = self._extract_text_from_tables_(preprocessed_image, tables)\n",
        "            all_tables_text.extend(texts)\n",
        "\n",
        "        logging.info(\"Completed table and text extraction process.\")\n",
        "        # Return the list of extracted texts from all tables.\n",
        "        return all_tables_text\n",
        "      except Exception as e:\n",
        "        logging.error(\"Error in extracting tables and text\", exc_info=True)\n",
        "        raise\n",
        "\n",
        "    def extracted_data(self) -> List[str]:\n",
        "      \"\"\"\n",
        "      Cleans and returns the extracted text data from tables in the document.\n",
        "\n",
        "      This method calls `extract_tables_and_text` to get the raw text from tables,\n",
        "      then cleans the text by normalizing spaces and removing excessive newlines.\n",
        "\n",
        "      Returns:\n",
        "          List[str]: A list of cleaned strings, each representing the text extracted\n",
        "                    and cleaned from a single table detected in the document.\n",
        "      \"\"\"\n",
        "      try:\n",
        "        # Log the start of the data extraction process\n",
        "        logging.info(\"Starting extracted data processing.\")\n",
        "\n",
        "        # Extract raw tables text\n",
        "        tables_text = self.extract_tables_and_text()\n",
        "\n",
        "        # Initialize an empty list to hold cleaned text data\n",
        "        answer=[]\n",
        "\n",
        "        # Iterate through each raw text extracted from tables\n",
        "        for text in tables_text:\n",
        "          # Replace multiple spaces or tabs with a single space\n",
        "          cleaned_string = re.sub(r'[ \\t]+', ' ', text)\n",
        "          cleaned_string = re.sub(r'\\n\\s*\\n', '', cleaned_string)\n",
        "          answer.append(cleaned_string)\n",
        "        logging.info(\"Completed data extraction and cleaning.\")\n",
        "        return answer\n",
        "\n",
        "      except Exception as e:\n",
        "          logging.error(\"Error in extracting data\", exc_info=True)\n",
        "          raise\n",
        "\n",
        "    def response(self, content: str) -> str:\n",
        "      \"\"\"\n",
        "      Processes the given content by formatting it into a key-value pair JSON-like structure using an AI assistant.\n",
        "\n",
        "      Args:\n",
        "          content (str): The input data that needs to be analyzed and formatted.\n",
        "\n",
        "      Returns:\n",
        "          str: The cleaned and formatted result as a JSON-like string, where keys without values are set to an empty string.\n",
        "      \"\"\"\n",
        "\n",
        "      try:\n",
        "\n",
        "        # Define the template for processing the input content\n",
        "        template = \"\"\"[INST]you are json formatter.your task analyze the given data{data} and must return answer as json.key doest have value return empty string.only generate json for given data's.all answers should be in  json format(for all data).[/INST]\"\"\"\n",
        "        # Assuming PromptTemplate and LLMChain are correctly defined and imported\n",
        "        prompt = PromptTemplate(template=template, input_variables=[\"data\"])\n",
        "        llm_chain = LLMChain(prompt=prompt, verbose=True, llm=self.llm)\n",
        "        # Run the language model chain with the provided content\n",
        "        result = llm_chain.run({\"data\":content})\n",
        "        # # Clean the result by removing the pattern specified\n",
        "        # pattern = r\"\\*[^*]*\\*\"\n",
        "        # cleaned_text = re.sub(pattern, '', result)\n",
        "        # # Log the completion of the cleaning process\n",
        "        # logging.info(\"Completed processing and cleaning the response.\")\n",
        "        # print(\"result\",result)\n",
        "        return result\n",
        "\n",
        "      except Exception as e:\n",
        "          logging.error(\"Error in response\", exc_info=True)\n",
        "          raise\n",
        "\n",
        "    def list_of_answer(self) -> List[str]:\n",
        "      \"\"\"\n",
        "      Processes extracted data to generate a list of answers after further cleaning and formatting.\n",
        "\n",
        "      This method iterates over the data extracted by `extracted_data`, processes each item using `response`,\n",
        "      and compiles the results into a final list.\n",
        "\n",
        "      Returns:\n",
        "          List[str]: A list of strings, each a processed and cleaned response based on the extracted data.\n",
        "      \"\"\"\n",
        "      try:\n",
        "        # Retrieve extracted data\n",
        "        answer=self.extracted_data()\n",
        "        # Initialize an empty list to hold the final processed results\n",
        "        final=[]\n",
        "        # Iterate over each item in the extracted data\n",
        "        for i in range(len(answer)):\n",
        "          result=self.response(answer[i])\n",
        "          final.append(result)\n",
        "        logging.info(\"Completed processing list of answers.\")\n",
        "        return final\n",
        "\n",
        "      except Exception as e:\n",
        "          logging.error(\"Error in list of answer\", exc_info=True)\n",
        "          raise\n",
        "\n",
        "    def extract_and_combine_json(self,text_list: List[str]) -> List[Dict[str, Any]]:\n",
        "      \"\"\"\n",
        "      Extracts JSON objects from a list of strings and combines them into a single list.\n",
        "\n",
        "      Each string in the input list is searched for JSON objects enclosed within ```json ... ``` markers.\n",
        "      All found JSON objects are parsed and combined into a list of dictionaries.\n",
        "\n",
        "      Args:\n",
        "          text_list: A list of strings, each potentially containing one or more JSON objects.\n",
        "\n",
        "      Returns:\n",
        "          A list of dictionaries, where each dictionary is a parsed JSON object found in the input text.\n",
        "\n",
        "      Note:\n",
        "          This function uses a specific pattern to identify JSON blocks within the text, which are enclosed in\n",
        "          triple backticks followed by 'json' keyword and assumes well-formed JSON objects.\n",
        "      \"\"\"\n",
        "      try:\n",
        "        # This pattern matches your JSON blocks specifically formatted in your example\n",
        "        pattern = r'```json\\n({.*?})\\n```'\n",
        "        combined_json_objects = []  # This will hold all your parsed JSON objects\n",
        "\n",
        "        for text in text_list:\n",
        "            # Find all JSON strings within the text\n",
        "            json_strings = re.findall(pattern, text, re.DOTALL)\n",
        "            for json_str in json_strings:\n",
        "                try:\n",
        "                    # Parse the JSON string and append the resulting object to your list\n",
        "                    json_obj = json.loads(json_str)\n",
        "                    combined_json_objects.append(json_obj)\n",
        "                except json.JSONDecodeError as e:\n",
        "                    print(f\"Failed to decode JSON: {e}\")\n",
        "        return combined_json_objects\n",
        "\n",
        "      except Exception as e:\n",
        "        print(f\"Error in extract_and_combine_json {e}\")\n",
        "\n",
        "    def key_value_pair(self) -> str:\n",
        "      \"\"\"\n",
        "      Extracts JSON objects from a list of text blocks, combines them, and returns the combined JSON as a string.\n",
        "\n",
        "      This method calls `list_of_answer` to retrieve a list of text blocks, each potentially containing JSON objects.\n",
        "      These blocks are then processed by `extract_and_combine_json` to extract and combine all JSON objects into a single structure.\n",
        "      Finally, it converts this structure into a nicely formatted JSON string.\n",
        "\n",
        "      Returns:\n",
        "          A string representation of the combined JSON objects, formatted with an indent of 2 spaces.\n",
        "      \"\"\"\n",
        "      try:\n",
        "        # Retrieve the list of text blocks that may contain JSON objects\n",
        "        list_of_text=self.list_of_answer()\n",
        "        # Extract and combine JSON objects from the text blocks\n",
        "        combined_json=self.extract_and_combine_json(list_of_text)\n",
        "        # Convert the combined JSON objects into a formatted string\n",
        "        key_value=json.dumps(combined_json, indent=2)\n",
        "        logging.info(\"Successfully combined JSON objects.\")\n",
        "        return key_value\n",
        "\n",
        "      except Exception as e:\n",
        "          logging.error(f\"An error occurred in key_value_pair: {e}\")\n",
        "          raise"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "X695svACeG7u",
        "outputId": "c03afda1-3ff5-4fbe-b1d2-f9975ccfc61c"
      },
      "execution_count": 11,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Overwriting utils.py\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "if __name__==\"__main__\":\n",
        "  pdf_path=\"YOUR_PDF_PATH\"\n",
        "  table=TableExtractor(pdf_path)\n",
        "  result=table.key_value_pair()\n",
        "  print(result)"
      ],
      "metadata": {
        "id": "BGv8WZpEePtC"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}