Ce que tu peux apprendre
Ce que tu ne peux pas apprendre
twiiter.py
import tweepy
from collections import deque
from threading import Thread
#Préparation minimale avant d'utiliser tweepy
CONSUMER_KEY='XXXXXXX'
CONSUMER_SECRET='XXXXXXXX'
ACCESS_TOKEN='XXXXXXXX'
ACCESS_SECRET='XXXXXXXX'
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET)
#Jusque là
class Format(object):
"""Class to provide with a variety of formats."""
def __init__(self, tweet):
"""Store some data of tweets in a hash-table. """
self.status = {
'text': tweet.text,
'author': tweet.author.screen_name,
'reply_name': tweet.in_reply_to_screen_name,
'created_at': tweet.created_at,
}
class StreamingFormat(Format):
def format(self):
"""Format tweet text readable with some information."""
author = self.status['author']
created_at = self.status['created_at']
string = '@\033[33m{}\033[0m:{}\n'.format(author, created_at)
string += self.status['text']
string += '\n'
return string
class Memory(deque):
"""Class to have hashed tweet contents."""
def __init__(self, capacity):
"""Define max capacity of self."""
super().__init__()
self.capacity = capacity
def append(self, item):
"""Append unless over capacity."""
if len(self) > self.capacity:
self.popleft
super().append(item)
class Producer(Thread):
"""Class to get tweets and avoid duplication. """
def __init__(self, queue, api, query):
""" Initialize memory to avoid duplication. """
super().__init__(target=self.main)
self.api = api
self.queue = queue
self.query = query
self.memory = Memory(1000)
def main(self):
""" Thread to add tweets to queue. """
while True:
for tweet in self.twigen():
if tweet is not None:
self.queue.put(tweet)
"""
https://syncer.jp/what-is-twitter-api-limit
Selon le site ci-dessus, Tweet est pris sur Twitter
Il y a une limite de profit, et si vous accédez à plus que cela,
Vous devez être prudent car il sera limité.
15 minutes pour rechercher dans tous les Tweets
Il y a une limite de 180 fois, donc une fois toutes les 5 secondes
Le rythme est maximum. Vous n'êtes pas obligé d'attaquer à la dernière minute, alors
Réglez-le toutes les 10 secondes environ.
"""
time.sleep(10)
def twigen(self):
"""Yield a tweet after hashing the tweet and memorizing it"""
import hashlib
tweets = self.api.search(q=self.query, count=100)
for tweet in tweets:
hashing = hashlib.md5(tweet.text.encode('utf-8'))
hashed_tweet = hashing.hexdigest()
if hashed_tweet in self.memory:
yield None
else:
self.memory.append(hashed_tweet)
yield tweet
class Consumer(Thread):
def __init__(self, queue):
super().__init__(target=self.main)
self.queue = queue
def main(self):
"""Take tweet from queue and print it."""
while True:
q = self.queue.get()
f = StreamingFormat(q)
print(f.format())
time.sleep(1)
class TwitterStreamer(object):
"""Class to print formatted tweet data on console."""
def __init__(self, api, query=None):
from queue import Queue
queue = Queue()
Producer(queue, api, query).start()
Consumer(queue).start()
if __name__ == '__main__':
opts = {
"api":tweepy.API(auth),
"query":'Plus tard'
}
TwitterStreamer(**opts)
La méthode d'acquisition de CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_SECRET est omise. Préparez-le car c'est absolument nécessaire
@decimal1010:2016-04-02 10:36:36
Mitsuanaguma est-il un ratel?
@anaguma86:2016-04-02 09:53:06
Un ratel qui s'enthousiasme pour la notation des dauphins à main au lieu des dauphins de bande.
@kyuurin525:2016-04-02 09:28:04
J'ai ouvert quelque chose et suis devenu captif de Ratel à la 5ème page
@tsube_kii:2016-04-02 08:05:31
Je ne savais pas quel genre d'animal était avec Kion dans Lion Guard, mais Ratel est un animal ...
Lorsqu'il est exécuté, il continue à être envoyé à la console comme ceci.
Recommended Posts