· 8 years ago · Apr 04, 2018, 07:08 AM
1import datetime
2import json
3import logging
4import os
5import string
6import time
7import uuid
8from copy import deepcopy
9from decimal import Decimal
10
11import Geohash
12import boto3
13import pymysql
14from aws_requests_auth.aws_auth import AWSRequestsAuth
15from boto3.dynamodb.conditions import Key
16from boto3.dynamodb.types import TypeDeserializer, TypeSerializer
17from elasticsearch import Elasticsearch, RequestsHttpConnection
18
19logger = logging.getLogger()
20logger.setLevel(logging.INFO)
21
22rds_host = os.environ['rds_host']
23user_name = os.environ['rds_user_name']
24password = os.environ['rds_password']
25db_name = os.environ['rds_db_name']
26
27dynamodb = boto3.resource('dynamodb')
28restaurant_table = dynamodb.Table('CF-restaurant_' + os.environ['environment'])
29menuItems_table = dynamodb.Table('CF-menuItems_' + os.environ['environment'])
30restaurant_mapper_table = dynamodb.Table('CF-restaurantMapper_' +
31 os.environ['environment'])
32menu_mapper_table = dynamodb.Table('CF-menuItemMapper_' +
33 os.environ['environment'])
34meals_table = dynamodb.Table('CF-meals_' + os.environ['environment'])
35meals_reported_table = dynamodb.Table(
36 'CF-mealReported_' + os.environ['environment'])
37images_reported_table = dynamodb.Table(
38 'CF-imagesReported_' + os.environ['environment'])
39images_table = dynamodb.Table('CF-images_' + os.environ['environment'])
40unpublished_image_table = dynamodb.Table('CF-imagesUnpublished_'
41 + os.environ['environment'])
42tags_table = dynamodb.Table('CF-tags_' + os.environ['environment'])
43users_table = dynamodb.Table('CF-users_' + os.environ['environment'])
44# Scan tags table to create a tag_id -> tag_name lookup map.
45tag_lookup = {}
46tags = tags_table.scan()
47if tags.get('ResponseMetadata', {}).get('HTTPStatusCode') == 200:
48 for item in tags.get('Items', []):
49 tag_lookup.update({item.get('tag_id'): item})
50
51serializer = TypeSerializer()
52deserializer = TypeDeserializer()
53
54lambda_client = boto3.client('lambda', region_name='us-east-1')
55
56awsauth = AWSRequestsAuth(aws_access_key=os.environ['aws_access_key'],
57 aws_secret_access_key=os.environ[
58 'aws_secret_access_key'],
59 aws_host=os.environ['host'],
60 aws_region='us-east-1',
61 aws_service='es')
62esClient = Elasticsearch(
63 hosts=[{'host': os.environ['host'], 'port': 443}],
64 http_auth=awsauth,
65 use_ssl=True,
66 verify_certs=True,
67 connection_class=RequestsHttpConnection)
68
69
70def create_menu_item(menu_detail):
71 """
72 Create menu item with given menu detail
73 :param menu_detail: dictionary of menu item detail
74 :return: created menu item id
75 """
76 try:
77 menu_response = menuItems_table.put_item(
78 Item=menu_detail)
79 if menu_response.get('ResponseMetadata', {}).get(
80 'HTTPStatusCode') == 200:
81 return True
82 else:
83 return False
84 except Exception as err:
85 logger.info("error in create menu item %s ", str(err))
86 return False
87
88
89def create_restaurant(restaurant_detail):
90 """
91 Create restaurant item with given restaurant detail
92 :param restaurant_detail: dictionary of restaurant detail
93 :return: created rentaurant id
94 """
95 try:
96 logger.info('restaurant_detail: {}'.format(restaurant_detail))
97 restaurant_response = restaurant_table.put_item(
98 Item=restaurant_detail)
99 if restaurant_response.get('ResponseMetadata', {}).get(
100 'HTTPStatusCode') == 200:
101 return True
102 else:
103 return False
104 except Exception as err:
105 logger.info("error in create restaurant entry %s ", str(err))
106 return False
107
108
109def update_count_of_tag(tags):
110 '''
111 :param tags: tags ids associated with menu_item
112 :return: success/Failure
113 '''
114 logger.info("tag ids to be updated : %s", str(tags))
115 try:
116 tag_ids = [tag for tag in tags]
117 with tags_table.batch_writer() as batch:
118 for tag_id in tag_ids:
119 if not tag_lookup.get(tag_id, {}).get('occurrence_count'):
120 occurrence_count = 0
121 else:
122 occurrence_count = tag_lookup.get(
123 tag_id, {}).get('occurrence_count')
124
125 tag_item = tag_lookup.get(tag_id).update(
126 {'occurrence_count': int(occurrence_count) + 1})
127 batch.put_item(
128 Item=tag_item
129 )
130 logger.error("successfully updated count of tags")
131 return True
132 except Exception as exc:
133 logger.error("Exception in batch write: err %s", str(exc))
134 return False
135
136
137def map_provider_menu_id(event, restaurant_id, image_url, timestamp,
138 validation_response):
139 """
140 Map between provider restaurant id with menu item
141 :param event: user input
142 :param restaurant_id: restaurant id
143 :return: Mapped menu item id
144 """
145 TWOPLACES = Decimal(10) ** -2
146 try:
147 # Query database to check if provider menu item id already exists.
148 response = menu_mapper_table.query(
149 KeyConditionExpression=Key('provider_menu_item_id').eq(
150 event.get('provider_restaurant_id') + '_' +
151 string.capwords(str(event.get('menu_item_name')).strip())),
152 ScanIndexForward=False,
153 Limit=1
154 )
155 new_image_url = None
156 tag_ids = [tag.get('tag_id') for tag in event.get('tags', {})]
157 # Return unique id of existing menu item.
158 if response.get('Items', []):
159 item = response['Items'][0]
160 menu_item_id = item.get('menu_item_id')
161 logger.info("menu item id = %s", str(menu_item_id))
162 response_menuitem = menuItems_table.query(
163 KeyConditionExpression=Key('menu_item_id').eq(menu_item_id),
164 ScanIndexForward=False,
165 Limit=1
166 )
167 # increment count of each tag
168 update_count_of_tag(tag_ids)
169 logger.info(response_menuitem)
170 logger.info("validation response = %s", str(validation_response))
171
172 if response_menuitem.get('ResponseMetadata', {}).get(
173 'HTTPStatusCode') == 200:
174 if response_menuitem.get('Items', []):
175 menu_item = response_menuitem.get('Items', [])[0]
176 new_image_url = menu_item.get('image_url')
177 logger.info("old image url = %s", new_image_url)
178 if validation_response:
179 logger.info("Got validation response true")
180 if not new_image_url or event.get('update_menu'):
181 logger.info("image url doesnot exist for menu item")
182 updated_list = []
183 updated_value = {}
184
185 elastic_input_menu_item = {
186 'restaurant_id': restaurant_id,
187 'menu_item_name': string.capwords(
188 str(event.get('menu_item_name')).strip()),
189 'provider_restaurant_id': event.get(
190 'provider_restaurant_id'),
191 'timestamp': timestamp,
192 'created_by': event.get('identity_id'),
193 'created_date': str(datetime.date.today())
194 }
195
196 # Save geohash values.
197 if event.get('longitude') and event.get('latitude'):
198 geohash = Geohash.encode(
199 float(event.get('latitude')),
200 float(event.get('longitude')))
201
202 elastic_input_menu_item.update(
203 {'geohash': geohash, 'geohash_5': geohash[:5],
204 'geohash_6': geohash[:6]})
205 updated_list.append("geohash = :geohash")
206 updated_value.update({':geohash': geohash})
207 updated_list.append("geohash_5 = :geohash_5")
208 updated_value.update({':geohash_5': geohash[:5]})
209 updated_list.append("geohash_6 = :geohash_6")
210 updated_value.update({':geohash_6': geohash[:6]})
211
212 if image_url:
213 updated_list.append("image_url = :updated_image")
214 updated_value.update({':updated_image': image_url})
215 elastic_input_menu_item.update(
216 {'image_url': image_url})
217 if event.get('tags'):
218 updated_list.append("tags = :updated_tags")
219 updated_value.update(
220 {':updated_tags': [tag.get('tag_id')
221 for tag in
222 event.get('tags')]})
223 elastic_input_menu_item.update(
224 {'tags': [tag.get('tag_id')
225 for tag in event.get('tags')]})
226 if event.get('price'):
227 price = event.get('price')
228 try:
229 price = str(Decimal(price).quantize(TWOPLACES))
230 except Exception as err:
231 logger.info("invalid price value %s ",
232 str(err))
233 price = event.get('price')
234 updated_list.append("price = :updated_price")
235 updated_value.update({':updated_price': price})
236 if event.get('currency'):
237 updated_list.append("currency = :updated_currency")
238 updated_value.update({
239 ':updated_currency': event.get('currency')})
240 elastic_input_menu_item.update(
241 {'currency': event.get('currency')})
242 if event.get('symbol'):
243 updated_list.append("symbol = :updated_symbol")
244 updated_value.update({
245 ':updated_symbol': event.get('symbol')})
246 elastic_input_menu_item.update(
247 {'symbol': event.get('symbol')})
248 if event.get('description'):
249 updated_list.append(
250 "description = :updated_description")
251 updated_value.update(
252 {':updated_description': event.get(
253 'description')})
254 elastic_input_menu_item.update(
255 {'description': event.get('description')})
256
257 updated_query = "SET " + ", ".join(updated_list)
258 logger.info(updated_query)
259 logger.info(updated_value)
260 menu_item_update = menuItems_table.update_item(
261 Key={
262 'menu_item_id': menu_item_id
263 },
264 UpdateExpression=updated_query,
265 ExpressionAttributeValues=updated_value
266 )
267 if menu_item_update.get('ResponseMetadata', {}).get(
268 'HTTPStatusCode') == 200:
269 elastic_input_menu_item.update({
270 'menu_item_id': menu_item_id})
271 print('Updating menu item in elastic search table')
272 print(elastic_input_menu_item)
273 response_image_save_activity = lambda_client.invoke(
274 FunctionName='CF-updateElastic:' + os.environ[
275 'alias'],
276 InvocationType='Event',
277 Payload=json.dumps(
278 {'menu_item': elastic_input_menu_item}))
279
280 return menu_item_id
281 else:
282 logger.info('got error in update menu item')
283 return None
284 else:
285 logger.info("image url exist for menu item")
286 return menu_item_id
287 else:
288 logger.info("got validation response as false")
289 return menu_item_id
290
291 else:
292 menu_item_id = str(uuid.uuid4())
293 # Create menu item in database.
294 response = restaurant_table.query(
295 KeyConditionExpression=Key('restaurant_id').eq(restaurant_id)
296 )
297 geohash = None
298 restaurant_name = None
299 if response.get('Items', []):
300 item = response['Items'][0]
301 restaurant_name = item.get('restaurant_name')
302 menu_detail = {
303 'menu_item_id': menu_item_id,
304 'restaurant_id': restaurant_id,
305 'menu_item_name': string.capwords(
306 str(event.get('menu_item_name')).strip()),
307 'timestamp': timestamp,
308 'created_by': event.get('identity_id'),
309 'created_date': str(datetime.date.today())
310 }
311 # Save geohash values.
312 if event.get('longitude') and event.get('latitude'):
313 geohash = Geohash.encode(
314 float(event.get('latitude')),
315 float(event.get('longitude')))
316
317 menu_detail.update({
318 'geohash': geohash,
319 'geohash_5': geohash[:5],
320 'geohash_6': geohash[:6]
321 })
322 if restaurant_name:
323 menu_detail.update({
324 'restaurant_name': restaurant_name
325 })
326 if image_url and validation_response:
327 menu_detail.update({
328 'image_url': image_url
329 })
330 if event.get('tags'):
331 menu_detail.update(
332 {
333 'tags': [tag.get('tag_id')
334 for tag in event.get('tags')]
335 })
336
337 if event.get('city'):
338 menu_detail.update({
339 'city': event.get('city')
340 })
341 if event.get('price'):
342 price = event.get('price')
343 try:
344 price = str(Decimal(price).quantize(TWOPLACES))
345 except Exception as err:
346 logger.info("invalid price value %s ", str(err))
347 price = event.get('price')
348 menu_detail.update({
349 'price': price
350 })
351 if event.get('currency'):
352 menu_detail.update({
353 'currency': event.get('currency')
354 })
355 if event.get('symbol'):
356 menu_detail.update({
357 'symbol': event.get('symbol')
358 })
359 if event.get('description'):
360 menu_detail.update({
361 'description': event.get('description')
362 })
363 if event.get('identity_id'):
364 menu_detail.update({
365 'created_by': event.get('identity_id')
366 })
367
368 # Map provider menu item id in database if it does not exist.
369 if create_menu_item(menu_detail):
370
371 # Add menu item to elastic search.
372 elastic_input_menu_item = deepcopy(menu_detail)
373 elastic_input_menu_item.update({
374 'provider_restaurant_id': event.get(
375 'provider_restaurant_id')})
376 print('Adding menu item to elastic search table: ')
377 print(elastic_input_menu_item)
378 response_image_save_activity = lambda_client.invoke(
379 FunctionName='CF-updateElastic:' + os.environ['alias'],
380 InvocationType='Event',
381 Payload=json.dumps({'menu_item': elastic_input_menu_item}))
382
383 menuitem_mapper_response = menu_mapper_table.put_item(
384 Item={'menu_item_id': menu_item_id,
385 'provider_menu_item_id': event.get(
386 'provider_restaurant_id') + '_' + string.capwords(
387 str(event.get('menu_item_name')).strip())
388 })
389 if menuitem_mapper_response.get('ResponseMetadata', {}).get(
390 'HTTPStatusCode') == 200:
391 # increment count of each tag
392 if tag_ids:
393 update_count_of_tag(tag_ids)
394 return menu_item_id
395 else:
396 return None
397
398 else:
399 return None
400 except Exception as err:
401 logger.info("error in map and get menu_id %s ", str(err))
402 return None
403
404
405def map_provider_restaurant_id(event, timestamp):
406 """
407 Map restaurant id and provider restaurant id.
408 :param event: user input
409 :param timestamp: time of restaurant creation
410 :return: Mapped restaurant id
411 """
412 try:
413 # Query database to check if provider restaurant id already exists.
414 response = restaurant_mapper_table.query(
415 KeyConditionExpression=Key('provider_restaurant_id').eq(
416 event.get('provider_restaurant_id')),
417 ScanIndexForward=False,
418 Limit=1
419 )
420
421 # return corresponding restaurant id for give provider id.
422 if response.get('Items', []):
423 item = response['Items'][0]
424 restaurant_id = item.get('restaurant_id')
425 return restaurant_id
426 # Create restaurant entry in database.
427 else:
428 restaurant_id = str(uuid.uuid4())
429 restaurant_detail = {
430 'restaurant_id': restaurant_id,
431 'timestamp': timestamp,
432 'created_by': event.get('identity_id'),
433 'created_date': str(datetime.date.today())
434 }
435 try:
436 geohash = Geohash.encode(
437 float(event.get('latitude')),
438 float(event.get('longitude')))
439 restaurant_detail.update(
440 {'geohash': geohash, 'geohash_5': geohash[:5],
441 'geohash_6': geohash[:6]})
442 except Exception as err:
443 logger.info("error in calculating restaurant geohash: %s ",
444 str(err))
445
446 # Map provider restaurant id in database if it does not exist.
447 if create_restaurant(restaurant_detail):
448 restaurant_mapper_detail = {'restaurant_id': restaurant_id,
449 'provider_restaurant_id': event.get(
450 'provider_restaurant_id'),
451 }
452 restaurant_mapper_response = restaurant_mapper_table.put_item(
453 Item=restaurant_mapper_detail)
454 if restaurant_mapper_response.get('ResponseMetadata', {}).get(
455 'HTTPStatusCode') == 200:
456 return restaurant_id
457 else:
458 return None
459
460 else:
461 return None
462 except Exception as err:
463 logger.info("error in mapping and get restaurant id %s ", str(err))
464 return None
465
466
467def update_elastic_search_db(menu_item_id, average_rating, reviews_count=None):
468 try:
469 update_data = {'average_rating': average_rating}
470 if reviews_count:
471 update_data.update({'reviews_count': reviews_count})
472
473 get_response = esClient.get(
474 index='menu_items_' + os.environ['environment'],
475 doc_type='menu_item',
476 id=menu_item_id)
477 print('Old Data {0}'.format(get_response.get('_source', {})))
478 response = esClient.update(
479 index='menu_items_' + os.environ['environment'],
480 doc_type='menu_item',
481 id=menu_item_id,
482 body={"doc": update_data},
483 refresh='true')
484 logger.info('Elastic Search Update Response {0}'.format(response))
485 get_response = esClient.get(
486 index='menu_items_' + os.environ['environment'],
487 doc_type='menu_item',
488 id=menu_item_id)
489 print('New Data {0}'.format(get_response.get('_source', {})))
490 except Exception as exc:
491 logger.error(
492 'Exception in "update_elastic_search_db": {0}'.format(exc))
493
494
495def update_news_feed(meal_id):
496 try:
497 conn = pymysql.connect(rds_host, user=user_name, passwd=password,
498 db=db_name, connect_timeout=5)
499 logger.info('RDS connection Info {0}'.format(conn))
500 current_time = int(time.time())
501 with conn.cursor() as cur:
502 cur.execute("SELECT * FROM newsFeed_" + os.environ['environment'] +
503 " WHERE meal_id = %s", args=[meal_id])
504 print('Data Before {0}'.format(cur.fetchall()))
505
506 cur.execute("UPDATE newsFeed_" + os.environ['environment'] +
507 " SET last_updated = %s" + " WHERE meal_id = %s",
508 args=[str(current_time), meal_id])
509 conn.commit()
510
511 cur.execute("SELECT * FROM newsFeed_" + os.environ['environment'] +
512 " WHERE meal_id = %s", args=[meal_id])
513 print('Data After {0}'.format(cur.fetchall()))
514 except Exception as exc:
515 logger.error('Exception in "update_news_feed": {0}'.format(exc))
516
517
518def editPost_handler(event, context):
519 try:
520 logger.info("Input Data for %s environment is %s",
521 str(os.environ['environment']), str(event))
522
523 image_url = event.get('image_url')
524 timestamp = int(time.time())
525 image_id = str(uuid.uuid4())
526 image_response = {}
527 restaurant_id = ""
528 menu_item_id = ""
529 validation_response = False
530 detailed_response = {}
531 if image_url:
532 if event.get('google_vision_status') != 'failure':
533 validation_response = True
534
535 if event.get('provider_restaurant_id') or event.get(
536 'restaurant_id'):
537 if event.get('restaurant_id'):
538 restaurant_id = event.get('restaurant_id')
539 if event.get('menu_item_id'):
540 menu_item_id = event.get('menu_item_id')
541 if event.get('provider_restaurant_id'):
542 restaurant_id = map_provider_restaurant_id(event, timestamp)
543 if restaurant_id:
544 menu_item_id = map_provider_menu_id(event,
545 restaurant_id,
546 image_url,
547 timestamp,
548 validation_response)
549 if not restaurant_id:
550 return {'status': 'failure',
551 'status_code': 500,
552 'message': 'Internal Server Error'}
553
554 if not menu_item_id:
555 return {'status': 'failure',
556 'status_code': 500,
557 'message': 'Internal Server Error'}
558
559 images = [image_id]
560 user_id = event.get('identity_id')
561
562 return {'status': 'success',
563 'message': 'Successfully updated post'}
564 except Exception as exc:
565 logger.error('Exception in editPost_handler {0}'.format(exc))
566 return {'status': 'failure', 'message': str(exc)}