#stdlib imports
from io import StringIO as _StringIO
from collections import Counter as _Counter, OrderedDict as _OrderedDict
from logging import getLogger as _getLogger, DEBUG as _DEBUG, INFO as _INFO, basicConfig as _basicConfig
from datetime import datetime as _datetime, timedelta as _timedelta, timezone as _timezone
import json as _json
import time as _time
import re as _re
import multiprocessing as _multiprocessing
import itertools as _itertools
import glob as _glob
from argparse import ArgumentParser as _ArgumentParser


#3rd party imports
from pandas import read_csv as _read_csv, pivot_table as _pivot_table, DataFrame as _DataFrame, Series as _Series, date_range as _date_range, to_numeric as _to_numeric, concat as _concat
from regex import sub as _sub, compile as _compile
from dateutil.relativedelta import relativedelta as _relativedelta
import pytz as _pytz
from influxdb import InfluxDBClient as _InfluxDBClient
from functools import reduce as _reduce
from random import randint as _randint
import json as _json
import time as _time
import re as _re

import numpy as _np
import itertools as _itertools
import requests as _requests
from requests.adapters import HTTPAdapter as _HTTPAdapter
from requests.packages.urllib3.util.retry import Retry as _Retry

#internal imports
from det.utilities.CSTime import CSTimeToDatetime as _CSTimeToDatetime
from det.logging.logging_framework import setup_logging as _setup_logging

def readData(filename):
	"""open filename and return binary data"""
	logger = _getLogger(__name__)
	logger.info("in readData")
	data = None
	with open(filename,'rb') as fid:
		data = fid.read()

	return data

def decodeData(inputData, encoding):
	"""decode data using encoding to return string data"""
	logger = _getLogger(__name__)
	logger.info("in decodeData")
	data = None
	try:
		data = inputData.decode(encoding).strip()
	except Exception as e:
		logger.error("Exception in decode:%s"%(e))
		raise(e)

	return data

def clean_sub_row(text, otsdbExcludeCharRegex, otsdbExcludeCharReplacer='_', excludeChars = ''):
	list1 = text.split('=')
	r = ''.join(list1[1:])
	list1[1] = _re.sub(otsdbExcludeCharRegex%(excludeChars), otsdbExcludeCharReplacer, str(r)).strip()
	l = ''.join(list1[0])
	list1[0] = _re.sub(otsdbExcludeCharRegex%(excludeChars), otsdbExcludeCharReplacer, str(l)).strip()
	list1 = [x for x in list1[:2] if x is not '']
	return ('='.join(list1))

def clean_row_part(z, otsdbExcludeCharRegex, otsdbExcludeCharReplacer='_', excludeChars = ''):
	for i in range(len(z)):
		z[i] = clean_sub_row(z[i], otsdbExcludeCharRegex, otsdbExcludeCharReplacer=otsdbExcludeCharReplacer, excludeChars = excludeChars)
	return z

def split(df, chunkSize=30):
	numberChunks = len(df) // chunkSize + 1
	return _np.array_split(df, numberChunks, axis=0)

def retry_session(
		retries=3,
		backoff_factor=0.3,
		status_forcelist=(500, 502, 504),
		session=None,
):
	session = session or _requests.Session()
	retry = _Retry(
		total=retries,
		read=retries,
		connect=retries,
		backoff_factor=backoff_factor,
		status_forcelist=status_forcelist,
	)
	adapter = _HTTPAdapter(max_retries=retry)
	session.mount('http://', adapter)
	session.mount('https://', adapter)
	return session

def processInputData(inputa, fieldSeparator = None, recordSeparator = None, otsdbExcludeCharRegex = '[^a-zA-Z\d\-_%s/]+', otsdbExcludeCharReplacer='_'):
	logger = _getLogger(__name__)
	logger.info("in processInputData")

	output = {}
	if fieldSeparator is None or recordSeparator is None:
		return output
	engine = 'c'	
	if len(recordSeparator) > 1:
		engine = 'python'
	lines = inputa.strip().split('\n')

	partsAll = [elem.strip().split(recordSeparator) for elem in lines]
	subfields = []
	for selector in range(3):
		subfields.append([elem[selector].split(fieldSeparator) for elem in partsAll])

	for i in range(len(subfields[0])):

		clean_row_part(subfields[0][i], otsdbExcludeCharRegex, otsdbExcludeCharReplacer=otsdbExcludeCharReplacer, excludeChars = '')
		clean_row_part(subfields[1][i], otsdbExcludeCharRegex, otsdbExcludeCharReplacer=otsdbExcludeCharReplacer, excludeChars = '')
		clean_row_part(subfields[2][i], otsdbExcludeCharRegex, otsdbExcludeCharReplacer=otsdbExcludeCharReplacer, excludeChars = '.')

	columns = [elem.split('=')[0] for elem in subfields[0][0]]
	times = [elem.split('=')[0] for elem in subfields[1][0]]
	values = [elem.split('=')[0] for elem in subfields[2][0]]
	dfAll = []
	for currentFieldIndex, currentField in enumerate(subfields):
		d1 = []
		logger.info(currentFieldIndex)
		for indexa,elema in enumerate(currentField):
			try:
				d1.append({elem.split('=')[0]:elem.split('=')[1]  for elem in elema })
			except Exception as e:
				logger.info(e)
				logger.debug(elema)
				logger.debug((d1[0].keys()))
				raise(e)
		dfAll.append(_DataFrame(d1))
	dfMaster = _concat(dfAll, axis=1)

	logger.info('before timestamp normalization')
	timestamp = normalizeTimeStamps(dfMaster, time_columns = times)
	dfMaster['timestamp'] = timestamp
	if logger.isEnabledFor(_DEBUG):
		logger.debug("type debug:%s"%(type(timestamp[0])))
		logger.debug("%s"%(timestamp[0]))
		logger.debug("type debug:%s"%(type(dfMaster['timestamp'][0])))
		logger.debug("%s"%(dfMaster['timestamp'][0]))
	output = {}
	output['df'] = dfMaster
	output['time_columns'] = times
	output['value_columns'] = values
	output['group_columns'] = columns
	logger.info("times=%s, values=%s, columns=%s"%(times, values, columns))
	logger.info("leave processInputData")
	return output

def findStartTime(timeColumnsDict = None, numberPoints = 0, timePeriod='D'):
	"""given time columns, end time and numberPoints return the python date time object"""
	logger = _getLogger(__name__)
	logger.info("in findStartTime")
	if 'month_id' in timeColumnsDict:
		logger.info("found month_id")
		timestamp = _CSTimeToDatetime(month_id=timeColumnsDict['month_id'] - numberPoints)
	elif 'week_id' in timeColumnsDict:
		logger.info("found week_id")
		timestamp = _CSTimeToDatetime(week_id=timeColumnsDict['week_id'] - numberPoints)
	elif 'hour_id' in timeColumnsDict and 'time_id' in timeColumnsDict:
		logger.info("found time_id and hour_id")
		timestamp = _CSTimeToDatetime(time_id=timeColumnsDict['time_id'], hour_id = timeColumnsDict['hour_id'] - numberPoints)
	elif 'time_id' in timeColumnsDict:
		logger.info("found time_id ")
		timestamp = _CSTimeToDatetime(time_id=timeColumnsDict['time_id'] - numberPoints)
	elif len(timeColumnsDict) == 1:  # assume it is already a timestamp
		#for single column it is assumed that the time column is seconds since epoch

		secondsSinceEpoch = timeColumnsDict[next(iter(timeColumnsDict))]
		utcDateTime = _datetime.fromtimestamp( secondsSinceEpoch, _timezone.utc)
		time_id = 0
		hour_id = 0
		week_id = 0
		month_id = 0
		#set the 1 that should be in the delta based on the time parameter
		needDelta = True
		if timePeriod == 'daily':
			time_id = -1*numberPoints
		elif timePeriod == 'hourly':
			hour_id = -1*numberPoints 
		elif timePeriod == 'monthly':
			month_id = -1*numberPoints
		elif timePeriod == 'weekly':
			week_id = -1*numberPoints
		elif timePeriod == 'unknown':
			needDelta = False
			logger.warn("unknown time type defaulting to start time as epoch")
			timestamp = _datetime(1970,1,1,0,0,0,0,_pytz.UTC)
		else:
			needDelta = False
			logger.warn("unsupported time type defaulting to start time as epoch")
			timestamp = _datetime(1970,1,1,0,0,0,0,_pytz.UTC)
		if needDelta:
			timestamp = utcDateTime + _timedelta(days=time_id,hours=hour_id,) + _relativedelta(months=month_id, weeks = week_id)
	return timestamp


def normalizeTimeStamps(df=None, time_columns=None):
	"""take in comscore time ids or assume time column is a timestamp and return a timestamp series"""
	logger = _getLogger(__name__)
	logger.info("in normalizeTimeStamps")
	timestamps = None
	# overall policy is to check for month id and if found use it alone
	# otherwise if time_id and hour_id present use those
	# otherwise use time_id else ensure time column is len 1, and assume it is already a timestamp
	if 'month_id' in time_columns:
		logger.info("found month_id")
		timestamps = df.month_id.apply(lambda z: _CSTimeToDatetime(month_id=z).timestamp())
	elif 'week_id' in time_columns:
		logger.info("found week_id")
		timestamps = df.week_id.apply(lambda z: _CSTimeToDatetime(week_id=z).timestamp())
	elif 'hour_id' in time_columns and 'time_id' in time_columns:
		logger.info("found time_id and hour_id")
		timestamps = df[['time_id', 'hour_id']].apply(
			lambda row: _CSTimeToDatetime(time_id=row['time_id'], hour_id=row['hour_id']).timestamp(), axis=1)
	elif 'time_id' in time_columns:
		logger.info("found time_id ")
		timestamps = df[['time_id']].apply(lambda row: _CSTimeToDatetime(time_id=row['time_id'],hour_id=0).timestamp(), axis=1)
	elif len(time_columns) == 1:  # assume it is already a timestamp
		timestamps = df[time_columns[0]].astype(int)
	else:
		raise Exception('Unrecognized time format')

	# log diagnostic type info
	if len(timestamps) > 0:
		logger.info("type info:%s" % (type(timestamps[0])))
		logger.info("%s" % (timestamps[0]))
	return timestamps

def buildMetricFromKV(d1):
	return ".".join([str(value[1]) for value in sorted(d1.items())])

def buildTagSearchFromKV(d1):
	z1 = ",".join(['%s=%s'%(str(value[0]),str(value[1])) for value in sorted(d1.items())])
	return '{' + z1 + '}'

def generateOpentsdbJsonPayloadAsMetrics(fields=None, metric=None, tags=None,time=None,tagsToKeep=None, compressTags = False, overrideMillisecond=False):
	"""input a processed data frame row and output the proper influx db format of a point"""
	points = []
	tagDict = {}
	tagsC = tags.copy()
	if tagsToKeep is None:
		tagDict = {'static':'static'}
	else:
		if compressTags:
			tagDict['static'] = ".".join([str(tagsC[elem]) for elem in tagsToKeep])
			tagsC = {}

		else:
			for elem in tagsToKeep:
				tagValue = tagsC.pop(elem)
				tagDict[elem] = tagValue

	tagMetricPart = ''
	if overrideMillisecond:
		tagDict = {'static':'static'}
	else:
		tagMetricPart = buildMetricFromKV(tagsC)
	if len(tagMetricPart)>0:
		tagMetricPart = '.'+tagMetricPart
	for i in fields:
		point = {}
		point['metric'] = "%s.%s%s"%(metric, i, tagMetricPart )
		point['timestamp'] = time
		point['value'] = fields[i]
		point['tags'] = tagDict
		points.append(point)
	return points

def generateInfluxJsonPayload(fields = None, measurement = None, tags = None, time = None):
	"""input a processed data frame row and output the proper influx db format of a point"""
	points = []
	point = {}
	fields['sorting_stamp'] = float(time)
	point['fields'] = fields
	point['measurement'] = measurement
	point['tags'] = tags
	point['time'] = time
	points.append(point)
	return points

def generateOpentsdbJsonPayload(fields=None, metric=None, tags=None, time=None):
	"""input a processed data frame row and output the proper influx db format of a point"""
	points = []

	for i in fields:
		point = {}
		point['metric'] = metric + "." + i
		point['timestamp'] = time
		point['value'] = fields[i]
		tagsC = tags.copy()
		point['tags'] = tagsC
		#        point['tags']['valueField'] = i
		points.append(point)
	return points

def writeDataFrameToInfluxDB(df=None, valueColumns = None, groupColumns = None, apiEntryPoint='cviadqat07.office.comscore.com',port=8086,database='Panel_Only_Mobile_Data', measurement = None, username = '', password = '', timestampCol = ['timestamp']):
	"""store dataframe into tsdb via client"""
	logger = _getLogger(__name__)
	logger.info("in writeDataFrameToInfluxDB")

	restructured_df = df[groupColumns + valueColumns + timestampCol]
	num_group_cols = len(groupColumns)
	num_value_cols = len(valueColumns)
	value_col_start_index = num_group_cols + 1
	value_col_end_index = num_group_cols + num_value_cols

	#create a client connection
	client = _InfluxDBClient(apiEntryPoint, port, username, password, database)
	#create the db if needed
	client.create_database(database)

	#verify that we have a valid measurement name
	if not (measurement and measurement.strip()):
		raise Exception('Measurement must not be empty')

	#add the dataframe row by row, store the result of each load
	result = []
	# Using itertuples as it gives speed improvement over iterrows.
	# increase depends on size of the input df but for O(1000) can improve performance by 2x for O(1e5) can improve by 100x
	for tup in restructured_df.itertuples():
		pj = generateInfluxJsonPayload(fields = (dict(zip(valueColumns, tup[value_col_start_index: value_col_end_index + 1]))), measurement=measurement, tags=dict(zip(groupColumns, tup[1:num_group_cols + 1])), time=int(tup[-1]))
		result.append(client.write_points(pj, time_precision='s'))
	return result

def assignUIDtoOpentsdb(df=None, valueColumns=None, groupColumns=None, apiEndPoint=None):
	#    tagk = valueColumns.copy()
	#    tagk.append("valueField")
	#    tagv = df[groupColumns].values.flatten().tolist()
	#    tagv.extend(list(df[groupColumns].columns.values))
	#    tagk = list(set(tagk))
	#    tagv = list(set(tagv))
	metrics = [metric + "." + m for m in valueColumns]
	tagk = list(df[groupColumns].columns.values)  # tagk
	tagv = df[groupColumns].values.flatten().tolist()
	dataJ = {"metric": metrics, "tagk": tagk, "tagv": tagv}
	r = _requests.post(url=apiEndPoint, data=_json.dumps(dataJ))
	return len(r.json())


def writeDataFrameToOpenTsdb(df=None,
							 valueColumns=None,
							 groupColumns=None,
							 apiEntryPoint='cviadqat07.office.comscore.com',
							 putApiEndPoint=None,
							 assignApiEndPoint=None,
							 port=None,
							 metric=None,
							 host_tag=False,
							 check_tsdb_alive=True,
							 send_metrics_batch_limit=50,
							 tagsToKeep=None,
							 max_queue_size=50000, compressTags = False, overrideMillisecond=False):
	"""store dataframe into tsdb via client"""
	logger = _getLogger(__name__)

	if not (metric and metric.strip()):
		raise Exception('Metric must not be empty')

	result = putAPIOpentsdb(df=df, valueColumns=valueColumns, groupColumns=groupColumns, metric=metric, putApiEndPoint=putApiEndPoint, tagsToKeep = tagsToKeep, compressTags = compressTags, overrideMillisecond=overrideMillisecond)
	return result


def putAPIOpentsdb(df=None,
					valueColumns=None,
					groupColumns=None,
					metric=None,
					tagsToKeep = None,
					putApiEndPoint=None, compressTags = False, overrideMillisecond=False, timestampCol='timestamp', tolerance=0.1):

	logger = _getLogger(__name__)
	pj = get_metric_names(dataframe=df, group_columns=groupColumns, value_columns=valueColumns, metric_prefix=metric, tagsToKeep=tagsToKeep)

	pjAll = list(_itertools.chain.from_iterable(pj))

	result = list()

	if len(pjAll) > 0:
		if logger.isEnabledFor(_DEBUG):
			logger.debug(len(pjAll))
			logger.debug(len(pjAll[0]))
			logger.debug(type(pjAll[0]))
			logger.debug(pjAll[0])
		else:
			if _randint(1,10001) < 10:

				logger.info(len(pjAll))
				logger.info(len(pjAll[0]))
				logger.info(type(pjAll[0]))
				logger.info(pjAll[0])

	else:
		logger.info('number of metrics generated are 0')
		logger.info('Not calling put API, returning empty list')
		return result


	session = retry_session(retries=5)
	r = session.post(url=putApiEndPoint, data=_json.dumps(pjAll)) #, timeout=8
	result.append(r.status_code)
	successful_puts = sum([elem < 300 for elem in result])
	logger.info('number 204:%d'%(successful_puts))

	if successful_puts == 0:
		raise ValueError('put api call was unsucessful')
	return result

def getTimeTaken(sTime, eTime):
	return f"{round(eTime - sTime)} seconds OR {round((eTime - sTime)/60)} minutes OR {round((eTime - sTime)/3600)} hours"


def mainProcessor(input_file_glob=None,apiEntryPointList = ['cviadqat07.office.comscore.com',], processes = None,metric = 'bID.gID.daySession.daily.input_dataExploded', tagsToKeep = None, compressTags = False, overrideMillisecond=False, useInflux = False, encoding = None, recordSeparator = None, fieldSeparator = None):
		
		
	logger = _getLogger(__name__)
	logger.info("in main")

	startTimeMain = _time.time()
	inputa = ""

	"""Read the file as string"""

	allFiles = sorted(_glob.glob(input_file_glob))
	logger.info('processing %s'%(str(allFiles)))
	counter = 0
	for file_ in allFiles:
		logger.info(f"Processing file: {file_}")
		inputa = readData(file_)

		inputData = decodeData(inputa, encoding)
		logger.info(f"Time taken to read the file: {getTimeTaken(startTimeMain, _time.time())}")

		data = processInputData(inputData, recordSeparator = recordSeparator, fieldSeparator = fieldSeparator)
		if overrideMillisecond:
			#increment timestamp field
			df = data['df']
			df['timestamp'] *= 1000
			milliVector = _np.arange(1,df.shape[0]+1)
			df['timestamp'] += milliVector



		splitedDF = split(data['df'], chunkSize=100)
		valueColumns = data['value_columns']
		groupColumns = data['group_columns']
		timeColumns = data['time_columns']
		
		logger.info(f"Starting put API call at: {_time.ctime()}")
		if useInflux:
			logger.info('before import into influx tsdb')
			#influx client needs leading http decorators removed
			r1 = _compile(r"^https?://(www\.)?")
			firstEntry = apiEntryPointList[0]
			firstEntry = r1.sub('',firstEntry)
			influxEntry = firstEntry.split(':')[0]
			influxPort = firstEntry.split(':')[1]
			cleanDataframe = data['df'].apply(_to_numeric, errors='ignore')
			logger.info('before import into influx tsdb %s'%(metric))
			result = writeDataFrameToInfluxDB(df=cleanDataframe,valueColumns = valueColumns, groupColumns = groupColumns,apiEntryPoint=influxEntry,port=influxPort,database=metric,measurement='autoload')
			logger.info('successfully imported %d of %d records into influxdb'%(sum(result),len(result)))

		else:
			host_tag = False
			check_tsdb_alive = False

			putApiEndPoint = [f"{apiEntryPoint}/api/put/" for apiEntryPoint in apiEntryPointList]
			assignApiEndPoint = [f"{apiEntryPoint}/api/uid/assign/"for apiEntryPoint in apiEntryPointList]
			logger.info(putApiEndPoint)
			logger.info(assignApiEndPoint)

			# result = [
			#	 writeDataFrameToOpenTsdb(
			#		 sdf.reset_index(drop=True),
			#		 valueColumns,
			#		 groupColumns,
			#		 apiEntryPoint,
			#		 putApiEndPoint,
			#		 assignApiEndPoint,
			#		 port=4248,
			#		 metric=metric,
			#		 host_tag=False,
			#		 check_tsdb_alive=False) for sdf in splitedDF
			# ]
			loadBalancerCount = len(putApiEndPoint)
			logger.info('before import into opentsdb %s'%(metric))
			with _multiprocessing.Pool(processes=processes*loadBalancerCount) as pool:
				result = pool.starmap_async(
					writeDataFrameToOpenTsdb,
					[(sdf.reset_index(drop=True), valueColumns, groupColumns, None, putApiEndPoint[indexa%loadBalancerCount], assignApiEndPoint[indexa%loadBalancerCount], None, metric, host_tag, check_tsdb_alive, 50,tagsToKeep, 50000, compressTags, overrideMillisecond) for indexa,sdf in enumerate(splitedDF)],
					chunksize=None,
					callback=None,
					error_callback=None)
				result.get(timeout=None)

		logger.info(f"End put API call at: {_time.ctime()}")
		logger.info(f"Time taken for main: {getTimeTaken(startTimeMain, _time.time())}")

def SweepingTest(metrics=None, endpoint=None, batchSize=1, printVolume = 10000):
	logger = _getLogger(__name__)
	#logger.info("in SweepingTest")#
	st = _time.time()
	allR = []
	printFactor = 0
	for queryIndex in range(0,len(metrics),batchSize):
		printFactor += batchSize
		allR.append(_requests.get('%s/api/query?start=2018/04/25-00:00:00&end=2018/06/12-00:00:00&%s'%(endpoint,"&".join(metrics[queryIndex:queryIndex+batchSize]))))
		if printFactor > printVolume:
			logger.info(queryIndex)
			logger.info('%s/api/query?start=2018/04/25-00:00:00&end=2018/06/12-00:00:00&%s'%(endpoint,"&".join(metrics[queryIndex:queryIndex+batchSize])))
			et = _time.time()
			logger.info(et-st)
			logger.info("total throughput %f"%((et-st)/(queryIndex+batchSize)))
			logger.debug(allR[-1].text)
			printFactor = 0
	et = _time.time()
	throughput = (et-st)/(len(metrics))
	logger.info("throughput=%f"%(throughput))
	return throughput,allR[-1]


def SweepingMetaTest(metaDF = None, totalRows = None, endpointsMap = None, batchSize=1, printVolume = 10000, testNumber=2, additionalSearchParameters = '', tagsToKeep = None):
	logger = _getLogger(__name__)
	metaInfo = []
	for endpoint in endpointsMap:
		logger.info('using %s'%(endpoint[0]))
		try:
			_requests.get('%s/api/dropcaches'%(endpoint[0]))
			df = metaDF['df']
			timeColumns = metaDF['time_columns']
			valueColumns = metaDF['value_columns']
			groupColumns = metaDF['group_columns']
			pj = [generateOpentsdbJsonPayloadAsMetrics(
						fields=dict(row[valueColumns]),
						metric=endpoint[testNumber],
						tags=dict(row[groupColumns]),
						time=row['timestamp'], tagsToKeep=tagsToKeep) for index,row in df[:totalRows].iterrows()]
			if tagsToKeep is None:
				metrics = ['m=none:%s%s'%(additionalSearchParameters,elem['metric']) for elem in _itertools.chain.from_iterable(pj)]
			else:
				metrics = ['m=none:%s%s%s'%(additionalSearchParameters,elem['metric'],buildTagSearchFromKV(elem['tags'],)) for elem in _itertools.chain.from_iterable(pj)]

			logger.info(pj[0][0])
			logger.info(metrics[0])
			results = SweepingTest(metrics = metrics, endpoint= endpoint[0], batchSize=batchSize, printVolume=printVolume)
			metaInfo.append(results)
		except _requests.exceptions.ConnectionError as e:
			logger.info("Catching Connection Error: %s: process next record"%(str(e)))
			metaInfo.append([e])
			continue
	return metaInfo


def get_metric_names(dataframe=None, group_columns=None, value_columns=None, timestamp_col=['timestamp'], metric_prefix=None, tagsToKeep=None):
	# Using itertuples as it gives ~100x speed improvement over iterrows.
	# The slicing/indexing of the tuple in itertuples below is based on the assumption that
	# the dataframe cols order is: [groupcols, timecols, value_cols, timestamp]

	restructured_df = dataframe[group_columns + value_columns + timestamp_col]
	num_group_cols = len(group_columns)
	num_value_cols = len(value_columns)
	value_col_start_index = num_group_cols + 1
	value_col_end_index = num_group_cols + num_value_cols

	metric_names = [generateOpentsdbJsonPayloadAsMetrics(
		fields=(dict(zip(value_columns, tup[value_col_start_index: value_col_end_index + 1]))),
		metric=metric_prefix,
		tags=dict(zip(group_columns, tup[1:num_group_cols + 1])),
		time=tup[-1], tagsToKeep=tagsToKeep) for tup in restructured_df.itertuples()]

	return metric_names


def merge_dicts(dict1, dict2):
	for k, v in _itertools.chain(dict2.items()):
		dict1[k].append(v)


def openTSDB_data_processor(metric_names=None, query_string=None, query_offset=20, tolerance=0.0):

	logger = _getLogger(__name__)
	logger.info('read failure tolerance is %s' %tolerance)	
	logger.info('number of metric names: %d' % (len(metric_names)))
	metrics = list(set(['m=none:' + elem['metric'] for elem in _itertools.chain.from_iterable(metric_names)]))
	logger.info('number of metrics as set: %d' % (len(metrics)))
	response_df_list = [(_requests.get('%s%s' % (query_string, "&".join(metrics[queryIndex:queryIndex + query_offset]))).json())
						for queryIndex in range(0, len(metrics), query_offset)]

	flattened_list = list(_itertools.chain(*response_df_list))
	logger.info('number of elements in response list: %d' % (len(flattened_list)))
	if len(metrics)>0:
		logger.info(str(metrics[:1]))
	super_dict = {'metric': [], 'dps': [], 'tags': [], 'aggregateTags': []}
	num_defective_records = 0
	num_correct_records = 0
	for ite in flattened_list:
		if ite == 'error' or type(ite) is not dict:
			num_defective_records += 1
		else:
			merge_dicts(super_dict, ite)
			num_correct_records += 1
	
	max_defective_records = num_defective_records * query_offset
	total_num_queries = max_defective_records + num_correct_records
	logger.info('number of read request errors are: %s' % num_defective_records)
	logger.info('max number of failed reads are: %s' % max_defective_records)
	logger.info('number of total queries sent are %s' % total_num_queries)
	if total_num_queries > 0:
		read_error_perc = max_defective_records / total_num_queries
	else:
		#in state where metrics were sent to tsdb for query, but no time series were returned. 
		logger.warn('total_num_queries is zero.  no input data found in tsdb')
		raise ValueError("NoRecordsFound")
	
	if read_error_perc > tolerance:
		raise ValueError('Stopping execution as read errors exceeded tolerance for num metrics queried that do not exist in OTSDB. Read error percent is %s and tolerance is %s' % (read_error_perc, tolerance))

	causal_data_frame = _DataFrame.from_dict(super_dict)

	new_df = causal_data_frame.drop(['dps', 'aggregateTags', 'tags'], 1).assign(
		**_DataFrame(causal_data_frame.dps.values.tolist()))
	transposed_df = new_df.T
	cleaned_causal_df = transposed_df.rename(
		columns=dict(zip(transposed_df.columns.tolist(), transposed_df.iloc[0].astype(str)))).drop(
		transposed_df.index[0])
	numeric_causal_df = cleaned_causal_df.apply(_to_numeric, errors='ignore')
	return numeric_causal_df


def format_1cs_df(input_df, group_col_list, time_col_list, value_col_list=list(), field_separator='\1xf', record_seperator='\1xe'):
	if len(value_col_list) > 0:
		record_separator_columns = [group_col_list[-1], time_col_list[-1], value_col_list[-1]]
	else:
		record_separator_columns = [group_col_list[-1], time_col_list[-1]]

	df = input_df.astype(str)
	df_col_list = df.columns.tolist()
	new_line_char = '\n'
	structure_df = _DataFrame(_np.tile(_np.array(df.columns), len(df.index)).reshape(len(df.index), -1), index=df.index, columns=df.columns) + '='
	key_value_df = structure_df.add(df)
	for index, col in enumerate(df_col_list):

		if index == len(df_col_list) - 1:
			key_value_df[col] = key_value_df[col].apply(lambda row_value: str(row_value) + new_line_char)
		elif col in record_separator_columns:
			key_value_df[col] = key_value_df[col].apply(lambda row_value: str(row_value) + record_seperator)
		else:
			key_value_df[col] = key_value_df[col].apply(lambda row_value: str(row_value) + field_separator)

	return key_value_df

def openTSDB_query_string(openTSDBEndpoint, start_time, end_time, api_name='query'):

	return "%s/api/%s?start=%s&end=%s&" %(openTSDBEndpoint,api_name,start_time,end_time)
	
if __name__ == '__main__':
	print('initiate main processing')
	parser = _ArgumentParser()
	parser.add_argument('input_file_glob',nargs='?', help='filename from which to read data')
	parser.add_argument('-processes', default=1, type=int,help='number of multiprocessing workers to start')
	parser.add_argument('-metric', default='testA', type=str,help='Base metric name')
	parser.add_argument('-group_id', default=1, type=int,help='comscore group tracking id')
	parser.add_argument('-business_id', default=1, type=int,help='comscore business tracking id')
	parser.add_argument('-record_separator', default='\x1e', type=str,help='record_separator to distinguish key columns, time columns, and value columns')
	parser.add_argument('-field_separator', default='\x1f', type=str,help='field_separator to distinguish individual k-v pairs in a record')
	parser.add_argument('-apiEndPoints', default='cviadqat07.office.comscore.com', type=str,help='comma separated list of api endpoints to issue requests on')
	parser.add_argument('-tagsToKeep', default=None, type=str,help='comma separated list of tags to keep as==in metric expansion')
	parser.add_argument('-compressTags', default=False, action="store_true",help='indicate whether to compress tag values to single kv pair')
	parser.add_argument('-overrideMillisecond', default=False, action="store_true",help='store data as single metric/tag static using millisecond field for different agg key')
	#parser.add_argument('-apiPorts', default='4242', type=str,help='api port to issue requests on')#
	args = parser.parse_args()
	apiEndPoints = args.apiEndPoints.split(',')
	tagsToKeep = None
	if args.tagsToKeep is not None:
		tagsToKeep = args.tagsToKeep.split(',')
	_setup_logging()
	logger = _getLogger(__name__)
	logger.info('start processing loop')
	filename = args.input_file_glob
	logger.info(filename)
	mainProcessor(input_file_glob = filename, processes = args.processes, metric = args.metric, apiEntryPointList = apiEndPoints, tagsToKeep = tagsToKeep, compressTags = args.compressTags, overrideMillisecond=args.overrideMillisecond)
