import requests
from bs4 import BeautifulSoup

base_url = 'https://example.com/products?page='
all_products = []

# 爬取前5頁
for page in range(1, 6):
    response = requests.get(f'{base_url}{page}')
    soup = BeautifulSoup(response.text, 'html.parser')
    
    products = soup.select('.product-card')
    for product in products:
        name = product.select_one('.product-name').text.strip()
        price = product.select_one('.product-price').text.strip()
        all_products.append({'name': name, 'price': price})
    
    print(f"已爬取第 {page} 頁，目前共有 {len(all_products)} 個產品")

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.common.action_chains import ActionChains
import time

driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))
driver.get('https://example.com/products')

# 設定目標產品數量
target_products = 100
products_found = 0

while products_found < target_products:
    # 取得當前產品元素
    products = driver.find_elements(By.CSS_SELECTOR, '.product-card')
    products_found = len(products)
    
    print(f"目前已找到 {products_found} 個產品")
    
    if products_found >= target_products:
        break
    
    # 滾動到頁面底部以觸發更多內容載入
    driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
    time.sleep(2)  # 等待新內容載入

# 處理找到的產品
for product in products[:target_products]:
    name = product.find_element(By.CSS_SELECTOR, '.product-name').text
    price = product.find_element(By.CSS_SELECTOR, '.product-price').text
    print(f"名稱: {name}, 價格: {price}")

driver.quit()

import requests
import random
import time

# 常見瀏覽器 User-Agent 列表
user_agents = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15',
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:89.0) Gecko/20100101 Firefox/89.0'
]

urls = [f'https://example.com/products?page={i}' for i in range(1, 11)]

for url in urls:
    # 隨機選擇 User-Agent
    headers = {'User-Agent': random.choice(user_agents)}
    
    # 新增隨機延遲，模擬人工瀏覽間隔
    time.sleep(random.uniform(3, 7))
    
    response = requests.get(url, headers=headers)
    print(f"抓取 {url} 狀態碼: {response.status_code}")
    
    # 處理資料...

import requests
from concurrent.futures import ThreadPoolExecutor
import random
import time

def fetch_url(url):
    # 新增隨機延遲
    time.sleep(random.uniform(1, 3))
    
    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',
        'Referer': 'https://example.com/'
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        return response.text
    except Exception as e:
        print(f"抓取 {url} 時發生錯誤: {e}")
        return None

# 準備要爬取的URL列表
urls = [f'https://example.com/products?page={i}' for i in range(1, 101)]

# 使用執行緒池限制並發請求數
with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor

pip freeze > requirements.txt

export PYVER=3.9 && sudo apt update && sudo apt upgrade -y && sudo apt install -y software-properties-common && sudo add-apt-repository ppa:deadsnakes/ppa -y && sudo apt update && sudo apt install -y python${PYVER} python${PYVER}-venv python${PYVER}-dev python3-pip && python${PYVER} -m venv venv && source venv/bin/activate && pip install --upgrade pip && [ -f requirements.txt ] && pip install -r requirements.txt

# 使用requests函式庫esponse = requests.get("https://example.com", timeout=20)

# 使用aiohttp函式庫同步）
timeout = aiohttp.ClientTimeout(total=60, sock_connect=10, sock_read=10)
async with aiohttp.ClientSession(timeout=timeout) as session:
    async with session.get(url) as response:
        return await response.text()

def parse_product_listings(url_base, max_pages=None):
    """生成器函式：逐頁解析產品列表"""
    page = 1
    while max_pages is None or page <= max_pages:
        url = f"{url_base}?page={page}"
        try:
            response = requests.get(url, timeout=15)
            if response.status_code != 200:
                logging.warning(f"頁面 {page} 回應狀態碼: {response.status_code}")
                break
                
            soup = BeautifulSoup(response.text, 'html.parser')
            products = soup.select('.product-item')
            
            if not products:
                break  # 沒有更多產品，結束迭代
                
            for product in products:
                # 解析每個產品資訊
                product_data = {
                    'name': product.select_one('.product-name').text.strip(),
                    'price': product.select_one('.price').text.strip(),
                    'url': product.select_one('a')['href']
                }
                yield product_data  # 每解析一個產品就立即回傳
                
            page += 1
            time.sleep(1)  # 禮貌性延遲
            
        except Exception as e:
            logging.error(f"解析頁面 {page} 時發生錯誤: {e}")
            break

import requests
from bs4 import BeautifulSoup


class StreamParser:
    def __init__(self, url):
        self.url = url

    def parse(self):
        """
        實作生成器爬蟲，逐項產出解析結果
        """
        response = requests.get(self.url)
        if response.status_code != 200:
            raise Exception(f"無法取得頁面，狀態碼: {response.status_code}")

        soup = BeautifulSoup(response.text, "html.parser")
        items = soup.select("div")

        for item in items:
            title = item.select_one("h1").get_text(strip=True) if item.select_one("h1") else "無標題"
            yield {
                "title": title,
                "content": item.get_text(strip=True)
            }


if __name__ == "__main__":
    parser = StreamParser("https://example.com")
    for data_item in parser.parse():
        print(data_item["title"], "--", data_item["content"])
        # 在這裡可以直接處理每筆資料，例如寫入資料函式庫送通知

import asyncio
import aiohttp
from aiohttp import ClientTimeout

# 限制同時進行的請求數量
semaphore = asyncio.Semaphore(10)


async def fetch(session, url):
    async with semaphore:
        try:
            async with session.get(url) as response:
                return await response.text()
        except Exception:
            # 使用指數退避策略處理失敗請求
            for delay in [1, 2, 4, 8]:
                await asyncio.sleep(delay)
                try:
                    async with session.get(url) as response:
                        return await response.text()
                except Exception:
                    continue
            return None


async def main(urls):
    # 設定各種超時引數，提高穩定性
    timeout = ClientTimeout(total=60, sock_connect=10, sock_read=10)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
        results = await asyncio.gather(*tasks)
        # 處理取得的資料
        for result in results:
            if result:
                # 這裡可以進行內容解析或其他處理
                print(f"成功取得內容，長度: {len(result)}")

# 測試用的URL列表
urls = ["https://timeweb.cloud"] * 100

asyncio.run(main(urls))

# 使用BeautifulSoup提取資料範例
from bs4 import BeautifulSoup
import requests

response = requests.get('https://example.com/products')
soup = BeautifulSoup(response.text, 'html.parser')

# 使用CSS選擇器提取產品資訊
products = []
for item in soup.select('.product-item'):
    product = {
        'name': item.select_one('.product-name').text.strip(),
        'price': item.select_one('.product-price').text.strip(),
        'rating': item.select_one('.rating')['data-value'] if item.select_one('.rating') else 'N/A'
    }
    products.append(product)

# 使用Selenium處理動態頁面範例
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

driver = webdriver.Chrome()
driver.get('https://example.com/dynamic-page')

# 等待動態內容載入
WebDriverWait(driver, 10).until(
    EC.presence_of_element_located((By.CLASS_NAME, 'content-loaded'))
)

# 提取載入後的內容
elements = driver.find_elements(By.CSS_SELECTOR, '.data-item')
results = [element.text for element in elements]

driver.quit()

# 直接請求API範例
import requests
import json

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Accept': 'application/json'
}

response = requests.get(
    'https://api.example.com/products', 
    params={'category': 'electronics', 'page': 1},
    headers=headers
)

data = response.json()
products = data['products']

# 健壯的錯誤處理範例
import requests
from requests.exceptions import RequestException
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def fetch_with_retry(url, max_retries=3, backoff_factor=1.5):
    """帶重試機制的請求函式"""
    retries = 0
    while retries < max_retries:
        try:
            response = requests.get(url, timeout=(3.05, 10))
            response.raise_for_status()  # 檢查HTTP錯誤
            return response
        except RequestException as e:
            wait_time = backoff_factor * (2 ** retries)
            logger.warning(f"請求失敗 ({url}): {e}, 等待 {wait_time} 秒後重試")
            time.sleep(wait_time)
            retries += 1
    
    # 所有重試都失敗
    logger.error(f"達到最大重試次數, 無法請求 {url}")
    return None

def parse_safely(html, selector, default="N/A"):
    """安全解析函式，處理選擇器不比對情況"""
    from bs4 import BeautifulSoup
    soup = BeautifulSoup(html, 'html.parser')
    try:
        element = soup.select_one(selector)
        if element:
            return element.text.strip()
        return default
    except Exception as e:
        logger.error(f"解析錯誤 ({selector}): {e}")
        return default

# 模組化爬蟲設計範例
class ProductScraper:
    # 可設定的選擇器字典，便於單獨更新
    SELECTORS = {
        'product_container': '.product-grid .item',
        'name': '.product-title a',
        'price': '.product-price .current',
        'stock': '.stock-info',
        'rating': '.rating-stars'
    }
    
    def __init__(self, url, selectors=None):
        self.url = url
        # 允許覆寫預設選擇器
        if selectors:
            self.SELECTORS.update(selectors)
    
    def scrape(self):
        # 爬蟲邏輯實作...
        pass