Python爬虫开发：Cookie池与定期清除的代码实现

小白学大数据

发布于 2025-07-17 17:09:50

9700

代码可运行

文章被收录于专栏：python进阶学习python进阶学习

运行总次数：0

代码可运行

一、Cookie池的概念与作用

（一）什么是Cookie池

Cookie池是一个存储多个有效Cookie的集合，通常用于模拟多个用户的行为，从而避免因单一Cookie被频繁使用而导致的封禁风险。Cookie池中的每个Cookie都可以独立地用于发起网络请求，从而分散风险并提高爬虫的稳定性。

（二）Cookie池的作用

规避反爬机制：许多网站会检测单一IP或单一Cookie的访问频率，如果访问过于频繁，可能会触发反爬机制。通过使用Cookie池，可以模拟多个用户的行为，降低被封禁的风险。
提高爬虫效率：Cookie池中的多个Cookie可以并行使用，从而提高数据采集的速度。
增强爬虫的鲁棒性：即使某个Cookie失效或被封禁，其他Cookie仍然可以继续工作，从而保证爬虫的整体稳定性。

二、Cookie池的构建与管理

（一）存储Cookie池

Cookie池的存储方式有多种选择，常见的包括内存存储、文件存储和数据库存储。对于大规模的爬虫项目，数据库存储是更合适的选择，因为它可以方便地管理大量Cookie，并支持高效的查询和更新操作。

示例：使用SQLite数据库存储Cookie池

import sqlite3
import json

# 初始化数据库
def init_db(db_file):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS cookies (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            cookie TEXT,
            last_used TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            status INTEGER DEFAULT 1
        )
    ''')
    conn.commit()
    conn.close()

# 保存Cookie到数据库
def save_cookie(db_file, cookie):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('INSERT INTO cookies (cookie) VALUES (?)', (json.dumps(cookie),))
    conn.commit()
    conn.close()

# 获取一个有效的Cookie
def get_valid_cookie(db_file):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('SELECT id, cookie FROM cookies WHERE status=1 ORDER BY last_used ASC LIMIT 1')
    result = cursor.fetchone()
    conn.close()
    if result:
        return result[0], json.loads(result[1])
    return None, None

# 更新Cookie的使用时间
def update_cookie_usage(db_file, cookie_id):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('UPDATE cookies SET last_used=CURRENT_TIMESTAMP WHERE id=?', (cookie_id,))
    conn.commit()
    conn.close()

# 标记Cookie为失效
def mark_cookie_invalid(db_file, cookie_id):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('UPDATE cookies SET status=0 WHERE id=?', (cookie_id,))
    conn.commit()
    conn.close()

（二）Cookie池的使用

在爬虫中使用Cookie池时，需要从Cookie池中获取一个有效的Cookie，并在请求完成后更新其使用时间。如果请求失败或Cookie失效，则需要将其标记为失效。

示例：使用Cookie池发起请求

import requests

# 数据库文件路径
db_file = 'cookies.db'

# 初始化数据库
init_db(db_file)

# 登录并保存Cookie
def login_and_save_cookie():
    response = requests.post('https://example.com/login', data={'username': 'user', 'password': 'pass'})
    if response.status_code == 200:
        save_cookie(db_file, response.cookies.get_dict())

# 使用Cookie池发起请求
def fetch_data():
    cookie_id, cookie = get_valid_cookie(db_file)
    if cookie:
        try:
            response = requests.get('https://example.com/data', cookies=cookie)
            if response.status_code == 200:
                update_cookie_usage(db_file, cookie_id)
                return response.text
            else:
                mark_cookie_invalid(db_file, cookie_id)
        except Exception as e:
            mark_cookie_invalid(db_file, cookie_id)
    return None

# 示例：登录并保存Cookie
login_and_save_cookie()

# 示例：使用Cookie池发起请求
data = fetch_data()
if data:
    print(data)
else:
    print("Failed to fetch data.")

三、定期清除失效的Cookie

随着时间的推移，Cookie可能会因为过期、失效或被网站封禁而无法使用。因此，定期清理失效的Cookie是确保Cookie池高效运行的关键步骤。

（一）基于时间的清理

可以设置一个定时任务，定期清理超过一定时间未使用的Cookie。

示例：基于时间的清理

Python

复制

import time

# 清理超过一定时间未使用的Cookie
def clean_expired_cookies(db_file, expire_time=3600):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('DELETE FROM cookies WHERE last_used < ?', (time.time() - expire_time,))
    conn.commit()
    conn.close()

# 示例：每小时清理一次
import schedule

def scheduled_clean():
    clean_expired_cookies(db_file, expire_time=3600)

schedule.every(1).hour.do(scheduled_clean)

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

（二）基于状态的清理

在每次请求后，根据响应状态判断Cookie是否失效。如果Cookie失效，则将其从Cookie池中移除。

示例：基于状态的清理

def fetch_data():
    cookie_id, cookie = get_valid_cookie(db_file)
    if cookie:
        try:
            response = requests.get('https://example.com/data', cookies=cookie)
            if response.status_code == 200:
                update_cookie_usage(db_file, cookie_id)
                return response.text
            else:
                mark_cookie_invalid(db_file, cookie_id)
        except Exception as e:
            mark_cookie_invalid(db_file, cookie_id)
    return None

四、完整代码实现

以下是结合Cookie池的构建、使用和定期清理的完整代码实现：

import sqlite3
import json
import requests
import time
import schedule

# 数据库文件路径
db_file = 'cookies.db'

# 初始化数据库
def init_db(db_file):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS cookies (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            cookie TEXT,
            last_used TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            status INTEGER DEFAULT 1
        )
    ''')
    conn.commit()
    conn.close()

# 保存Cookie到数据库
def save_cookie(db_file, cookie):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('INSERT INTO cookies (cookie) VALUES (?)', (json.dumps(cookie),))
    conn.commit()
    conn.close()

# 获取一个有效的Cookie
def get_valid_cookie(db_file):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('SELECT id, cookie FROM cookies WHERE status=1 ORDER BY last_used ASC LIMIT 1')
    result = cursor.fetchone()
    conn.close()
    if result:
        return result[0], json.loads(result[1])
    return None, None

# 更新Cookie的使用时间
def update_cookie_usage(db_file, cookie_id):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('UPDATE cookies SET last_used=CURRENT_TIMESTAMP WHERE id=?', (cookie_id,))
    conn.commit()
    conn.close()

# 标记Cookie为失效
def mark_cookie_invalid(db_file, cookie_id):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('UPDATE cookies SET status=0 WHERE id=?', (cookie_id,))
    conn.commit()
    conn.close()

# 清理超过一定时间未使用的Cookie
def clean_expired_cookies(db_file, expire_time=3600):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('DELETE FROM cookies WHERE last_used < ?', (time.time() - expire_time,))
    conn.commit()
    conn.close()

# 登录并保存Cookie
def login_and_save_cookie():
    response = requests.post('https://example.com/login', data={'username': 'user', 'password': 'pass'})
    if response.status_code == 200:
        save_cookie(db_file, response.cookies.get_dict())

# 使用Cookie池发起请求
def fetch_data():
    cookie_id, cookie = get_valid_cookie(db_file)
    if cookie:
        try:
            response = requests.get('https://example.com/data', cookies=cookie)
            if response.status_code == 200:
                update_cookie_usage(db_file, cookie_id)
                return response.text
            else:
                mark_cookie_invalid(db_file, cookie_id)
        except Exception as e:
            mark_cookie_invalid(db_file, cookie_id)
    return None

# 初始化数据库
init_db(db_file)

# 示例：登录并保存Cookie
login_and_save_cookie()

# 示例：使用Cookie池发起请求
data = fetch_data()
if data:
    print(data)
else:
    print("Failed to fetch data.")

# 定时清理失效的Cookie
def scheduled_clean():
    clean_expired_cookies(db_file, expire_time=3600)

schedule.every(1).hour.do(scheduled_clean)

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

五、总结

通过构建一个高效的Cookie池并定期清理失效的Cookie，可以显著提高Python爬虫的稳定性和效率。Cookie池不仅可以帮助爬虫规避反爬机制，还能通过模拟多个用户的行为分散风险。定期清理失效的Cookie可以确保Cookie池始终处于最佳状态，从而提高爬虫的可靠性。本文提供的代码示例展示了如何实现Cookie池的存储、使用和清理，开发者可以根据实际需求进行调整和优化。

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2025-07-10，如有侵权请联系 cloudcommunity@tencent.com 删除

数据库