目标站点
https://dexscreener.com
目标数据
列表页数据
问题
- 看到网上有不少推荐使用该库的
aiowebsocket，测试下来该库无法实现，也是耽误了我一天的时间，各种调试，一开始只是以为是我没有过风控的问题，其实并不是，问题在库上；该项目已经4年没有更新了，废掉了，不建议使用
# from aiowebsocket.converses import AioWebSocket
#https://github.com/asyncins/aiowebsocket
- Sec-WebSocket-Key 加密参数；该参数是根据一定规则随机生成的，只要按照规则生成即可，不按规则就无法连接成功； 代码示例
function generateWebSocketKey() {
// 生成一个16字节的随机值
const buffer = new Uint8Array(16);
window.crypto.getRandomValues(buffer);
// 将随机值进行Base64编码
const key = btoa(String.fromCharCode.apply(null, buffer));
return key;
}
import os
import base64
def generate_websocket_key():
# 生成一个16字节的随机值
buffer = os.urandom(16)
# 将随机值进行Base64编码
key = base64.b64encode(buffer).decode('utf-8')
return key
print(generate_websocket_key())
- headers 是需要对UA进行校验的
解决代码示例
class ListPage:
def __init__(self):
self.base_url = 'wss://io.dexscreener.com/dex/screener/pairs/h24/{}?rankBy[key]=txns&rankBy[order]=desc&filters[liquidity][min]=100000&filters[marketCap][min]=200000&filters[txns][h24][min]=100'
def generate_websocket_key(self):
# 生成一个16字节的随机值
buffer = os.urandom(16)
# 将随机值进行Base64编码
key = base64.b64encode(buffer).decode('utf-8')
return key
def open_connection(self,num):
"""
建立连接
"""
header = {
# 用户唯一性校验+风控识别校验，可以理解为mac地址的作用，该值有一定的规范，是随机生成的，但是得符合一定的标准，在标准之外的随机数是不被认可的，无法建立连接
"Sec-WebSocket-Key": self.generate_websocket_key(),
# 和http请求的用途一样，用于校验客户端角色；主要风控
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36"
}
ws = websocket.WebSocket(sslopt={"cert_reqs": ssl.CERT_NONE})
# remote = 'wss://io.dexscreener.com/dex/screener/pairs/h24/4??rankBy[key]=txns&rankBy[order]=desc&filters[liquidity][min]=100000&filters[marketCap][min]=200000&filters[txns][h24][min]=100'
url = self.base_url.format(num)
logger.info(f'request {num}: {url}')
# 必须得携带头部信息，Python代码实现的wss通信自动生成的Sec-WebSocket-Key 很多服务器会不支持
ws.connect(url, header=header)
return ws
def get_data(self, num):
"""
发起请求获取数据
"""
retry_num = 0
while True:
retry_num += 1
if retry_num > 3:
logger.error('重试超过次数！')
return
try:
ws = self.open_connection(num)
# message = input("Enter message: ")
# 测了下好像不会过期(约20个小时) 该行代码有可能报错
ws.send('pong')
break
except:
logger.warning('wss通信失败，重试...')
time.sleep(1)
response = ws.recv()
logger.info(f'response: {num}')
data = json.loads(response)
Top comments (0)