<commit_before>from __future__ import print_function

import os
import time
import json
import boto3
from datetime import datetime
# Cannot be imported?
#from botocore.errorfactory import ExpiredIteratorException
from ..common.conf_reader import ConfReader


class KinesisIngester(ConfReader):
  """KinesisIngester
  """

  def __init__(self, global_conf, prefix="", pid=None):
    """KinesisIngester constructor

    :param global_conf: configuration file or dictionary
    :type global_conf: str, dict
    :param prefix: prefix in configuration file
    :type prefix: str
    :type prefix: str
    :param pid: process id
    :type pid: int
    """
    # When running as deamon, save process id
    self.pid = pid
    self.verbose = 1

    super(KinesisIngester, self).__init__(global_conf, prefix)

    # Set print prefix
    self.set_pp(pp=self.get_param("pp"))
    print('[{}: log] verbose level is: {}'.format(self.pp, self.verbose))

    # Initialize attributes
    self.client = None
    self.shard_iters = dict()
    self.shard_infos = dict()
    self.stream_name = self.get_required_param('stream_name')

    # Initialize everything
    self.init_consumer()


  def set_pp(self, pp=None):
    """Set pretty print name

    :param pp: pretty print name, default will be `KinesisIngester`
    :type pp: str
    """
    if pp is not None:
      self.pp = pp
    else:
      self.pp = "KinesisIngester"

  def get_shard_infos_filename(self):
    """Get shard infos filename.

    :return: shard infos filename
    :rtype: string
    """
    return self.get_param('shard_infos_filename', self.pp+'_'+self.stream_name+'.json')

  def get_shard_iterator(self, shard_id):
    """Get shard iterator

    :param shard_id: shard identifier
    :return: shard iterator
    """
    shard_iterator_type = self.get_param('shard_iterator_type', "TRIM_HORIZON")
    # Try to get iterator based on latest processed sequence number if available
    if shard_id in self.shard_infos:
      sqn = self.shard_infos[shard_id]['sqn']
      shard_iterator = self.client.get_shard_iterator(StreamName=self.stream_name,
                                                      ShardId=shard_id,
                                                      StartingSequenceNumber=sqn,
                                                      ShardIteratorType='AFTER_SEQUENCE_NUMBER')
    else:
      shard_iterator = self.client.get_shard_iterator(StreamName=self.stream_name,
                                                      ShardId=shard_id,
                                                      ShardIteratorType=shard_iterator_type)
    return shard_iterator['ShardIterator']


  def init_client(self):
    """Initialize Kinesis client.
    """
    region_name = self.get_required_param('region_name')
    aws_profile = self.get_param('aws_profile', None)
    # This is mostly to be able to test locally
    endpoint_url = self.get_param('endpoint_url', None)
    verify = self.get_param('verify_certificates', True)
    use_ssl = self.get_param('use_ssl', True)

    # # Should we get these in another way?
    # aws_access_key_id = self.get_param('aws_access_key_id', None)
    # aws_secret_access_key = self.get_param('aws_secret_access_key', None)
    #
    # # Are there other parameters we should pass?
    # self.client = boto3.client('kinesis', region_name=region_name, endpoint_url=endpoint_url,
    #                            aws_access_key_id=aws_access_key_id,
    #                            aws_secret_access_key=aws_secret_access_key,<commit_msg>catch shard iterator issue on restart<commit_after>from __future__ import print_function

import os
import time
import json
import boto3
from datetime import datetime
# Cannot be imported?
#from botocore.errorfactory import ExpiredIteratorException
from ..common.conf_reader import ConfReader


class KinesisIngester(ConfReader):
  """KinesisIngester
  """

  def __init__(self, global_conf, prefix="", pid=None):
    """KinesisIngester constructor

    :param global_conf: configuration file or dictionary
    :type global_conf: str, dict
    :param prefix: prefix in configuration file
    :type prefix: str
    :type prefix: str
    :param pid: process id
    :type pid: int
    """
    # When running as deamon, save process id
    self.pid = pid
    self.verbose = 1

    super(KinesisIngester, self).__init__(global_conf, prefix)

    # Set print prefix
    self.set_pp(pp=self.get_param("pp"))
    print('[{}: log] verbose level is: {}'.format(self.pp, self.verbose))

    # Initialize attributes
    self.client = None
    self.shard_iters = dict()
    self.shard_infos = dict()
    self.stream_name = self.get_required_param('stream_name')

    # Initialize everything
    self.init_consumer()


  def set_pp(self, pp=None):
    """Set pretty print name

    :param pp: pretty print name, default will be `KinesisIngester`
    :type pp: str
    """
    if pp is not None:
      self.pp = pp
    else:
      self.pp = "KinesisIngester"

  def get_shard_infos_filename(self):
    """Get shard infos filename.

    :return: shard infos filename
    :rtype: string
    """
    return self.get_param('shard_infos_filename', self.pp+'_'+self.stream_name+'.json')

  def get_shard_iterator(self, shard_id):
    """Get shard iterator

    :param shard_id: shard identifier
    :return: shard iterator
    """
    shard_iterator_type = self.get_param('shard_iterator_type', "TRIM_HORIZON")
    # Try to get iterator based on latest processed sequence number if available
    if shard_id in self.shard_infos:
      try:
        sqn = self.shard_infos[shard_id]['sqn']
        shard_iterator = self.client.get_shard_iterator(StreamName=self.stream_name,
                                                        ShardId=shard_id,
                                                        StartingSequenceNumber=sqn,
                                                        ShardIteratorType='AFTER_SEQUENCE_NUMBER')
        return shard_iterator['ShardIterator']
      except Exception as inst:
        msg = "[{}.get_shard_iterator] Could not initialize from previous SequenceNumber {}. {}"
        print(msg.format(self.pp, sqn, inst))
      # Could fail with
      #botocore.errorfactory.InvalidArgumentException: An error occurred (InvalidArgumentException) when calling the GetShardIterator operation: StartingSequenceNumber 49592949124142737608129152630877400786080442094840709122 used in GetShardIterator on shard shardId-000000000000 in stream test-local-kinesis-caltech101 under account 000000000000 is invalid because it did not come from this stream.

    shard_iterator = self.client.get_shard_iterator(StreamName=self.stream_name,
                                                      ShardId=shard_id,
                                                      ShardIteratorType=shard_iterator_type)
    return shard_iterator['ShardIterator']


  def init_client(self):
    """Initialize Kinesis client.
    """
    region_name = self.get_required_param('region_name')
    aws_profile = self.get_param('aws_profile', None)
    # This is mostly to be able to test locally
    endpoint_url = self.get_param('endpoint_url', None)
    verify = self.get_param('verify_certificates', True)
    use_ssl = self.get_param('use_ssl', True)

    # # Should we get these in another way?
    # aws_access_key_id = self.get_param('aws_access_key_id', None)
    # aws_secret_access_key = self.get_param('aws_secret_access_key', None)
    #
    # # Are there other parameters we should pass?
    # self.client = boto3.client('kinesis', region_name=region_name, endpoint_url=endpoint_url,
    #                            aws_access_key_id=aws_access_key_id,
    #                            aws_secret_access_key=aws_secret_access_key,
{'fraction_non_alphanumeric': 0.08747745426436486, 'fraction_numerical': 0.011466116980159752, 'mean_word_length': 2.8016650342801177, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '412023', 'n_tokens_mistral': 2367, 'n_tokens_neox': 2258, 'n_words': 556}