Commit b88e7d95 authored by edne's avatar edne
Browse files

Retrieving old material

parents
__pycache__
*.json
from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream
access_token = "XXXXXXX"
access_token_secret = "XXXXXXX"
consumer_key = "XXXXXXX"
consumer_secret = "XXXXXXX"
from secret import access_token, access_token_secret
from secret import consumer_key, consumer_secret
class StdOutListener(StreamListener):
def on_data(self, data):
print(data)
return True
def on_error(self, status):
print(status)
if __name__ == '__main__':
l = StdOutListener()
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
stream = Stream(auth, l)
stream.filter(track=['python', 'javascript', 'ruby'])
# Politecnico Open unix Labs (https://poul.org)
# Edoardo Negri (http://edne.net)
from tweepy import OAuthHandler
from tweepy.streaming import StreamListener
from tweepy import Stream
import json
# per poter usare le API di Twitter è necessario registare l'app
# https://apps.twitter.com/
access_token = "XXXXXXX"
access_token_secret = "XXXXXXX"
consumer_key = "XXXXXXX"
consumer_secret = "XXXXXXX"
# autenticazione
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
class Listener(StreamListener):
"""Listener che salva sul file tweets.json"""
def on_data(self, json_data):
"""funzione chiamata all'arrivo di un tweet"""
with open("tweets.json", "a") as f:
f.write(json_data)
# print(json_data)
def on_error(self, status):
"""funzione chiamata in caso di errore"""
print(status)
listener = Listener()
stream = Stream(auth, listener)
# decommentare per far partire lo stream
# stream.filter(track=["python"])
# ctrl-C per uccidere il processo
# Apro il file con i tweet salvati
with open("tweets.json") as f:
tweets_json = f.readlines() # lista di stringhe
def load(s):
"""funione per convertire una stringa (json) in un dizionario"""
try:
return json.loads(s)
except ValueError:
# se la conversione non riesce ritorna None
return None
# applico load all'intera lista
tweets_data = [load(tw_j) for tw_j in tweets_json]
# rimuovo gli elementi che sono None
# (l'operatore if si aspetta una variabile di tipo logico, i valori 0, [] e
# None vengono convertiti a False, tutto il resto a True)
tweets_data = [tw for tw in tweets_data if tw]
#
# Disegno il grafico a torta lingue e numero tweet
#
from collections import Counter
from matplotlib import pyplot as plt # http://matplotlib.org/
# lista linguaggi (ripetuti)
langs = [tw["lang"] for tw in tweets_data]
# dizionario {lingua: n_occorrenze}
counted = Counter(langs)
# dizionario con solo le lingue con più di 100 tweet
top = {k: v for (k, v) in counted.items() if v > 100}
# lista con i numeri dei tweet nelle antre lingue
others = [v for (k, v) in counted.items() if v <= 100]
# aggiungo il campo "others" al dizionario
others_sum = sum(others)
top["others"] = others_sum
sizes = list(top.values())
labels = list(top.keys())
# http://matplotlib.org/examples/pie_and_polar_charts/pie_demo_features.html
plt.pie(sizes, labels=labels)
plt.show(block=True)
#!/usr/bin/env python3
##
# per poter usare le API di Twitter è necessario registare l'app
# https://apps.twitter.com/
access_token = "XXXXXXX"
access_token_secret = "XXXXXXX"
consumer_key = "XXXXXXX"
consumer_secret = "XXXXXXX"
##
##
# salvati in un moulo esterno
from secret import access_token, access_token_secret
from secret import consumer_key, consumer_secret
##
##
# mi autentico
from tweepy import OAuthHandler
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
##
##
from tweepy.streaming import StreamListener
from tweepy import Stream
import json
# classe per "ascoltare" i tweets
# http://docs.tweepy.org/en/latest/streaming_how_to.html
class Listener(StreamListener):
def on_data(self, json_data):
# stringa contenente json
# print(json_data)
data = json.loads(json_data)
print(data["text"])
return True
def on_error(self, status):
print(status)
listener = Listener()
stream = Stream(auth, listener)
##
##
# posso filtrare in base a parole contenute
stream.filter(track=['python'])
##
##
# salviamo il .json in un file
class SavingListener(StreamListener):
def __init__(self, filename):
self.filename = filename
def on_data(self, json_data):
with open(self.filename, "a") as f:
f.write(json_data)
print(json_data)
return True
def on_error(self, status):
print(status)
listener = SavingListener("test.json")
stream = Stream(auth, listener)
stream.filter(track=['python'])
##
##
# leggiamo il .json
with open("test.json") as f:
lines = f.readlines()
print(len(lines), "tweets")
tweets = [json.loads(line) for line in lines]
# ma esplode, dobbiamo gestire il casi di tweet invalido:
def load(json_string):
try:
return json.loads(json_string)
except ValueError:
return None
#
# parentesi sulle comprehension
#
tweets = [load(line) for line in lines]
tweets = [tw for tw in tweets if tw]
for tw in tweets:
if tw:
print(tw["text"])
##
##
# solo le lingue dei tweet
print([tw["lang"] for tw in tweets])
from collections import Counter
# Counter restituisce un dizionario {"linguaggio": n_apparizioni}
# il metodo .items() di un dizionario da una lista di coppie (chiave, valore)
couples = Counter([tw["lang"] for tw in tweets]).items()
keys = [cp[0] for cp in couples]
values = [cp[1] for cp in couples]
# queste due comprehension possono essere sostituite da:
keys, value = zip(*couples)
# ma è qualcosa di avanzato...
import matplotlib.pyplot as plt
plt.bar(range(len(keys)), values, align='center')
plt.xticks(range(len(keys)), keys)
plt.show(block=True)
##
##
# visto come generare il grafico di un attrubuto dei tweet, possiamo fare una
# funzione per farne uno qualsiasi
def plot(tweets, key):
couples = Counter([tw[key] for tw in tweets]).items()
keys = [cp[0] for cp in couples]
values = [cp[1] for cp in couples]
plt.bar(range(len(keys)), values, align='center')
plt.xticks(range(len(keys)), keys)
plt.show(block=True)
print(tweets[0].keys())
plot(tweets, "source")
##
##
# si possono fare altre analisi, come per esempio la massima sorgente dei tweet
counted = Counter([tw["source"] for tw in tweets])
src_max = max(counted, key=counted.get)
print(src_max, counted[src_max])
##
##
# altri esempi, se avanza tempo
##
consumer_key = 'xxxxxxxxxxxxxxxxxxxxxxxxx'
consumer_secret = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
access_token = 'xxxxxxxxxx-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
access_token_secret = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
#!/usr/bin/env python
from secret import access_token, access_token_secret
from secret import consumer_key, consumer_secret
from tweepy import OAuthHandler, Stream
from tweepy.streaming import StreamListener
import json
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
class SavingListener(StreamListener):
def __init__(self, filename):
self.filename = filename
def on_data(self, data_json):
with open(self.filename, "a") as f:
f.write(data_json)
print(data_json)
def on_error(self, status):
print(status)
listener = SavingListener("out.json")
stream = Stream(auth, listener)
# stream.filter(track=["python"])
with open("out.json") as f:
lines = f.readlines()
# print(json.loads(lines[0]))
def load(s):
try:
return json.loads(s)
except ValueError:
print(s)
return None
tweets = [load(l) for l in lines]
tweets = [tw for tw in tweets if tw is not None]
print(len(tweets), "tweets")
langs = [tw["lang"] for tw in tweets]
# print(langs)
from collections import Counter
d = Counter(langs)
top = {lang: n for lang, n in d.items() if n > 100}
others_n = [n for n in d.values() if n <= 100]
top["others"] = sum(others_n)
from matplotlib import pyplot as plt
labels = list(top.keys())
sizes = list(top.values())
# print(labels)
# print(sizes)
# plt.pie(sizes, labels=labels)
# plt.show(block=True)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment