<commit_before>//
// Treasure Data Bulk-Import Tool in Java
//
// Copyright (C) 2012 - 2013 <NAME>
//
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//
//        http://www.apache.org/licenses/LICENSE-2.0
//
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.
//
package com.treasure_data.td_import.source;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.logging.Logger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.ListObjectsRequest;
import com.amazonaws.services.s3.model.ObjectListing;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import com.treasure_data.td_import.Configuration;

/**
 * td import:prepare "s3://access_key_id:secret_access_key@/bucket/key_prefix"
 */
public class S3Source extends Source {
    private static final Logger LOG = Logger.getLogger(S3Source.class.getName());

    public static List<Source> createSources(SourceDesc desc) {
        String rawPath = desc.getPath();
        String[] elm = rawPath.split("/");

        String bucket = elm[0];
        String basePath = rawPath.substring(bucket.length() + 1, rawPath.length());

        AmazonS3Client client = createAmazonS3Client(desc);
        List<S3ObjectSummary> s3objects = getSources(client, bucket, basePath);
        List<Source> srcs = new ArrayList<Source>();
        for (S3ObjectSummary s3object : s3objects) {
            LOG.info(String.format("create s3-src s3object=%s, rawPath=%s",
                    s3object.getKey(), rawPath));
            srcs.add(new S3Source(createAmazonS3Client(desc), rawPath, s3object));
        }

        return srcs;
    }

    static AmazonS3Client createAmazonS3Client(SourceDesc desc) {
        String accessKey = desc.getUser();
        if (accessKey == null || accessKey.isEmpty()) {
            throw new IllegalArgumentException("S3 AccessKey is null or empty.");
        }
        String secretAccessKey = desc.getPassword();
        if (secretAccessKey == null || secretAccessKey.isEmpty()) {
            throw new IllegalArgumentException("S3 SecretAccessKey is null or empty.");
        }
        AWSCredentials credentials = new BasicAWSCredentials(accessKey, secretAccessKey);

        ClientConfiguration conf = new ClientConfiguration();
        conf.setProtocol(Configuration.BI_PREPARE_S3_PROTOCOL);
        conf.setMaxConnections(Configuration.BI_PREPARE_S3_MAX_CONNECTIONS);
        conf.setMaxErrorRetry(Configuration.BI_PREPARE_S3_MAX_ERRORRETRY);
        conf.setSocketTimeout(Configuration.BI_PREPARE_S3_SOCKET_TIMEOUT);

        return new AmazonS3Client(credentials, conf);
    }

    static List<S3ObjectSummary> getSources(AmazonS3Client client, String bucket, String basePath) {
        String prefix;
        int index = basePath.indexOf('*');
        if (index >= 0) {
            prefix = basePath.substring(0, index);
        } else {
            prefix = basePath;
        }

        LOG.info(String.format("list s3 files by client %s: bucket=%s, basePath=%s, prefix=%s",
                client, bucket, basePath, prefix));

        List<S3ObjectSummary> s3objects = new ArrayList<S3ObjectSummary>();
        String lastKey = prefix;
        do {
            ObjectListing listing = client.listObjects(new ListObjectsRequest(
                    bucket, prefix, lastKey, null, 1024));
            for(S3ObjectSummary s3object : listing.getObjectSummaries()) {
                s3objects.add(s3object);<commit_msg>changed S3Source class -- fixed bug: when a user specifies a S3 key (not use wildcard), he cannot download the file.<commit_after>//
// Treasure Data Bulk-Import Tool in Java
//
// Copyright (C) 2012 - 2013 <NAME>
//
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//
//        http://www.apache.org/licenses/LICENSE-2.0
//
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.
//
package com.treasure_data.td_import.source;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.logging.Logger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.ListObjectsRequest;
import com.amazonaws.services.s3.model.ObjectListing;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import com.treasure_data.td_import.Configuration;

/**
 * td import:prepare "s3://access_key_id:secret_access_key@/bucket/key_prefix"
 */
public class S3Source extends Source {
    private static final Logger LOG = Logger.getLogger(S3Source.class.getName());

    public static List<Source> createSources(SourceDesc desc) {
        String rawPath = desc.getPath();
        String[] elm = rawPath.split("/");

        String bucket = elm[0];
        String basePath = rawPath.substring(bucket.length() + 1, rawPath.length());

        AmazonS3Client client = createAmazonS3Client(desc);
        List<S3ObjectSummary> s3objects = getSources(client, bucket, basePath);
        List<Source> srcs = new ArrayList<Source>();
        for (S3ObjectSummary s3object : s3objects) {
            LOG.info(String.format("create s3-src s3object=%s, rawPath=%s",
                    s3object.getKey(), rawPath));
            srcs.add(new S3Source(createAmazonS3Client(desc), rawPath, s3object));
        }

        return srcs;
    }

    static AmazonS3Client createAmazonS3Client(SourceDesc desc) {
        String accessKey = desc.getUser();
        if (accessKey == null || accessKey.isEmpty()) {
            throw new IllegalArgumentException("S3 AccessKey is null or empty.");
        }
        String secretAccessKey = desc.getPassword();
        if (secretAccessKey == null || secretAccessKey.isEmpty()) {
            throw new IllegalArgumentException("S3 SecretAccessKey is null or empty.");
        }
        AWSCredentials credentials = new BasicAWSCredentials(accessKey, secretAccessKey);

        ClientConfiguration conf = new ClientConfiguration();
        conf.setProtocol(Configuration.BI_PREPARE_S3_PROTOCOL);
        conf.setMaxConnections(Configuration.BI_PREPARE_S3_MAX_CONNECTIONS);
        conf.setMaxErrorRetry(Configuration.BI_PREPARE_S3_MAX_ERRORRETRY);
        conf.setSocketTimeout(Configuration.BI_PREPARE_S3_SOCKET_TIMEOUT);

        return new AmazonS3Client(credentials, conf);
    }

    static List<S3ObjectSummary> getSources(AmazonS3Client client, String bucket, String basePath) {
        String prefix;
        int index = basePath.indexOf('*');
        if (index >= 0) {
            prefix = basePath.substring(0, index);
        } else {
            ObjectMetadata om = client.getObjectMetadata(bucket, basePath);
            S3ObjectSummary s3object = new S3ObjectSummary();
            s3object.setBucketName(bucket);
            s3object.setKey(basePath);
            s3object.setSize(om.getContentLength());

            return Arrays.asList(s3object);
        }

        LOG.info(String.format("list s3 files by client %s: bucket=%s, basePath=%s, prefix=%s",
                client, bucket, basePath, prefix));

        List<S3ObjectSummary> s3objects = new ArrayList<S3ObjectSummary>();
        String lastKey = prefix;
        do {
            ObjectListing listing = client.listObjects(new ListObjectsRequest(
                    bucket, prefix, lastKey, null, 1024));
            for(S3ObjectSummary s3object : listing.getObjectSummaries()) {
                s3objects.add(s3object);
{'fraction_non_alphanumeric': 0.10193577263063668, 'fraction_numerical': 0.01544142329640819, 'mean_word_length': 3.228003784295175, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '980273', 'n_tokens_mistral': 2540, 'n_tokens_neox': 2497, 'n_words': 622}