<commit_before>//-------------------------------------------------------------------------------
// Copyright IBM Corp. 2015
//
// 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.
//-------------------------------------------------------------------------------

'use strict';

var request = require("request");
var _ = require("lodash");
var async = require("async");
var pipesSDK = require('pipes-sdk');
var connectorExt = pipesSDK.connectorExt;
var bluemixHelperConfig = require('bluemix-helper-config');
var configManager = bluemixHelperConfig.configManager;
var vcapServices = bluemixHelperConfig.vcapServices;

var concurrency = 20;	//Async queue concurrency
/**
 * Pipes Connector for Reddit
 */
function redditConnector( parentDirPath ){
	//Call constructor from super class
	connectorExt.call(this, "reddit", "Reddit AMA", {
		copyToDashDb: false,
		extraRequiredFields: "amaURL",
		useOAuth: false,
		useCustomTables: true
	});
	
	this.getTablePrefix = function(){
		return "reddit";
	}
	
	this.getCloudantDbName = function(pipe, table){
		return pipe.name + "_" + table.name;
	}
	
	var watsonToneAnalyzerService = vcapServices.getService( configManager.get("WATSON_TONE_ANALYZER") || "tone analyzer" );
	
	this.fetchRecords = function( table, pushRecordFn, done, pipeRunStep, pipeRunStats, logger, pipe, pipeRunner ){
		if ( !watsonToneAnalyzerService ){
			return done("Unable to find Watson Tone Analyzer Service");
		}
		if ( !pipe.amaURL ){
			return done("Missing field amaURL");
		}
		console.log("Connecting to AMA with url " + pipe.amaURL );
		request.get(pipe.amaURL, {"json": true},
			function( err, response, body ){
				if ( err ){
					console.log("Unable to fetch AMA: " + err );
					return done( err );
				}
				console.log("response: " + body);
				//Create an async queue to process the record with Watson Tone Analyzer
				var q = async.queue( function( post, callback ){
					//Call Tone analyzer to get sentiments from the post
					request.post( watsonToneAnalyzerService.credentials.url + "/v1/tone",{
						'auth': {
						  'user': watsonToneAnalyzerService.credentials.username,
						  'pass': <PASSWORD>,
						  'sendImmediately': true
						},
						'json': {
							'text': post.selftext || post.body
						}
					}, function( err, response, body ){
						if ( err ){
							console.log("Error accessing Watson Tone Analyzer");
							return callback( err );
						}

	        			pushRecordFn({
							author: post.author,
							created: post.created_utc,
							edited: post.edited,
							id: post.id,
							num_comments: post.num_comments,
							title: post.title || "",
							text: post.selftext || post.body,
							sentiment: body
						});
						return callback();
		        	});
					
				}, concurrency );
				
				_.forEach( body, function( listing ){
					if ( listing.kind == "Listing" && listing.hasOwnProperty("data" ) ){
						if ( listing.data.hasOwnProperty("children") ){
							//Get the children from the data fields, only process the top posts for now
							_.forEach( listing.data.children, function( post ){
								if ( post.hasOwnProperty("data") ){
									q.push( post.data )
								}
							});
						}						
					}
				})
				
				//Add a drain function
				q.drain = function(){
					return done();
				}
			}
		);
	};
}

//Extend event Emitter
require('util').inherits(redditConnector, connectorExt);

module.exports = new redditConnector();<commit_msg>better logging when tone analyzer service is not found<commit_after>//-------------------------------------------------------------------------------
// Copyright IBM Corp. 2015
//
// 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.
//-------------------------------------------------------------------------------

'use strict';

var request = require("request");
var _ = require("lodash");
var async = require("async");
var pipesSDK = require('pipes-sdk');
var connectorExt = pipesSDK.connectorExt;
var bluemixHelperConfig = require('bluemix-helper-config');
var configManager = bluemixHelperConfig.configManager;
var vcapServices = bluemixHelperConfig.vcapServices;

var concurrency = 20;	//Async queue concurrency
/**
 * Pipes Connector for Reddit
 */
function redditConnector( parentDirPath ){
	//Call constructor from super class
	connectorExt.call(this, "reddit", "Reddit AMA", {
		copyToDashDb: false,
		extraRequiredFields: "amaURL",
		useOAuth: false,
		useCustomTables: true
	});
	
	this.getTablePrefix = function(){
		return "reddit";
	}
	
	this.getCloudantDbName = function(pipe, table){
		return pipe.name + "_" + table.name;
	}
	
	var watsonToneAnalyzerService = vcapServices.getService( configManager.get("WATSON_TONE_ANALYZER") || "tone analyzer" );
	
	this.fetchRecords = function( table, pushRecordFn, done, pipeRunStep, pipeRunStats, logger, pipe, pipeRunner ){
		if ( !watsonToneAnalyzerService ){
			var msg = "Unable to find Watson Tone Analyzer Service";
			console.log( msg );
			return done( msg );
		}
		if ( !pipe.amaURL ){
			return done("Missing field amaURL");
		}
		console.log("Connecting to AMA with url " + pipe.amaURL );
		request.get(pipe.amaURL, {"json": true},
			function( err, response, body ){
				if ( err ){
					console.log("Unable to fetch AMA: " + err );
					return done( err );
				}
				//console.log("response: " + body);
				//Create an async queue to process the record with Watson Tone Analyzer
				var q = async.queue( function( post, callback ){
					//Call Tone analyzer to get sentiments from the post
					request.post( watsonToneAnalyzerService.credentials.url + "/v1/tone",{
						'auth': {
						  'user': watsonToneAnalyzerService.credentials.username,
						  'pass': watsonToneAnalyzerService.credentials.password,
						  'sendImmediately': true
						},
						'json': {
							'text': post.selftext || post.body
						}
					}, function( err, response, body ){
						if ( err ){
							console.log("Error accessing Watson Tone Analyzer");
							return callback( err );
						}

	        			pushRecordFn({
							author: post.author,
							created: post.created_utc,
							edited: post.edited,
							id: post.id,
							num_comments: post.num_comments,
							title: post.title || "",
							text: post.selftext || post.body,
							sentiment: body
						});
						return callback();
		        	});
					
				}, concurrency );
				
				_.forEach( body, function( listing ){
					if ( listing.kind == "Listing" && listing.hasOwnProperty("data" ) ){
						if ( listing.data.hasOwnProperty("children") ){
							//Get the children from the data fields, only process the top posts for now
							_.forEach( listing.data.children, function( post ){
								if ( post.hasOwnProperty("data") ){
									q.push( post.data )
								}
							});
						}						
					}
				})
				
				//Add a drain function
				q.drain = function(){
					return done();
				}
			}
		);
	};
}

//Extend event Emitter
require('util').inherits(redditConnector, connectorExt);

module.exports = new redditConnector();
{'fraction_non_alphanumeric': 0.1579474342928661, 'fraction_numerical': 0.002753441802252816, 'mean_word_length': 2.56900401965163, 'pattern_counts': {'":': 2, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3336332', 'n_tokens_mistral': 2871, 'n_tokens_neox': 2242, 'n_words': 671}