# coding=utf-8 |
# 刷企业直通车 |
import re |
import random |
import sys |
import time |
import datetime |
import threading |
from random import choice |
import requests |
import bs4 |
def get_ip(): |
"""获取代理IP""" |
url = "https://www.xicidaili.com/nn/" |
headers = { "Accept" : "text/html,application/xhtml+xml,application/xml;" , |
"Accept-Encoding" : "gzip, deflate, sdch" , |
"Accept-Language" : "zh-CN,zh;q=0.8,en;q=0.6" , |
"Referer" : "http://www.xicidaili.com" , |
"User-Agent" : "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/42.0.2311.90 Safari/537.36" |
} |
r = requests.get(url, headers = headers) |
soup = bs4.BeautifulSoup(r.text, 'html.parser' ) |
data = soup.table.find_all( "td" ) |
ip_compile = re. compile (r '<td>(\d+\.\d+\.\d+\.\d+)</td>' ) # 匹配IP |
port_compile = re. compile (r '<td>(\d+)</td>' ) # 匹配端口 |
ip = re.findall(ip_compile, str (data)) # 获取所有IP |
port = re.findall(port_compile, str (data)) # 获取所有端口 |
return [ ":" .join(i) for i in zip (ip, port)] # 组合IP+端口,如:115.112.88.23:8080 |
# 设置 user-agent列表,每次请求时,可在此列表中随机挑选一个user-agnet |
uas = [ |
"Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:17.0; Baiduspider-ads) Gecko/17.0 Firefox/17.0" , |
"Mozilla/5.0 (Windows; U; Windows NT 5.1; zh-CN; rv:1.9b4) Gecko/2008030317 Firefox/3.0b4" , |
"Mozilla/5.0 (Windows; U; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727; BIDUBrowser 7.6)" , |
"Mozilla/5.0 (Windows NT 6.3; WOW64; Trident/7.0; rv:11.0) like Gecko" , |
"Mozilla/5.0 (Windows NT 6.3; WOW64; rv:46.0) Gecko/20100101 Firefox/46.0" , |
"Mozilla/5.0 (Windows NT 6.3; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/45.0.2454.99 Safari/537.36" , |
"Mozilla/5.0 (Windows NT 6.3; Win64; x64; Trident/7.0; Touch; LCJB; rv:11.0) like Gecko" , |
] |
def get_url(code = 0 , ips = []): |
""" |
投票 |
如果因为代理IP不可用造成投票失败,则会自动换一个代理IP后继续投 |
""" |
try : |
ip = choice(ips) |
except : |
return False |
else : |
proxies = { |
"http" : ip, |
} |
headers2 = { "Accept" : "text/html,application/xhtml+xml,application/xml;" , |
"Accept-Encoding" : "gzip, deflate, sdch" , |
"Accept-Language" : "zh-CN,zh;q=0.8,en;q=0.6" , |
"Referer" : "", |
"User-Agent" : choice(uas), |
} |
try : |
hz_url = "http://ztc.jx96871.com:9082/message/goMessage.do?d5f4634d-5022-4935-9g20-312800ac264c" # 某投票网站的地址,这里不用真实的域名 |
hz_url2 = "http://ztc.jx96871.com:9082/policysearch/list?F_PFID=d5f4634d-5022-4935-9g20-312800ac264c" |
requests.get(hz_url2, headers = headers2, proxies = proxies) |
except requests.exceptions.ConnectionError: |
print ( "ConnectionError" ) |
if not ips: |
print ( "not ip" ) |
sys.exit() |
# 删除不可用的代理IP |
if ip in ips: |
ips.remove(ip) |
# 重新请求URL |
get_url(code, ips) |
else : |
if ip in ips: |
ips.remove(ip) |
date = datetime.datetime.now().strftime( '%H:%M:%S' ) |
print (u "第%s次 [%s] [%s]:投票 (剩余可用代理IP数:%s)" % (code, date, ip, len (ips))) |
ips = [] |
for i in range ( 3000 ): |
if i % 100 = = 0 : |
ips.extend(get_ip()) |
print (u "IP数量:%s" % len (ips)) |
# 启用线程,隔1秒产生一个线程,可控制时间加快投票速度 ,time.sleep的最小单位是毫秒 |
t1 = threading.Thread(target = get_url, args = (i, ips)) |
t1.start() |
time.sleep( 3 ) |
初级程序员
by: 野望 发表于:2020-09-20 16:41:10 顶(0) | 踩(0) 回复
没有那个模块
回复评论