告别手动切换IP：动态住宅代理自动化调度方案（附Python实战代码）

import requests
import time
import random
from typing import Optional

# 动态住宅代理配置
PROXY_API = "动态住宅代理获取API"
# 请求超时时间
TIMEOUT = 15
# 最大重试次数
MAX_RETRY = 3

def get_dynamic_proxy() -> Optional[dict]:
    """
    自动获取高质量动态住宅代理
    :return: 代理字典
    """
    try:
        resp = requests.get(PROXY_API, timeout=TIMEOUT)
        proxy_data = resp.json()
        # 适配主流代理返回格式，按需微调
        proxy_ip = proxy_data.get("ip")
        proxy_port = proxy_data.get("port")
        if proxy_ip and proxy_port:
            proxy = {
                "http": f"http://{proxy_ip}:{proxy_port}",
                "https": f"http://{proxy_ip}:{proxy_port}"
            }
            return proxy
    except Exception as e:
        print(f"获取代理节点失败：{str(e)}，即将重新获取...")
        time.sleep(1)
        return get_dynamic_proxy()
    return None

def auto_request(url: str, retry=0):
    """
    带自动换IP、重试机制的请求函数
    :param url: 请求地址
    :param retry: 重试次数
    :return: 响应内容
    """
    # 每次请求/重试都重新获取新代理，实现动态换IP
    proxy = get_dynamic_proxy()
    if not proxy:
        return None

    try:
        # 模拟真实浏览器请求头
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
        }
        response = requests.get(url, headers=headers, proxies=proxy, timeout=TIMEOUT)
        
        # 捕获限流、封禁状态码，主动触发换IP重试
        if response.status_code in [403, 429, 406]:
            raise Exception(f"触发风控，状态码：{response.status_code}")
        
        print(f"请求成功，当前代理IP：{proxy['http']}")
        return response.text
    
    except Exception as e:
        if retry < MAX_RETRY:
            print(f"请求失败：{str(e)}，第{retry+1}次重试换IP...")
            time.sleep(random.uniform(1, 3))
            return auto_request(url, retry+1)
        else:
            print("重试次数耗尽，当前任务跳过")
            return None

def batch_task_runner(task_url_list: list):
    """
    批量任务无人值守执行器
    :param task_url_list: 待采集URL列表
    """
    for idx, url in enumerate(task_url_list):
        print(f"\n正在执行第{idx+1}个任务：{url}")
        result = auto_request(url)
        # 此处可拓展：数据解析、入库、保存逻辑
        if result:
            pass
        # 随机休眠，模拟真人访问节奏，规避风控
        time.sleep(random.uniform(2, 5))

if __name__ == "__main__":
    # 待执行的批量任务列表
    task_urls = [
        "https://example.com/test1",
        "https://example.com/test2",
        "https://example.com/test3"
    ]
    # 无人值守批量跑任务，夜间可挂机运行
    batch_task_runner(task_urls)
    print("本轮批量采集任务执行完成")