Open
Description
https://github.com/dxstiny/cevlib/blob/6fa393b99e515247dd57b397a8a2a9f65e383958/cevlib/match.py#L220
def _getParameter(self, link: str, parameter: str) -> Optional[str]:
try:
return "0"
# TODO working?? (not indexable)
#return re.search(f"(?<={parameter}=)([A-Za-z0-9]*)(?=&)?", link)[0]
except:
return None
def _getLinks(self, contains: str) -> List[str]:
eligibleLinks = [ ]
for umbracoLink in self._umbracoLinks:
if contains in umbracoLink:
eligibleLinks.append("https://" + umbracoLink.replace("amp;", ""))
return eligibleLinks
def _getLink(self, contains: str, index: int = 0) -> str:
links = self._getLinks(contains)
if len(links) > index:
return links[index]
return ""
async def _getForm(self) -> JObject:
if self._formCache is None:
async with aiohttp.ClientSession() as client:
async with client.get(self._getLink("GetFormComponent")) as resp:
self._formCache = json.loads(await resp.json(content_type=None))
return self._formCache
async def _getMatchId(self) -> Optional[int]:
try:
async with aiohttp.ClientSession() as client:
async with client.get(self._getLink("livescorehero")) as resp:
jdata = await resp.json(content_type=None)
return int(jdata.get("MatchId"))
except: # pylint: disable=bare-except
return None
async def _requestLiveScoresJson(self, useCache: bool = True) -> JObject:
if useCache and self._liveScoresCache:
return self._liveScoresCache
async with aiohttp.ClientSession() as client:
async with client.get("https://www.cev.eu/LiveScores.json") as resp:
jdata = DictEx(await resp.json(content_type=None))
self._liveScoresCache = jdata
return jdata
async def _requestLiveScoresJsonByMatchSafe(self, useCache: bool = True) -> Optional[JObject]:
return await (self._requestLiveScoresJsonByMatchId(useCache) if not self._invalidMatchCentre
else self._requestLiveScoresJsonByMatchCentreLink(useCache))
async def _requestLiveScoresJsonByMatchCentreLink(self,
useCache: bool = True) -> Optional[JObject]:
assert self._matchCentreLink
jdata = DictEx(await self._requestLiveScoresJson(useCache))
for competition in jdata.ensure("competitions", list):
cdex = DictEx(competition)
for match in cdex.ensure("matches", list):
mdex = DictEx(match)
if mdex.ensure("matchCentreLink", str) == self._matchCentreLink:
self._finished = mdex.ensure("matchState_String", str) == "FINISHED"
mdex["competition"] = { "name": cdex.tryGet("competitionName", str),
"id": cdex.tryGet("competitionId", int) }
return mdex
return await self._tryGetFinishedGameData()
async def _requestLiveScoresJsonByMatchId(self, useCache: bool = True) -> Optional[JObject]:
assert self._matchId
jdata = DictEx(await self._requestLiveScoresJson(useCache))
for competition in jdata.ensure("competitions", list):
for match in DictEx(competition).ensure("matches", list):
mdex = DictEx(match)
if mdex.ensure("matchId", int) == self._matchId:
self._finished = mdex.ensure("matchState_String", str) == "FINISHED"
return mdex
return await self._tryGetFinishedGameData()
async def _tryGetFinishedGameData(self, trulyFinished: bool = True) -> Optional[JObject]:
if self._invalidMatchCentre:
return None
async with aiohttp.ClientSession() as client: