<commit_before># encoding=utf-8
'''Base class for URL shortener services'''

import datetime
import logging
import re
import sys
import time

from requests.exceptions import ConnectionError

from terroroftinytown.client import alphabet, VERSION
from terroroftinytown.client.errors import (UnhandledStatusCode,
    UnexpectedNoResult, ScraperError, PleaseRetry)
from terroroftinytown.services.status import URLStatus
from terroroftinytown.six.moves import html_parser
import terroroftinytown


__all__ = ['BaseService', 'registry']
DEFAULT_USER_AGENT = 'ArchiveTeam TerrorOfTinyTown/{0} ({1})'.format(
    VERSION, datetime.datetime.utcnow())


class BaseService(object):
    def __init__(self, params):
        self.params = params
        self.logger = logging.getLogger(self.__class__.__name__)
        self.current_shortcode = None
        self.user_agent = DEFAULT_USER_AGENT

    def wait(self):
        sleep_time = self.params['request_delay']
        time.sleep(sleep_time)

    def transform_sequence_num(self, sequence_number):
        return alphabet.int_to_str(
            sequence_number, self.params['alphabet']
        )

    def scrape_one(self, sequence_number):
        self.current_shortcode = shortcode = self.transform_sequence_num(sequence_number)
        url = self.params['url_template'].format(shortcode=shortcode)

        self.logger.info('Requesting %s', url)

        response = self.fetch_url(url)
        url_status, result_url, encoding = self.process_response(response)

        if url_status == URLStatus.ok:
            assert result_url is not None
            self.logger.info('Got a result.')
            self.logger.debug('%s %s', result_url, response.encoding)

            return {
                'shortcode': shortcode,
                'url': result_url,
                'encoding': encoding or 'latin-1'
            }

    def fetch_url(self, url, method=None):
        # this import is moved here so that tracker can import
        # registry without installing requests
        import requests

        assert method in (None, 'get', 'head'), method

        headers = {
            'User-Agent': self.user_agent,
        }

        try:
            if method == 'get' or self.params['method'] == 'get':
                response = requests.get(
                    url, allow_redirects=False, headers=headers)
            else:
                response = requests.head(
                    url, allow_redirects=False, headers=headers)
        except ConnectionError as e:
            return self.process_connection_error(e)

        return response

    def process_response(self, response):
        status_code = response.status_code

        if status_code in self.params['redirect_codes']:
            return self.process_redirect(response)
        elif status_code in self.params['no_redirect_codes']:
            return self.process_no_redirect(response)
        elif status_code in self.params['unavailable_codes']:
            return self.process_unavailable(response)
        elif status_code in self.params['banned_codes']:
            return self.process_banned(response)
        else:
            return self.process_unknown_code(response)

    def process_redirect(self, response):
        if 'Location' in response.headers:
            result_url = response.headers['Location']
<commit_msg>services.base: Add timeout arg to requests<commit_after># encoding=utf-8
'''Base class for URL shortener services'''

import datetime
import logging
import re
import sys
import time

from requests.exceptions import ConnectionError

from terroroftinytown.client import alphabet, VERSION
from terroroftinytown.client.errors import (UnhandledStatusCode,
    UnexpectedNoResult, ScraperError, PleaseRetry)
from terroroftinytown.services.status import URLStatus
from terroroftinytown.six.moves import html_parser
import terroroftinytown


__all__ = ['BaseService', 'registry']
DEFAULT_USER_AGENT = 'ArchiveTeam TerrorOfTinyTown/{0} ({1})'.format(
    VERSION, datetime.datetime.utcnow())


class BaseService(object):
    def __init__(self, params):
        self.params = params
        self.logger = logging.getLogger(self.__class__.__name__)
        self.current_shortcode = None
        self.user_agent = DEFAULT_USER_AGENT

    def wait(self):
        sleep_time = self.params['request_delay']
        time.sleep(sleep_time)

    def transform_sequence_num(self, sequence_number):
        return alphabet.int_to_str(
            sequence_number, self.params['alphabet']
        )

    def scrape_one(self, sequence_number):
        self.current_shortcode = shortcode = self.transform_sequence_num(sequence_number)
        url = self.params['url_template'].format(shortcode=shortcode)

        self.logger.info('Requesting %s', url)

        response = self.fetch_url(url)
        url_status, result_url, encoding = self.process_response(response)

        if url_status == URLStatus.ok:
            assert result_url is not None
            self.logger.info('Got a result.')
            self.logger.debug('%s %s', result_url, response.encoding)

            return {
                'shortcode': shortcode,
                'url': result_url,
                'encoding': encoding or 'latin-1'
            }

    def fetch_url(self, url, method=None):
        # this import is moved here so that tracker can import
        # registry without installing requests
        import requests

        assert method in (None, 'get', 'head'), method

        headers = {
            'User-Agent': self.user_agent,
        }

        try:
            if method == 'get' or self.params['method'] == 'get':
                response = requests.get(
                    url, allow_redirects=False, headers=headers, timeout=60)
            else:
                response = requests.head(
                    url, allow_redirects=False, headers=headers, timeout=60)
        except ConnectionError as e:
            return self.process_connection_error(e)

        return response

    def process_response(self, response):
        status_code = response.status_code

        if status_code in self.params['redirect_codes']:
            return self.process_redirect(response)
        elif status_code in self.params['no_redirect_codes']:
            return self.process_no_redirect(response)
        elif status_code in self.params['unavailable_codes']:
            return self.process_unavailable(response)
        elif status_code in self.params['banned_codes']:
            return self.process_banned(response)
        else:
            return self.process_unknown_code(response)

    def process_redirect(self, response):
        if 'Location' in response.headers:
            result_url = response.headers['Location']

{'fraction_non_alphanumeric': 0.08825707093143788, 'fraction_numerical': 0.001776988005330964, 'mean_word_length': 2.6886947023484433, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '4900472', 'n_tokens_mistral': 1847, 'n_tokens_neox': 1792, 'n_words': 439}