Don’t be a lonely document

input {
 twitter {
  consumer_key => "foo"
  consumer_secret => "bar"
  oauth_token => "baz"
  oauth_token_secret => "qux"
  keywords => ["graphconnect", "neo4j", "GraphConnect"]
 }
}
output {
  kafka {
 codec => plain {
    format => "%{message}"
 }
 topic_id => "tweets"
  }
}

import tweepy
import time
import json

ckey = 'foo'
csecret = 'bar'
atoken = 'baz'
asecret = 'qux'

OAUTH_KEYS = {'consumer_key': ckey, 'consumer_secret':csecret,
 'access_token_key':atoken, 'access_token_secret':asecret}
auth = tweepy.OAuthHandler(OAUTH_KEYS['consumer_key'], OAUTH_KEYS['consumer_secret'])
api = tweepy.API(auth)

def limit_handled(cursor):
 while True:
  try:
   yield cursor.next()
  except tweepy.TweepError as e:
   print(e.error_msg)
   time.sleep(15 * 60)

def search(keyword):
 # Extract the first "xxx" tweets related to "fast car"
 with open('tweets_friday.json', 'a') as the_file:
  for tweet in limit_handled(tweepy.Cursor(api.search, q=keyword, since='2017-05-09').items()):
   the_file.write(json.dumps(tweet._json) + '\n')

import json
from neo4j.v1 import GraphDatabase
import time

def store_tweet(tx, tweet):
 neo4j_params = {"user_id": tweet['user']['id'],
     "user_name": tweet['user']['name'],
     "tweet_id": tweet['id'],
     "tweet_text": tweet['text'],
     "tweet_time": time.strftime('%Y-%m-%d %H:%M:%S', time.strptime(tweet['created_at'],'%a %b %d %H:%M:%S +0000 %Y')),
     "mentions": tweet['entities']['user_mentions']
       }
 tx.run("""
   MERGE (u:User {uid: $user_id})
     on create set u.name = $user_name
   MERGE (t:Tweet {uid: $tweet_id})
     on create set t.text = $tweet_text, t.time = $tweet_time
   MERGE (u)-[:TWEETS]->(t)
   WITH t, $mentions as mentions
   unwind mentions as mention
   MERGE (u:User {uid: mention.id}) on create set u.name = mention.name
   MERGE (t)-[:MENTIONS]->(u)
   """, neo4j_params)

def process_file(file_name):
 with GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "test")) as driver:
  with open(file_name, 'r') as the_file:
   with driver.session() as session:
    with session.begin_transaction() as tx:
     for line in the_file:
      tweet = json.loads(line)
      store_tweet(tx, tweet)

      if 'retweeted_status' in tweet:
       store_tweet(tx, tweet['retweeted_status'])
       retweet_data = {
        'tweet_id': tweet['id'],
        "retweet_id": tweet['retweeted_status']['id']
       }

       tx.run("""
       MATCH (t:Tweet {uid: $tweet_id})
       MATCH (r:Tweet {uid: $retweet_id})
       MERGE (t)-[:RE_TWEETS]->(r)
       """, retweet_data)

process_file('tweets_friday2.json')

MATCH p=()-[r:MENTIONS]->() RETURN p LIMIT 50

MATCH (t:Tweet)-[r:MENTIONS]->(mentioned:User)
RETURN mentioned.name, count(r) as numberOfMentions
order by numberOfMentions desc
limit 10

MATCH (u:User)-[:TWEETS]->(t:Tweet)
RETURN u.name as user, count(t) as numberOfTweets
order by numberOfTweets desc
limit 10

MATCH (u:User)-[:TWEETS]->(t:Tweet)
where u.name = "Hakaishin Hokutosei"
RETURN t.text
limit 100

MATCH (u:User)-[:TWEETS]->(t:Tweet)
where u.name = "Hakaishin Hokutosei"
and not (t)-[:RE_TWEETS]->()
RETURN count(t)

MATCH (u)-[r1:TWEETS]->(t)
where not (t)-[:RE_TWEETS]->()
optional match (u)-[:TWEETS]->(rt)-[r2:RE_TWEETS]->()
RETURN u.name, count(distinct r2) as numberOfReTweets, count(distinct r1) as numberOfTweets
order by numberOfTweets desc

MATCH (rt)-[r2:RE_TWEETS]->(t)[:TWEETS]-(u)
RETURN u.name AS user, t.text, count(rt) AS numberOfRetweets
ORDER BY numberOfRetweets DESC

MATCH (rt)-[r2:RE_TWEETS]->(t)[:TWEETS]-(u)
RETURN u.name as user, count(rt) as numberOfRetweets
order by numberOfRetweets desc

Rik van Bruggen