多语言爬虫实现网站价格监控

import requests
from bs4 import BeautifulSoup
import csv
from time import sleep
​
def scrape_static_page(url):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
    }
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, 'html.parser')
        # 假设价格在类名为 price 的元素中
        price = soup.find('span', class_='price').text.strip()
        return price
    except Exception as e:
        print(f"Error: {e}")
        return None
​
# 保存到CSV
def save_to_csv(data):
    with open('prices.csv', 'a', newline='') as f:
        writer = csv.writer(f)
        writer.writerow([data['timestamp'], data['price']])
​
# 定时监控
if __name__ == "__main__":
    target_url = "https://example-product-page.com"
    while True:
        current_price = scrape_static_page(target_url)
        if current_price:
            save_to_csv({'timestamp': datetime.now(), 'price': current_price})
        sleep(3600)  # 每小时执行一次

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
import time
​
def scrape_dynamic_page(url):
    options = Options()
    options.headless = True  # 无头模式
    driver = webdriver.Chrome(options=options)
    driver.get(url)
    time.sleep(3)  # 等待页面加载
    # 假设价格通过JavaScript动态加载
    price_element = driver.find_element(By.CSS_SELECTOR, ".dynamic-price")
    price = price_element.text
    driver.quit()
    return price
​
# 使用方式同上（可结合定时任务）

# 在Scrapy项目的spiders目录下创建 price_spider.py
import scrapy
from datetime import datetime
​
class PriceSpider(scrapy.Spider):
    name = "price_monitor"
    start_urls = ["https://example-product-page.com"]
​
    def parse(self, response):
        price = response.css('span.price::text').get()
        yield {
            'timestamp': datetime.now().isoformat(),
            'price': price.strip(),
            'url': response.url
        }
​
# 运行命令：scrapy crawl price_monitor -o prices.json

const puppeteer = require('puppeteer');
const fs = require('fs');
​
async function scrapePrice(url) {
    const browser = await puppeteer.launch({ headless: true });
    const page = await browser.newPage();
    await page.goto(url, { waitUntil: 'networkidle2' });
    const price = await page.$eval('.price', el => el.textContent.trim());
    await browser.close();
    return price;
}
​
// 定时任务
setInterval(async () => {
    const price = await scrapePrice('https://example-product-page.com');
    fs.appendFileSync('prices.csv', `${new Date().toISOString()},${price}\n`);
}, 3600 * 1000);

package main
​
import (
    "fmt"
    "github.com/gocolly/colly"
    "log"
    "time"
)
​
func main() {
    c := colly.NewCollector(
        colly.UserAgent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"),
    )
​
    c.OnHTML(".price", func(e *colly.HTMLElement) {
        fmt.Printf("Price: %s at %s\n", e.Text, time.Now().Format(time.RFC3339))
    })
​
    c.OnError(func(r *colly.Response, err error) {
        log.Println("Request failed:", err)
    })
​
    for {
        c.Visit("https://example-product-page.com")
        time.Sleep(1 * time.Hour)
    }
}