Mercurial > public > tweet-analysis

comparison dependencies/python/aws_controller.py @ 8:6541622b6127

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
add tweet analysis method
author Dennis Concepcion Martin <dennisconcepcionmartin@gmail.com>
date Fri, 17 Sep 2021 21:10:02 +0200
parents
children
comparison
equal deleted inserted replaced
7:1b1296559c31 8:6541622b6127
1 import boto3
2 import base64
3 import json
4 from botocore.exceptions import ClientError
5
6
7 class AwsSecretsManager:
8 @staticmethod
9 def get_secret(secret_name, region_name='eu-west-2'):
10 """
11 Get Secret Keys from AWS Secrets Manager
12 :return:
13 """
14
15 # Create a Secrets Manager client
16 session = boto3.session.Session()
17 client = session.client(
18 service_name='secretsmanager',
19 region_name=region_name
20 )
21
22 try:
23 get_secret_value_response = client.get_secret_value(SecretId=secret_name)
24 except ClientError as e:
25 if e.response['Error']['Code'] == 'DecryptionFailureException':
26 # Secrets Manager can't decrypt the protected secret text using the provided KMS key.
27 raise e
28 elif e.response['Error']['Code'] == 'InternalServiceErrorException':
29 # An error occurred on the server side.
30 raise e
31 elif e.response['Error']['Code'] == 'InvalidParameterException':
32 # You provided an invalid value for a parameter.
33 raise e
34 elif e.response['Error']['Code'] == 'InvalidRequestException':
35 # You provided a parameter value that is not valid for the current state of the resource.
36 raise e
37 elif e.response['Error']['Code'] == 'ResourceNotFoundException':
38 # AWS Can't find the resource that you asked for.
39 raise e
40 else:
41 # Decrypts secret using the associated KMS CMK.
42 # Depending on whether the secret is a string or binary, one of these fields will be populated.
43 if 'SecretString' in get_secret_value_response:
44 secret = get_secret_value_response['SecretString']
45 else:
46 secret = base64.b64decode(get_secret_value_response['SecretBinary'])
47
48 return json.loads(secret)
49
50
51 class AwsComprehend:
52
53 @staticmethod
54 def get_sentiment(tweets):
55 """
56 :param tweets: list (string), required
57 :return: dict
58 """
59 # Create a Comprehend client
60 session = boto3.session.Session()
61 comprehend = session.client(
62 service_name='comprehend',
63 region_name='eu-west-2'
64 )
65
66 response = comprehend.batch_detect_sentiment(
67 TextList=tweets,
68 LanguageCode='en'
69 )
70
71 return response