Siehe @ tyokuyokus "Ich habe ein Skript erstellt, das auf die Tweets eines bestimmten Benutzers auf Twitter zurückgeht und die veröffentlichten Bilder sofort speichert" Es ist nicht funktionalisiert und die Verschachtelung ist ziemlich tief, daher denke ich, dass es verbessert werden kann. Ich dachte, dass es überarbeitet wurde. Ich habe das Refactoring-Ergebnis kommentiert, werde es aber erläutern, da es weiter überprüft wurde. Ich selbst überarbeite durch Versuch und Irrtum. Wenn Sie also andere gute Ideen haben, würde ich mich freuen, wenn Sie einen Kommentar abgeben könnten.
Machen Sie zuerst den tiefsten Teil des Nestes zu einer Funktion. Ist der Verarbeitungsinhalt "Speichern der gefundenen Bilddaten in einer Datei von einem Ende"? Da es sich um einen Crawling-Prozess handelt, lautet der Funktionsname Crawl.
def crawl():
for image, filename in images():
path = os.path.join(SAVE_DIRECTORY, filename)
with open(path, 'wb') as localfile:
localfile.write(image)
Ich habe noch keine Bildfunktion, aber wenn es eine Funktion gibt, die Bilddaten und Dateinamen nacheinander abruft (auflistet), ist es nur eine Aufgabe, sie in einer Datei zu speichern. Lassen Sie uns nun eine Bildfunktion erstellen, die dies tut. Implementieren Sie zunächst nur den Vorgang zum Abrufen von Bilddaten.
def images():
for url in image_url():
image = urllib.urlopen(url)
yield image.read()
image.close()
Es gibt noch keine image_url-Funktion, aber wenn es eine Funktion gibt, die die URL der Bilddaten auflistet, ist es nur eine Aufgabe, die Bilddaten am URL-Ziel zu lesen und zu erhalten. Wenn es zurückgegeben wird, wird das Ergebnis einmal zurückgegeben und der Prozess endet. Mithilfe einer Generatorfunktion, die Yield verwendet, können die Ergebnisse jedoch nacheinander benachrichtigt werden. Fügen Sie dieser Funktion einen Prozess hinzu, um auch den Dateinamen zu benachrichtigen. Es weicht vom Prinzip der Einzelverantwortung des SOLID-Prinzips ab, aber ...
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()
Der Dateiname wird aus der Tweet-Zeit erstellt. Da das Bild jedoch nicht die Tweet-Zeit enthält, lasse ich mich von der Funktion image_urls informieren. Da mehrere Bilder in einen Tweet eingefügt werden können, wird eine Tweet-Zeit mit mehreren Bildern verknüpft. Bitte teilen Sie mir die Tweet-Zeit mit dem Python-Standard datetime mit und konvertieren Sie sie in "Jahr / Monat / Tag_Stunde / Minute / Sekunde", die für den Dateinamen verwendet wird Zu Die Erweiterung des Dateinamens wurde im ursprünglichen Programm auf ".jpg " festgelegt, aber es gibt auch ".png " usw., daher werde ich die in url enthaltene Erweiterung herausnehmen und verwenden. Als nächstes folgt die Funktion 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
Der Tweet enthält Medieninformationen wie Bilder zusammen mit einem kurzen Satz (Text). Wenn eine Medienfunktion vorhanden ist, die die Medieninformationen auflistet, können Sie nur die Bild-URL vom Typ "Foto" aus den Medieninformationen extrahieren und benachrichtigen. Ich werde. Da die Medieninformationen nicht die Tweet-Zeit enthalten, werden wir die Medienfunktion bitten, uns auch die Tweet-Zeit mitzuteilen. Als nächstes kommt die Medienfunktion.
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
Wenn es eine Tweets-Funktion gibt, die Tweets der Reihe nach auflistet, können Sie die Tweet-Zeit und die Medien abrufen und benachrichtigen. Es ist möglich, dass keine Informationen zu erweiterten_Einträgen oder Medien vorhanden sind. Wenn jedoch keine Informationen vorhanden sind, wird ein leerer Taple benachrichtigt, um Sie darüber zu informieren, dass kein Bild vorhanden ist. Schließlich funktionieren die Tweets.
USER_NAME = 'Benutzername, den Sie erhalten möchten'
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"]
Sie müssen die Twitter-API verwenden, um Tweets zu erhalten, und Sie müssen sich im Voraus als Benutzer bei Twitter registrieren, um den Schlüssel und das Token für den Zugriff auf die Twitter-API zu erhalten. Schreiben Sie diese Informationen als Variable in eine Datei namens twitkey.py. Die Programmquelle und vertrauliche Informationen werden getrennt. Im ursprünglichen Skript handelt es sich um eine Wörterbuchvariable, Sie können jedoch auch einfach eine Variable zuweisen.
twitkey.py
#coding: UTF-8
CONSUMER_KEY = ""
CONSUMER_SECRET = ""
ACCESS_TOKEN = ""
ACCESS_TOKEN_SECRET = ""
Immerhin wurde die Verschachtelung in der for-Anweisung vertieft, aber sie wurde flacher gemacht, indem sie mit Ertrag zu einer Generatorfunktion gemacht wurde. Durch die Trennung der Funktionen wird der Funktionsname auch zu einem Kommentar, der den Verarbeitungsinhalt angibt, und ich denke, es wird einfacher sein, die Verarbeitung zu verstehen.
Schließlich werde ich das gesamte Skript einschließlich des Hauptprozesses, des Imports und des Drucks der Fortschrittsanzeige veröffentlichen.
#!/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 = 'Benutzername, den Sie erhalten möchten'
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