如何在Python中代理转发GET请求？含m3u8播放列表场景

Got it, let's walk through two solid approaches to build this proxy—one with Flask for simplicity, and another with aiohttp for better performance with streaming content like m3u8 playlists.

方案1：基于Flask的同步代理（适合快速搭建） #

这个方案用Flask处理客户端请求，借助requests库转发到目标流媒体站点，适合入门或低并发场景。

步骤1：安装依赖 #

pip install flask requests

步骤2：实现代码 #

from flask import Flask, request, Response
import requests

app = Flask(__name__)

# 替换为你的目标流媒体站点基础URL
TARGET_STREAMING_URL = "https://your-target-stream-site.com"

@app.route('/', defaults={'path': ''}, methods=['GET'])
@app.route('/<path:path>', methods=['GET'])
def proxy(path):
    # 拼接完整目标URL，保留客户端的查询参数
    target_url = f"{TARGET_STREAMING_URL}/{path}"
    if request.query_string:
        target_url += f"?{request.query_string.decode('utf-8')}"
    
    # 转发客户端请求头，调整Host避免目标站点拒绝
    headers = dict(request.headers)
    headers.pop('Host', None)
    headers['Host'] = TARGET_STREAMING_URL.split('//')[1]

    try:
        # 启用流式响应，避免大文件占用过多内存（适配m3u8流媒体）
        response = requests.get(target_url, headers=headers, stream=True)
        
        # 把目标站点的响应原封不动返回给客户端
        return Response(
            response.iter_content(chunk_size=1024),
            status=response.status_code,
            headers=dict(response.headers)
        )
    except requests.exceptions.RequestException as e:
        return f"Proxy error: {str(e)}", 500

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

关键细节说明 #

• 流式响应：用stream=True和iter_content确保m3u8分片这类大内容不会一次性加载到内存，适配流媒体场景。
• 请求头处理：转发时调整Host头，避免目标站点因来源不匹配拒绝请求。
• 查询参数保留：确保客户端传递的播放参数（如token、清晰度选项）能正确传递到目标站点。

方案2：基于aiohttp的异步代理（适合高并发流媒体） #

如果你的代理需要处理大量并发请求（比如多人同时访问m3u8），异步的aiohttp会比同步Flask更高效，不会阻塞线程。

步骤1：安装依赖 #

pip install aiohttp

步骤2：实现代码 #

from aiohttp import web, ClientSession

# 替换为你的目标流媒体站点基础URL
TARGET_STREAMING_URL = "https://your-target-stream-site.com"

async def proxy_handler(request):
    # 构建目标URL，保留路径和查询参数
    path = request.match_info.get('path', '')
    target_url = f"{TARGET_STREAMING_URL}/{path}"
    if request.query_string:
        target_url += f"?{request.query_string}"
    
    # 转发请求头，调整Host
    headers = dict(request.headers)
    headers.pop('Host', None)
    headers['Host'] = TARGET_STREAMING_URL.split('//')[1]

    async with ClientSession() as session:
        try:
            # 异步发送GET请求，启用流式响应
            async with session.get(target_url, headers=headers) as resp:
                # 把目标站点的响应原封不动返回给客户端
                return web.Response(
                    status=resp.status,
                    headers=resp.headers,
                    body=resp.content
                )
        except Exception as e:
            return web.Response(text=f"Proxy error: {str(e)}", status=500)

app = web.Application()
app.add_routes([web.get('/{path:.*}', proxy_handler)])

if __name__ == '__main__':
    web.run_app(app, host='0.0.0.0', port=5000)

额外优化：m3u8内容修正 #

如果目标站点返回的m3u8播放列表里是相对路径或原站点绝对路径，客户端可能无法通过代理访问分片。可以在返回前自动修正链接：

# 仅针对.m3u8请求处理内容
if path.endswith('.m3u8'):
    m3u8_content = await resp.text()
    # 替换原站点URL为你的代理服务器URL
    modified_content = m3u8_content.replace(TARGET_STREAMING_URL, "http://your-proxy-server.com")
    return web.Response(
        status=resp.status,
        headers=resp.headers,
        text=modified_content
    )

把这段代码插入到异步代理的async with session.get(...) as resp:块中，就能自动修正m3u8里的分片链接，确保客户端通过代理正常加载流媒体。

内容的提问来源于stack exchange，提问作者Ofek Ezon