1. 使用天氣 API
現在我們已經掌握了基礎知識，讓我們來看看一個更逼真的場景。想像一下，我們需要每30分鐘收集一次天氣數據。為此，我們會使用天氣數據的 API。當然，對於學習目的，真實 API 的使用可能有限，所以我們將假設它的運作方式。
通過 OpenWeather API 獲取當前天氣
這個範例展示了如何使用 requests 通過 OpenWeather 的 API 獲取特定城市的天氣數據。
import schedule
import time
import requests
def fetch_weather(city):
api_key = "YOUR_API_KEY" # 用您的 OpenWeather API 金鑰替換
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"{city} 的當前溫度: {temperature}°C")
print(f"天氣描述: {weather_description}")
except requests.exceptions.RequestException as e:
print(f"獲取天氣數據時出錯: {e}")
def fetch_weather_of_london():
fetch_weather("London")
# 設定每30分鐘執行一次此任務
schedule.every(30).minutes.do(fetch_weather_of_london)
while True:
schedule.run_pending()
time.sleep(1)
在這裡，我們向 OpenWeather 的 API 發送 GET 請求以獲取指定城市的當前天氣。在 JSON 回應中，我們提取溫度和天氣描述，並將其打印到螢幕上。別忘了將 YOUR_API_KEY 替換為您的 API 金鑰。
通過 API 獲取當前匯率
在這個範例中，我們將使用 requests 通過 API 獲取當前匯率。
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() # 確保請求成功
data = response.json()
usd_to_eur = data["rates"]["EUR"]
print(f"USD 到 EUR 的當前匯率: {usd_to_eur}")
except requests.exceptions.RequestException as e:
print(f"獲取數據時發生錯誤: {e}")
# 設定每10分鐘執行一次此任務
schedule.every(10).minutes.do(fetch_exchange_rate)
while True:
schedule.run_pending()
time.sleep(1)
在這裡，我們向貨幣匯率 API 發送 GET 請求，並以 JSON 格式接收數據。我們從 JSON 回應中提取 USD 到 EUR 的匯率，並將其打印到螢幕上。此腳本可以適配以收集其他貨幣對的數據，只需更改 data["rates"] 中的鍵。
使用這個腳本，我們可以持續收集天氣和匯率的數據。不錯的開始，對吧？
實際應用場景
數據收集的自動化在不同的場景中都很有用：
- 伺服器狀態監測: 自動檢查伺服器狀態可以提早發現問題並進行預防。
- 社交媒體數據收集: 不斷分析趨勢和品牌提及。
- 匯率追踪: 匯率變化可能對企業或個人需求很有用。
2. 自動網頁數據收集範例
如果我們想收集網頁上的數據怎麼辦？比如說，定期檢查新聞是否更新。為此，我們將使用 BeautifulSoup 和 requests。
從網頁收集數據
假設我們有一個網站，想要從中收集新聞標題。我們可以這樣做:
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)
在這個範例中，我們可以看到，每60分鐘，我們的腳本將檢查網頁並打印新聞標題。這使得獲取最新信息更加輕鬆。
從網站收集新聞標題
在這個範例中，我們將使用 requests 下載 HTML 網頁，並使用 BeautifulSoup 解析新聞標題。
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') # 查找所有 <h3> 標籤 (通常是標題)
print("BBC 最新新聞標題:")
for headline in headlines[:5]: # 取前五個標題
print("-", headline.get_text(strip=True))
except requests.exceptions.RequestException as e:
print(f"獲取數據時發生錯誤: {e}")
fetch_news_headlines()
在這裡，我們下載了 BBC News 的頁面，並使用 BeautifulSoup 查找所有 <h3> 標籤，其中包含新聞標題。我們打印出前5個標題，並使用 strip=True 刪除多餘的空格和符號。
從線上商店收集商品價格數據
這個範例展示了如何從電子商務網站 (例如 Amazon 或其他商城) 獲取商品價格數據。我們使用 requests 發送請求，並使用 BeautifulSoup 進行價格解析。
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"商品: {product_name}")
print(f"價格: {price} 元")
except requests.exceptions.RequestException as e:
print(f"獲取數據時發生錯誤: {e}")
except AttributeError:
print("找不到商品或價格信息")
# 商品鏈接範例
fetch_product_price("https://www.amazon.com/dp/B08N5WRWNW")
在這個範例中，我們發送帶有 User-Agent 標頭的 GET 請求，以避免被封鎖。接著，我們使用 BeautifulSoup 根據 id="productTitle" 查找商品名稱，並根據類別 a-price-whole 獲取價格。我們使用 strip=True 清除多餘空格。
GO TO FULL VERSION