Tự động hóa các tác vụ dựa trên thời gian để thu thập dữ liệu định kỳ

1. Làm việc với API thời tiết

Giờ đây, khi chúng ta đã nắm vững cơ bản, hãy xem qua một kịch bản thực tế. Hãy tưởng tượng chúng ta cần thu thập dữ liệu thời tiết mỗi 30 phút. Để làm điều này, chúng ta sẽ sử dụng API thu thập dữ liệu thời tiết. Đương nhiên, trong mục đích học tập, việc sử dụng API thực có thể bị hạn chế, vì vậy hãy tưởng tượng cách nó sẽ hoạt động.

Thu thập thời tiết hiện tại qua OpenWeather API

Ví dụ này cho thấy cách sử dụng requests để lấy dữ liệu thời tiết trong một thành phố cụ thể, sử dụng API OpenWeather.

import schedule
import time
import requests

def fetch_weather(city):
    api_key = "YOUR_API_KEY"  # Thay thế bằng API key của bạn từ 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"Nhiệt độ hiện tại ở {city}: {temperature}°C")
        print(f"Mô tả thời tiết: {weather_description}")
    except requests.exceptions.RequestException as e:
        print(f"Lỗi khi lấy dữ liệu thời tiết: {e}")

def fetch_weather_of_london():
    fetch_weather("London")

# Thiết lập tác vụ thực thi mỗi 30 phút
schedule.every(30).minutes.do(fetch_weather_of_london)

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

Ở đây, chúng ta gửi một GET request đến API OpenWeather để lấy dữ liệu thời tiết hiện tại cho một thành phố cụ thể. Trong JSON response, chúng ta lấy nhiệt độ và mô tả thời tiết, sau đó in chúng ra màn hình. Đừng quên thay YOUR_API_KEY bằng API key của bạn.

Thu thập tỷ giá hối đoái hiện tại qua API

Trong ví dụ này, chúng ta sẽ sử dụng requests để thu thập tỷ giá hối đoái hiện tại qua 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()  # Kiểm tra yêu cầu thành công
        data = response.json()
        usd_to_eur = data["rates"]["EUR"]
        print(f"Tỷ giá hiện tại USD sang EUR: {usd_to_eur}")
    except requests.exceptions.RequestException as e:
        print(f"Đã xảy ra lỗi khi lấy dữ liệu: {e}")

# Lên lịch tác vụ thực hiện mỗi 10 phút
schedule.every(10).minutes.do(fetch_exchange_rate)

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

Ở đây, chúng ta gửi một GET request đến API tỷ giá hối đoái và nhận dữ liệu ở định dạng JSON. Tỷ giá USD sang EUR được lấy từ response JSON và in ra màn hình. Script này có thể được tùy chỉnh để thu thập dữ liệu của các cặp tiền khác, bằng cách thay đổi key trong data["rates"].

Với script này, chúng ta có thể liên tục thu thập dữ liệu về thời tiết và tỷ giá hối đoái. Không tệ để bắt đầu, phải không?

Trường hợp sử dụng thực tế

Tự động hóa thu thập dữ liệu có thể hữu ích trong nhiều trường hợp thực tế:

• Giám sát trạng thái máy chủ: Kiểm tra tự động trạng thái máy chủ có thể giúp phát hiện và dự báo sự cố trước khi chúng xảy ra.
• Thu thập dữ liệu mạng xã hội: Phân tích liên tục xu hướng và đề cập đến thương hiệu.
• Theo dõi tỷ giá hối đoái: Thay đổi tỷ giá có thể hữu ích cho doanh nghiệp hoặc nhu cầu cá nhân.

2. Ví dụ thu thập dữ liệu web tự động

Nếu chúng ta muốn thu thập dữ liệu từ một trang web thì sao? Chẳng hạn, kiểm tra định kỳ các tin tức mới. Trong trường hợp này, chúng ta sẽ sử dụng BeautifulSoup và requests.

Thu thập dữ liệu từ trang web

Giả sử chúng ta có một trang web muốn lấy tiêu đề tin tức. Đây là cách thực hiện:

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)

Trong ví dụ này, mỗi 60 phút script của chúng ta sẽ kiểm tra trang web và in tiêu đề tin tức. Điều này làm cho việc nhận thông tin cập nhật trở nên dễ dàng hơn.

Thu thập tiêu đề tin tức từ trang web

Trong ví dụ này, chúng ta sử dụng requests để tải trang HTML và BeautifulSoup để phân tích tiêu đề tin tức.

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')  # Tìm tất cả thẻ <h3> (thường chứa tiêu đề)
        print("Các tiêu đề tin tức mới nhất trên BBC:")
        for headline in headlines[:5]:  # Lấy 5 tiêu đề đầu tiên
            print("-", headline.get_text(strip=True))
    except requests.exceptions.RequestException as e:
        print(f"Lỗi khi lấy dữ liệu: {e}")

fetch_news_headlines()

Ở đây chúng ta tải trang tin tức BBC và sử dụng BeautifulSoup để tìm tất cả thẻ <h3>, nơi chứa tiêu đề tin tức. Chúng ta in ra 5 tiêu đề đầu tiên và loại bỏ khoảng trắng thừa bằng strip=True.

Thu thập dữ liệu giá sản phẩm từ trang thương mại điện tử

Ví dụ này chỉ ra cách lấy thông tin về giá sản phẩm từ một trang thương mại điện tử (như Amazon hoặc trang khác). Chúng ta sẽ sử dụng requests để lấy trang và BeautifulSoup để phân tích giá cả.

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"Sản phẩm: {product_name}")
        print(f"Giá: {price} VNĐ")
    except requests.exceptions.RequestException as e:
        print(f"Lỗi khi lấy dữ liệu: {e}")
    except AttributeError:
        print("Không thể tìm thông tin về sản phẩm hoặc giá cả")

# Ví dụ link sản phẩm
fetch_product_price("https://www.amazon.com/dp/B08N5WRWNW")

Trong ví dụ này, chúng ta gửi một GET request với header User-Agent để tránh bị block. Sau đó, sử dụng BeautifulSoup, chúng ta tìm tên sản phẩm bằng id="productTitle" và giá sản phẩm bằng class a-price-whole. Chúng ta sử dụng strip=True để loại bỏ khoảng trắng thừa.