最新阿里|腾讯|华为云函数搭建代理池

前言：

• 这段时间在写爬虫项目时，需要代理池绕过反爬，但是看了一下没有便宜一点的代理商家，于是想起来很久以前看到的通过使用云厂商提供的云函数服务来实现代理池的文章，于是就试了一下，发现目前只有阿里云可以实现，但是阿里云变换IP很慢，发很多次请求都用的是同一个IP，但是腾讯云（本文使用Web函数搭建）和华为云（本文使用华为新版云函数搭建）目前不支持API网关触发，所以理论不能使用了，这篇文章就是介绍如何可以在这三个云厂商中都能搭建代理池
• 利用云函数搭建代理池原因：就是正常搭建的云函数如果要访问互联网，它是可以设置为没有固定公网IP地址，当云函数每次都触发且需要访问互联网时，都会给它一个闲置的随机IP，所以就可以实现每次触发函数时都有一个不同的IP，于是就可以搭建代理池了

1 阿里云

• 对于阿里云搭建代理池的话，这个项目当前还可以正常使用，可以根据这个项目搭建
• https://github.com/shimmeris/SCFProxy/blob/main/README_zh.md

这边我也给出函数代码，可以用这个代码直接搭建

# -*- coding: utf8 -*-
import json
from base64 import b64decode, b64encode

import urllib3
urllib3.disable_warnings()


def handler(environ: dict, start_response):
    try:
        request_body_size = int(environ.get('CONTENT_LENGTH', 0))
    except (ValueError):
        request_body_size = 0
    request_body = environ['wsgi.input'].read(request_body_size)

    kwargs = json.loads(request_body.decode("utf-8"))
    kwargs['body'] = b64decode(kwargs['body'])

    http = urllib3.PoolManager(cert_reqs="CERT_NONE")
    # Prohibit automatic redirect to avoid network errors such as connection reset
    r = http.request(**kwargs, retries=False, decode_content=False)

    response = {
        "headers": {k.lower(): v.lower() for k, v in r.headers.items()},
        "status_code": r.status,
        "content": b64encode(r._body).decode('utf-8')
    }

    status = '200 OK'
    response_headers = [('Content-type', 'text/json')]
    start_response(status, response_headers)
    return [json.dumps(response)]

2 腾讯云

2.1 点击函数服务

• 链接：https://console.cloud.tencent.com/scf/list?rid=8&ns=default

2.2 新建函数

• 如下填写

• 函数代码

# -*- coding: utf-8 -*-
import json
from base64 import b64encode, b64decode
import time
import urllib3
from flask import Flask, request, jsonify

urllib3.disable_warnings()
app = Flask(__name__)


@app.route("/proxy", methods=["POST"])
def proxy_request():
    try:
        request_data = request.get_json()

        # 检查必要参数
        body = b64decode(request_data.get('body', '')).decode() if 'body' in request_data else None,
        # 解析请求参数
        required_fields = ['method', 'url']
        if not all(field in request_data for field in required_fields):
            raise ValueError(f"请求必须包含字段: {required_fields}")

        # 2. 准备请求参数
        kwargs = {
            'method': request_data['method'].upper(),
            'url': request_data['url'],
            'headers': request_data.get('headers', {}),
            'body': body,
            'timeout': urllib3.Timeout(connect=5.0, read=10.0)
        }

        # 3. 发送请求（禁用SSL验证仅用于测试，生产环境应配置证书）
        http = urllib3.PoolManager(cert_reqs="CERT_NONE")
        start_time = time.time()
        response = http.request(**kwargs)
        elapsed = time.time() - start_time

        # 4. 构造响应
        proxy_response = {
            "status_code": response.status,
            "headers": {k.lower(): v for k, v in response.headers.items()},
            "body": b64encode(response.data).decode('utf-8') if response.data else "",
            "elapsed": f"{elapsed:.3f}s"
        }

        return jsonify(proxy_response)

    except Exception as e:
        proxy_response = {
            "status_code": 500,
            "headers": {},
            "body": json.dumps({"error": str(e)}),
            "elapsed": "0s"
        }
        return jsonify(proxy_response)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=9000, debug=True)

2.3 新建终端

2.4 安装依赖包

pip install urllib3==1.26.18 -t .

2.5 查看URL

2.6 测试

2.7 base64解码

2.8 使用python调用

import json
import base64
import random
import string

import requests
import urllib3
url = "https://bigfish.ap-beijing.tencentscf.com/proxy"

# 需要发送的数据
data = {
    "method": "POST",
    "url": "https://ipinfo.io",
    "headers": {"Content-Type": "application/json"},
    "body": base64.b64encode(json.dumps({"username": "123456"}).encode()).decode()  # Base64 编码 body
    # "body": "eyJ1c2VybmFtZSI6ICIxMjM0NTYifQ=="
}
# 发送 POST 请求
response = requests.post(url, json=data)
# 解析返回的数据
if response.status_code == 200:
    res_json = response.json()
    decoded_content = base64.b64decode(res_json["body"]).decode()
    print("解码后的内容:", decoded_content)
else:
    print(f"请求失败，状态码: {response.status_code}")
    print("返回内容:", response.text)

3 华为云

3.1 新建函数

• 链接：https://developer.huawei.com/consumer/cn/service/josp/agc/index.html#/myProject/461323198429810208/9249519184595931747

3.2 上传代码

• 代码地址

通过网盘分享的文件：hwy.zip
链接: https://pan.baidu.com/s/1hrsrshlortj_KaJNpdOjSQ 提取码: bxah 
--来自百度网盘超级会员v2的分享

• 如果想自己制作文件，请查看后面备注

3.3 测试

3.4 发布版本

• 好像发布版本之后才能使用url

3.5 查看URL

3.6 URL测试

3.7 使用python调用

import json
import base64
import random
import string

import requests
import urllib3
url = "https://bigfish.ap-beijing.tencentscf.com/proxy"

# 需要发送的数据
data = {
    "method": "POST",
    "url": "https://ipinfo.io",
    "headers": {"Content-Type": "application/json"},
    "body": base64.b64encode(json.dumps({"username": "123456"}).encode()).decode()  # Base64 编码 body
    # "body": "eyJ1c2VybmFtZSI6ICIxMjM0NTYifQ=="
}
# 发送 POST 请求
response = requests.post(url, json=data)
# 解析返回的数据
if response.status_code == 200:
    res_json = response.json()
    decoded_content = base64.b64decode(res_json["body"]).decode()
    print("解码后的内容:", decoded_content)
else:
    print(f"请求失败，状态码: {response.status_code}")
    print("返回内容:", response.text)

备注

• 制作文件文档：https://developer.huawei.com/consumer/cn/doc/AppGallery-connect-Guides/developfunc-python-0000001512919930#section475324014138
• 附上述华为云函数代码

# -*- coding: utf8 -*-
from pythonRT import LogLevel, Context
import json
from base64 import b64decode, b64encode
import requests
import urllib3
urllib3.disable_warnings()
import time

def test(environ: dict, start_response):
    resp = requests.get("http://ipinfo.io").json()
    return resp

def myHandler(event: dict, context: Context):
    try:
        request_data = event

        # 检查必要参数
        body = b64decode(request_data.get('body', '')).decode() if 'body' in request_data else None
        # 解析请求参数
        required_fields = ['method', 'url']
        if not all(field in request_data for field in required_fields):
            raise ValueError(f"请求必须包含字段: {required_fields}")

        # 2. 准备请求参数
        kwargs = {
            'method': request_data['method'].upper(),
            'url': request_data['url'],
            'headers': request_data.get('headers', {}),
            'body': body,
            'timeout': urllib3.Timeout(connect=5.0, read=10.0)
        }

        # 3. 发送请求（禁用SSL验证仅用于测试，生产环境应配置证书）
        http = urllib3.PoolManager(cert_reqs="CERT_NONE")
        start_time = time.time()
        response = http.request(**kwargs)
        elapsed = time.time() - start_time

        # 4. 构造响应
        proxy_response = {
            "status_code": response.status,
            "headers": {k.lower(): v for k, v in response.headers.items()},
            "body": b64encode(response.data).decode('utf-8') if response.data else "",
            "elapsed": f"{elapsed:.3f}s"
        }

        return proxy_response

    except Exception as e:
        return {
            "status_code": 500,
            "headers": {},
            "body": json.dumps({"error": str(e)}),
            "elapsed": "0s"
        }