[github/fretlink/tap-google-sheets.git] / tap_google_sheets / sync.py

import time
import math
import json
from datetime import datetime, timedelta
import pytz
import singer
from singer import metrics, metadata, Transformer, utils
from singer.utils import strptime_to_utc, strftime
from tap_google_sheets.streams import STREAMS
from tap_google_sheets.schema import get_sheet_metadata

LOGGER = singer.get_logger()


def write_schema(catalog, stream_name):
    stream = catalog.get_stream(stream_name)
    schema = stream.schema.to_dict()
    try:
        singer.write_schema(stream_name, schema, stream.key_properties)
    except OSError as err:
        LOGGER.info('OS Error writing schema for: {}'.format(stream_name))
        raise err


def write_record(stream_name, record, time_extracted):
    try:
        singer.messages.write_record(stream_name, record, time_extracted=time_extracted)
    except OSError as err:
        LOGGER.info('OS Error writing record for: {}'.format(stream_name))
        LOGGER.info('record: {}'.format(record))
        raise err


def get_bookmark(state, stream, default):
    if (state is None) or ('bookmarks' not in state):
        return default
    return (
        state
        .get('bookmarks', {})
        .get(stream, default)
    )


def write_bookmark(state, stream, value):
    if 'bookmarks' not in state:
        state['bookmarks'] = {}
    state['bookmarks'][stream] = value
    LOGGER.info('Write state for stream: {}, value: {}'.format(stream, value))
    singer.write_state(state)


# Transform/validate batch of records w/ schema and sent to target
def process_records(catalog,
                    stream_name,
                    records,
                    time_extracted):
    stream = catalog.get_stream(stream_name)
    schema = stream.schema.to_dict()
    stream_metadata = metadata.to_map(stream.metadata)
    with metrics.record_counter(stream_name) as counter:
        for record in records:
            # Transform record for Singer.io
            with Transformer() as transformer:
                transformed_record = transformer.transform(
                    record,
                    schema,
                    stream_metadata)
                write_record(stream_name, transformed_record, time_extracted=time_extracted)
                counter.increment()
        return counter.value


def sync_stream(stream_name, selected_streams, catalog, state, records, time_extracted=None):
    # Should sheets_loaded be synced?
    if stream_name in selected_streams:
        LOGGER.info('STARTED Syncing {}'.format(stream_name))
        update_currently_syncing(state, stream_name)
        selected_fields = get_selected_fields(catalog, stream_name)
        LOGGER.info('Stream: {}, selected_fields: {}'.format(stream_name, selected_fields))
        write_schema(catalog, stream_name)
        if not time_extracted:
            time_extracted = utils.now()
        record_count = process_records(
            catalog=catalog,
            stream_name=stream_name,
            records=records,
            time_extracted=time_extracted)
        LOGGER.info('FINISHED Syncing {}, Total Records: {}'.format(stream_name, record_count))
        update_currently_syncing(state, None)


# Currently syncing sets the stream currently being delivered in the state.
# If the integration is interrupted, this state property is used to identify
#  the starting point to continue from.
# Reference: https://github.com/singer-io/singer-python/blob/master/singer/bookmarks.py#L41-L46
def update_currently_syncing(state, stream_name):
    if (stream_name is None) and ('currently_syncing' in state):
        del state['currently_syncing']
    else:
        singer.set_currently_syncing(state, stream_name)
    singer.write_state(state)


# List selected fields from stream catalog
def get_selected_fields(catalog, stream_name):
    stream = catalog.get_stream(stream_name)
    mdata = metadata.to_map(stream.metadata)
    mdata_list = singer.metadata.to_list(mdata)
    selected_fields = []
    for entry in mdata_list:
        field = None
        try:
            field = entry['breadcrumb'][1]
            if entry.get('metadata', {}).get('selected', False):
                selected_fields.append(field)
        except IndexError:
            pass
    return selected_fields


def get_data(stream_name,
             endpoint_config,
             client,
             spreadsheet_id,
             range_rows=None):
    if not range_rows:
        range_rows = ''
    path = endpoint_config.get('path', stream_name).replace(
        '{spreadsheet_id}', spreadsheet_id).replace('{sheet_title}', stream_name).replace(
            '{range_rows}', range_rows)
    params = endpoint_config.get('params', {})
    api = endpoint_config.get('api', 'sheets')
    querystring = '&'.join(['%s=%s' % (key, value) for (key, value) in params.items()]).replace(
        '{sheet_title}', stream_name)
    data = {}
    time_extracted = utils.now()
    data = client.get(
        path=path,
        api=api,
        params=querystring,
        endpoint=stream_name)
    return data, time_extracted


# Tranform file_metadata: remove nodes from lastModifyingUser, format as array
def transform_file_metadata(file_metadata):
    # Convert to dict
    file_metadata_tf = json.loads(json.dumps(file_metadata))
    # Remove keys
    if file_metadata_tf.get('lastModifyingUser'):
        file_metadata_tf['lastModifyingUser'].pop('photoLink', None)
        file_metadata_tf['lastModifyingUser'].pop('me', None)
        file_metadata_tf['lastModifyingUser'].pop('permissionId', None)
    # Add record to an array of 1
    file_metadata_arr = []
    file_metadata_arr.append(file_metadata_tf)
    return file_metadata_arr


# Tranform spreadsheet_metadata: remove defaultFormat and sheets nodes, format as array
def transform_spreadsheet_metadata(spreadsheet_metadata):
    # Convert to dict
    spreadsheet_metadata_tf = json.loads(json.dumps(spreadsheet_metadata))
    # Remove keys: defaultFormat and sheets (sheets will come in sheet_metadata)
    if spreadsheet_metadata_tf.get('properties'):
        spreadsheet_metadata_tf['properties'].pop('defaultFormat', None)
    spreadsheet_metadata_tf.pop('sheets', None)
    # Add record to an array of 1
    spreadsheet_metadata_arr = []
    spreadsheet_metadata_arr.append(spreadsheet_metadata_tf)
    return spreadsheet_metadata_arr


# Tranform spreadsheet_metadata: add spreadsheetId, sheetUrl, and columns metadata
def transform_sheet_metadata(spreadsheet_id, sheet, columns):
    # Convert to properties to dict
    sheet_metadata = sheet.get('properties')
    sheet_metadata_tf = json.loads(json.dumps(sheet_metadata))
    sheet_id = sheet_metadata_tf.get('sheetId')
    sheet_url = 'https://docs.google.com/spreadsheets/d/{}/edit#gid={}'.format(
        spreadsheet_id, sheet_id)
    sheet_metadata_tf['spreadsheetId'] = spreadsheet_id
    sheet_metadata_tf['sheetUrl'] = sheet_url
    sheet_metadata_tf['columns'] = columns
    return sheet_metadata_tf


# Convert Excel Date Serial Number (excel_date_sn) to datetime string
# timezone_str: defaults to UTC (which we assume is the timezone for ALL datetimes)
def excel_to_dttm_str(excel_date_sn, timezone_str=None):
    if not timezone_str:
        timezone_str = 'UTC'
    tzn = pytz.timezone(timezone_str)
    sec_per_day = 86400
    excel_epoch = 25569 # 1970-01-01T00:00:00Z
    epoch_sec = math.floor((excel_date_sn - excel_epoch) * sec_per_day)
    epoch_dttm = datetime(1970, 1, 1)
    excel_dttm = epoch_dttm + timedelta(seconds=epoch_sec)
    utc_dttm = tzn.localize(excel_dttm).astimezone(pytz.utc)
    utc_dttm_str = strftime(utc_dttm)
    return utc_dttm_str


# Transform sheet_data: add spreadsheet_id, sheet_id, and row, convert dates/times
#  Convert from array of values to JSON with column names as keys
def transform_sheet_data(spreadsheet_id, sheet_id, from_row, columns, sheet_data_rows):
    sheet_data_tf = []
    is_last_row = False
    row_num = from_row
    # Create sorted list of columns based on columnIndex
    cols = sorted(columns, key=lambda i: i['columnIndex'])

    # LOGGER.info('sheet_data_rows: {}'.format(sheet_data_rows))
    for row in sheet_data_rows:
        # If empty row, return sheet_data_tf w/ is_last_row and row_num - 1
        if row == []:
            is_last_row = True
            return sheet_data_tf, row_num - 1, is_last_row
        sheet_data_row_tf = {}
        # Add spreadsheet_id, sheet_id, and row
        sheet_data_row_tf['__sdc_spreadsheet_id'] = spreadsheet_id
        sheet_data_row_tf['__sdc_sheet_id'] = sheet_id
        sheet_data_row_tf['__sdc_row'] = row_num
        col_num = 1
        for value in row:
            # Select column metadata based on column index
            col = cols[col_num - 1]
            col_skipped = col.get('columnSkipped')
            if not col_skipped:
                col_name = col.get('columnName')
                col_type = col.get('columnType')
                # Convert dates/times from Lotus Notes Serial Numbers
                if col_type == 'numberType.DATE_TIME':
                    if isinstance(value, (int, float)):
                        col_val = excel_to_dttm_str(value)
                    else:
                        col_val = str(value)
                elif col_type == 'numberType.DATE':
                    if isinstance(value, (int, float)):
                        col_val = excel_to_dttm_str(value)[:10]
                    else:
                        col_val = str(value)
                elif col_type == 'numberType.TIME':
                    if isinstance(value, (int, float)):
                        try:
                            total_secs = value * 86400 # seconds in day
                            col_val = str(timedelta(seconds=total_secs))
                        except ValueError:
                            col_val = str(value)
                    else:
                        col_val = str(value)
                elif col_type == 'numberType':
                    if isinstance(value, int):
                        col_val = int(value)
                    else:
                        try:
                            col_val = float(value)
                        except ValueError:
                            col_val = str(value)
                elif col_type == 'stringValue':
                    col_val = str(value)
                elif col_type == 'boolValue':
                    if isinstance(value, bool):
                        col_val = value
                    elif isinstance(value, str):
                        if value.lower() in ('true', 't', 'yes', 'y'):
                            col_val = True
                        elif value.lower() in ('false', 'f', 'no', 'n'):
                            col_val = False
                        else:
                            col_val = str(value)
                    elif isinstance(value, int):
                        if value in (1, -1):
                            col_val = True
                        elif value == 0:
                            col_val = False
                        else:
                            col_val = str(value)

                else:
                    col_val = value
                sheet_data_row_tf[col_name] = col_val
            col_num = col_num + 1
        sheet_data_tf.append(sheet_data_row_tf)
        row_num = row_num + 1
    return sheet_data_tf, row_num, is_last_row


def sync(client, config, catalog, state):
    start_date = config.get('start_date')
    spreadsheet_id = config.get('spreadsheet_id')

    # Get selected_streams from catalog, based on state last_stream
    #   last_stream = Previous currently synced stream, if the load was interrupted
    last_stream = singer.get_currently_syncing(state)
    LOGGER.info('last/currently syncing stream: {}'.format(last_stream))
    selected_streams = []
    for stream in catalog.get_selected_streams(state):
        selected_streams.append(stream.stream)
    LOGGER.info('selected_streams: {}'.format(selected_streams))

    if not selected_streams:
        return

    # FILE_METADATA
    file_metadata = {}
    stream_name = 'file_metadata'
    file_metadata_config = STREAMS.get(stream_name)

    # GET file_metadata
    LOGGER.info('GET file_meatadata')
    file_metadata, time_extracted = get_data(stream_name=stream_name,
                                             endpoint_config=file_metadata_config,
                                             client=client,
                                             spreadsheet_id=spreadsheet_id)
    # Transform file_metadata
    LOGGER.info('Transform file_meatadata')
    file_metadata_tf = transform_file_metadata(file_metadata)
    # LOGGER.info('file_metadata_tf = {}'.format(file_metadata_tf))

    # Check if file has changed, if not break (return to __init__)
    last_datetime = strptime_to_utc(get_bookmark(state, stream_name, start_date))
    this_datetime = strptime_to_utc(file_metadata.get('modifiedTime'))
    LOGGER.info('last_datetime = {}, this_datetime = {}'.format(last_datetime, this_datetime))
    if this_datetime <= last_datetime:
        LOGGER.info('this_datetime <= last_datetime, FILE NOT CHANGED. EXITING.')
        return
    # Sync file_metadata if selected
    sync_stream(stream_name, selected_streams, catalog, state, file_metadata_tf, time_extracted)
    write_bookmark(state, stream_name, strftime(this_datetime))

    # SPREADSHEET_METADATA
    spreadsheet_metadata = {}
    stream_name = 'spreadsheet_metadata'
    spreadsheet_metadata_config = STREAMS.get(stream_name)

    # GET spreadsheet_metadata
    LOGGER.info('GET spreadsheet_meatadata')
    spreadsheet_metadata, ss_time_extracted = get_data(
        stream_name=stream_name,
        endpoint_config=spreadsheet_metadata_config,
        client=client,
        spreadsheet_id=spreadsheet_id)

    # Transform spreadsheet_metadata
    LOGGER.info('Transform spreadsheet_meatadata')
    spreadsheet_metadata_tf = transform_spreadsheet_metadata(spreadsheet_metadata)

    # Sync spreadsheet_metadata if selected
    sync_stream(stream_name, selected_streams, catalog, state, spreadsheet_metadata_tf, \
        ss_time_extracted)

    # SHEET_METADATA and SHEET_DATA
    sheets = spreadsheet_metadata.get('sheets')
    sheet_metadata = []
    sheets_loaded = []
    sheets_loaded_config = STREAMS['sheets_loaded']
    if sheets:
        # Loop thru sheets (worksheet tabs) in spreadsheet
        for sheet in sheets:
            sheet_title = sheet.get('properties', {}).get('title')
            sheet_id = sheet.get('properties', {}).get('sheetId')

            # GET sheet_metadata and columns
            sheet_schema, columns = get_sheet_metadata(sheet, spreadsheet_id, client)
            LOGGER.info('sheet_schema: {}'.format(sheet_schema))

            # Transform sheet_metadata
            sheet_metadata_tf = transform_sheet_metadata(spreadsheet_id, sheet, columns)
            # LOGGER.info('sheet_metadata_tf = {}'.format(sheet_metadata_tf))
            sheet_metadata.append(sheet_metadata_tf)

            # SHEET_DATA
            # Should this worksheet tab be synced?
            if sheet_title in selected_streams:
                LOGGER.info('STARTED Syncing Sheet {}'.format(sheet_title))
                update_currently_syncing(state, sheet_title)
                selected_fields = get_selected_fields(catalog, sheet_title)
                LOGGER.info('Stream: {}, selected_fields: {}'.format(sheet_title, selected_fields))
                write_schema(catalog, sheet_title)

                # Determine max range of columns and rows for "paging" through the data
                sheet_last_col_index = 1
                sheet_last_col_letter = 'A'
                for col in columns:
                    col_index = col.get('columnIndex')
                    col_letter = col.get('columnLetter')
                    if col_index > sheet_last_col_index:
                        sheet_last_col_index = col_index
                        sheet_last_col_letter = col_letter
                sheet_max_row = sheet.get('properties').get('gridProperties', {}).get('rowCount')

                # Initialize paging for 1st batch
                is_last_row = False
                batch_rows = 200
                from_row = 2
                if sheet_max_row < batch_rows:
                    to_row = sheet_max_row
                else:
                    to_row = batch_rows

                # Loop thru batches (each having 200 rows of data)
                while not is_last_row and from_row < sheet_max_row and to_row <= sheet_max_row:
                    range_rows = 'A{}:{}{}'.format(from_row, sheet_last_col_letter, to_row)

                    # GET sheet_data for a worksheet tab
                    sheet_data, time_extracted = get_data(
                        stream_name=sheet_title,
                        endpoint_config=sheets_loaded_config,
                        client=client,
                        spreadsheet_id=spreadsheet_id,
                        range_rows=range_rows)
                    # Data is returned as a list of arrays, an array of values for each row
                    sheet_data_rows = sheet_data.get('values')

                    # Transform batch of rows to JSON with keys for each column
                    sheet_data_tf, row_num, is_last_row = transform_sheet_data(
                        spreadsheet_id=spreadsheet_id,
                        sheet_id=sheet_id,
                        from_row=from_row,
                        columns=columns,
                        sheet_data_rows=sheet_data_rows)
                    if row_num < to_row:
                        is_last_row = True

                    # Process records, send batch of records to target
                    record_count = process_records(
                        catalog=catalog,
                        stream_name=sheet_title,
                        records=sheet_data_tf,
                        time_extracted=ss_time_extracted)
                    LOGGER.info('Sheet: {}, ecords processed: {}'.format(
                        sheet_title, record_count))

                    # Update paging from/to_row for next batch
                    from_row = to_row + 1
                    if to_row + batch_rows > sheet_max_row:
                        to_row = sheet_max_row
                    else:
                        to_row = to_row + batch_rows

                # SHEETS_LOADED
                # Add sheet to sheets_loaded
                sheet_loaded = {}
                sheet_loaded['spreadsheetId'] = spreadsheet_id
                sheet_loaded['sheetId'] = sheet_id
                sheet_loaded['title'] = sheet_title
                sheet_loaded['loadDate'] = strftime(utils.now())
                sheet_loaded['lastRowNumber'] = row_num
                sheets_loaded.append(sheet_loaded)

                # Emit a Singer ACTIVATE_VERSION message after each sheet is complete.
                # This forces hard deletes on the data downstream if fewer records are sent.
                # https://github.com/singer-io/singer-python/blob/master/singer/messages.py#L137
                activate_version_message = singer.ActivateVersionMessage(
                    stream=sheet_title,
                    version=int(time.time() * 1000))
                singer.write_message(activate_version_message)

                LOGGER.info('FINISHED Syncing Sheet {}, Total Rows: {}'.format(
                    sheet_title, row_num - 1))

    stream_name = 'sheet_metadata'
    # Sync sheet_metadata if selected
    sync_stream(stream_name, selected_streams, catalog, state, sheet_metadata)

    stream_name = 'sheets_loaded'
    # Sync sheet_metadata if selected
    sync_stream(stream_name, selected_streams, catalog, state, sheets_loaded)

    return
Commit	Line	Data
89643ba6 JH	1	import time
89643ba6 JH	2	import math
89643ba6	3	import json
da690bda	4	from datetime import datetime, timedelta
99424fee JH	5	import pytz
99424fee JH	6	import singer
89643ba6 JH	7	from singer import metrics, metadata, Transformer, utils
89643ba6 JH	8	from singer.utils import strptime_to_utc, strftime
89643ba6 JH	9	from tap_google_sheets.streams import STREAMS
	10	from tap_google_sheets.schema import get_sheet_metadata
	11
	12	LOGGER = singer.get_logger()
	13
	14
	15	def write_schema(catalog, stream_name):
	16	stream = catalog.get_stream(stream_name)
	17	schema = stream.schema.to_dict()
	18	try:
	19	singer.write_schema(stream_name, schema, stream.key_properties)
	20	except OSError as err:
	21	LOGGER.info('OS Error writing schema for: {}'.format(stream_name))
	22	raise err
	23
	24
	25	def write_record(stream_name, record, time_extracted):
	26	try:
	27	singer.messages.write_record(stream_name, record, time_extracted=time_extracted)
	28	except OSError as err:
	29	LOGGER.info('OS Error writing record for: {}'.format(stream_name))
	30	LOGGER.info('record: {}'.format(record))
	31	raise err
	32
	33
	34	def get_bookmark(state, stream, default):
	35	if (state is None) or ('bookmarks' not in state):
	36	return default
	37	return (
	38	state
	39	.get('bookmarks', {})
	40	.get(stream, default)
	41	)
	42
	43
	44	def write_bookmark(state, stream, value):
	45	if 'bookmarks' not in state:
	46	state['bookmarks'] = {}
	47	state['bookmarks'][stream] = value
	48	LOGGER.info('Write state for stream: {}, value: {}'.format(stream, value))
	49	singer.write_state(state)
	50
	51
da690bda JH	52	# Transform/validate batch of records w/ schema and sent to target
da690bda JH	53	def process_records(catalog,
89643ba6 JH	54	stream_name,
89643ba6 JH	55	records,
da690bda	56	time_extracted):
89643ba6 JH	57	stream = catalog.get_stream(stream_name)
	58	schema = stream.schema.to_dict()
	59	stream_metadata = metadata.to_map(stream.metadata)
89643ba6 JH	60	with metrics.record_counter(stream_name) as counter:
89643ba6 JH	61	for record in records:
89643ba6 JH	62	# Transform record for Singer.io
	63	with Transformer() as transformer:
	64	transformed_record = transformer.transform(
	65	record,
	66	schema,
	67	stream_metadata)
da690bda JH	68	write_record(stream_name, transformed_record, time_extracted=time_extracted)
	69	counter.increment()
	70	return counter.value
	71
	72
99424fee	73	def sync_stream(stream_name, selected_streams, catalog, state, records, time_extracted=None):
da690bda JH	74	# Should sheets_loaded be synced?
	75	if stream_name in selected_streams:
	76	LOGGER.info('STARTED Syncing {}'.format(stream_name))
	77	update_currently_syncing(state, stream_name)
99424fee JH	78	selected_fields = get_selected_fields(catalog, stream_name)
99424fee JH	79	LOGGER.info('Stream: {}, selected_fields: {}'.format(stream_name, selected_fields))
da690bda	80	write_schema(catalog, stream_name)
99424fee JH	81	if not time_extracted:
99424fee JH	82	time_extracted = utils.now()
da690bda JH	83	record_count = process_records(
	84	catalog=catalog,
	85	stream_name=stream_name,
	86	records=records,
99424fee	87	time_extracted=time_extracted)
da690bda JH	88	LOGGER.info('FINISHED Syncing {}, Total Records: {}'.format(stream_name, record_count))
da690bda JH	89	update_currently_syncing(state, None)
89643ba6 JH	90
	91
	92	# Currently syncing sets the stream currently being delivered in the state.
	93	# If the integration is interrupted, this state property is used to identify
	94	# the starting point to continue from.
	95	# Reference: https://github.com/singer-io/singer-python/blob/master/singer/bookmarks.py#L41-L46
	96	def update_currently_syncing(state, stream_name):
	97	if (stream_name is None) and ('currently_syncing' in state):
	98	del state['currently_syncing']
	99	else:
	100	singer.set_currently_syncing(state, stream_name)
	101	singer.write_state(state)
	102
	103
	104	# List selected fields from stream catalog
	105	def get_selected_fields(catalog, stream_name):
	106	stream = catalog.get_stream(stream_name)
	107	mdata = metadata.to_map(stream.metadata)
	108	mdata_list = singer.metadata.to_list(mdata)
	109	selected_fields = []
	110	for entry in mdata_list:
99424fee	111	field = None
89643ba6	112	try:
99424fee	113	field = entry['breadcrumb'][1]
89643ba6 JH	114	if entry.get('metadata', {}).get('selected', False):
	115	selected_fields.append(field)
	116	except IndexError:
	117	pass
	118	return selected_fields
	119
	120
	121	def get_data(stream_name,
	122	endpoint_config,
	123	client,
	124	spreadsheet_id,
	125	range_rows=None):
	126	if not range_rows:
	127	range_rows = ''
	128	path = endpoint_config.get('path', stream_name).replace(
	129	'{spreadsheet_id}', spreadsheet_id).replace('{sheet_title}', stream_name).replace(
	130	'{range_rows}', range_rows)
	131	params = endpoint_config.get('params', {})
	132	api = endpoint_config.get('api', 'sheets')
	133	querystring = '&'.join(['%s=%s' % (key, value) for (key, value) in params.items()]).replace(
	134	'{sheet_title}', stream_name)
	135	data = {}
da690bda	136	time_extracted = utils.now()
89643ba6 JH	137	data = client.get(
	138	path=path,
	139	api=api,
	140	params=querystring,
	141	endpoint=stream_name)
da690bda	142	return data, time_extracted
89643ba6 JH	143
89643ba6 JH	144
da690bda	145	# Tranform file_metadata: remove nodes from lastModifyingUser, format as array
89643ba6 JH	146	def transform_file_metadata(file_metadata):
	147	# Convert to dict
	148	file_metadata_tf = json.loads(json.dumps(file_metadata))
	149	# Remove keys
	150	if file_metadata_tf.get('lastModifyingUser'):
	151	file_metadata_tf['lastModifyingUser'].pop('photoLink', None)
	152	file_metadata_tf['lastModifyingUser'].pop('me', None)
	153	file_metadata_tf['lastModifyingUser'].pop('permissionId', None)
	154	# Add record to an array of 1
	155	file_metadata_arr = []
	156	file_metadata_arr.append(file_metadata_tf)
	157	return file_metadata_arr
	158
	159
da690bda	160	# Tranform spreadsheet_metadata: remove defaultFormat and sheets nodes, format as array
89643ba6 JH	161	def transform_spreadsheet_metadata(spreadsheet_metadata):
	162	# Convert to dict
	163	spreadsheet_metadata_tf = json.loads(json.dumps(spreadsheet_metadata))
da690bda	164	# Remove keys: defaultFormat and sheets (sheets will come in sheet_metadata)
89643ba6 JH	165	if spreadsheet_metadata_tf.get('properties'):
	166	spreadsheet_metadata_tf['properties'].pop('defaultFormat', None)
	167	spreadsheet_metadata_tf.pop('sheets', None)
	168	# Add record to an array of 1
	169	spreadsheet_metadata_arr = []
	170	spreadsheet_metadata_arr.append(spreadsheet_metadata_tf)
	171	return spreadsheet_metadata_arr
	172
	173
da690bda	174	# Tranform spreadsheet_metadata: add spreadsheetId, sheetUrl, and columns metadata
89643ba6 JH	175	def transform_sheet_metadata(spreadsheet_id, sheet, columns):
	176	# Convert to properties to dict
	177	sheet_metadata = sheet.get('properties')
99424fee	178	sheet_metadata_tf = json.loads(json.dumps(sheet_metadata))
89643ba6 JH	179	sheet_id = sheet_metadata_tf.get('sheetId')
	180	sheet_url = 'https://docs.google.com/spreadsheets/d/{}/edit#gid={}'.format(
	181	spreadsheet_id, sheet_id)
	182	sheet_metadata_tf['spreadsheetId'] = spreadsheet_id
	183	sheet_metadata_tf['sheetUrl'] = sheet_url
	184	sheet_metadata_tf['columns'] = columns
	185	return sheet_metadata_tf
	186
	187
da690bda JH	188	# Convert Excel Date Serial Number (excel_date_sn) to datetime string
	189	# timezone_str: defaults to UTC (which we assume is the timezone for ALL datetimes)
	190	def excel_to_dttm_str(excel_date_sn, timezone_str=None):
	191	if not timezone_str:
	192	timezone_str = 'UTC'
99424fee	193	tzn = pytz.timezone(timezone_str)
da690bda JH	194	sec_per_day = 86400
	195	excel_epoch = 25569 # 1970-01-01T00:00:00Z
	196	epoch_sec = math.floor((excel_date_sn - excel_epoch) * sec_per_day)
	197	epoch_dttm = datetime(1970, 1, 1)
	198	excel_dttm = epoch_dttm + timedelta(seconds=epoch_sec)
99424fee	199	utc_dttm = tzn.localize(excel_dttm).astimezone(pytz.utc)
da690bda JH	200	utc_dttm_str = strftime(utc_dttm)
	201	return utc_dttm_str
	202
	203
	204	# Transform sheet_data: add spreadsheet_id, sheet_id, and row, convert dates/times
	205	# Convert from array of values to JSON with column names as keys
	206	def transform_sheet_data(spreadsheet_id, sheet_id, from_row, columns, sheet_data_rows):
	207	sheet_data_tf = []
	208	is_last_row = False
	209	row_num = from_row
	210	# Create sorted list of columns based on columnIndex
99424fee	211	cols = sorted(columns, key=lambda i: i['columnIndex'])
da690bda JH	212
	213	# LOGGER.info('sheet_data_rows: {}'.format(sheet_data_rows))
	214	for row in sheet_data_rows:
	215	# If empty row, return sheet_data_tf w/ is_last_row and row_num - 1
	216	if row == []:
	217	is_last_row = True
	218	return sheet_data_tf, row_num - 1, is_last_row
	219	sheet_data_row_tf = {}
	220	# Add spreadsheet_id, sheet_id, and row
	221	sheet_data_row_tf['__sdc_spreadsheet_id'] = spreadsheet_id
	222	sheet_data_row_tf['__sdc_sheet_id'] = sheet_id
	223	sheet_data_row_tf['__sdc_row'] = row_num
	224	col_num = 1
	225	for value in row:
	226	# Select column metadata based on column index
	227	col = cols[col_num - 1]
	228	col_skipped = col.get('columnSkipped')
	229	if not col_skipped:
	230	col_name = col.get('columnName')
	231	col_type = col.get('columnType')
	232	# Convert dates/times from Lotus Notes Serial Numbers
	233	if col_type == 'numberType.DATE_TIME':
99424fee	234	if isinstance(value, (int, float)):
da690bda JH	235	col_val = excel_to_dttm_str(value)
	236	else:
	237	col_val = str(value)
	238	elif col_type == 'numberType.DATE':
99424fee	239	if isinstance(value, (int, float)):
da690bda JH	240	col_val = excel_to_dttm_str(value)[:10]
	241	else:
	242	col_val = str(value)
	243	elif col_type == 'numberType.TIME':
99424fee	244	if isinstance(value, (int, float)):
da690bda JH	245	try:
	246	total_secs = value * 86400 # seconds in day
	247	col_val = str(timedelta(seconds=total_secs))
	248	except ValueError:
	249	col_val = str(value)
	250	else:
	251	col_val = str(value)
	252	elif col_type == 'numberType':
	253	if isinstance(value, int):
	254	col_val = int(value)
	255	else:
	256	try:
	257	col_val = float(value)
	258	except ValueError:
	259	col_val = str(value)
	260	elif col_type == 'stringValue':
	261	col_val = str(value)
	262	elif col_type == 'boolValue':
	263	if isinstance(value, bool):
	264	col_val = value
	265	elif isinstance(value, str):
	266	if value.lower() in ('true', 't', 'yes', 'y'):
	267	col_val = True
	268	elif value.lower() in ('false', 'f', 'no', 'n'):
	269	col_val = False
	270	else:
	271	col_val = str(value)
	272	elif isinstance(value, int):
99424fee	273	if value in (1, -1):
da690bda JH	274	col_val = True
	275	elif value == 0:
	276	col_val = False
	277	else:
	278	col_val = str(value)
	279
	280	else:
	281	col_val = value
	282	sheet_data_row_tf[col_name] = col_val
	283	col_num = col_num + 1
	284	sheet_data_tf.append(sheet_data_row_tf)
	285	row_num = row_num + 1
	286	return sheet_data_tf, row_num, is_last_row
	287
	288
89643ba6 JH	289	def sync(client, config, catalog, state):
	290	start_date = config.get('start_date')
	291	spreadsheet_id = config.get('spreadsheet_id')
	292
	293	# Get selected_streams from catalog, based on state last_stream
	294	# last_stream = Previous currently synced stream, if the load was interrupted
	295	last_stream = singer.get_currently_syncing(state)
	296	LOGGER.info('last/currently syncing stream: {}'.format(last_stream))
	297	selected_streams = []
	298	for stream in catalog.get_selected_streams(state):
	299	selected_streams.append(stream.stream)
	300	LOGGER.info('selected_streams: {}'.format(selected_streams))
	301
	302	if not selected_streams:
	303	return
	304
da690bda	305	# FILE_METADATA
89643ba6	306	file_metadata = {}
da690bda JH	307	stream_name = 'file_metadata'
	308	file_metadata_config = STREAMS.get(stream_name)
	309
	310	# GET file_metadata
	311	LOGGER.info('GET file_meatadata')
	312	file_metadata, time_extracted = get_data(stream_name=stream_name,
	313	endpoint_config=file_metadata_config,
	314	client=client,
	315	spreadsheet_id=spreadsheet_id)
	316	# Transform file_metadata
	317	LOGGER.info('Transform file_meatadata')
89643ba6 JH	318	file_metadata_tf = transform_file_metadata(file_metadata)
89643ba6 JH	319	# LOGGER.info('file_metadata_tf = {}'.format(file_metadata_tf))
da690bda JH	320
	321	# Check if file has changed, if not break (return to __init__)
	322	last_datetime = strptime_to_utc(get_bookmark(state, stream_name, start_date))
89643ba6 JH	323	this_datetime = strptime_to_utc(file_metadata.get('modifiedTime'))
	324	LOGGER.info('last_datetime = {}, this_datetime = {}'.format(last_datetime, this_datetime))
	325	if this_datetime <= last_datetime:
	326	LOGGER.info('this_datetime <= last_datetime, FILE NOT CHANGED. EXITING.')
99424fee JH	327	return
	328	# Sync file_metadata if selected
	329	sync_stream(stream_name, selected_streams, catalog, state, file_metadata_tf, time_extracted)
	330	write_bookmark(state, stream_name, strftime(this_datetime))
da690bda JH	331
da690bda JH	332	# SPREADSHEET_METADATA
89643ba6	333	spreadsheet_metadata = {}
da690bda JH	334	stream_name = 'spreadsheet_metadata'
	335	spreadsheet_metadata_config = STREAMS.get(stream_name)
	336
	337	# GET spreadsheet_metadata
	338	LOGGER.info('GET spreadsheet_meatadata')
99424fee	339	spreadsheet_metadata, ss_time_extracted = get_data(
da690bda JH	340	stream_name=stream_name,
	341	endpoint_config=spreadsheet_metadata_config,
	342	client=client,
	343	spreadsheet_id=spreadsheet_id)
	344
	345	# Transform spreadsheet_metadata
	346	LOGGER.info('Transform spreadsheet_meatadata')
89643ba6	347	spreadsheet_metadata_tf = transform_spreadsheet_metadata(spreadsheet_metadata)
89643ba6	348
da690bda	349	# Sync spreadsheet_metadata if selected
99424fee JH	350	sync_stream(stream_name, selected_streams, catalog, state, spreadsheet_metadata_tf, \
99424fee JH	351	ss_time_extracted)
da690bda JH	352
da690bda JH	353	# SHEET_METADATA and SHEET_DATA
89643ba6 JH	354	sheets = spreadsheet_metadata.get('sheets')
	355	sheet_metadata = []
	356	sheets_loaded = []
	357	sheets_loaded_config = STREAMS['sheets_loaded']
	358	if sheets:
da690bda	359	# Loop thru sheets (worksheet tabs) in spreadsheet
89643ba6 JH	360	for sheet in sheets:
89643ba6 JH	361	sheet_title = sheet.get('properties', {}).get('title')
da690bda JH	362	sheet_id = sheet.get('properties', {}).get('sheetId')
	363
	364	# GET sheet_metadata and columns
89643ba6	365	sheet_schema, columns = get_sheet_metadata(sheet, spreadsheet_id, client)
99424fee	366	LOGGER.info('sheet_schema: {}'.format(sheet_schema))
da690bda JH	367
da690bda JH	368	# Transform sheet_metadata
89643ba6 JH	369	sheet_metadata_tf = transform_sheet_metadata(spreadsheet_id, sheet, columns)
	370	# LOGGER.info('sheet_metadata_tf = {}'.format(sheet_metadata_tf))
	371	sheet_metadata.append(sheet_metadata_tf)
	372
da690bda JH	373	# SHEET_DATA
	374	# Should this worksheet tab be synced?
	375	if sheet_title in selected_streams:
	376	LOGGER.info('STARTED Syncing Sheet {}'.format(sheet_title))
	377	update_currently_syncing(state, sheet_title)
99424fee JH	378	selected_fields = get_selected_fields(catalog, sheet_title)
99424fee JH	379	LOGGER.info('Stream: {}, selected_fields: {}'.format(sheet_title, selected_fields))
da690bda JH	380	write_schema(catalog, sheet_title)
	381
	382	# Determine max range of columns and rows for "paging" through the data
	383	sheet_last_col_index = 1
	384	sheet_last_col_letter = 'A'
	385	for col in columns:
	386	col_index = col.get('columnIndex')
	387	col_letter = col.get('columnLetter')
	388	if col_index > sheet_last_col_index:
	389	sheet_last_col_index = col_index
	390	sheet_last_col_letter = col_letter
	391	sheet_max_row = sheet.get('properties').get('gridProperties', {}).get('rowCount')
	392
	393	# Initialize paging for 1st batch
	394	is_last_row = False
	395	batch_rows = 200
	396	from_row = 2
	397	if sheet_max_row < batch_rows:
	398	to_row = sheet_max_row
	399	else:
	400	to_row = batch_rows
	401
	402	# Loop thru batches (each having 200 rows of data)
	403	while not is_last_row and from_row < sheet_max_row and to_row <= sheet_max_row:
	404	range_rows = 'A{}:{}{}'.format(from_row, sheet_last_col_letter, to_row)
99424fee	405
da690bda JH	406	# GET sheet_data for a worksheet tab
	407	sheet_data, time_extracted = get_data(
	408	stream_name=sheet_title,
	409	endpoint_config=sheets_loaded_config,
	410	client=client,
	411	spreadsheet_id=spreadsheet_id,
	412	range_rows=range_rows)
	413	# Data is returned as a list of arrays, an array of values for each row
	414	sheet_data_rows = sheet_data.get('values')
	415
	416	# Transform batch of rows to JSON with keys for each column
	417	sheet_data_tf, row_num, is_last_row = transform_sheet_data(
	418	spreadsheet_id=spreadsheet_id,
	419	sheet_id=sheet_id,
	420	from_row=from_row,
	421	columns=columns,
	422	sheet_data_rows=sheet_data_rows)
	423	if row_num < to_row:
	424	is_last_row = True
	425
	426	# Process records, send batch of records to target
	427	record_count = process_records(
	428	catalog=catalog,
	429	stream_name=sheet_title,
	430	records=sheet_data_tf,
	431	time_extracted=ss_time_extracted)
99424fee JH	432	LOGGER.info('Sheet: {}, ecords processed: {}'.format(
	433	sheet_title, record_count))
	434
da690bda JH	435	# Update paging from/to_row for next batch
	436	from_row = to_row + 1
	437	if to_row + batch_rows > sheet_max_row:
	438	to_row = sheet_max_row
	439	else:
	440	to_row = to_row + batch_rows
	441
	442	# SHEETS_LOADED
	443	# Add sheet to sheets_loaded
	444	sheet_loaded = {}
	445	sheet_loaded['spreadsheetId'] = spreadsheet_id
	446	sheet_loaded['sheetId'] = sheet_id
	447	sheet_loaded['title'] = sheet_title
	448	sheet_loaded['loadDate'] = strftime(utils.now())
	449	sheet_loaded['lastRowNumber'] = row_num
	450	sheets_loaded.append(sheet_loaded)
	451
	452	# Emit a Singer ACTIVATE_VERSION message after each sheet is complete.
99424fee JH	453	# This forces hard deletes on the data downstream if fewer records are sent.
99424fee JH	454	# https://github.com/singer-io/singer-python/blob/master/singer/messages.py#L137
da690bda JH	455	activate_version_message = singer.ActivateVersionMessage(
	456	stream=sheet_title,
	457	version=int(time.time() * 1000))
	458	singer.write_message(activate_version_message)
	459
	460	LOGGER.info('FINISHED Syncing Sheet {}, Total Rows: {}'.format(
	461	sheet_title, row_num - 1))
da690bda JH	462
	463	stream_name = 'sheet_metadata'
	464	# Sync sheet_metadata if selected
	465	sync_stream(stream_name, selected_streams, catalog, state, sheet_metadata)
99424fee	466
da690bda JH	467	stream_name = 'sheets_loaded'
	468	# Sync sheet_metadata if selected
	469	sync_stream(stream_name, selected_streams, catalog, state, sheets_loaded)
99424fee JH	470
99424fee JH	471	return