J'utilise habituellement C # et Java pour le travail, mais j'ai toujours été intéressé par Python. L'analyse de données et l'apprentissage automatique sont si populaires que j'ai décidé de saisir cette opportunité pour étudier Python! !!
Pour Python, récemment publié [Introduction Python3](http://www.amazon.co.jp/%E5%85%A5%E9%96%80-Python-3-Bill-Lubanovic/dp/4873117380/ref = sr_1_1? ie = UTF8 & qid = 1457003044 & sr = 8-1 & keywords =% E5% 85% A5% E9% 96% 80 Python3) J'apprends.
Cette fois, plusieurs personnes l'ont déjà présenté, mais je vais le présenter car j'ai écrit un programme pour enregistrer les résultats de la recherche Twitter dans MongoDB. Je serais très heureux si vous pouviez approfondir d'autres choses comme celle-ci! !!
Définissez les paramètres qui conviennent à votre environnement.
config.py
# coding=utf-8
# write code...
# mongodb
HOST = 'localhost'
PORT = 27017
DB_NAME = 'twitter-archive'
COLLECTION_NAME = 'tweets'
# twitter
CONSUMER_KEY = ''
CONSUMER_SECRET = ''
ACCESS_TOKEN_KEY = ''
ACCESS_TOKEN_SECRET = ''
J'ai décidé de gérer les mots-clés spécifiés lors de la recherche Twitter dans un fichier YAML.
keywords.yml
#Définissez les mots-clés de recherche Twitter sous forme de liste.
#Ce qui suit est un exemple.
- 'Hamburger'
- 'base-ball'
- 'Noël'
J'ai créé une classe wrapper tout en étudiant comment utiliser la journalisation. Il y a beaucoup de choses que je ne comprends pas encore, donc j'étudie les paramètres détaillés, mais j'ai confirmé que je pouvais générer des journaux.
logger.py
import logging
from logging.handlers import TimedRotatingFileHandler
# coding=utf-8
# write code...
class Logger:
def __init__(self, log_type):
logger = logging.getLogger(log_type)
logger.setLevel(logging.DEBUG)
#Je veux tourner tous les jours, mais je ne l'ai pas encore fait. .. ..
handler = TimedRotatingFileHandler(filename='archive.log', when='D', backupCount=30)
formatter = logging.Formatter('[%(asctime)s] %(name)s %(levelname)s %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
self.logger = logger
def info(self, msg, *args, **kwargs):
self.logger.info(msg, *args, **kwargs)
def debug(self, msg, *args, **kwargs):
self.logger.debug(msg, *args, **kwargs)
def error(self, msg, *args, **kwargs):
self.logger.error(msg, *args, **kwargs)
def exception(self, msg, *args, exc_info=True, **kwargs):
self.logger.exception(msg, *args, exc_info, **kwargs)
Je pense commencer un lot une fois par semaine et accumuler régulièrement des tweets. J'ai eu plus de mal à comprendre les spécifications de l'API Twitter que ce à quoi je m'attendais. J'ai réfléchi à la façon de contrôler l'utilisation de since_id et max_id pour ne pas avoir de tweets en double. Comment était-ce bon de le faire? .. ..
archive.py
import sys
import config
import yaml
from tweepy import *
from tweepy.parsers import JSONParser
from pymongo import *
from logger import Logger
# coding: UTF-8
# write code...
def archive():
#Lisez la liste des mots-clés de recherche dans le fichier YAML et générez une chaîne pour la recherche OR.
with open('keywords.yml', 'r') as file:
keywords = yaml.load(file)
query_string = ' OR '.join(keywords)
#Initialisation de l'objet de sortie de journal
logger = Logger('archive')
#Générer un client pour la recherche Twitter
auth = OAuthHandler(config.CONSUMER_KEY, config.CONSUMER_SECRET)
auth.set_access_token(config.ACCESS_TOKEN_KEY, config.ACCESS_TOKEN_SECRET)
#Je veux recevoir le résultat dans JSON, alors définissez JSONParser.
#Même si la limite de recherche est atteinte, la bibliothèque fera de son mieux. Devrait être.
twitter_client = API(auth, parser=JSONParser(), wait_on_rate_limit=True, wait_on_rate_limit_notify=True)
if twitter_client is None:
logger.error('la certification a échoué.')
sys.exit(-1)
#Initialisez la collection mongodb pour stocker les tweets
client = MongoClient(config.HOST, config.PORT)
tweet_collection = client[config.DB_NAME][config.COLLECTION_NAME]
#Obtenez les derniers tweets des tweets acquis et définissez-le pour obtenir l'identifiant et plus tard des tweets.
last_tweet = tweet_collection.find_one(sort=[('id', DESCENDING)])
since_id = None if last_tweet is None else last_tweet['id']
#Lors de la première recherche, max_Ne pas définir d'identifiant-Réglez 1.
max_id = -1
# tweet_le nombre est max_tweet_Lorsque le nombre est atteint, la recherche se termine.
# max_tweet_Définissez une valeur élevée pour count.
tweet_count = 0
max_tweet_count = 100000
logger.info('maximum{0}Recueillez des tweets individuels.'.format(max_tweet_count))
while tweet_count < max_tweet_count:
try:
params = {
'q': query_string,
'count': 100,
'lang': 'ja'
}
# max_id et depuis_Ne transmettez id comme paramètre que s'il est défini.
if max_id > 0:
params['max_id'] = str(max_id - 1)
if since_id is not None:
params['since_id'] = since_id
search_result = twitter_client.search(**params)
statuses = search_result['statuses']
#Vérifiez si vous pouvez rechercher jusqu'au bout
if statuses is None or len(statuses) == 0:
logger.info('Le tweet n'a pas été trouvé.')
break
tweet_count += len(statuses)
logger.debug('{0}J'ai les tweets.'.format(tweet_count))
result = tweet_collection.insert_many([status for status in statuses])
logger.debug('Je l'ai sauvegardé dans MongoDB. L'ID est{0}est.'.format(result))
#Mettez à jour avec le dernier ID de Tweet que vous avez obtenu.
max_id = statuses[-1]['id']
except (TypeError, TweepError) as e:
print(str(e))
logger.exception(str(e))
break
if __name__ == '__main__':
archive()
Je n'ai pas encore maîtrisé Python du tout, mais je pensais que c'était un langage dans lequel je pourrais écrire ce que je voulais faire. Je vais continuer à étudier. À l'avenir, j'essaierai d'analyser les tweets collectés en utilisant la bibliothèque pour l'analyse des données! !!
Recommended Posts