当前位置：首页 > article >正文

python网络爬虫进阶

article 2024/11/30 3:17:40

当你掌握了 Python 网络爬虫的基本概念和技术后，可以进一步学习一些高级技术和最佳实践，以提高爬虫的效率、稳定性和安全性。以下是一些进阶的网络爬虫技术和技巧：

1. 异步请求

使用 `aiohttp` 库

异步请求可以显著提高爬虫的性能，特别是当需要同时处理大量请求时。

import aiohttp
import asyncio

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

async def main():
    urls = [
        'https://example.com/page1',
        'https://example.com/page2',
        'https://example.com/page3'
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(result)

asyncio.run(main())

2. 动态页面爬取

使用 `Selenium` 库

对于包含 JavaScript 动态加载内容的页面，可以使用 Selenium 模拟浏览器行为。

from selenium import webdriver
from selenium.webdriver.common.by import By

# 启动浏览器
driver = webdriver.Chrome()

# 访问页面
driver.get('https://example.com')

# 等待元素加载
element = driver.find_element(By.ID, 'content')

# 获取页面内容
html_content = driver.page_source

# 关闭浏览器
driver.quit()

3. 分布式爬虫

使用 `Scrapy` 框架

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}

4. 数据清洗和预处理

使用 `pandas` 库

pandas 是一个强大的数据处理库，可以用于数据清洗和预处理。

import pandas as pd

data = {
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [30, 25, 35],
    'email': ['alice@example.com', 'bob@example.com', 'charlie@example.com']
}

df = pd.DataFrame(data)

# 数据清洗
df['age'] = df['age'].fillna(0)  # 填充缺失值
df['email'] = df['email'].str.lower()  # 转换为小写

print(df)

5. 数据存储优化

使用 `SQLAlchemy` 库

SQLAlchemy 是一个 ORM（对象关系映射）库，可以简化数据库操作。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# 插入数据
new_user = User(name='Alice', age=30)
session.add(new_user)
session.commit()

# 查询数据
users = session.query(User).all()
for user in users:
    print(user.name, user.age)

session.close()

6. 反爬虫技术应对

使用 `requests_html` 库

requests_html 是一个强大的库，支持渲染 JavaScript 页面。

from requests_html import HTMLSession

session = HTMLSession()
response = session.get('https://example.com')
response.html.render()  # 渲染 JavaScript 页面
print(response.html.html)

使用 `Cloudflare` 绕过

有些网站使用 Cloudflare 防护，可以使用 cloudscraper 库绕过。

import cloudscraper

scraper = cloudscraper.create_scraper()
response = scraper.get('https://example.com')
print(response.text)

7. 日志记录

使用 `logging` 模块

记录爬虫运行的日志，便于调试和监控。

import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def main():
    logging.info('开始爬取')
    # 爬取逻辑
    logging.info('爬取结束')

main()

8. 定时任务

使用 `APScheduler` 库

定期执行爬虫任务。

from apscheduler.schedulers.blocking import BlockingScheduler

def job():
    logging.info('开始定时任务')
    # 爬取逻辑
    logging.info('定时任务结束')

scheduler = BlockingScheduler()
scheduler.add_job(job, 'interval', hours=1)  # 每小时执行一次
scheduler.start()

9. 数据可视化

使用 `Matplotlib` 或 `Seaborn` 库

将爬取的数据进行可视化展示。

import matplotlib.pyplot as plt
import seaborn as sns

data = {
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [30, 25, 35]
}

df = pd.DataFrame(data)

sns.barplot(x='name', y='age', data=df)
plt.show()

10. 安全性

使用 `pycryptodome` 库

对敏感数据进行加密处理。

from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from base64 import b64encode, b64decode

key = b'sixteen byte key'
cipher = AES.new(key, AES.MODE_CBC)

def encrypt(plain_text):
    ct_bytes = cipher.encrypt(pad(plain_text.encode(), AES.block_size))
    iv = b64encode(cipher.iv).decode('utf-8')
    ct = b64encode(ct_bytes).decode('utf-8')
    return iv, ct

def decrypt(iv, ct):
    iv = b64decode(iv)
    ct = b64decode(ct)
    cipher = AES.new(key, AES.MODE_CBC, iv)
    pt = unpad(cipher.decrypt(ct), AES.block_size).decode('utf-8')
    return pt

iv, ct = encrypt('Hello, World!')
print(f'IV: {iv}, Cipher Text: {ct}')
print(f'Decrypted: {decrypt(iv, ct)}')