Python网络爬虫技术高阶用法

import requests
from bs4 import BeautifulSoup

# 发起 HTTP 请求
response = requests.get('https://example.com')
# 解析 HTML 内容
soup = BeautifulSoup(response.text, 'html.parser')

# 提取特定元素
title = soup.find('title').text
print(title)

# 爬虫示例代码，需在 Scrapy 项目中使用
import scrapy

class ExampleSpider(scrapy.Spider):
    name = 'example'
    start_urls = ['https://example.com']

    def parse(self, response):
        title = response.css('title::text').get()
        yield {'title': title}

from selenium import webdriver

# 初始化 WebDriver
driver = webdriver.Chrome()

# 打开动态网页
driver.get('https://example.com')

# 等待页面完全加载
driver.implicitly_wait(5)

# 获取网页源代码
html = driver.page_source

# 关闭浏览器
driver.quit()

import asyncio
from pyppeteer import launch

async def main():
    browser = await launch()
    page = await browser.newPage()
    await page.goto('https://example.com')
    content = await page.content()
    print(content)
    await browser.close()

asyncio.get_event_loop().run_until_complete(main())

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'
}

response = requests.get('https://example.com', headers=headers)

proxies = {
    'http': 'http://10.10.1.10:3128',
    'https': 'http://10.10.1.10:1080',
}

response = requests.get('https://example.com', proxies=proxies)

# 使用会话保持
session = requests.Session()

# 设置初始的 cookie
session.cookies.set('name', 'value')

# 发送带有 cookie 的请求
response = session.get('https://example.com')

# pipelines.py 示例
import pymongo

class MongoPipeline:

    def open_spider(self, spider):
        self.client = pymongo.MongoClient("mongodb://localhost:27017/")
        self.db = self.client["example_db"]

    def close_spider(self, spider):
        self.client.close()

    def process_item(self, item, spider):
        self.db.example_collection.insert_one(dict(item))
        return item

# 在 Scrapy 项目中使用 scrapy-redis 进行分布式爬虫
# 安装 scrapy-redis 并配置 settings.py
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, 'https://example.com')
        print(html)

asyncio.run(main())

from concurrent.futures import ThreadPoolExecutor

def fetch(url):
    response = requests.get(url)
    return response.text

with ThreadPoolExecutor(max_workers=5) as executor:
    results = executor.map(fetch, ['https://example.com'] * 5)
    for result in results:
        print(result)

import pymysql

# 连接 MySQL 数据库
connection = pymysql.connect(host='localhost',
                             user='user',
                             password='passwd',
                             db='database')

# 插入数据
with connection.cursor() as cursor:
    sql = "INSERT INTO `table` (`column1`, `column2`) VALUES (%s, %s)"
    cursor.execute(sql, ('value1', 'value2'))
    connection.commit()

import csv

# 写入 CSV 文件
with open('data.csv', mode='w') as file:
    writer = csv.writer(file)
    writer.writerow(['column1', 'column2'])
    writer.writerow(['value1', 'value2'])

import requests
from bs4 import BeautifulSoup
import csv

# 发送 HTTP 请求
response = requests.get('https://example.com/products')

# 解析 HTML 内容
soup = BeautifulSoup(response.text, 'html.parser')

# 提取商品信息
products = soup.find_all('div', class_='product')

# 写入 CSV 文件
with open('products.csv', mode='w') as file:
    writer = csv.writer(file)
    writer.writerow(['Product Name', 'Price'])

    for product in products:
        name = product.find('h2').text
        price = product.find('span', class_='price').text
        writer.writerow([name, price])