第五课：代理池

在使用 Requests 爬虫或者在命令行中使用网络请求命令（比如 curl）时，可以手动指定代理服务器，从而实现请求被转发的效果。

（一）requests 代理

一个普通的爬取百度首页的爬虫如下：

import requests

headers={'User-Agent':
         'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36 Edg/119.0.0.0'}
response=requests.get('http://www.baidu.com/',headers=headers)
print("访问结果：",response.status_code)
print(response.text[:500])

但是现在我们直接使用了自己的 IP 去爬。而对于某些网站而言，如果短时间内访问的次数太多，就会触发警报，限制这个 IP 的访问请求，从而导致爬取失败。因此可以使用 **代理（Proxy）**的方式，转发本地的访问请求。例如使用一个网上的公共代理IP地址：43.248.63.89:80

import requests

proxy = "43.248.63.89:80"
headers={'User-Agent':
         'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36 Edg/119.0.0.0'}
proxies = {
        "http": f"http://{proxy}",
        "https": f"http://{proxy}",
    }
response=requests.get('http://www.baidu.com/',headers=headers)
print("访问结果：",response.status_code)
print(response.text[:500])

现在，我们对百度的请求就是通过这个节点转发的。同样的道理，如果我们有很多的代理IP，并在爬虫过程中不停地切换IP，那么就可以大大降低被检测到的概率。这样的代理IP集合被称为 代理池（ProxyPool）

（二）代理地址的分类

我们可以从三个维度来划分代理：

1. 按协议类型分类

类型	描述	应用场景
HTTP 代理	只代理 HTTP 请求，不支持 HTTPS	适合请求纯 HTTP 网站
HTTPS 代理（也叫 HTTP CONNECT）	支持 HTTPS，能建立加密隧道	最常见，适用于抓 HTTPS 网站，如 Google、百度
SOCKS5 代理	底层代理协议，支持任意 TCP（甚至 UDP）连接，不限于 HTTP 协议	最强大，适用于翻墙、P2P、SSH、邮件等任意流量
SOCKS4	SOCKS5 的早期版本，不支持认证	基本已淘汰

说明：

requests 默认支持 HTTP/HTTPS

SOCKS5 需要用第三方包（如 requests[socks] 或 PySocks）

2. 按 IP 类型分类

类型	描述	优点	缺点
数据中心代理（Datacenter Proxy）	来自云服务器（如 AWS、阿里云）	快速、便宜、易得	容易被识别为爬虫
住宅代理（Residential Proxy）	分配给真实家庭用户的 IP，ISP（互联网服务供应商）提供	不易被封，模拟真实用户	贵，有时较慢
移动代理（Mobile Proxy）	来自 3G/4G/5G 蜂窝网络的设备	最难识别，非常稳定	最贵，资源稀缺

一般公开免费代理几乎都是数据中心代理。

3. 按匿名等级分类

类型	描述	特点
透明代理（Transparent Proxy）	会把你的真实 IP 传给目标网站（如 HTTP header 中）	不推荐爬虫
匿名代理（Anonymous Proxy）	不传真实 IP，但会告诉目标服务器你用了代理	对付一般反爬可以
高匿名代理（Elite Proxy）	不传真实 IP，也不会透露你用了代理	最推荐，最安全

举个例子：比如你在 spys.one 上看到这一行：

45.58.233.115:3128	HTTP	NOA	United States	匿名（Anonymous）	高匿名（Elite）

你可以这样理解：

3128：这是端口号，常见于 Squid HTTP 代理
HTTP：这是协议类型（不能用于 HTTPS 或 SOCKS）
NOA：表示 匿名级别（Non-Anonymous, Anonymous, Elite）
United States：代理所在国家

（三）公开代理池

事实上在互联网上，我们可以找到很多免费的公开代理源

1. 免费节点的来源

类型	描述
人为公开分享的节点	出于技术测试、流量推广、数据收集，或“免费试用”引导付费购买等目的而公开共享的代理节点。
开放的 HTTP/HTTPS/SOCKS 代理端口	一些服务器配置不当，暴露了无认证的公共代理端口，如常见的 8080、3128、1080 等。常见软件包括：`Squid`、`Mikrotik`、`OpenProxy`、`CCProxy` 等。有时也可能是公司或学校忘记设置访问限制。
肉鸡代理（Botnet Proxies）	黑产通过恶意软件感染设备（如路由器、PC），将其变为代理中转节点（“肉鸡”），并将这些代理发布为“免费节点”。这类代理风险极高，可能涉及非法行为，使用者也可能被追踪。

2. 获取免费节点

这里列出了部分收集免费代理的网站：

国外：

国内：

纯文本：

github 开源代理池工具：

danger

这些代理很多不稳定，而且存在隐私泄露或法律风险，不建议用于敏感操作。可以作为学习/实验用途。
免费代理中 能访问 Google 的比例极低，你可能测试 50 个都找不到 1 个。

3. 筛选免费节点

你可以用爬虫写个自动抓取脚本定期收集这些网站上的公开 IP 列表。但是获取到的 IP 大部分是不可用的。

网站上显示的 “节点在线” 并不是指 “代理正常”，而是指 “该服务器正常工作并可以 ping 通”。

代理 IP 能 ping 通 ≠ 它能作为代理用”。ping 成功，只说明：

IP 地址本身存在
对方服务器的 ICMP 协议响应是开放的

但代理能不能用，还取决于：

目标端口（如 3128）是否开放
该端口是否真的在跑代理服务（如 Squid）
代理服务是否允许你访问（是否限制来源 IP）
代理服务是否允许访问你目标网站（比如 http://www.google.com）

这些条件必须全部满足才可以

可以先写个小脚本检查单个节点的代理可用性：

import requests
def test_proxy(proxy):
    # 测试国外节点
    URL_list = ['http://www.google.com','http://www.baidu.com']
    for URL in URL_list:
        try:
            response = requests.get(URL, proxies={
                "http": proxy,
                "https": proxy
            }, timeout=5)
            print(f"可以访问 {URL}:{response.status_code ==200}")
        except Exception as e:
            print(f"[✗] 无法连接 {URL}")
    
# 示例代理
proxy = '119.28.51.117:80'
test_proxy(f"http://{proxy}")

也可以批量爬取并测试 IP 节点，这里以上面提供的比较好处理的 ”纯文本“ 类型为例

获取节点列表

import requests
import aiohttp
import asyncio
import time
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36'
}

def get_free_proxies(url,file):
    try:
        response = requests.get(url,headers=headers, timeout=3)
        proxy_list = response.text.split('\n')
        with open(file,'a',encoding='utf-8') as f:
            for proxy in proxy_list:
                f.write(proxy+'\n')
        print(f"[✓] 成功连接：{url}")
        return len(proxy_list)
    except Exception as e:
        print(f"[✗] 无法连接到 {url}")
        return 0

异步检测单个代理是否可用

async def check_proxy(session, proxy,testurl):
    proxy_url = f"http://{proxy}"
    try:
        async with session.get(testurl, proxy=proxy_url, headers=headers, timeout=6) as response:
            if response.status == 200:
                print(f"[✓] 有效代理：{proxy}")
                return proxy
    except Exception as e:
        print(f"[✗] 无效代理：{proxy}，错误类型：{type(e).__name__}，错误信息：{str(e)}")

    return None

异步主函数

async def main(original_file,testurl):
    tasks = []
    connector = aiohttp.TCPConnector(ssl=False)
    timeout = aiohttp.ClientTimeout(total=10)
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        with open(original_file,'r',encoding='utf-8') as f:
            for line in f:
                proxy = line.replace('\n','')
                task = asyncio.create_task(check_proxy(session, proxy,testurl))
                tasks.append(task)
        results = await asyncio.gather(*tasks)

    # 过滤掉 None 的结果
    valid_proxies = list(filter(None, results))
    return valid_proxies

主函数

def main_get_proxy(original_file,test_url):
    url_list = [
        'https://www.proxy-list.download/api/v1/get?type=http',
        'https://api.proxyscrape.com/?request=displayproxies&proxytype=http&timeout=5000',
        'https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt',
        'https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt',
        'https://api.openproxylist.xyz/http.txt',
        'https://proxy.scdn.io/text.php'
                ]

    # 清空文件
    with open(original_file,'a+',encoding='utf-8') as test:
        test.truncate(0)
    print('文件初始化完成，开始爬取原始节点：')

    # 爬取原始节点
    number = 0
    for url in url_list:
        number+= get_free_proxies(url,original_file)
    print(f"共获取到了{ number} 个节点,已保存到{original_file}")

    print("开始检测代理可用性：")
    time.sleep(3)
    start = time.time()
    valid_proxies = asyncio.run(main(original_file,test_url))
    print(f"\n✅ 检测完毕，用时：{time.time() - start:.2f} 秒")
    return valid_proxies

调用方式

if __name__=="__main__":
    # 节点获取
    original_nodes_file = './original_nodes.txt'
    test_url = 'http://www.github.com'

    valid_proxies = main_get_proxy(original_nodes_file,test_url)

    # 保存到文件
    save_file  = f"./{test_url.split('//')[1]}.txt"
    with open(save_file, 'w') as f:
        for p in valid_proxies:
            f.write(p + '\n')
    print(f'可用节点已经保存到：{save_file}')

    print("\n 可用代理列表：")
    for p in valid_proxies:
        print(p)

这里写的这个函数 main_get_proxy 可以获取对指定网站（如 www.baidu.com）可代理访问的所有节点，并返回一个列表 valid_proxies

4. 使用Proxypool

现在已经获得了一个 valid_proxies 列表，里面包含了所有可以成功代理访问 Baidu 的节点。如何使用这个 Proxypool 呢？

使用 requests + 随机代理访问百度

import random
import requests

# 假设 valid_proxies 是像这样的列表
# valid_proxies = ['http://123.123.123.123:8080', 'http://111.111.111.111:3128', ...]

def get_random_proxy():
    return random.choice(valid_proxies)

def fetch_with_proxy(url):
    proxy = get_random_proxy()
    proxies = {
        "http": proxy,
        "https": proxy,
    }
    try:
        response = requests.get(url, proxies=proxies, timeout=5)
        print(f"[✓] Success with proxy: {proxy}")
        return response.text
    except Exception as e:
        print(f"[✗] Failed with proxy: {proxy} | {e}")
        return None

# 示例：访问百度
html = fetch_with_proxy("http://www.baidu.com")

或者循环使用所有代理（失败就换一个）

def fetch_with_proxypool(url):
    for proxy in valid_proxies:
        try:
            proxies = {"http": proxy, "https": proxy}
            response = requests.get(url, proxies=proxies, timeout=5)
            print(f"[✓] Success with proxy: {proxy}")
            return response.text
        except:
            print(f"[✗] Failed with proxy: {proxy}")
    print("所有代理都失败了。")
    return None

# 使用
html = fetch_with_proxypool("http://www.baidu.com")

你还可以封装成一个 ProxyPool 类：

class ProxyPool:
    def __init__(self, proxy_list):
        self.proxies = proxy_list
        self.index = 0

    def get_next(self):
        if not self.proxies:
            raise Exception("代理池为空！")
        proxy = self.proxies[self.index]
        self.index = (self.index + 1) % len(self.proxies)
        return proxy

# 使用
pool = ProxyPool(valid_proxies)

def fetch(url):
    for _ in range(len(valid_proxies)):
        proxy = pool.get_next()
        try:
            response = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=5)
            print(f"[✓] 成功代理：{proxy}")
            return response.text
        except:
            print(f"[✗] 失败代理：{proxy}")
    print("全部代理失败。")
    return None

（四）代理蜜罐

为了诱捕滥用公开代理的攻击者和爬虫，这里我们在 Vultr 上购买了一台 Ubuntu24 云服务器，作为代理的话 1核1GB 已经够用。我们将在服务器上部署一个 代理监听蜜罐，记录代理使用者的 IP、访问的 URL、请求内容等信息。并且只允许HTTP 代理，遇到 HTTPS 则拒绝代理。这是因为 HTTPS 中间过程是加密的，没有记录价值。

1. 安装 `mitmproxy`

我们使用 mitmproxy 来实现这个目标，这是一个功能强大且非常适合搭建 HTTP/HTTPS 代理蜜罐的工具。

执行以下命令安装：

sudo apt update
sudo apt install python3-pip -y

创建一个新的虚拟环境

python3 -m venv env1

激活虚拟环境

source env1/bin/activate

在虚拟环境中安装 mitmproxy

pip install --upgrade pip
pip install mitmproxy

检查版本

mitmproxy --version

创建 mitmproxy 配置文件目录

mkdir -p ~/.mitmproxy

创建配置文件 config.yaml，

vim ~/.mitmproxy/config.yaml

填入以下内容

block_global: false
listen_host: 0.0.0.0
listen_port: 8080
mode:
  - regular

你需要开放服务器的 8080 端口让外部 IP 可以访问：

sudo ufw allow 8080
sudo ufw enable

2. 制作透明代理

运行 mitmproxy 的 HTTP 透明代理模式（监听 8080 端口）

我们用 mitmdump（mitmproxy 的无界面版本）+ 日志记录脚本实现监听。

创建自定义脚本 proxy_logger.py

from mitmproxy import http
from datetime import datetime

def request(flow: http.HTTPFlow) -> None:
    # 这是 mitmproxy 提供的 事件钩子函数（event hook），在每个 HTTP 请求被代理接收到时调用。
    # flow 是一个 HTTPFlow 对象，它包含请求的所有信息，例如 IP、URL、请求体、请求头等等
    
    client_ip = flow.client_conn.address[0]
    # flow.client_conn.address 是一个元组，格式为 (IP地址, 端口号)，这里只取 IP 部分。
    method = flow.request.method
    # 请求方法，如 `GET`, `POST`, `PUT`, `DELETE` 等。
    url = flow.request.pretty_url
    # 获取完整 URL（包括协议、主机、端口、路径和查询参数）。pretty_url 会自动格式化为标准可读形式。
    headers = dict(flow.request.headers)
    # 请求头，是一个多重字典对象，这里转换为普通 `dict` 方便打印和处理。
    body = flow.request.get_text()
    # 如果请求有 body（如 POST 表单或 JSON），获取文本内容。

    # 日志记录内容拼接。这部分就是拼接最终写入日志的格式。
    log = (
        f"\n[请求时间] {datetime.now()}\n"
        f"[客户端 IP] {client_ip}\n"
        f"[请求方法] {method}\n"
        f"[请求 URL] {url}\n"
        f"[请求头] {headers}\n"
        f"[请求内容] {body}\n"
        f"{'-'*60}\n"
    )

    # 写入日志文件
    with open("proxy_log.txt", "a", encoding="utf-8") as f:
        f.write(log)

启动代理监听蜜罐

mitmdump -s proxy_logger.py

此时 mitmdump 会自动加载 ~/.mitmproxy/config.yaml 中的配置

整个流程如下：

有人通过代理访问 HTTP 网站（比如访问你服务器的 45.32.219.173:8080）。
mitmdump 捕获这个 HTTP 请求并触发 request(flow) 函数。
函数提取请求中的 IP、方法、URL、头部和正文。写入 proxy_log.txt

注意：

flow 对象中还能获取响应、TLS、连接信息等。
response(flow)、error(flow) 是另外两个钩子函数，分别处理响应和异常。
你甚至可以修改请求或响应内容，对蜜罐行为做出伪造响应。

3. 测试代理

在你的本地电脑上，配置代理为你服务器的 IP 和端口 8080：

import requests
def test_proxy(proxy):
    # 测试国外节点
    # URL_list = ['http://www.google.com','http://www.baidu.com','http://www.github.com']
    URL_list = ['http://example.com', 'http://neverssl.com']  # 只使用 HTTP 的网站
    for URL in URL_list:
        try:
            response = requests.get(URL, proxies={
                "http": proxy,
                "https": proxy
            }, timeout=5)
            print(f"可以访问 {URL}:{response.status_code ==200}")
        except Exception as e:
            print(f"[✗] 无法连接 {URL}, 错误：{type(e).__name__}：{e}")
    
# 示例代理
proxy = '45.32.219.173:8080'
test_proxy(f"http://{proxy}")

连接都成功

4. 记录地理地址

接下来我们来实现：记录每次请求的地理位置（国家、省份、城市等）。

我们使用百度提供的 IP归属地查询API

接口示例：https://opendata.baidu.com/api.php?co=&resource_id=6006&oe=utf8&query=51.158.68.133

响应：

{
  "status": "0",
  "t": "",
  "set_cache_time": "",
  "data": [
    {
      "ExtendedLocation": "",
      "OriginQuery": "51.158.68.133",
      "SchemaVer": "",
      "appinfo": "",
      "disp_type": 0,
      "fetchkey": "51.158.68.133",
      "location": "法国巴黎",
      "origip": "51.158.68.133",
      "origipquery": "51.158.68.133",
      "resourceid": "6006",
      "role_id": 0,
      "schemaID": "",
      "shareImage": 1,
      "showLikeShare": 1,
      "showlamp": "1",
      "strategyData": {

      },
      "titlecont": "IP地址查询",
      "tplt": "ip"
    }
  ]
}

因此我们要先写一个函数，写入 get_online_location.py

import requests

ip_cache = {}
def get_online_location(ip):
    if ip in ip_cache:
        return ip_cache[ip]

    try:
        url = f"https://opendata.baidu.com/api.php?co=&resource_id=6006&oe=utf8&query={ip}"
        resp = requests.get(url, timeout=3)
        data = resp.json()

        if data["status"] == "0" and "data" in data and len(data["data"]) > 0:
            location = data["data"][0].get("location", "Unknown")
        else:
            location = "Unknown"

    except Exception:
        location = "Unknown"

    ip_cache[ip] = location
    return location


if __name__=="_main__":
    print(get_online_location("51.158.68.133"))
    # 输出：法国巴黎

总文件 proxy_logger.py

from mitmproxy import http
from datetime import datetime
import os
from get_online_location import get_online_location


# 保证 logs 目录存在
log_dir = "logs"
os.makedirs(log_dir, exist_ok=True)
log_filename = os.path.join(log_dir, datetime.now().strftime("%Y-%m-%d") + ".log")  # 每天一个日志文件
with open(log_filename,'a+',encoding='utf-8') as test: # 文件清空
    test.truncate(0)
print("文件初始化完成")
print("开始监听……")

def request(flow: http.HTTPFlow) -> None:
    client_ip = flow.client_conn.address[0]
    method = flow.request.method
    url = flow.request.pretty_url
    headers = dict(flow.request.headers)
    body = flow.request.get_text()

    # 尝试查询地理位置
    location = get_online_location(client_ip)

    # 记录日志
    log = (
        f"\n[请求时间] {datetime.now()}\n"
        f"[客户端 IP] {client_ip}\n"
        f"[地理位置] {location}\n"
        f"[请求方法] {method}\n"
        f"[请求 URL] {url}\n"
        f"[请求头] {headers}\n"
        f"[请求内容] {body}\n"
        f"{'-'*60}\n"
    )
    print(f"[{datetime.now().strftime('%H:%M:%S')}][{client_ip}] {location} -> {method} {url}")


    with open(log_filename, "a", encoding="utf-8") as f:
        f.write(log)

启动命令

mitmdump -s proxy_logger.py --quiet

5. 数据库化

接下来安装一个轻量数据库，在记录日志的同时存入数据库

由于需求是轻量日志存储，SQLite 是理想选择：

单个 .db 文件存储，零配置
Python 自带 sqlite3 模块，无需安装依赖
性能足以应对中低流量蜜罐

设计的表名：proxy_logs 字段建议如下：

字段名	类型	含义
`id`	INTEGER	自增主键
`time`	TEXT	请求时间（ISO 格式）
`client_ip`	TEXT	请求源 IP
`location`	TEXT	地理位置
`method`	TEXT	HTTP 请求方法
`url`	TEXT	完整 URL
`headers`	TEXT	请求头 JSON 字符串
`body`	TEXT	请求体内容

出于函数化思想，我们创建一个文件 database_operation.py ,里面包含所有的数据库操作函数

当前的文件结构如下：

.
├── proxy_logger.py           # 主逻辑脚本
├── database_operation.py     # 数据库操作模块
├── get_online_location.py    # 获取地理位置
└── logs/
	├── 2025-03-15.log    	  # 日志文件
    └── proxy_logs.db         # SQLite 数据库文件

database_operation.py 中包含这几个函数

初始化数据库并创建表（如不存在)

# database_operation.py
import sqlite3
import json
import os

DB_PATH = os.path.join("logs", "proxy_logs.db")

def init_db():
    os.makedirs("logs", exist_ok=True)
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS proxy_logs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            time TEXT,
            client_ip TEXT,
            location TEXT,
            method TEXT,
            url TEXT,
            headers TEXT,
            body TEXT
        )
    ''')
    conn.commit()
    conn.close()

清空日志文件内容

def clear_db():
    """清空日志表内容"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute('DELETE FROM proxy_logs')
    conn.commit()
    conn.close()

插入日志记录

def insert_log(time, client_ip, location, method, url, headers, body):
    """插入一条日志记录"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute('''
        INSERT INTO proxy_logs (time, client_ip, location, method, url, headers, body)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    ''', (
        time,
        client_ip,
        location,
        method,
        url,
        json.dumps(headers, ensure_ascii=False),
        body
    ))
    conn.commit()
    conn.close()

主文件只需这样引入并使用这几个函数：

from mitmproxy import http
from datetime import datetime
import os
from get_online_location import get_online_location
from database_operation import *


# 保证 logs 目录存在
log_dir = "logs"
os.makedirs(log_dir, exist_ok=True)
log_filename = os.path.join(log_dir, datetime.now().strftime("%Y-%m-%d") + ".log")  # 每天一个日志文件
with open(log_filename,'a+',encoding='utf-8') as test: # 文件清空
    test.truncate(0)
print("文件初始化完成")
print("开始监听……")


# 初始化数据库（如果首次启动）
init_db()

# 询问是否清空数据库
if input("是否清空数据库(y 为清空): ") == 'y':
    clear_db()
    print('数据库清空完毕')
else:
    print("数据库保留")



def request(flow: http.HTTPFlow) -> None:
    client_ip = flow.client_conn.address[0]
    method = flow.request.method
    url = flow.request.pretty_url
    headers = dict(flow.request.headers)
    body = flow.request.get_text()

    # 尝试查询地理位置
    location = get_online_location(client_ip)

    # 记录日志
    log = (
        f"\n[请求时间] {datetime.now()}\n"
        f"[客户端 IP] {client_ip}\n"
        f"[地理位置] {location}\n"
        f"[请求方法] {method}\n"
        f"[请求 URL] {url}\n"
        f"[请求头] {headers}\n"
        f"[请求内容] {body}\n"
        f"{'-'*60}\n"
    )
    insert_log(
        time=datetime.now().isoformat(),
        client_ip=client_ip,
        location=location,
        method=method,
        url=url,
        headers=headers,
        body=body
    )
    print(f"[{datetime.now().strftime('%H:%M:%S')}][{client_ip}] {location} -> {method} {url}")


    with open(log_filename, "a", encoding="utf-8") as f:
        f.write(log)

6. 在线可视化

接下来使用 Flask 读取 /root/box/logs/proxy_logs.db 数据库文件中的日志记录，并以网页方式展示日志内容，实现基本的数据可视化。我们从最基础的可视化开始做起，比如表格展示所有访问日志。

当前目录结构

├── app.py                  # Flask 主应用
├── templates/
│   └── index.html          # 日志展示页面
└── static/   

其中，主文件 app.py（默认展示最近 100 条访问日志）

from flask import Flask, render_template
import sqlite3
import os

app = Flask(__name__)
DB_PATH = os.path.join("../box/logs", "proxy_logs.db")

@app.route('/')
def index():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute('SELECT time, client_ip, location, method, url FROM proxy_logs ORDER BY id DESC LIMIT 100')
    logs = cursor.fetchall()
    conn.close()
    
    return render_template("index.html", logs=logs)

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

然后在 templates/index.html中创建以下内容：

<!DOCTYPE html>
<html lang="zh">
<head>
    <meta charset="UTF-8">
    <title>代理蜜罐日志展示</title>
    <style>
        table { width: 100%; border-collapse: collapse; font-family: Arial; }
        th, td { border: 1px solid #ccc; padding: 8px; text-align: left; font-size: 14px; }
        th { background-color: #f2f2f2; }
        body { margin: 30px; }
        h1 { font-size: 24px; }
    </style>
</head>
<body>
    <h1>最近访问日志</h1>
    <table>
        <tr>
            <th>时间</th>
            <th>IP地址</th>
            <th>地理位置</th>
            <th>请求方法</th>
            <th>请求 URL</th>
        </tr>
        {% for row in logs %}
        <tr>
            <td>{{ row[0] }}</td>
            <td>{{ row[1] }}</td>
            <td>{{ row[2] }}</td>
            <td>{{ row[3] }}</td>
            <td>{{ row[4] }}</td>
        </tr>
        {% endfor %}
    </table>
</body>
</html>

启动命令：

python3 app.py

记得打开防火墙

（一）requests 代理​

（二）代理地址的分类​

1. 按协议类型分类​

2. 按 IP 类型分类​

3. 按匿名等级分类​

（三）公开代理池​

1. 免费节点的来源​

2. 获取免费节点​

3. 筛选免费节点​

4. 使用Proxypool​

（四）代理蜜罐​

1. 安装 mitmproxy​

2. 制作透明代理​

3. 测试代理​

4. 记录地理地址​

5. 数据库化​

6. 在线可视化​