Voir @ tyokuyoku's "J'ai créé un script qui renvoie aux tweets d'un utilisateur spécifique sur Twitter et enregistre les images publiées à la fois" Ce n'est pas fonctionnalisé et l'emboîtement est assez profond, donc je pense qu'il peut être amélioré. Je pensais que c'était refactorisé. J'ai commenté le résultat du refactoring, mais je vais l'expliquer car il a été revu plus en détail. Je suis moi-même en train de refactoriser par essais et erreurs, donc si vous avez d'autres bonnes idées, je vous serais reconnaissant si vous pouviez commenter.
Tout d'abord, faites de la partie la plus profonde du nid une fonction. Le contenu de traitement "enregistre-t-il les données d'image trouvées dans un fichier à une extrémité"? Puisqu'il s'agit d'un processus d'analyse, le nom de la fonction est crawl.
def crawl():
for image, filename in images():
path = os.path.join(SAVE_DIRECTORY, filename)
with open(path, 'wb') as localfile:
localfile.write(image)
Je n'ai pas encore de fonction d'images, mais s'il existe une fonction qui récupère (répertorie) les données d'image et les noms de fichiers les uns après les autres, c'est juste un travail pour les enregistrer dans un fichier. Maintenant, créons une fonction d'images qui fait cela. Tout d'abord, implémentez uniquement le processus de récupération des données d'image.
def images():
for url in image_url():
image = urllib.urlopen(url)
yield image.read()
image.close()
Il n'y a pas encore de fonction image_url, mais s'il existe une fonction qui énumère l'url des données d'image, c'est juste un travail pour lire les données d'image à la destination de l'url et les produire. S'il s'agit d'un retour, le résultat sera renvoyé une fois et le processus se terminera, mais en utilisant une fonction de générateur qui utilise yield, les résultats peuvent être notifiés les uns après les autres. Ajoutez un processus à cette fonction qui notifie également le nom du fichier. Il s'écarte du principe de responsabilité unique du principe SOLID, mais ...
def images():
for urls, timestamp in image_urls():
date_time = timestamp.strftime("%Y%m%d_%H%M%S")
for index, url in enumerate(urls):
_, ext = os.path.splitext(url)
filename = '%s_%s_%s%s' % (USER_NAME, date_time, index, ext)
image = urllib.urlopen(url)
yield image.read(), filename
image.close()
Le nom du fichier est créé à partir de l'heure du tweet, mais comme l'image n'inclut pas l'heure du tweet, je vais laisser la fonction image_urls me le dire.
Étant donné que plusieurs images peuvent être collées sur un tweet, une seule fois sera liée à plusieurs images. Veuillez me dire l'heure du tweet avec le standard python datetime et le convertir en année / mois / jour_heure / minute / seconde utilisé pour le nom du fichier. À
L'extension du nom de fichier a été fixée à "" .jpg "dans le programme d'origine, mais il y a aussi" .png "` etc., donc je vais retirer l'extension incluse dans l'url et l'utiliser.
Vient ensuite la fonction image_urls.
def image_urls():
for media, timestamp in medias():
urls = [content["media_url_https"]
for content in media
if content.get("type") == "photo"]
yield urls, timestamp
Le tweet contient des informations multimédias telles que des images accompagnées d'une courte phrase (texte), et s'il existe une fonction médias qui énumère les informations multimédias, il est possible d'extraire uniquement l'URL d'image de type «photo» des informations multimédias et de la notifier. Je vais. Puisque les informations médiatiques n'incluent pas l'heure du tweet, nous demanderons à la fonction média de nous indiquer également l'heure du tweet. Vient ensuite la fonction medias.
def medias():
for tweet in tweets():
created_at = tweet["created_at"]
timestamp = dateutil.parser.parse(created_at)
extended_entities = tweet.get("extended_entities", {})
media = extended_entities.get("media", ())
yield media, timestamp
S'il existe une fonction tweets qui énumère les tweets dans l'ordre, vous pouvez récupérer l'heure et les médias du tweet et les notifier. Il est possible qu'il n'y ait pas d'informations sur les entrées_étendues ou sur les médias, mais s'il n'y a pas d'informations, un taple vide est notifié pour vous faire savoir qu'il n'y a pas d'image. Enfin, la fonction tweets.
USER_NAME = 'Nom d'utilisateur que vous souhaitez obtenir'
NUM_OF_TWEET_AT_ONCE = 200 # max: 200
NUM_OF_TIMES_TO_CRAWL = 16 # max: 3200 / NUM_OF_TWEET_AT_ONCE
SAVE_DIRECTORY = os.path.join('.', 'images')
def tweets():
import twitkey
twitter = OAuth1Session(twitkey.CONSUMER_KEY,
twitkey.CONSUMER_SECRET,
twitkey.ACCESS_TOKEN,
twitkey.ACCESS_TOKEN_SECRET)
url = ("https://api.twitter.com/1.1/statuses/user_timeline.json"
"?screen_name=%s&include_rts=false" % USER_NAME)
params = {"count": NUM_OF_TWEET_AT_ONCE}
for i in range(NUM_OF_TIMES_TO_CRAWL):
req = twitter.get(url, params=params)
if req.status_code != requests.codes.ok:
return
timeline = json.loads(req.text)
for tweet in timeline:
yield tweet
params["max_id"] = tweet["id"]
Vous devez utiliser l'API Twitter pour obtenir des tweets, et vous devez vous inscrire en tant qu'utilisateur avec Twitter à l'avance pour obtenir la clé et le jeton pour accéder à l'API Twitter. Écrivez ces informations sous forme de variable dans un fichier appelé twitkey.py. La source du programme et les informations confidentielles sont séparées. Dans le script d'origine, il s'agit d'une variable de dictionnaire, mais vous pouvez également simplement affecter une variable.
twitkey.py
#coding: UTF-8
CONSUMER_KEY = ""
CONSUMER_SECRET = ""
ACCESS_TOKEN = ""
ACCESS_TOKEN_SECRET = ""
Après tout, l'imbrication a été approfondie dans l'instruction for, mais elle a été rendue moins profonde en en faisant une fonction de générateur utilisant le rendement. En séparant les fonctions, le nom de la fonction devient également un commentaire qui indique le contenu du traitement, et je pense qu'il sera plus facile de comprendre le traitement.
Enfin, je publierai l'intégralité du script, y compris le processus principal, l'importation et l'impression d'affichage de la progression.
#!/usr/bin/env python2
# -*- coding:utf-8 -*-
import sys
import os.path
import dateutil.parser
import urllib
import requests
import json
from requests_oauthlib import OAuth1Session
USER_NAME = 'Nom d'utilisateur que vous souhaitez obtenir'
NUM_OF_TWEET_AT_ONCE = 200 # max: 200
NUM_OF_TIMES_TO_CRAWL = 16 # max: 3200 / NUM_OF_TWEET_AT_ONCE
SAVE_DIRECTORY = os.path.join('.', 'images')
def tweets():
import twitkey
twitter = OAuth1Session(twitkey.CONSUMER_KEY,
twitkey.CONSUMER_SECRET,
twitkey.ACCESS_TOKEN,
twitkey.ACCESS_TOKEN_SECRET)
url = ("https://api.twitter.com/1.1/statuses/user_timeline.json"
"?screen_name=%s&include_rts=false" % USER_NAME)
params = {"count": NUM_OF_TWEET_AT_ONCE}
for i in range(NUM_OF_TIMES_TO_CRAWL):
req = twitter.get(url, params=params)
if req.status_code != requests.codes.ok:
print "ERROR:", req.status_code
return
timeline = json.loads(req.text)
for tweet in timeline:
print "TWEET:", tweet["text"]
yield tweet
params["max_id"] = tweet["id"]
def medias():
for tweet in tweets():
created_at = tweet["created_at"]
timestamp = dateutil.parser.parse(created_at)
extended_entities = tweet.get("extended_entities", {})
media = extended_entities.get("media", ())
print "CREATE:", created_at
yield media, timestamp
def image_urls():
for media, timestamp in medias():
urls = [content["media_url_https"]
for content in media
if content.get("type") == "photo"]
print "IMAGE:", len(urls)
yield urls, timestamp
def images():
for urls, timestamp in image_urls():
date_time = timestamp.strftime("%Y%m%d_%H%M%S")
for index, url in enumerate(urls):
_, ext = os.path.splitext(url)
filename = '%s_%s_%s%s' % (USER_NAME, date_time, index, ext)
image = urllib.urlopen(url)
print "URL:", url
yield image.read(), filename
image.close()
def crawl():
for image, filename in images():
path = os.path.join(SAVE_DIRECTORY, filename)
print "SAVE:", path
with open(path, 'wb') as localfile:
localfile.write(image)
if __name__ == '__main__':
if len(sys.argv) > 1:
USER_NAME = sys.argv[1]
crawl()
Recommended Posts