<commit_before>import boto3
import datetime
import logging
import time

from botocore.exceptions import ClientError
from botocore.exceptions import NoCredentialsError
from cached_property import cached_property


logger = logging.getLogger(__file__)
PERCENTAGE_APPROVED_REQUIREMENT_ID = "000000000000000000L0"
LOCALE_REQUIREMENT_ID = "00000000000000000071"
MAX_SUPPORTED_BATCH_SIZE = 100


class MTurkServiceException(Exception):
    """Custom exception type"""


class DuplicateQualificationNameError(MTurkServiceException):
    """A Qualification with the given name already exists."""


class QualificationNotFoundException(MTurkServiceException):
    """A Qualification searched for by name does not exist."""


class WorkerLacksQualification(MTurkServiceException):
    """The worker has not been granted a given Qualification."""


class RevokedQualification(MTurkServiceException):
    """The Qualification has been revoked for this worker."""


class SNSService(object):
    """Handles AWS SNS subscriptions"""

    def __init__(
        self, aws_access_key_id, aws_secret_access_key, region_name, max_wait_secs=0
    ):
        self.aws_key = aws_access_key_id
        self.aws_secret = aws_secret_access_key
        self.region_name = region_name
        self.max_wait_secs = max_wait_secs

    @cached_property
    def _sns(self):
        session = boto3.session.Session(
            aws_access_key_id=self.aws_key,
            aws_secret_access_key=self.aws_secret,
            region_name=self.region_name,
        )
        return session.client("sns")

    def confirm_subscription(self, token, topic):
        logger.warning("Confirming SNS subsription.")
        self._sns.confirm_subscription(
            # AuthenticateOnUnsubscribe=True,  # Should we care?
            Token=token,
            TopicArn=topic,
        )

    def create_subscription(self, experiment_id, notification_url):
        logger.warning("Creating new SNS subscription...")
        protocol = "https" if notification_url.startswith("https") else "http"
        topic = self._sns.create_topic(Name=experiment_id)
        notification_url = "https://webhook.site/9a8abb0a-2e1e-4feb-9962-f35d7c1f1d4e"
        subscription = self._sns.subscribe(
            TopicArn=topic["TopicArn"],
            Protocol=protocol,
            Endpoint=notification_url,
            ReturnSubscriptionArn=True,  # So we can start polling the status
        )
        start = time.time()
        while self._awaiting_confirmation(subscription) and self._time_remains(start):
            logger.warning("Awaiting SNS subscription confirmation...")
            time.sleep(1)
        logger.warning("Subscription complete or timed out.")
        return topic["TopicArn"]

    def cancel_subscription(self, experiment_id):
        logger.warning("Cancelling SNS subscription")
        sns_topic = self._get_sns_topic_for_experiment(experiment_id)
        self._sns.delete_topic(TopicArn=sns_topic["TopicArn"])
        return True

    def _awaiting_confirmation(self, subscription):
        report = self._sns.get_subscription_attributes(
            SubscriptionArn=subscription["SubscriptionArn"]
        )
        status = report["Attributes"]["PendingConfirmation"]
        return status == "true"

    def _get_sns_topic_for_experiment(self, experiment_id):
        all_topics = self._sns.list_topics()["Topics"]
        experiment_topics = [
            t for t in all_topics if t["TopicArn"].endswith(experiment_id)
        ]
        return experiment_topics[0]

    def _time_remains(self, start):
        elapsed = time.time() - start
        return elapsed < self.max_wait_secs


class MTurkService(object):
    """Facade for Amazon Mechanical Turk services provided via the boto3
       library.
    """

    def __init__(
        self,
        aws_access_key_id,
        aws_secret_access_key,
        region_name,
        sns=None,
        sandbox=True,
        max_wait_secs=0,
        subscribe=True,
    ):
        self.aws_key = aws_access_key_id
        self.aws_secret = aws_secret_access_key
        self.region_name = region_name
        self.is_sandbox = sandbox
        self.max_wait_secs = max_wait_secs
        self.do_subscribe = subscribe

    @cached_property
    def mturk(self):
        session = boto3.session.Session(
            aws_access_key_id=self.aws_key,
            aws_secret_access_key=self.aws_secret,
            region_name=self.region_name,
        )
        return session.client(
            "mturk", endpoint_url=self.host, region_name=self.region_name
        )

    @cached_property
    def sns(self):
        return SNSService(
            aws_access_key_id=self.aws_key,
            aws_secret_access_key=self.aws_secret,
            region_name=self.region_name,
            max_wait_secs=60,
        )

    @property
    def host(self):
        if self.is_sandbox:
            template = u"https://mturk-requester-sandbox.{}.amazonaws.com"
        else:
            template = u"https://mturk-requester.{}.amazonaws.com"
        return template.format(self.region_name)

    def check_credentials(self):
        """Verifies key/secret/host combination by making a balance inquiry"""
        try:
            return bool(self.mturk.get_account_balance())
        except NoCredentialsError:
            raise MTurkServiceException("No AWS credentials set!")
        except ClientError:
            raise MTurkServiceException("Invalid AWS credentials!")
        except Exception as ex:
            raise MTurkServiceException(
                "Error checking credentials: {}".format(str(ex))
            )

    def confirm_subscription(self, token, topic):
        """Called by the MTurkRecruiter Flask route"""
        self.sns.confirm_subscription(Token=token, TopicArn=topic)

    def register_hit_type(
        self, title, description, reward, duration_hours, keywords, qualifications
    ):
        """Register HIT Type for this HIT and return the type's ID, which
        is required for creating a HIT.
        """
        reward = str(reward)
        duration_secs = int(datetime.timedelta(hours=duration_hours).total_seconds())
        hit_type = self.mturk.create_hit_type(
            Title=title,
            Description=description,
            Reward=reward,
            AssignmentDurationInSeconds=duration_secs,
            Keywords=",".join(keywords),<commit_msg>Bugfix<commit_after>import boto3
import datetime
import logging
import time

from botocore.exceptions import ClientError
from botocore.exceptions import NoCredentialsError
from cached_property import cached_property


logger = logging.getLogger(__file__)
PERCENTAGE_APPROVED_REQUIREMENT_ID = "000000000000000000L0"
LOCALE_REQUIREMENT_ID = "00000000000000000071"
MAX_SUPPORTED_BATCH_SIZE = 100


class MTurkServiceException(Exception):
    """Custom exception type"""


class DuplicateQualificationNameError(MTurkServiceException):
    """A Qualification with the given name already exists."""


class QualificationNotFoundException(MTurkServiceException):
    """A Qualification searched for by name does not exist."""


class WorkerLacksQualification(MTurkServiceException):
    """The worker has not been granted a given Qualification."""


class RevokedQualification(MTurkServiceException):
    """The Qualification has been revoked for this worker."""


class SNSService(object):
    """Handles AWS SNS subscriptions"""

    def __init__(
        self, aws_access_key_id, aws_secret_access_key, region_name, max_wait_secs=0
    ):
        self.aws_key = aws_access_key_id
        self.aws_secret = aws_secret_access_key
        self.region_name = region_name
        self.max_wait_secs = max_wait_secs

    @cached_property
    def _sns(self):
        session = boto3.session.Session(
            aws_access_key_id=self.aws_key,
            aws_secret_access_key=self.aws_secret,
            region_name=self.region_name,
        )
        return session.client("sns")

    def confirm_subscription(self, token, topic):
        logger.warning("Confirming SNS subsription.")
        self._sns.confirm_subscription(
            # AuthenticateOnUnsubscribe=True,  # Should we care?
            Token=token,
            TopicArn=topic,
        )

    def create_subscription(self, experiment_id, notification_url):
        logger.warning("Creating new SNS subscription...")
        protocol = "https" if notification_url.startswith("https") else "http"
        topic = self._sns.create_topic(Name=experiment_id)
        notification_url = "https://webhook.site/9a8abb0a-2e1e-4feb-9962-f35d7c1f1d4e"
        subscription = self._sns.subscribe(
            TopicArn=topic["TopicArn"],
            Protocol=protocol,
            Endpoint=notification_url,
            ReturnSubscriptionArn=True,  # So we can start polling the status
        )
        start = time.time()
        while self._awaiting_confirmation(subscription) and self._time_remains(start):
            logger.warning("Awaiting SNS subscription confirmation...")
            time.sleep(1)
        logger.warning("Subscription complete or timed out.")
        return topic["TopicArn"]

    def cancel_subscription(self, experiment_id):
        logger.warning("Cancelling SNS subscription")
        sns_topic = self._get_sns_topic_for_experiment(experiment_id)
        self._sns.delete_topic(TopicArn=sns_topic["TopicArn"])
        return True

    def _awaiting_confirmation(self, subscription):
        report = self._sns.get_subscription_attributes(
            SubscriptionArn=subscription["SubscriptionArn"]
        )
        status = report["Attributes"]["PendingConfirmation"]
        return status == "true"

    def _get_sns_topic_for_experiment(self, experiment_id):
        all_topics = self._sns.list_topics()["Topics"]
        experiment_topics = [
            t for t in all_topics if t["TopicArn"].endswith(experiment_id)
        ]
        return experiment_topics[0]

    def _time_remains(self, start):
        elapsed = time.time() - start
        return elapsed < self.max_wait_secs


class MTurkService(object):
    """Facade for Amazon Mechanical Turk services provided via the boto3
       library.
    """

    def __init__(
        self,
        aws_access_key_id,
        aws_secret_access_key,
        region_name,
        sns=None,
        sandbox=True,
        max_wait_secs=0,
        subscribe=True,
    ):
        self.aws_key = aws_access_key_id
        self.aws_secret = aws_secret_access_key
        self.region_name = region_name
        self.is_sandbox = sandbox
        self.max_wait_secs = max_wait_secs
        self.do_subscribe = subscribe

    @cached_property
    def mturk(self):
        session = boto3.session.Session(
            aws_access_key_id=self.aws_key,
            aws_secret_access_key=self.aws_secret,
            region_name=self.region_name,
        )
        return session.client(
            "mturk", endpoint_url=self.host, region_name=self.region_name
        )

    @cached_property
    def sns(self):
        return SNSService(
            aws_access_key_id=self.aws_key,
            aws_secret_access_key=self.aws_secret,
            region_name=self.region_name,
            max_wait_secs=60,
        )

    @property
    def host(self):
        if self.is_sandbox:
            template = u"https://mturk-requester-sandbox.{}.amazonaws.com"
        else:
            template = u"https://mturk-requester.{}.amazonaws.com"
        return template.format(self.region_name)

    def check_credentials(self):
        """Verifies key/secret/host combination by making a balance inquiry"""
        try:
            return bool(self.mturk.get_account_balance())
        except NoCredentialsError:
            raise MTurkServiceException("No AWS credentials set!")
        except ClientError:
            raise MTurkServiceException("Invalid AWS credentials!")
        except Exception as ex:
            raise MTurkServiceException(
                "Error checking credentials: {}".format(str(ex))
            )

    def confirm_subscription(self, token, topic):
        """Called by the MTurkRecruiter Flask route"""
        self.sns.confirm_subscription(token=token, topic=topic)

    def register_hit_type(
        self, title, description, reward, duration_hours, keywords, qualifications
    ):
        """Register HIT Type for this HIT and return the type's ID, which
        is required for creating a HIT.
        """
        reward = str(reward)
        duration_secs = int(datetime.timedelta(hours=duration_hours).total_seconds())
        hit_type = self.mturk.create_hit_type(
            Title=title,
            Description=description,
            Reward=reward,
            AssignmentDurationInSeconds=duration_secs,
            Keywords=",".join(keywords),
{'fraction_non_alphanumeric': 0.08498672082487112, 'fraction_numerical': 0.01062334010310889, 'mean_word_length': 2.7867494824016563, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 6, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, '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': '5652856', 'n_tokens_mistral': 3824, 'n_tokens_neox': 3697, 'n_words': 739}