<commit_before>from multiprocessing import Pool
import echo_listener_settings as settings
from boto import sqs
from boto.sqs.message import RawMessage, Message
from boto.s3.connection import S3Connection
from boto.s3.key import Key
import json
import os.path
import sys
import redis
import time
import random
import string
import datetime

class AgnosticMessage(RawMessage):
	"""
	A message might originate from SNS or SQS. If from SNS then it will have a wrapper on it.
	"""
	
	def get_effective_message(self):
		b = json.loads(str(self.get_body()))
		if 'Type' in b and b['Type'] == "Notification":
			return json.loads(b['Message'])
		return b

def main():
	if len(sys.argv) < 7:
		showUsage()
		return
	
	redisHost = sys.argv[1]
	redisPort = int(sys.argv[2])
	redisDB = int(sys.argv[3])

	region = sys.argv[4]
	inputQueueName = sys.argv[5]
	errorQueueName = sys.argv[6]
	
	input_queue = get_queue(region, inputQueueName)

	input_queue.set_message_class(AgnosticMessage)
	
	num_pool_workers = settings.NUM_POOL_WORKERS
	messages_per_fetch = settings.MESSAGES_PER_FETCH

	pool = Pool(num_pool_workers, initializer=workerSetup, initargs=(redisHost, redisPort, redisDB, region, errorQueueName))

	while True:
			messages = input_queue.get_messages(num_messages=messages_per_fetch, visibility_timeout=120, wait_time_seconds=20)
			if len(messages) > 0:
					pool.map(process_message, messages)

def workerSetup(redisHost, redisPort, redisDB, region, errorQueueName):
	global s3Connection
	s3Connection = S3Connection()
	
	global redisClient
	redisClient = redis.Redis(host=redisHost, port=redisPort, db=redisDB)
	
	global errorQueue
	errorQueue = get_queue(region, errorQueueName)

def showUsage():
	print "Usage: echo_listener.py <Redis IP> <Redis Port> <Redis DB> <AWS region> <AWS input queue name> <AWS error queue name>"
	print "Example: echo_listener.py 172.17.0.2 6379 0 eu-west-1 echo-eu-west-1a echo-eu-west-1a-errors"

def process_message(message):
	# console_log("process_message called")
	
	message_body = message.get_effective_message()
	
	# console_log("message type=" + message_body['_type'])
	
	try:<commit_msg>pick up credentials from secrets if they exist, otherwise use ambient credentials according to boto config.<commit_after>from multiprocessing import Pool
import echo_listener_settings as settings
from boto import sqs
from boto.sqs.message import RawMessage, Message
from boto.s3.connection import S3Connection
from boto.s3.key import Key
import json
import os.path
import sys
import redis
import time
import random
import string
import datetime

class AgnosticMessage(RawMessage):
	"""
	A message might originate from SNS or SQS. If from SNS then it will have a wrapper on it.
	"""
	
	def get_effective_message(self):
		b = json.loads(str(self.get_body()))
		if 'Type' in b and b['Type'] == "Notification":
			return json.loads(b['Message'])
		return b

def main():
	if len(sys.argv) < 7:
		showUsage()
		return
	
	redisHost = sys.argv[1]
	redisPort = int(sys.argv[2])
	redisDB = int(sys.argv[3])

	region = sys.argv[4]
	inputQueueName = sys.argv[5]
	errorQueueName = sys.argv[6]
	
	input_queue = get_queue(region, inputQueueName)

	input_queue.set_message_class(AgnosticMessage)
	
	num_pool_workers = settings.NUM_POOL_WORKERS
	messages_per_fetch = settings.MESSAGES_PER_FETCH

	pool = Pool(num_pool_workers, initializer=workerSetup, initargs=(redisHost, redisPort, redisDB, region, errorQueueName))

	while True:
			messages = input_queue.get_messages(num_messages=messages_per_fetch, visibility_timeout=120, wait_time_seconds=20)
			if len(messages) > 0:
					pool.map(process_message, messages)

def workerSetup(redisHost, redisPort, redisDB, region, errorQueueName):
	global s3Connection
	if hasattr(settings, "AWS_ACCESS_KEY_ID") && hasattr(settings, "AWS_SECRET_ACCESS_KEY"):
		s3Connection = S3Connection(aws_access_key_id=settings.AWS_ACCESS_KEY_ID, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY)
	else:
		s3Connection = S3Connection()
	
	global redisClient
	redisClient = redis.Redis(host=redisHost, port=redisPort, db=redisDB)
	
	global errorQueue
	errorQueue = get_queue(region, errorQueueName)

def showUsage():
	print "Usage: echo_listener.py <Redis IP> <Redis Port> <Redis DB> <AWS region> <AWS input queue name> <AWS error queue name>"
	print "Example: echo_listener.py 172.17.0.2 6379 0 eu-west-1 echo-eu-west-1a echo-eu-west-1a-errors"

def process_message(message):
	# console_log("process_message called")
	
	message_body = message.get_effective_message()
	
	# console_log("message type=" + message_body['_type'])
	
	try:
{'fraction_non_alphanumeric': 0.09881941407958024, 'fraction_numerical': 0.01530389156099694, 'mean_word_length': 5.6400580551523944, 'pattern_counts': {'":': 2, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2174609', 'n_tokens_mistral': 1651, 'n_tokens_neox': 1543, 'n_words': 391}