关键词：
import asyncio
import websockets
import aiohttp
import random
import json
import re
from collections import Counter
import sys
class Scraper(object):
class __Messenger__(object):
def __init__(self, teams, lock):
self.__vals__ = dict()
self.__teams__ = ' vs '.format(teams[0], teams[1])
self.__lock__ = lock
async def announce(self, name, values):
self.__lock__.acquire()
prev = (1.0, 1.0)
if name in self.__vals__:
prev = self.__vals__[name]
self.__vals__[name] = values
print(self.__teams__, 'Updated:', name, prev, '->', values, self.__vals__)
self.__lock__.release()
__msngr__ = None
def __init__(self, name, teams, lock):
self.__source__ = name
self.__teams__ = teams
self.__scores__ = (0, 0)
lock.acquire()
if not Scraper.__msngr__:
Scraper.__msngr__ = Scraper.__Messenger__(teams, lock)
lock.release()
async def __announce__(self, nvalue):
nvalue = (round(nvalue[0], 4), round(nvalue[1], 4))
if self.__scores__ != nvalue:
self.__scores__ = nvalue
await self.__msngr__.announce(self.__source__, self.__scores__)
class CSGOpositive(Scraper):
def __init__(self, teams, lock):
self.__ws_url__ = 'wss://ws.csgopositive.com:2096/socket.io/?EIO=3&transport=websocket'
self.__web_url__ = 'https://csgopositive.com/en/'
super().__init__('csgopositive.com', teams, lock)
async def start(self):
data = None
async with aiohttp.ClientSession() as session:
async with session.get(self.__web_url__) as resp:
data = await resp.text()
ids = []
teams_index = []
for team in self.__teams__:
res = [(m.start(0), m.end(0)) for m in re.finditer('<span class="team_name"></span>'.format(team), data)]
for start, end in res:
idx = data.rfind('data-id="', 0, start)
idx = re.search('\d+', data[idx:idx+20])
if idx:
ids.append(idx.group(0))
teams_index.append((start, team))
c = Counter(ids)
self.__game_id__ = int(c.most_common(1)[0][0])
self.__koef_table__ =
self.__teams__[0]: 'koef_1',
self.__teams__[1]: 'koef_2'
index = [m.start(0) for m in re.finditer('<div class="event(( hot)|())" data-app_id="730" data-id="">'.format(self.__game_id__), data)][0]
positions = []
scores = []
for team in self.__teams__:
idx = data.index(team, index, -1)
numerical_item = '<span class="sum">'
score_idx = data.index(numerical_item, idx) + len(numerical_item)
scores.append(await self.__convert__(data[score_idx:score_idx + 4]))
positions.append((team, idx))
self.__koef_table__[min(positions, key=lambda x: x[1])[0]] = 'koef_1'
self.__koef_table__[max(positions, key=lambda x: x[1])[0]] = 'koef_2'
await self.__announce__(tuple(scores))
async def loop(self):
async with websockets.connect(self.__ws_url__) as websocket:
await websocket.send('2probe')
while 'pigs' != 'fly':
msg = await websocket.recv()
resp = await self.__response__(msg)
if resp:
await websocket.send(resp)
if msg[:2] == '42' and 'koef_change' in msg:
data = json.loads(msg[2:])[1]
if int(data['id']) == self.__game_id__:
scores = []
for team in self.__teams__:
param = self.__koef_table__[team]
scores.append(float(data[param]))
await self.__announce__(tuple(scores))
async def __response__(self, msg):
if msg == '3probe':
return '5'
elif random.random() < 0.2:
return '2'
return None
async def __convert__(self, value):
return 100.0 / float(re.search('\d+', value).group(0))
class Pinnacle(Scraper):
def __init__(self, teams, lock):
self.__api_url__ = 'https://www.pinnacle.com/webapi/1.17/api/v1/guestlines/deadball/12//moneyline'
self.__league_id__ = None
self.__leagues__ = [
197700,
200992,
200987,
197375
]
super().__init__('pinnacle.com', teams, lock)
async def start(self):
for idx in self.__leagues__:
data = await self.__send_request__(idx)
res = await self.__parse__(data)
if res:
self.__league_id__ = idx
await self.__announce__(res)
break
async def loop(self):
while 'pigs' != 'fly':
data = await self.__send_request__(self.__league_id__)
res = await self.__parse__(data)
await self.__announce__(res)
async def __parse__(self, data):
team_set = set(self.__teams__)
for league in data['Leagues']:
for event in league['Events']:
participants = set([x['Name'] for x in event['Participants']])
if team_set == participants:
scores = [None, None]
subscores = [(x['Name'], x['MoneyLine']) for x in event['Participants']]
for name, score in subscores:
converted = await self.__convert__(data['OddsType'], score)
idx = self.__teams__.index(name)
scores[idx] = converted
return tuple(scores)
return None
async def __send_request__(self, idx):
async with aiohttp.ClientSession() as session:
async with session.get(self.__api_url__.format(idx)) as resp:
return await resp.json()
async def __convert__(self, stype, value):
if stype == 'american':
if value > 0:
return (float(value) / 100) + 1
else:
return (100.0 / (-value)) + 1
if stype == 'percentage':
if value > 1:
return 1.0 / (value / 100)
else:
return 1.0 / value
return value
class Egb(Scraper):
def __init__(self, teams, lock):
self.__api_url__ = 'https://egb.com/bets'
super().__init__('egb.com', teams, lock)
async def start(self):
data = await self.__send_request__()
res = await self.__parse__(data)
if res:
await self.__announce__(res)
async def loop(self):
while 'pigs' != 'fly':
data = await self.__send_request__()
res = await self.__parse__(data)
await self.__announce__(res)
async def __parse__(self, data):
team_set = set(self.__teams__)
for bet in data['bets']:
names = set([bet['gamer_1']['nick'], bet['gamer_2']['nick']])
if names == team_set:
subscores = [
(bet['gamer_1']['nick'], bet['coef_1']),
(bet['gamer_2']['nick'], bet['coef_2'])
]
scores = [None, None]
for name, score in subscores:
idx = self.__teams__.index(name)
scores[idx] = float(score)
return tuple(scores)
return None
async def __send_request__(self):
headers =
'accept-encoding': 'gzip, deflate, br',
'accept-language': 'lt,en-GB;q=0.9,en;q=0.8',
'accept': 'application/json, text/javascript, */*; q=0.01',
'referer': 'https://egb.com/play/simple_bets/csgo',
'authority': 'egb.com',
'x-requested-with': 'XMLHttpRequest',
async with aiohttp.ClientSession() as session:
async with session.get(self.__api_url__, headers=headers) as resp:
return await resp.json()
def main(teams):
ioloop = asyncio.get_event_loop()
lock = asyncio.Semaphore(1)
csgopositive = CSGOpositive(teams, lock)
pinnacle = Pinnacle(teams, lock)
egb = Egb(teams, lock)
start_tasks = [
ioloop.create_task(pinnacle.start()),
ioloop.create_task(csgopositive.start()),
ioloop.create_task(egb.start())
]
wait_tasks = asyncio.wait(start_tasks)
ioloop.run_until_complete(wait_tasks)
loop_tasks = [
ioloop.create_task(pinnacle.loop()),
ioloop.create_task(csgopositive.loop()),
ioloop.create_task(egb.loop())
]
wait_tasks = asyncio.wait(loop_tasks)
ioloop.run_until_complete(wait_tasks)
if __name__ == '__main__':
if len(sys.argv) < 3:
print('Usage:')
print('./scraper.py <TEAM_1> <TEAM_2>')
exit(1)
main([sys.argv[1], sys.argv[2]])
aiohttp==3.0.9
async-timeout==2.0.1
attrs==17.4.0
certifi==2018.1.18
chardet==3.0.4
idna==2.6
idna-ssl==1.0.1
multidict==4.1.0
requests==2.18.4
six==1.11.0
urllib3==1.22
websocket-client==0.47.0
websockets==4.0.1
wincertstore==0.2
yarl==1.1.1### Info:
CS:GO betting info gatherer. It pulls data from these sources:
- csgopositive.com
- egb.com
- pinnacle.com
### Usage:
> python ./scraper.py <TEAM_NAME_ONE> <TEAM_NAME_TWO>
Example:
> python ./scraper.py 'Windigo' 'ALTERNATE aTTaX'