专业做蛋糕视频网站,wordpress插件 采集,精品网站建设比较好,常州网络排名优化爬取cos猎人
数据库管理主要分为4个模块#xff0c;代理获取模块#xff0c;代理储存模块#xff0c;代理测试模块#xff0c;爬取模块 cos猎人已经倒闭#xff0c;所以放出爬虫源码
api.py 为爬虫评分提供接口支持
import requests
import concurrent.futures
import …爬取cos猎人
数据库管理主要分为4个模块代理获取模块代理储存模块代理测试模块爬取模块 cos猎人已经倒闭所以放出爬虫源码
api.py 为爬虫评分提供接口支持
import requests
import concurrent.futures
import redis
import random
import flask # 导入flask模块
from flask import request # 获取url地址中查询参数
from flask import jsonify # 可以把对象转换为字符串
REDIS_HOST 127.0.0.1
REDIS_PORT 6379
REDIS_DATABASE 0
REDISOBJECT proxysss时间间隔配置
GETTER_PROXY 60*5
VERIFY_PROXY 60*3class RedisClient:def __init__(self, hostREDIS_HOST, portREDIS_PORT, dbREDIS_DATABASE):self.db redis.Redis(hosthost, portport, dbdb, decode_responsesTrue)def exists(self, proxy):判断传入代理有没有存输到数据库有TRUE,没有Falseis比not优先级高return not self.db.zscore(REDISOBJECT, proxy) is Nonedef add(self, proxy, score10):添加代理到数据库设置初始分数为10分决定是否加入新代理if not self.exists(proxy):return self.db.zadd(REDISOBJECT, {proxy: score})def random(self):随机选择一个代理尝试获取评分为100分的代理获取指定范围的代理如果数据库没有代理就提示数据库为空proxies self.db.zrangebyscore(REDISOBJECT, 100, 100)if len(proxies):return random.choice(proxies)proxies self.db.zrangebyscore(REDISOBJECT, 1, 99)if len(proxies):return random.choice(proxies)print(-----数据库为空----)def decrease(self, proxy):传入代理如果检测不过关降低代理分数self.db.zincrby(REDISOBJECT, -10, proxy)score self.db.zscore(REDISOBJECT, proxy) # 查询分数if score 0:self.db.zrem(REDISOBJECT, proxy) # 删除代理def max(self, proxy):检测代理可用就将代理设置最大分数return self.db.zadd(REDISOBJECT, {proxy: 100})def count(self):获取数据库中代理的数量return self.db.zcard(REDISOBJECT)def all(self):获取所有代理返回列表proxies self.db.zrangebyscore(REDISOBJECT,1,100)if proxies:return proxieselse:print(-----数据库无代理----)def count_for_num(self,number):指定数量获取代理返回一个列表all_proxies self.all()proxies random.sample(all_proxies,knumber)#随机取数据不重样return proxiesdef get_proxy():return requests.get(http://127.0.0.1:5010/all).json()def delete_proxy(proxy):requests.get(http://127.0.0.1:5010/delete/?proxy{}.format(proxy))# getHtml()
# def verify_thread_pool():
# 线程池检测代理
# 1.从数据库中取到所有代理
# 2.用线程池检测代理
# proxies_list client.all() # 列表
# with concurrent.futures.ThreadPoolExecutor(max_workers10) as executor:
# for proxy in proxies_list:
# executor.submit(verify_proxy, proxy)#
#
# TEST_URL https://www.baidu.com/
# headers {User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36}
#
#
#
#
# def verify_proxy(proxy):
# 检测代理是否可用
# proxies {
# http: http:// proxy,
# https: https:// proxy
# }
# try:
# response requests.get(urlTEST_URL, headersheaders, proxiesproxies, timeout2)
# if response.status_code in [200, 206, 302]:
# #判断请求返回的状态码是否成功
# 请求成功设为100分,调用max
# 请求不成功将代理降分,调用decrease
# client.max(proxy)
# print(***代理可用***, proxy)
# else:
# client.decrease(proxy)
# print(--状态码不合法--, proxy)
# except:
# 请求超时表示代理不可用
# client.decrease(proxy)
# print(请求超时)
#
# # 检测速度太慢引入多任务多线程
# def verify_thread_pool():
# 线程池检测代理
# 1.从数据库中取到所有代理
# 2.用线程池检测代理
# proxies_list client.all() # 列表
# with concurrent.futures.ThreadPoolExecutor(max_workers10) as executor:
# for proxy in proxies_list:
# executor.submit(verify_proxy, proxy)
#
# if __name__ __main__:
# # proxy [
# # 45.234.63.220:999,
# # 60.168.255.69:8060,
# # 65.108.27.185:3128,
# # 62.162.91.205:3129,
# # 37.34.236.15:80
# # ]
# # for pro in proxy:
# # verify_proxy(pro)
# verify_thread_pool()getter.py从数据库抽取一个代理
import requests
def get_one_proxy():return requests.get(http://127.0.0.1:5000/all)
print(get_one_proxy().text)sever.py搭建本地服务器供调用
import flask # 导入flask模块
from api import RedisClient
from flask import request # 获取url地址中查询参数
from flask import jsonify # 可以把对象转换为字符串app flask.Flask(__name__)client RedisClient()
app.route(/)
# 将下面的函数挂载到路由
def index():视图函数http://demo.spiderpy.cn/get/视图函数返回的数据只能返回字符串类型的数据return h2欢迎来到代理池/h2app.route(/get)
def get_proxy():随机获取一个代理,调用数据库random模块one_proxy client.random()return one_proxyapp.route(/getcount)
def get_any_proxy():获取指定数量一个代理,调用数据库的 count_for_num()拿到查询参数的值又可能用户没有传递查询参数num返回为空num request.args.get(num, )if not num:没有获取到查询参数num 1else:num int(num)any_proxy client.count_for_num(num)return jsonify(any_proxy)app.route(/getnum)
def get_count_proxy():获取所有代理数量调用数据库count方法count_proxy client.count()return f代理可用的数量为:{count_proxy}个app.route(/getall)
def get_all_proxy():获取所有代理调用数据库的all()all_proxy client.all()return jsonify(all_proxy)if __name__ __main__:运行实例化的app对象app.run()test_self.py和tests.py对已经储存的代理质量进行检测
记不清哪个效果更好
import timeimport requests
from api import RedisClient
clients RedisClient()
def get_proxy():return requests.get(http://127.0.0.1:5000/getall)a get_proxy()
a a.json()
# print(a)
# for b in a:
# print(b)
# print(type(a))def getHtml():# retry_count 1for proxy in a:# print(proxy)try:html requests.get(http://www.example.com, proxies{http: http://{}.format(proxy)},timeout4)# print(html.text)if html.status_code in [200, 206, 302]:print(proxy,:可以使用)clients.add(proxy)# 使用代理访问except Exception:print(代理不可用, proxy)clients.decrease(proxy)# 删除代理池中代理# delete_proxy(proxy)
while True:getHtml()time.sleep(60*2)
进程池爬取cos猎人.py 主爬虫代码
from typing import List, Any
from concurrent.futures import ThreadPoolExecutor
import requests
import os
from lxml import etree
import re
if not os.path.exists(./img):os.makedirs(img)
def get_one_proxy():return requests.get(http://127.0.0.1:5000/get)
proxies get_one_proxy().text
# proxies def down_img(img_url):for urls in img_url:response requests.get(urlurls, headersheaders)name urls.split(/)[-1]with open(./img/f{name}, wb) as f:f.write(response.content)headers {user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36 Edg/112.0.1722.58,referer: https://www.coshunter.pro/simo
}a 252#354
while a 355:url fhttps://www.coshunter.pro/shop/buy/page/{a}res requests.get(url, headersheaders, proxies{http: http://{}.format(proxies)})res.encoding utf-8html re.findall(ra classlink-block href(.*?)/a,res.text)urls html[:-1]# print(urls)for i in urls:res requests.get(i, headersheaders, proxies{http: http://{}.format(proxies)})img_url re.findall(rfigure classwp-block-image.*src(.*?),res.text)print(img_url)with ThreadPoolExecutor(10) as t:t.submit(down_img,img_url)print(a)a 1
