强化学习驱动的智能限速：动态请求间隔的终极方案

import time
import random
import requests
from bs4 import BeautifulSoup
import pandas as pd
from collections import defaultdict

class DynamicRateLimiter:
    def __init__(self, intervals=[1, 2, 3], alpha=0.5, gamma=0.9, epsilon=0.1):
        self.q_table = {i: 0.0 for i in intervals}
        self.intervals = intervals
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon
        self.last_action = random.choice(intervals)

    def choose(self):
        # ε-贪婪策略
        if random.random() < self.epsilon:
            self.last_action = random.choice(self.intervals)
        else:
            self.last_action = max(self.q_table, key=self.q_table.get)
        return self.last_action

    def feedback(self, reward):
        old_value = self.q_table[self.last_action]
        self.q_table[self.last_action] = old_value + self.alpha * (reward + self.gamma * max(self.q_table.values()) - old_value)

def get_headers_and_cookies():
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36",
    }
    cookies = {
        "TrackID": "1XXXXXXXXXXXXXXX",  # 示例Cookie，可使用浏览器F12复制
    }
    return headers, cookies

def get_proxies():
    # 爬虫代理配置（参考亿牛云示例）
    proxy_host = "proxy.16yun.cn"
    proxy_port = "3100"
    proxy_user = "16YUN"
    proxy_pass = "16IP"
    
    proxy_meta = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
    proxies = {
        "http": proxy_meta,
        "https": proxy_meta,
    }
    return proxies

def search_jd(keyword, limiter):
    base_url = f"https://search.jd.com/Search?keyword={keyword}&enc=utf-8"
    headers, cookies = get_headers_and_cookies()
    proxies = get_proxies()

    while True:
        wait_time = limiter.choose()
        time.sleep(wait_time)

        try:
            response = requests.get(base_url, headers=headers, cookies=cookies, proxies=proxies, timeout=10)
            if response.status_code == 200:
                limiter.feedback(1)  # 成功，正向奖励
                return response.text
            else:
                limiter.feedback(-1)  # 失败，负向奖励
        except Exception as e:
            print(f"请求失败: {e}")
            limiter.feedback(-1)

def parse_products(html):
    soup = BeautifulSoup(html, "html.parser")
    items = soup.select("li.gl-item")
    results = []

    for item in items[:10]:  # 只处理前10个
        title_tag = item.select_one("div.p-name em")
        price_tag = item.select_one("div.p-price i")
        link_tag = item.select_one("div.p-name a")

        title = title_tag.text.strip() if title_tag else "N/A"
        price = price_tag.text.strip() if price_tag else "N/A"
        link = "https:" + link_tag.get("href") if link_tag else "N/A"

        spec = title.split(" ")[:2]  # 简化为前两项关键词
        results.append({
            "商品名称": title,
            "价格": price,
            "规格": " / ".join(spec),
            "链接": link
        })
    return results

def main():
    keyword = "笔记本电脑"
    limiter = DynamicRateLimiter()
    html = search_jd(keyword, limiter)
    product_list = parse_products(html)

    df = pd.DataFrame(product_list)
    df.to_csv("jd_top10_products.csv", index=False, encoding="utf-8-sig")
    print("已保存数据：")
    print(df)

if __name__ == "__main__":
    main()