Skip to content

Textract Handler API

auris_tools.textractHandler

TextractHandler

Handler for Amazon Textract operations to extract text from documents.

This class provides methods to interact with AWS Textract service for text extraction from documents stored in S3.

Source code in auris_tools/textractHandler.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
class TextractHandler:
    """
    Handler for Amazon Textract operations to extract text from documents.

    This class provides methods to interact with AWS Textract service for
    text extraction from documents stored in S3.
    """

    def __init__(self, config=None):
        """
        Initialize the Textract handler with AWS configuration.

        Args:
            config: An AWSConfiguration object, or None to use environment variables
        """
        if config is None:
            config = AWSConfiguration()

        # Create a boto3 session with the configuration
        session = boto3.session.Session(**config.get_boto3_session_args())

        # Create a Textract client with additional configuration if needed
        self.client = session.client('textract', **config.get_client_args())
        logging.info(f'Initialized Textract client in region {config.region}')

    def start_job(self, s3_bucket_name, object_name):
        """
        Start an asynchronous text detection job for a document in S3.

        Args:
            s3_bucket_name: Name of the S3 bucket containing the document
            object_name: Object key of the document in the S3 bucket

        Returns:
            str: The JobId of the started Textract job

        Raises:
            Exception: If there is an error starting the job
        """
        try:
            response = self.client.start_document_text_detection(
                DocumentLocation={
                    'S3Object': {'Bucket': s3_bucket_name, 'Name': object_name}
                }
            )
            job_id = response['JobId']
            logging.info(
                f'Started Textract job {job_id} for {s3_bucket_name}/{object_name}'
            )
            return job_id
        except Exception as e:
            logging.error(
                f'Error starting Textract job for {s3_bucket_name}/{object_name}: {str(e)}'
            )
            raise

    def get_job_status(self, job_id):
        """
        Get the status of a Textract job.

        Args:
            job_id: ID of the Textract job

        Returns:
            str: The job status (e.g., 'IN_PROGRESS', 'SUCCEEDED', 'FAILED')
        """
        try:
            response = self.client.get_document_text_detection(JobId=job_id)
            status = response['JobStatus']
            logging.info(f'Textract job {job_id} status: {status}')
            return status
        except Exception as e:
            logging.error(
                f'Error getting status for Textract job {job_id}: {str(e)}'
            )
            raise

    def is_job_complete(self, job_id):
        """
        Check if a Textract job has completed.

        Args:
            job_id: ID of the Textract job

        Returns:
            str: The job status
        """
        time.sleep(1)  # Avoid rate limiting
        return self.get_job_status(job_id)

    def get_job_results(self, job_id):
        """
        Get the results of a completed Textract job.

        This method handles pagination of results automatically.

        Args:
            job_id: ID of the Textract job

        Returns:
            list: List of response pages from Textract
        """
        pages = []
        next_token = None

        try:
            # Get first page
            response = self.client.get_document_text_detection(JobId=job_id)
            pages.append(response)
            logging.info(f'Received page 1 of results for job {job_id}')

            # Get next token if available
            if 'NextToken' in response:
                next_token = response['NextToken']

            # Get additional pages if available
            page_num = 2
            while next_token:
                time.sleep(1)  # Avoid rate limiting
                response = self.client.get_document_text_detection(
                    JobId=job_id, NextToken=next_token
                )
                pages.append(response)
                logging.info(
                    f'Received page {page_num} of results for job {job_id}'
                )
                page_num += 1

                next_token = response.get('NextToken')

            return pages
        except Exception as e:
            logging.error(
                f'Error getting results for Textract job {job_id}: {str(e)}'
            )
            raise

    def get_full_text(self, response):
        """
        Extract the full text from Textract response pages.

        Args:
            response: List of response pages from Textract

        Returns:
            str: The full extracted text as a string
        """
        try:
            text_lines = []
            for result_page in response:
                for item in result_page.get('Blocks', []):
                    if item.get('BlockType') == 'LINE':
                        text_lines.append(item.get('Text', ''))

            full_text = ' '.join(text_lines)
            return full_text
        except Exception as e:
            logging.error(
                f'Error extracting full text from Textract response: {str(e)}'
            )
            return ''
__init__ 
__init__(config = None)

Initialize the Textract handler with AWS configuration.

Parameters:

Name Type Description Default
config

An AWSConfiguration object, or None to use environment variables

 None
Source code in auris_tools/textractHandler.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def __init__(self, config=None):
    """
    Initialize the Textract handler with AWS configuration.

    Args:
        config: An AWSConfiguration object, or None to use environment variables
    """
    if config is None:
        config = AWSConfiguration()

    # Create a boto3 session with the configuration
    session = boto3.session.Session(**config.get_boto3_session_args())

    # Create a Textract client with additional configuration if needed
    self.client = session.client('textract', **config.get_client_args())
    logging.info(f'Initialized Textract client in region {config.region}')
get_full_text 
get_full_text(response)

Extract the full text from Textract response pages.

Parameters:

Name Type Description Default
response

List of response pages from Textract

 required

Returns:

Name Type Description
str

The full extracted text as a string
Source code in auris_tools/textractHandler.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def get_full_text(self, response):
    """
    Extract the full text from Textract response pages.

    Args:
        response: List of response pages from Textract

    Returns:
        str: The full extracted text as a string
    """
    try:
        text_lines = []
        for result_page in response:
            for item in result_page.get('Blocks', []):
                if item.get('BlockType') == 'LINE':
                    text_lines.append(item.get('Text', ''))

        full_text = ' '.join(text_lines)
        return full_text
    except Exception as e:
        logging.error(
            f'Error extracting full text from Textract response: {str(e)}'
        )
        return ''
get_job_results 
get_job_results(job_id)

Get the results of a completed Textract job.

This method handles pagination of results automatically.

Parameters:

Name Type Description Default
job_id

ID of the Textract job

 required

Returns:

Name Type Description
list

List of response pages from Textract
Source code in auris_tools/textractHandler.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def get_job_results(self, job_id):
    """
    Get the results of a completed Textract job.

    This method handles pagination of results automatically.

    Args:
        job_id: ID of the Textract job

    Returns:
        list: List of response pages from Textract
    """
    pages = []
    next_token = None

    try:
        # Get first page
        response = self.client.get_document_text_detection(JobId=job_id)
        pages.append(response)
        logging.info(f'Received page 1 of results for job {job_id}')

        # Get next token if available
        if 'NextToken' in response:
            next_token = response['NextToken']

        # Get additional pages if available
        page_num = 2
        while next_token:
            time.sleep(1)  # Avoid rate limiting
            response = self.client.get_document_text_detection(
                JobId=job_id, NextToken=next_token
            )
            pages.append(response)
            logging.info(
                f'Received page {page_num} of results for job {job_id}'
            )
            page_num += 1

            next_token = response.get('NextToken')

        return pages
    except Exception as e:
        logging.error(
            f'Error getting results for Textract job {job_id}: {str(e)}'
        )
        raise
get_job_status 
get_job_status(job_id)

Get the status of a Textract job.

Parameters:

Name Type Description Default
job_id

ID of the Textract job

 required

Returns:

Name Type Description
str

The job status (e.g., 'IN_PROGRESS', 'SUCCEEDED', 'FAILED')
Source code in auris_tools/textractHandler.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def get_job_status(self, job_id):
    """
    Get the status of a Textract job.

    Args:
        job_id: ID of the Textract job

    Returns:
        str: The job status (e.g., 'IN_PROGRESS', 'SUCCEEDED', 'FAILED')
    """
    try:
        response = self.client.get_document_text_detection(JobId=job_id)
        status = response['JobStatus']
        logging.info(f'Textract job {job_id} status: {status}')
        return status
    except Exception as e:
        logging.error(
            f'Error getting status for Textract job {job_id}: {str(e)}'
        )
        raise
is_job_complete 
is_job_complete(job_id)

Check if a Textract job has completed.

Parameters:

Name Type Description Default
job_id

ID of the Textract job

 required

Returns:

Name Type Description
str

The job status
Source code in auris_tools/textractHandler.py 
86
87
88
89
90
91
92
93
94
95
96
97
def is_job_complete(self, job_id):
    """
    Check if a Textract job has completed.

    Args:
        job_id: ID of the Textract job

    Returns:
        str: The job status
    """
    time.sleep(1)  # Avoid rate limiting
    return self.get_job_status(job_id)
start_job 
start_job(s3_bucket_name, object_name)

Start an asynchronous text detection job for a document in S3.

Parameters:

Name Type Description Default
s3_bucket_name

Name of the S3 bucket containing the document

 required
object_name

Object key of the document in the S3 bucket

 required

Returns:

Name Type Description
str

The JobId of the started Textract job

Raises:

Type Description
Exception

If there is an error starting the job
Source code in auris_tools/textractHandler.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def start_job(self, s3_bucket_name, object_name):
    """
    Start an asynchronous text detection job for a document in S3.

    Args:
        s3_bucket_name: Name of the S3 bucket containing the document
        object_name: Object key of the document in the S3 bucket

    Returns:
        str: The JobId of the started Textract job

    Raises:
        Exception: If there is an error starting the job
    """
    try:
        response = self.client.start_document_text_detection(
            DocumentLocation={
                'S3Object': {'Bucket': s3_bucket_name, 'Name': object_name}
            }
        )
        job_id = response['JobId']
        logging.info(
            f'Started Textract job {job_id} for {s3_bucket_name}/{object_name}'
        )
        return job_id
    except Exception as e:
        logging.error(
            f'Error starting Textract job for {s3_bucket_name}/{object_name}: {str(e)}'
        )
        raise