import pymysql# 获取数据库连接
db = pymysql.connect(host='localhost', user='root', passwd='123456', port=3306, db='proxy')
# 获取游标
cursor = db.cursor()
# 若无 proxy 数据库可执行如下操作
# cursor.execute("create database proxy default character set utf8")
# 删除之前创建的表格,不然新的数据会直接添加在旧数据之后,造成数据冗余
cursor.execute("drop table proxy_pool")
create_sql = "create table if not exists proxy_pool(" \"代理IP地址 varchar(255) not null," \"端口 int not null," \"匿名度 text not null," \"类型 varchar(255) not null," \"支持 varchar(255) not null," \"位置 text not null," \"响应速度 text not null," \"最后验证时间 varchar(255) not null)engine=innodb default charset=utf8;"
cursor.execute(create_sql)

proxy_dir = 'proxy_pool.txt'
# 以追加模式写入
with open(proxy_dir, 'a', encoding='utf-8') as f:# json.dumps序列化时对中文默认使用的ascii编码,想输出真正的中文需要指定ensure_ascii=Falsef.write(json.dumps(content, ensure_ascii=False) + '\n')

# @Author : Yy_Rose
import requests
from lxml import etree
import json
from requests.exceptions import RequestException
import time
import pymysql# 获取数据库连接
db = pymysql.connect(host='localhost', user='root', passwd='123456', port=3306, db='proxy')
# 获取游标
cursor = db.cursor()
# 若无 proxy 数据库可执行如下操作
# cursor.execute("create database proxy default character set utf8")
# 删除之前创建的表格,不然新的数据会直接添加在旧数据之后,造成数据冗余
cursor.execute("drop table proxy_pool")
create_sql = "create table if not exists proxy_pool(" \"代理IP地址 varchar(255) not null," \"端口 int not null," \"匿名度 text not null," \"类型 varchar(255) not null," \"支持 varchar(255) not null," \"位置 text not null," \"响应速度 text not null," \"最后验证时间 varchar(255) not null)engine=innodb default charset=utf8;"
# 执行数据库操作
cursor.execute(create_sql)# 构建请求头
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64)'' AppleWebKit/537.36 (KHTML, like Gecko)'' Chrome/86.0.4240.198 Safari/537.36','Cookie': 'Hm_lvt_c4dd741ab3585e047d56cf99ebbbe102=1640671968,''1641362375;'' Hm_lpvt_c4dd741ab3585e047d56cf99ebbbe102=1641362386'
}# 获取 HTTPResponse 类型的对象
def get_one_page(url):try:response = requests.get(url, headers=headers)if response.status_code == 200:# 编码response.encoding = 'gbk'return response.textreturn Noneexcept RequestException:return None# 解析获取内容
def parse_one_page(proxy_pool, html, ip_and_port):# 将 html 字符串转换为 _ELEMENT 对象content = etree.HTML(html)# 第一行为 th 标签内容，故从第二行开始tr_cnotent_list = content.xpath("//div[@id='list']//tr")[1:]for proxy_infos in tr_cnotent_list:# strip() 方法用于移除字符串头尾指定的字符（默认为空格或换行符）或字符序列proxy_pool.append([proxy_infos[0].text.strip(),proxy_infos[1].text.strip(),proxy_infos[2].text.strip(),proxy_infos[3].text.strip(),proxy_infos[4].text.strip(),proxy_infos[5].text.strip(),proxy_infos[6].text.strip(),proxy_infos[7].text.strip()])ip_and_port.append(proxy_infos[0].text.strip() + ':' + proxy_infos[1].text.strip())ip = proxy_infos[0].text.strip()port = proxy_infos[1].text.strip()anonymity = proxy_infos[2].text.strip()ip_type = proxy_infos[3].text.strip()support_type = proxy_infos[4].text.strip()position = proxy_infos[5].text.strip()response_speed = proxy_infos[6].text.strip()verify_time = proxy_infos[7].text.strip()# 将获取到的数据插入到数据库中insert_into = ("INSERT INTO proxy_pool(代理IP地址,端口,匿名度,类型,支持,位置,响应速度,最后验证时间)""VALUES(%s,%s,%s,%s,%s,%s,%s,%s);")data_into = (ip, port, anonymity, ip_type, support_type, position,response_speed,verify_time)cursor.execute(insert_into, data_into)# 将 insert 语句提交给数据库db.commit()# 写入本地 txt
def wirte_to_txt(content):proxy_dir = 'proxy_pool.txt'# 以追加模式写入with open(proxy_dir, 'a', encoding='utf-8') as f:# json.dumps序列化时对中文默认使用的ascii编码,想输出真正的中文需要指定ensure_ascii=Falsef.write(json.dumps(content, ensure_ascii=False) + '\n')# 打印内容到控制台
def print_list(proxy_list):for i in range(10):print(proxy_list[i])wirte_to_txt(proxy_list[i])# 测试IP是否可用
def ip_test(ip_and_port_list, usable_ip_list):url = "https://www.baidu.com/"try:for ip in ip_and_port_list:try:response = requests.get(url, proxy={'https': ip},headers=headers, timeout=0.5)if response.status_code == 200:usable_ip_list.append(ip)print("该代理ip有效:" + ip)else:print("该代理ip无效:" + ip)except:print("该代理ip无效:" + ip)except:return Nonedef main(offset):# 分页爬取url = 'http://www.ip3366.net/?stype=1&page=' + str(offset)html = get_one_page(url)proxy_pool = []usable_ip = []ip_port = []parse_one_page(proxy_pool, html, ip_port)print("以下为查询到的数据:")print_list(proxy_pool)print('=' * 50)ip_test(ip_port, usable_ip)print('=' * 50)print("有效的代理ip为:")# 判断列表是否为空if usable_ip:for usable in usable_ip:print(usable, end=' ')else:print(f"很遗憾,第{offset}页的代理IP全都无效")print('=' * 50)if __name__ == '__main__':num = int(input("请输出想查询多少页ip:"))# 共10页,100条记录,range()为左闭右开for page in range(1, num + 1):main(offset=page)time.sleep(2)

insert_into = ("INSERT INTO proxy_pool(代理IP地址,端口,匿名度,类型,支持,位置,响应速度,最后验证时间)""VALUES(?,?,?,?,?,?,?,?);")

insert_into = ("INSERT INTO proxy_pool(代理IP地址,端口,匿名度,类型,支持,位置,响应速度,最后验证时间)""VALUES(%s,%s,%s,%s,%s,%s,%s,%s);")