<commit_before># -*- coding: utf-8 -*-
#
# Telefónica Digital - Product Development and Innovation
#
# THIS CODE AND INFORMATION ARE PROVIDED 'AS IS' WITHOUT WARRANTY OF ANY KIND,
# EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A PARTICULAR PURPOSE.
#
# Copyright (c) Telefónica Investigación y Desarrollo S.A.U.
# All rights reserved.
#
import argparse
import logging as log
import os.path
import time
from urlparse import urlparse, urljoin

import requests

import cosmos.config as c
from cosmos.util import ExitWithError, ResponseError
from cosmos.tables import format_table
from cosmos.routes import Routes


READ_PERM = 4
WRITE_PERM = 2
EXEC_PERM = 1

UNITS = [
    ('K', 1024),
    ('M', 1048576),
    ('G', 1073741824),
    ('T', 1099511627776),
]

BUFFER_SIZE = 4096


class WebHdfsClient(object):

    def __init__(self, webhdfs_uri, username, client=requests):
        self.webhdfs_uri = webhdfs_uri
        self.username = username
        self.client = client

    def opParams(self, operation, **kwargs):
        return dict(kwargs.items() + [
            ('op', operation),
            ('user.name', self.username),
        ])

    def put_file(self, local_file, remote_path):
        namenode_url = self.to_http(remote_path)
        redirect = self.client.put(namenode_url, allow_redirects=False,
                                   params=self.opParams('CREATE'))
        if redirect.status_code == 201:
            raise ExitWithError(
                -1, "WebHDFS uploads are not supported on 1-machine " +
                "clusters: an empty file has been created. See #862.")
        if redirect.status_code != 307:
            raise ResponseError('Not redirected by the WebHDFS frontend',
                                redirect)

        datanode_url = redirect.headers['Location']
        upload = self.client.put(datanode_url, data=local_file)
        if upload.status_code != 201:
            raise ResponseError('Cannot upload file to %s' % datanode_url,
                                upload)

    def list_path(self, path):
        """Lists a directory or check a file status. Returns a list of status
        objects as defined in
        http://hadoop.apache.org/docs/r1.0.4/webhdfs.html#FileStatus
        or None if the path does not exits.
        """
        r = self.client.get(self.to_http(path),
                            params=self.opParams('LISTSTATUS'))
        if r.status_code == 200:
            return r.json()["FileStatuses"]["FileStatus"]
        elif r.status_code == 404:
            return None
        else:
            raise ResponseError('Cannot list directory %s' % path, r)

    def path_type(self, path):
        """Checks the type of a path returning either 'DIRECTORY', 'FILE' or
        'NONE'."""
        status = self.list_path(path)
        if status is None:
            return 'NONE'
        if len(status) == 1 and status[0]["pathSuffix"] == '':
            return status[0]["type"]
        return 'DIRECTORY'

    def get_file(self, remote_path, out_file):
        r = self.client.get(self.to_http(remote_path), stream=True,
                            params=self.opParams('OPEN'))
        if r.status_code == 404:
            raise ExitWithError(404, 'File %s does not exist' % remote_path)
        elif r.status_code != 200:
            raise ResponseError('Cannot download file %s' % remote_path, r)
        buf = r.raw.read(BUFFER_SIZE)
        written = 0
        while (len(buf) > 0):
            written += len(buf)
            out_file.write(buf)
            buf = r.raw.read(BUFFER_SIZE)
        return written

    def delete_path(self, path, recursive=False):
        r = self.client.delete(self.to_http(path), params=self.opParams(
            'DELETE', recursive=str(recursive).lower()))
        if r.status_code != 200:
            raise ResponseError('Cannot delete path %s' % path, r)
        return r.json()["boolean"]

    def to_http(self, path):
        """Translates a simple path to the related WebHDFS HTTP URL."""
        if path.startswith("/"):
            rel_path = path[1:]
        else:
            rel_path = path
        return urljoin('http://' + urlparse(self.webhdfs_uri).netloc,
                       '/webhdfs/v1/user/%s/%s' % (self.username, rel_path))


def webhdfs_client_from_config(config):
    r = requests.get(Routes(config.api_url).storage, auth=config.credentials)
    if r.status_code != 200:
        raise ResponseError("Cannot get WebHDFS details", r)
    details = r.json()
    log.info("Webhdfs at %s for user %s" % (details["location"],
                                            details["user"]))
    return WebHdfsClient(details["location"], details["user"])


@c.with_config
def put_file(args, config):
    """Upload a file to HDFS"""
    client = webhdfs_client_from_config(config)
    if args.remote_path.endswith('/'):
        target_path = args.remote_path
    else:
        remote_type = client.path_type(args.remote_path)
        if remote_type == 'DIRECTORY':
            target_path = os.path.join(args.remote_path,
                                       os.path.split(args.local_file)[-1])
            log.info("Path %s is an existing directory, uploading to %s" %
                     (args.remote_path, target_path))
        elif remote_type == 'FILE':
            raise ExitWithError(-1, "Path %s already exists" % args.remote_path)
        else:
            target_path = args.remote_path<commit_msg>Handle block replication error on file upload<commit_after># -*- coding: utf-8 -*-
#
# Telefónica Digital - Product Development and Innovation
#
# THIS CODE AND INFORMATION ARE PROVIDED 'AS IS' WITHOUT WARRANTY OF ANY KIND,
# EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A PARTICULAR PURPOSE.
#
# Copyright (c) Telefónica Investigación y Desarrollo S.A.U.
# All rights reserved.
#
import argparse
import logging as log
import os.path
import time
from urlparse import urlparse, urljoin

import requests

import cosmos.config as c
from cosmos.util import ExitWithError, ResponseError
from cosmos.tables import format_table
from cosmos.routes import Routes


READ_PERM = 4
WRITE_PERM = 2
EXEC_PERM = 1

UNITS = [
    ('K', 1024),
    ('M', 1048576),
    ('G', 1073741824),
    ('T', 1099511627776),
]

BUFFER_SIZE = 4096


class WebHdfsClient(object):

    def __init__(self, webhdfs_uri, username, client=requests):
        self.webhdfs_uri = webhdfs_uri
        self.username = username
        self.client = client

    def opParams(self, operation, **kwargs):
        return dict(kwargs.items() + [
            ('op', operation),
            ('user.name', self.username),
        ])

    def put_file(self, local_file, remote_path):
        namenode_url = self.to_http(remote_path)
        redirect = self.client.put(namenode_url, allow_redirects=False,
                                   params=self.opParams('CREATE'))
        if redirect.status_code == 201:
            raise ExitWithError(
                -1, "WebHDFS uploads are not supported on 1-machine " +
                "clusters: an empty file has been created. See #862.")
        if self.__is_replication_exception(redirect):
            raise ExitWithError(-1, 'Cannot replicate file %s blocks' %
                                remote_path)
        elif redirect.status_code != 307:
            raise ResponseError('Not redirected by the WebHDFS frontend',
                                redirect)

        datanode_url = redirect.headers['Location']
        upload = self.client.put(datanode_url, data=local_file)
        if upload.status_code != 201:
            raise ResponseError('Cannot upload file to %s' % datanode_url,
                                upload)

    def list_path(self, path):
        """Lists a directory or check a file status. Returns a list of status
        objects as defined in
        http://hadoop.apache.org/docs/r1.0.4/webhdfs.html#FileStatus
        or None if the path does not exits.
        """
        r = self.client.get(self.to_http(path),
                            params=self.opParams('LISTSTATUS'))
        if r.status_code == 200:
            return r.json()["FileStatuses"]["FileStatus"]
        elif r.status_code == 404:
            return None
        else:
            raise ResponseError('Cannot list directory %s' % path, r)

    def path_type(self, path):
        """Checks the type of a path returning either 'DIRECTORY', 'FILE' or
        'NONE'."""
        status = self.list_path(path)
        if status is None:
            return 'NONE'
        if len(status) == 1 and status[0]["pathSuffix"] == '':
            return status[0]["type"]
        return 'DIRECTORY'

    def get_file(self, remote_path, out_file):
        r = self.client.get(self.to_http(remote_path), stream=True,
                            params=self.opParams('OPEN'))
        if r.status_code == 404:
            raise ExitWithError(404, 'File %s does not exist' % remote_path)
        elif r.status_code != 200:
            raise ResponseError('Cannot download file %s' % remote_path, r)
        buf = r.raw.read(BUFFER_SIZE)
        written = 0
        while (len(buf) > 0):
            written += len(buf)
            out_file.write(buf)
            buf = r.raw.read(BUFFER_SIZE)
        return written

    def delete_path(self, path, recursive=False):
        r = self.client.delete(self.to_http(path), params=self.opParams(
            'DELETE', recursive=str(recursive).lower()))
        if r.status_code != 200:
            raise ResponseError('Cannot delete path %s' % path, r)
        return r.json()["boolean"]

    def to_http(self, path):
        """Translates a simple path to the related WebHDFS HTTP URL."""
        if path.startswith("/"):
            rel_path = path[1:]
        else:
            rel_path = path
        return urljoin('http://' + urlparse(self.webhdfs_uri).netloc,
                       '/webhdfs/v1/user/%s/%s' % (self.username, rel_path))

    def __is_replication_exception(self, response):
        if response.status_code != 500:
            return False
        try:
            exception = response.json()
        except ValueError:
            return False
        return (exception.get('RemoteException', {}).get('exception') ==
                'ArrayIndexOutOfBoundsException')


def webhdfs_client_from_config(config):
    r = requests.get(Routes(config.api_url).storage, auth=config.credentials)
    if r.status_code != 200:
        raise ResponseError("Cannot get WebHDFS details", r)
    details = r.json()
    log.info("Webhdfs at %s for user %s" % (details["location"],
                                            details["user"]))
    return WebHdfsClient(details["location"], details["user"])


@c.with_config
def put_file(args, config):
    """Upload a file to HDFS"""
    client = webhdfs_client_from_config(config)
    if args.remote_path.endswith('/'):
        target_path = args.remote_path
    else:
        remote_type = client.path_type(args.remote_path)
        if remote_type == 'DIRECTORY':
            target_path = os.path.join(args.remote_path,
                                       os.path.split(args.local_file)[-1])
            log.info("Path %s is an existing directory, uploading to %s" %
                     (args.remote_path, target_path))
        elif remote_type == 'FILE':
            raise ExitWithError(-1, "Path %s already exists" % args.remote_path)
        else:
            target_path = args.remote_path
{'fraction_non_alphanumeric': 0.10905793972886625, 'fraction_numerical': 0.0157153959070892, 'mean_word_length': 2.2433492019042287, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7432890', 'n_tokens_mistral': 3530, 'n_tokens_neox': 3309, 'n_words': 899}