<commit_before>#!/usr/bin/env node
var debug             = require('debug')('github:worker');
var base              = require('taskcluster-base');
var Promise           = require('promise');
var exchanges         = require('../lib/exchanges');
var worker            = require('../lib/worker');
var _                 = require('lodash');
var taskcluster       = require('taskcluster-client');
var GitHubAPI         = require('github');

/** Launch worker */
var launch = async function(profile) {
  debug("Launching with profile: %s", profile);

  // Load configuration
  var cfg = base.config({
    defaults:     require('../config/defaults'),
    profile:      require('../config/' + profile),
    envs: [
      'pulse_username',
      'pulse_password',
      'taskclusterGithub_publishMetaData',
      'taskcluster_credentials_clientId',
      'taskcluster_credentials_accessToken',
      'aws_accessKeyId',
      'aws_secretAccessKey',
      'influx_connectionString',
      'webhook_secret'
    ],
    filename:     'taskcluster-github'
  });

  // Create a single connection to the GithubAPI to pass around
  var githubAPI = new GitHubAPI(cfg.get('github:config'));
  githubAPI.authenticate(cfg.get('github:credentials'));

  // A context to be passed into message handlers
  let context = {cfg, githubAPI};

  // Create a default stats drain, which just prints to stdout
  let statsDrain = {
      addPoint: (...args) => {debug("stats:", args)}
  }

  // Create InfluxDB connection for submitting statistics
  let influxConnectionString = cfg.get('influx:connectionString')
  if (influxConnectionString) {
      statsDrain = new base.stats.Influx({
        connectionString:   influxConnectionString,
        maxDelay:           cfg.get('influx:maxDelay'),
        maxPendingPoints:   cfg.get('influx:maxPendingPoints')
      });
  } else {
      debug("Missing influx_connectionString: stats collection disabled.")
  }

  // Start monitoring the process
  base.stats.startProcessUsageReporting({
    drain:      statsDrain,
    component:  cfg.get('taskclusterGithub:statsComponent'),
    process:    'worker'
  });

  let pulseCredentials = cfg.get('pulse')
  if (pulseCredentials.username && pulseCredentials.password) {
    var pullRequestListener = new taskcluster.PulseListener({
      queueName:  profile,
      credentials: {
        username: pulseCredentials.username,
        password: <PASSWORD>
      }
    });

    let exchangeReference = exchanges.reference({
      exchangePrefix:   cfg.get('taskclusterGithub:exchangePrefix'),
      credentials:      cfg.get('pulse')
    });

    let GitHubEvents = taskcluster.createClient(exchangeReference);
    let githubEvents = new GitHubEvents();
    // Listen for, and handle, WebHook triggered events: i.e. pull requests
    await pullRequestListener.bind(githubEvents.pullRequest(
      {organization: '*', repository: '*', action: 'opened'}));

    // Listen for, and handle, changes in graph/task state: to reset status
    // messages, send notifications, etc....
    let schedulerEvents = new taskcluster.SchedulerEvents();
    let route = 'route.taskcluster-github.*.*.*';
    pullRequestListener.bind(schedulerEvents.taskGraphRunning(route));
    pullRequestListener.bind(schedulerEvents.taskGraphBlocked(route));
    pullRequestListener.bind(schedulerEvents.taskGraphFinished(route));

    pullRequestListener.on('message', function(message) {
      if (message.payload.action == 'opened') {
        worker.pullRequestHandler(message, context);
      } else {
        worker.graphStateChangeHandler(message, context);
      }
    });

    await pullRequestListener.resume();

  } else {
    throw "Missing pulse credentials"
  }
};

// If worker.js is executed start the worker
if (!module.parent) {
  // Find configuration profile
  var profile = process.argv[2];
  if (!profile) {
    console.log("Usage: worker.js [profile]");
    console.error("ERROR: No configuration profile is provided");
  }
  // Launch with given profile
  launch(profile).then(function() {
    debug("Launched worker successfully");
  }).catch(function(err) {
    debug("Failed to start worker, err: %s, as JSON: %j", err, err, err.stack);
    // If we didn't launch the worker we should crash
    process.exit(1);
  });
}
<commit_msg>Throw out non-pull github actions.<commit_after>#!/usr/bin/env node
var debug             = require('debug')('github:worker');
var base              = require('taskcluster-base');
var Promise           = require('promise');
var exchanges         = require('../lib/exchanges');
var worker            = require('../lib/worker');
var _                 = require('lodash');
var taskcluster       = require('taskcluster-client');
var GitHubAPI         = require('github');

/** Launch worker */
var launch = async function(profile) {
  debug("Launching with profile: %s", profile);

  // Load configuration
  var cfg = base.config({
    defaults:     require('../config/defaults'),
    profile:      require('../config/' + profile),
    envs: [
      'pulse_username',
      'pulse_password',
      'taskclusterGithub_publishMetaData',
      'taskcluster_credentials_clientId',
      'taskcluster_credentials_accessToken',
      'aws_accessKeyId',
      'aws_secretAccessKey',
      'influx_connectionString',
      'webhook_secret'
    ],
    filename:     'taskcluster-github'
  });

  // Create a single connection to the GithubAPI to pass around
  var githubAPI = new GitHubAPI(cfg.get('github:config'));
  githubAPI.authenticate(cfg.get('github:credentials'));

  // A context to be passed into message handlers
  let context = {cfg, githubAPI};

  // Create a default stats drain, which just prints to stdout
  let statsDrain = {
      addPoint: (...args) => {debug("stats:", args)}
  }

  // Create InfluxDB connection for submitting statistics
  let influxConnectionString = cfg.get('influx:connectionString')
  if (influxConnectionString) {
      statsDrain = new base.stats.Influx({
        connectionString:   influxConnectionString,
        maxDelay:           cfg.get('influx:maxDelay'),
        maxPendingPoints:   cfg.get('influx:maxPendingPoints')
      });
  } else {
      debug("Missing influx_connectionString: stats collection disabled.")
  }

  // Start monitoring the process
  base.stats.startProcessUsageReporting({
    drain:      statsDrain,
    component:  cfg.get('taskclusterGithub:statsComponent'),
    process:    'worker'
  });

  let pulseCredentials = cfg.get('pulse')
  if (pulseCredentials.username && pulseCredentials.password) {
    var pullRequestListener = new taskcluster.PulseListener({
      queueName:  profile,
      credentials: {
        username: pulseCredentials.username,
        password: <PASSWORD>
      }
    });

    let exchangeReference = exchanges.reference({
      exchangePrefix:   cfg.get('taskclusterGithub:exchangePrefix'),
      credentials:      cfg.get('pulse')
    });

    let GitHubEvents = taskcluster.createClient(exchangeReference);
    let githubEvents = new GitHubEvents();
    // Listen for, and handle, WebHook triggered events: i.e. pull requests
    await pullRequestListener.bind(githubEvents.pullRequest(
      {organization: '*', repository: '*', action: 'opened'}));

    // Listen for, and handle, changes in graph/task state: to reset status
    // messages, send notifications, etc....
    let schedulerEvents = new taskcluster.SchedulerEvents();
    let route = 'route.taskcluster-github.*.*.*';
    pullRequestListener.bind(schedulerEvents.taskGraphRunning(route));
    pullRequestListener.bind(schedulerEvents.taskGraphBlocked(route));
    pullRequestListener.bind(schedulerEvents.taskGraphFinished(route));

    // Route recieved messages to an appropriate handler
    pullRequestListener.on('message', function(message) {
      if (message.payload.action == 'opened') {
        worker.pullRequestHandler(message, context);
      } else if !message.payload.action {
        worker.graphStateChangeHandler(message, context);
      }
    });

    await pullRequestListener.resume();

  } else {
    throw "Missing pulse credentials"
  }
};

// If worker.js is executed start the worker
if (!module.parent) {
  // Find configuration profile
  var profile = process.argv[2];
  if (!profile) {
    console.log("Usage: worker.js [profile]");
    console.error("ERROR: No configuration profile is provided");
  }
  // Launch with given profile
  launch(profile).then(function() {
    debug("Launched worker successfully");
  }).catch(function(err) {
    debug("Failed to start worker, err: %s, as JSON: %j", err, err, err.stack);
    // If we didn't launch the worker we should crash
    process.exit(1);
  });
}

{'fraction_non_alphanumeric': 0.12550420652299182, 'fraction_numerical': 0.00046098882102109026, 'mean_word_length': 3.2769837358304583, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '6665754', 'n_tokens_mistral': 2429, 'n_tokens_neox': 2357, 'n_words': 581}