streamer.js

gistfile1.txt

var Twit = require('twit');
var util = require('util');
var sqlite3 = require('sqlite3').verbose();
var $ = require('jquery');

// Prototype Methods
String.prototype.double_quotes = function()
{
return this.replace(/[\\"]/g, '""');
}

var Streamer = {};

Streamer.Process = function(db_name, keywords, TwitObj)
  {
  var self = this;
  this.db = null; 
  
  this.init = function()
    {
    self.stream = TwitObj.stream('statuses/filter', { track: keywords });
    
    // load databases
    self.db = new sqlite3.Database(db_name);
    self.db_integrity();
    
    // fetch incoming tweets
    self.stream.on('tweet', function (tweet) {
      process.store_tweet(tweet);
      });
    
    console.log("Erfolgreich");
    }
    
  
    
  this.db_integrity = function()
    {
    self.db.run("CREATE TABLE IF NOT EXISTS Tweet (Id INTEGER PRIMARY KEY, favorited INTEGER, truncated INTEGER, created_at VARCHAR(128), text VARCHAR(512), retweet_count INTEGER, retweeted INTEGER, in_reply_to_status_id INTEGER, in_reply_to_user_id INTEGER, user_id INTEGER, FOREIGN KEY(user_id) REFERENCES User(Id) )");
 
  self.db.run("CREATE TABLE IF NOT EXISTS User (Id INTEGER PRIMARY KEY, name VARCHAR(128), profile_image_url VARCHAR(128), created_at VARCHAR(128), location VARCHAR(128), followers_count INTEGER, favourites_count INTEGER, lang VARCHAR(2), listed_count INTEGER, description VARCHAR(512), statuses_count INTEGER, friends_count INTEGER, protected INTEGER, verified INTEGER, admin_unit VARCHAR(64), admin_unit2 VARCHAR(64), admin_unit3 VARCHAR(64)), email VARCHAR(256)");
    }    
    
  this.cv_boolean = function(val)
    {
    if (val == false) return 0;
    else return 1;
    }
    
  this.store_tweet = function(item)
  {
     
    var p = item;
    
    // fetches location and then calls function to store user
    self.get_user_admin_unit(p.user);
    
    
    var tweet = {};
    
    tweet.Id = p.id;
    tweet.favorited = self.cv_boolean(p.favorited);
    tweet.truncated = self.cv_boolean(p.truncated);
    tweet.created_at = p.created_at;
    tweet.in_reply_to_user_id = p.in_reply_to_user_id;
    tweet.text = p.text.double_quotes();

    tweet.retweet_count = p.retweet_count;
    tweet.tweet_id = p.id;
    tweet.retweeted = self.cv_boolean(p.retweeted);
 
    tweet.in_reply_to_status_id = p.in_reply_to_status_id;
 
    tweet.user_id = p.user.id;
   
     var stm = 'INSERT OR IGNORE INTO Tweet(Id, favorited, truncated, created_at, in_reply_to_user_id, text, retweet_count, retweeted, in_reply_to_user_id, user_id) VALUES(' + tweet.Id + ',' + tweet.favorited + ',' + tweet.truncated + ',"' + tweet.created_at + '",' + tweet.in_reply_to_user_id + ',"' + tweet.text + '",' + tweet.retweet_count + ',' + tweet.retweeted + ',' + tweet.in_reply_to_user_id + ',' + tweet.user_id + ')';
     
    // console.log(stm);

    self.db.run(stm);
    
  }  
  
  
this.get_user_admin_unit = function(user)
  {
  var s = 'http://api.geonames.org/postalCodeSearchJSON?placename=' +
  user.location + '&username=planeshifter';
  user.admin_unit ="";
  
      $.getJSON(
      s,
      function(data){  
      
      if (data.postalCodes)
        {
        if (data.postalCodes[0] != null)
          {
            
          var admin_unit = data.postalCodes[0].adminName1;
          var admin_unit2 = data.postalCodes[0].adminName2;
          var admin_unit3 = data.postalCodes[0].adminName3;
          
          user.admin_unit = admin_unit;
          if(admin_unit2 === undefined) user.admin_unit2 = " ";
            else user.admin_unit2 = admin_unit2;
          if(admin_unit3 === undefined) user.admin_unit3 = " ";
            else user.admin_unit3 = admin_unit3;
          }
  
        }
        
      self.store_user(user);
  });
  
  }
  
this.store_user = function(user)
{
if (user.id)
  {
  var guy = {};
  guy.id = user.id;
  guy.name = user.name;
  guy.profile_image_url = user.profile_image_url;
  guy.created_at = user.created_at;
  guy.location = user.location;
  guy.followers_count = user.followers_count;
  guy.favourites_count = user.favourites_count;
  guy.lang = user.lang;
  guy.listed_count = user.listed_count;
  guy.description = user.description;
  guy.statuses_count = user.statuses_count;
  guy.friends_count = user.friends_count;
  guy.protected = self.cv_boolean(user.proteted);
  guy.verified = self.cv_boolean(user.verified);
  
  guy.admin_unit = user.admin_unit;
  guy.admin_unit2 = user.admin_unit2;
  guy.admin_unit3 = user.admin_unit3;
  
    
  var stm = 'INSERT OR IGNORE INTO User(Id, name, profile_image_url, created_at, location, followers_count, favourites_count, lang, listed_count, description, statuses_count, friends_count, protected, verified, admin_unit, admin_unit2, admin_unit3) VALUES(' + guy.id +',"' + guy.name + '","' + guy.profile_image_url + '","' + guy.created_at + '","' + guy.location + '",' + guy.followers_count + ',' + guy.favourites_count + ',"' + guy.lang + '",' + guy.listed_count + ',"' + guy.description + '",' + guy.statuses_count + ',' + guy.friends_count + ',' + guy.protected + ',' + guy.verified + ',"' + guy.admin_unit + '","' + guy.admin_unit2 + '","' + guy.admin_unit3 + '")';
  
  // console.log(stm);
  
  self.db.run(stm);
  }
}


  
  
  self.init();
  }
  
  
  
var T = new Twit({
    consumer_key:         'AiMqJjr09o5jN0jPq4VViA'
  , consumer_secret:      'njgULZTuPBRjCkogYQS1lyDRqllUoFt4LyentZ1u7zo'
  , access_token:         '206795751-lQUs1Lck75BjvbmkoYhZ7Vkafpw9j7lQHoOGgsNi'
  , access_token_secret:  'Ojh1P00DBi2QDojEa8lvavPdOpvIaHhxbk4o9Npw'
})  

// Parameters of Streamer.Process:
// 1. db_name : name of the database - here streamer.db
// 2. keywords : array of keywords - here ["Obama","Merkel"]
// 3. Twit object (contains Twitter credentials) - here T (only one open stream  
// per account)
var process = new Streamer.Process("streamer.db",["Obama","Merkel"],T);