Automatisierung von zeitbasierten Aufgaben für regelmäßiges Datensammeln

1. Arbeiten mit einer Wetter-API

Jetzt, wo wir die Grundlagen gelernt haben, schauen wir uns ein realistisches Szenario an. Stell dir vor, wir müssen alle 30 Minuten Wetterdaten sammeln. Dafür nutzen wir eine Wetter-API. Natürlich ist die Nutzung einer echten API in Lernzwecken oft eingeschränkt, daher simulieren wir, wie das aussehen könnte.

Aktuelles Wetter mit der OpenWeather API abrufen

Dieses Beispiel zeigt, wie man mit requests Wetterdaten für eine bestimmte Stadt mit der OpenWeather API abruft.

import schedule
import time
import requests

def fetch_weather(city):
    api_key = "YOUR_API_KEY"  # Ersetze durch deinen API-Schlüssel von OpenWeather
    url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric"

    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()
        temperature = data["main"]["temp"]
        weather_description = data["weather"][0]["description"]
        print(f"Aktuelle Temperatur in {city}: {temperature}°C")
        print(f"Wetterbeschreibung: {weather_description}")
    except requests.exceptions.RequestException as e:
        print(f"Fehler beim Abrufen der Wetterdaten: {e}")

def fetch_weather_of_london():
    fetch_weather("London")

# Einstellung der Aufgabe alle 30 Minuten
schedule.every(30).minutes.do(fetch_weather_of_london)

while True:
    schedule.run_pending()
    time.sleep(1)

Hier senden wir eine GET-Anfrage an die OpenWeather API, um die aktuellen Wetterdaten für eine bestimmte Stadt abzurufen. Aus der JSON-Antwort entnehmen wir die Temperatur und die Wetterbeschreibung und geben sie aus. Vergiss nicht, YOUR_API_KEY durch deinen API-Schlüssel zu ersetzen.

Aktuelle Wechselkurse mit einer API abrufen

In diesem Beispiel nutzen wir requests, um die aktuellen Wechselkurse über eine API abzurufen.

import schedule
import time
import requests

def fetch_exchange_rate():
    url = "https://api.exchangerate-api.com/v4/latest/USD"
    try:
        response = requests.get(url)
        response.raise_for_status()  # Überprüfen, ob die Anfrage erfolgreich war
        data = response.json()
        usd_to_eur = data["rates"]["EUR"]
        print(f"Aktueller USD zu EUR Wechselkurs: {usd_to_eur}")
    except requests.exceptions.RequestException as e:
        print(f"Fehler beim Abrufen der Daten: {e}")

# Einstellung der Aufgabe alle 10 Minuten
schedule.every(10).minutes.do(fetch_exchange_rate)

while True:
    schedule.run_pending()
    time.sleep(1)

Hier senden wir eine GET-Anfrage an die API für Wechselkurse und erhalten die Daten im JSON-Format. Der Wechselkurs von USD zu EUR wird aus der JSON-Antwort extrahiert und ausgegeben. Dieses Skript kann angepasst werden, um Daten für andere Währungspaare zu sammeln, indem der Schlüssel in data["rates"] geändert wird.

Mit diesen Skripten können wir kontinuierlich Wetter- und Währungsdaten sammeln. Nicht schlecht für den Anfang, oder?

Reale Anwendungsszenarien

Die Automatisierung der Datensammlung kann in verschiedenen Szenarien nützlich sein:

• Serverzustandsüberwachung: Automatische Serverzustandsprüfungen können Probleme erkennen und verhindern, bevor sie auftreten.
• Datensammlung zu sozialen Medien: Kontinuierliche Analyse von Trends und Markenerwähnungen.
• Überwachung von Wechselkursen: Änderungen der Wechselkurse können für Unternehmen oder den persönlichen Gebrauch nützlich sein.

2. Beispiel für automatisierte Web-Datensammlung

Was, wenn wir Daten von einer Webseite sammeln wollen? Zum Beispiel regelmäßig nach Nachrichten schauen. Hier kommt uns BeautifulSoup und requests zur Hilfe.

Daten von einer Webseite sammeln

Angenommen, wir möchten von einer Webseite Nachrichtentitel abrufen. So könnte man das machen:

import requests
from bs4 import BeautifulSoup

def fetch_news():
    response = requests.get("http://example.com/news")
    soup = BeautifulSoup(response.content, 'html.parser')
    for headline in soup.find_all('h2', class_='news'):
        print(headline.text)

schedule.every().hour.do(fetch_news)

while True:
    schedule.run_pending()
    time.sleep(1)

In diesem Beispiel überprüft unser Skript alle 60 Minuten die Webseite und gibt die Nachrichtentitel aus. Das vereinfacht den Prozess, aktuelle Informationen zu erhalten.

Nachrichtentitel von einer Webseite abrufen

In diesem Beispiel nutzen wir requests, um die HTML-Webseite herunterzuladen, und BeautifulSoup, um die Nachrichtentitel zu parsen.

import requests
from bs4 import BeautifulSoup

def fetch_news_headlines():
    url = "https://www.bbc.com/news"
    try:
        response = requests.get(url)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, 'html.parser')
        
        headlines = soup.find_all('h3')  # Suche nach allen <h3>-Tags (wo oft die Titel stehen)
        print("Letzte Nachrichten auf BBC:")
        for headline in headlines[:5]:  # Erste 5 Titel nehmen
            print("-", headline.get_text(strip=True))
    except requests.exceptions.RequestException as e:
        print(f"Fehler beim Abrufen der Daten: {e}")

fetch_news_headlines()

Hier laden wir die BBC-News-Seite herunter und suchen mit BeautifulSoup nach allen <h3>-Tags, wo die Nachrichtentitel stehen. Wir geben die ersten 5 Titel aus und entfernen unnötige Leerzeichen und Zeichen mit strip=True.

Preisdaten von einem Online-Shop abrufen

Dieses Beispiel zeigt, wie man Preisdaten von einer Webseite eines Online-Shops (z. B. Amazon oder einem anderen Shop) extrahiert. Wir nutzen requests, um die Seite abzufragen, und BeautifulSoup, um die Preise zu parsen.

import requests
from bs4 import BeautifulSoup

def fetch_product_price(url):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Safari/537.36"
    }
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.text, 'html.parser')
        product_name = soup.find('span', {'id': 'productTitle'}).get_text(strip=True)
        price = soup.find('span', {'class': 'a-price-whole'}).get_text(strip=True)
        
        print(f"Produkt: {product_name}")
        print(f"Preis: {price} €")
    except requests.exceptions.RequestException as e:
        print(f"Fehler beim Abrufen der Daten: {e}")
    except AttributeError:
        print("Produkt- oder Preisinformationen konnten nicht gefunden werden")

# Beispiel-Link zu einem Produkt
fetch_product_price("https://www.amazon.com/dp/B08N5WRWNW")

In diesem Beispiel senden wir eine GET-Anfrage mit einem User-Agent-Header, um Blockierungen zu umgehen. Danach suchen wir mit BeautifulSoup den Produktnamen anhand des id="productTitle" und den Produktpreis anhand der Klasse a-price-whole. Mit strip=True entfernen wir unnötige Leerzeichen.