non-cdk-version/convertTextractOutIntoTextHelper.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip3 install boto3 boto3_helpers pynamodb"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "import boto3\n",
    "import json\n",
    "import time\n",
    "import threading\n",
    "from botocore.config import Config\n",
    "from pynamodb.models import Model\n",
    "from pynamodb.attributes import UnicodeAttribute, NumberAttribute"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "logger = logging.getLogger(__name__)\n",
    "config = Config(retries = {'max_attempts': 10,'mode': 'adaptive'})\n",
    "\n",
    "dynamoDBMaxlistCount = 200 #  Max number of rows to pull at a time from DynamoDB and store in memory\n",
    "threadCountforTextExtract = 50 # Number of threads used to call Textract\n",
    "\n",
    "_tracking_table = \"s3ObjectNamesforTextract\" # name of DynamoDB table used to track objects that have been sent to Textract\n",
    "_textractFolder = \"textract_output/\" # name of folder that Textract created when it wrote results out to S3\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"DynamoDB Class\"\"\"\n",
    "class DocumentObjStatusModel(Model):\n",
    "\n",
    "    class Meta:\n",
    "        table_name = _tracking_table\n",
    "        region = boto3.Session().region_name\n",
    "\n",
    "    objectName = UnicodeAttribute(hash_key=True)\n",
    "    bucketName = UnicodeAttribute(null=True)\n",
    "    createDate = NumberAttribute(null=True)\n",
    "    txJobId = UnicodeAttribute(null=True)\n",
    "    outputbucketName = UnicodeAttribute(null=True)\n",
    "    outputTextObjName = UnicodeAttribute(null=True, default=\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"S3 Class for retrieving Object Names\"\"\"\n",
    "class S3Profile:\n",
    "    def __init__(self, bucketName, prefixName):\n",
    "        self.bucketName = bucketName\n",
    "        self.prefixName = prefixName\n",
    "        self.s3_client = boto3.client('s3')\n",
    "        self.profile_object_list = []\n",
    "        paginator = self.s3_client.get_paginator('list_objects_v2')\n",
    "        pages = paginator.paginate(Bucket=self.bucketName, Prefix=self.prefixName)\n",
    "        for page in pages:\n",
    "            for content in page.get('Contents'):\n",
    "                file = content.get('Key')\n",
    "                self.profile_object_list.append({\"fileObj\": file})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_s3_client():\n",
    "    return boto3.client('s3', config=config)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extractText(objRow, s3):\n",
    "    \n",
    "    outPutFolder = _textractFolder + objRow.txJobId\n",
    "    outPutTextObjName = outPutFolder\n",
    "    objRow.outputTextObjName = outPutTextObjName\n",
    "\n",
    "    txJSONoutputFiles = []\n",
    "    s3_profile = S3Profile(objRow.outputbucketName , objRow.outputTextObjName)\n",
    "    try:\n",
    "        for item in s3_profile.profile_object_list:\n",
    "            if item['fileObj'].split(\"/\")[-1].isnumeric():\n",
    "                txJSONoutputFiles.append(item['fileObj'])\n",
    "\n",
    "        txJSONoutputFiles.sort()\n",
    "        fileText = \"\"\n",
    "        for file in txJSONoutputFiles:\n",
    "            jsonContent = json.loads(s3.get_object(Bucket=objRow.outputbucketName, Key=file)['Body'].read().decode('utf-8'))\n",
    "            for block in jsonContent[\"Blocks\"]:\n",
    "                if block[\"BlockType\"] == \"LINE\":\n",
    "                    fileText += block[\"Text\"] + \"\\n\"    \n",
    "\n",
    "        objRow.outputTextObjName = objRow.outputTextObjName + \"/\" + objRow.objectName + \".txt\"\n",
    "        #print(objRow.outputTextObjName)\n",
    "    except Exception as e:\n",
    "        logger.error(e + \"Unable to parse text from JSON \" + objRow.objectName)\n",
    "        print (\"Unable to parse text from JSON \" + objRow.objectName)      \n",
    "        objRow.outputTextObjName = \"-1\"\n",
    "\n",
    "\n",
    "    try:\n",
    "        # write fileText content out to S3\n",
    "        s3.put_object(\n",
    "            Body=fileText, \n",
    "            Bucket=objRow.outputbucketName, \n",
    "            Key=objRow.outputTextObjName\n",
    "        )\n",
    "    except Exception as e:\n",
    "        logger.error(e)\n",
    "        print(\"Unable to write to S3\")\n",
    "\n",
    "    \n",
    "    try:\n",
    "        # update DynamoDB table with text file info\n",
    "        objRow.save()   \n",
    "    except Exception as e:\n",
    "        logger.error(e)\n",
    "        print(\"Unable to update DynamoDB table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def orchestrateTextExtraction():\n",
    "    print(\"Convert JSON from Textract into blobs of Text and save to S3\")\n",
    "    totalDocs = 0\n",
    "    try:\n",
    "        while True:\n",
    "            txtThreads = []\n",
    "            threadCounter = 0\n",
    "            objRows = DocumentObjStatusModel.scan(DocumentObjStatusModel.outputTextObjName.does_not_exist(), limit=dynamoDBMaxlistCount)\n",
    "            exitLoop = True\n",
    "            for objRow in objRows:\n",
    "                exitLoop = False\n",
    "                if (len(objRow.outputTextObjName)==0 and len(objRow.txJobId) > 2):\n",
    "                    txtThreads.append(threading.Thread(name=\"Thread - \" + str(threadCounter), target=extractText, args=(objRow, create_s3_client(),)))\n",
    "                    threadCounter+=1\n",
    "                    totalDocs+=1\n",
    "                    if threadCounter == threadCountforTextExtract:\n",
    "                        for thread in txtThreads:\n",
    "                            thread.start()\n",
    "                        for thread in txtThreads:\n",
    "                            thread.join()\n",
    "                        threadCounter = 0\n",
    "                        txtThreads.clear()\n",
    "\n",
    "            if len(txtThreads) > 0:\n",
    "                for thread in txtThreads:\n",
    "                    thread.start()\n",
    "                for thread in txtThreads:\n",
    "                    thread.join()\n",
    "            \n",
    "            if exitLoop: \n",
    "                print (f\"Total text documents created {totalDocs}\")\n",
    "                break\n",
    "\n",
    "    except Exception as e:\n",
    "        logger.error(e)\n",
    "        print (\"Unable to run script\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "now = time.perf_counter()\n",
    "print(\"started\")   \n",
    "    \n",
    "orchestrateTextExtraction()\n",
    "    \n",
    "print(f\"completed in - {time.perf_counter()-now} seconds\")"
   ]
  }
 ],
 "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.10.1"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}