fix ilrifiutologo_it request rate limit (#2501)

* fix request rate limit

* instance type check

* headers update

* reformatting

---------

Co-authored-by: 5ila5 <5ila5@users.noreply.github.com>
This commit is contained in:
Fabio Garavini
2024-08-20 12:30:32 +02:00
committed by GitHub
parent 761b1c750f
commit fbdcd5f6e0

View File

@@ -1,4 +1,6 @@
import datetime
import json
import logging
import requests
from waste_collection_schedule import Collection
@@ -12,9 +14,15 @@ TEST_CASES = {
"Test2": {"town": "Faenza", "street": "VIA AUGUSTO RIGHI", "house_number": 1},
}
API_URL = "https://webapp-ambiente.gruppohera.it/rifiutologo/rifiutologoweb"
API_URL = "https://www.ilrifiutologo.it/ajax/archivio_ilrifiutologo_ajax.php"
API_URL_BACKEND = "https://webapp-ambiente.gruppohera.it/rifiutologo/rifiutologoweb"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:129.0) Gecko/20100101 Firefox/129.0"
"Accept": "*/*",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36",
"Host": "www.ilrifiutologo.it",
"Origin": "https://www.ilrifiutologo.it",
"Referer": "http://www.ilrifiutologo.it/casa_rifiutologo/",
"X-Requested-With": "XMLHttpRequest",
}
ICON_MAP = {
@@ -28,64 +36,98 @@ ICON_MAP = {
"Carta e cartone": "mdi:newspaper",
}
_LOGGER = logging.getLogger(__name__)
class Source:
def __init__(
self, town: str, street: str, house_number: int | str
): # argX correspond to the args dict in the source configuration
self._comune = town
self._indirizzo = street
self._civico = house_number
self.comune_id = None
self.inirizzo_id = None
self.civico_id = None
def fetch(self):
comuni = api_get_request(relative_path="/getComuni.php")
comune_id = None
_LOGGER.debug(
"getComuni.php: %d s - %s - %s",
comuni.elapsed.total_seconds(),
comuni.status_code,
comuni.reason,
)
for city in comuni.json():
if city.get("name").upper() == self._comune.upper():
comune_id = city.get("id", "")
if comuni.status_code != 200 or not isinstance(comuni.json(), list):
raise Exception("Errore durante il recupero dei comuni")
for citta in comuni.json():
if citta.get("name").upper() == town.upper():
self.comune_id = citta.get("id", "")
break
if comune_id is None:
if self.comune_id is None:
raise Exception("Comune non trovato")
indirizzi = api_get_request(
relative_path="/getIndirizzi.php", params={"idComune": comune_id}
relative_path="/getIndirizzi.php", params={"idComune": self.comune_id}
)
_LOGGER.debug(
"getIndirizzi.php: %d s - %s - %s",
indirizzi.elapsed.total_seconds(),
indirizzi.status_code,
indirizzi.reason,
)
inirizzo_id = None
for street in indirizzi.json():
if street.get("indirizzo") == self._indirizzo.upper():
inirizzo_id = street.get("id", "")
if indirizzi.status_code != 200 or not isinstance(indirizzi.json(), list):
raise Exception("Errore durante il recupero degli indirizzi")
for strada in indirizzi.json():
if strada.get("indirizzo") == street.upper():
self.inirizzo_id = strada.get("id", "")
break
if inirizzo_id is None:
if self.inirizzo_id is None:
raise Exception("Strada non trovata")
numeri_civici = api_get_request(
relative_path="/getNumeriCivici.php",
params={"idComune": comune_id, "idIndirizzo": inirizzo_id},
params={"idComune": self.comune_id, "idIndirizzo": self.inirizzo_id},
)
_LOGGER.debug(
"getNumeriCivici.php: %d s - %s - %s",
numeri_civici.elapsed.total_seconds(),
numeri_civici.status_code,
numeri_civici.reason,
)
civio_id = None
for number in numeri_civici.json():
if number.get("numeroCivico") == str(self._civico):
civio_id = number.get("id", "")
if numeri_civici.status_code != 200 or not isinstance(
numeri_civici.json(), list
):
raise Exception("Errore durante il recupero dei numeri civici")
for civico in numeri_civici.json():
if civico.get("numeroCivico") == str(house_number):
self.civico_id = civico.get("id", "")
break
if civio_id is None:
if self.civico_id is None:
raise Exception("Civico non trovato")
def fetch(self):
r = api_get_request(
relative_path="/getCalendarioPap.php",
params={
"idComune": comune_id,
"idIndirizzo": inirizzo_id,
"idCivico": civio_id,
"idComune": self.comune_id,
"idIndirizzo": self.inirizzo_id,
"idCivico": self.civico_id,
"isBusiness": "0",
"date": datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
"giorniDaMostrare": 31,
},
)
_LOGGER.debug(
"getCalendarioPap.php: %d s - %s - %s",
r.elapsed.total_seconds(),
r.status_code,
r.reason,
)
if r.status_code != 200:
if r.status_code != 200 or not isinstance(r.json(), dict):
raise Exception("Errore durante il recupero del calendario")
calendar = r.json().get("calendario", [])
@@ -110,4 +152,13 @@ class Source:
def api_get_request(relative_path, params=None):
return requests.get(url=API_URL + relative_path, params=params, headers=HEADERS)
_LOGGER.debug("%s [GET] params: %s", relative_path, params)
return requests.post(
url=API_URL,
data={
"url": API_URL_BACKEND + relative_path,
"type": "GET",
"parameters": json.dumps(params) if params else None,
},
headers=HEADERS,
)