FilesExpand file tree

 master

/

lambda_textract_detect_text.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

82 lines (68 loc) · 2.25 KB

 master

/

lambda_textract_detect_text.py

Copy path

Top

File metadata and controls

• Code
• Blame

82 lines (68 loc) · 2.25 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

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

"""

-*- coding: utf-8 -*-

========================

AWS Lambda

========================

Contributor: Chirag Rathod (Srce Cde)

========================

"""

import sys

import traceback

import logging

import json

import uuid

import boto3

from urllib.parse import unquote_plus

logger = logging.getLogger()

logger.setLevel(logging.INFO)

def process_error() -> dict:

ex_type, ex_value, ex_traceback = sys.exc_info()

traceback_string = traceback.format_exception(ex_type, ex_value, ex_traceback)

error_msg = json.dumps(

{

"errorType": ex_type.__name__,

"errorMessage": str(ex_value),

"stackTrace": traceback_string,

}

)

return error_msg

def extract_text(response: dict, extract_by="LINE") -> list:

text = []

for block in response["Blocks"]:

if block["BlockType"] == extract_by:

text.append(block["Text"])

return text

def lambda_handler(event, context):

textract = boto3.client("textract")

s3 = boto3.client("s3")

try:

if "Records" in event:

file_obj = event["Records"][0]

bucketname = str(file_obj["s3"]["bucket"]["name"])

filename = unquote_plus(str(file_obj["s3"]["object"]["key"]))

logging.info(f"Bucket: {bucketname} ::: Key: {filename}")

response = textract.detect_document_text(

Document={

"S3Object": {

"Bucket": bucketname,

"Name": filename,

}

}

)

logging.info(json.dumps(response))

# change LINE by WORD if you want word level extraction

raw_text = extract_text(response, extract_by="LINE")

logging.info(raw_text)

s3.put_object(

Bucket=bucketname,

Key=f"output/{filename.split('/')[-1]}_{uuid.uuid4().hex}.txt",

Body=str("\n".join(raw_text)),

)

return {

"statusCode": 200,

"body": json.dumps("Document processed successfully!"),

}

except:

error_msg = process_error()

logger.error(error_msg)

return {"statusCode": 500, "body": json.dumps("Error processing the document!")}