Skip to content

Commit

Permalink
[#64] Captcha failed issue
Browse files Browse the repository at this point in the history
  • Loading branch information
Stéphane Senart committed May 4, 2024
1 parent bc256cc commit 85ad84a
Showing 1 changed file with 86 additions and 41 deletions.
127 changes: 86 additions & 41 deletions pygazpar/datasource.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@
import glob
import os
import json
import time
import pandas as pd
import http.cookiejar
from abc import ABC, abstractmethod
from typing import Any, List, Dict, cast, Optional
from requests import Session
Expand All @@ -11,15 +13,22 @@
from pygazpar.excelparser import ExcelParser
from pygazpar.jsonparser import JsonParser

AUTH_NONCE_URL = "https://monespace.grdf.fr/client/particulier/accueil"
LOGIN_URL = "https://login.monespace.grdf.fr/sofit-account-api/api/v1/auth"
LOGIN_HEADER = {"domain": "grdf.fr"}
LOGIN_PAYLOAD = """{{
"email": "{0}",
SESSION_TOKEN_URL = "https://connexion.grdf.fr/api/v1/authn"
SESSION_TOKEN_PAYLOAD = """{{
"username": "{0}",
"password": "{1}",
"capp": "meg",
"goto": "https://sofa-connexion.grdf.fr:443/openam/oauth2/externeGrdf/authorize?response_type=code&scope=openid%20profile%20email%20infotravaux%20%2Fv1%2Faccreditation%20%2Fv1%2Faccreditations%20%2Fdigiconso%2Fv1%20%2Fdigiconso%2Fv1%2Fconsommations%20new_meg&client_id=prod_espaceclient&state=0&redirect_uri=https%3A%2F%2Fmonespace.grdf.fr%2F_codexch&nonce={2}&by_pass_okta=1&capp=meg"}}"""

"options": {{
"multiOptionalFactorEnroll": "false",
"warnBeforePasswordExpired": "false"
}}
}}"""

AUTH_TOKEN_URL = "https://connexion.grdf.fr/login/sessionCookieRedirect"
AUTH_TOKEN_PARAMS = """{{
"checkAccountSetupComplete": "true",
"token": "{0}",
"redirectUrl": "https://monespace.grdf.fr"
}}"""

Logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -50,49 +59,52 @@ def __init__(self, username: str, password: str):
# ------------------------------------------------------
def load(self, pceIdentifier: str, startDate: date, endDate: date, frequencies: Optional[List[Frequency]] = None) -> MeterReadingsByFrequency:

session = Session()

session.headers.update(LOGIN_HEADER)
auth_token = self._login(self.__username, self.__password)

self._login(session, self.__username, self.__password)

res = self._loadFromSession(session, pceIdentifier, startDate, endDate, frequencies)
res = self._loadFromSession(auth_token, pceIdentifier, startDate, endDate, frequencies)

Logger.debug("The data update terminates normally")

return res

# ------------------------------------------------------
def _login(self, session: Session, username: str, password: str):
def _login(self, username: str, password: str) -> str:

# Get auth_nonce token.
session.get(AUTH_NONCE_URL)
if "auth_nonce" not in session.cookies:
raise Exception("Login error: Cannot get auth_nonce token")
auth_nonce = session.cookies.get("auth_nonce")
session = Session()
session.headers.update({"domain": "grdf.fr"})
session.headers.update({"Content-Type": "application/json"})
session.headers.update({"X-Requested-With": "XMLHttpRequest"})

# Build the login payload as a json string.
payload = LOGIN_PAYLOAD.format(username, password, auth_nonce)
payload = SESSION_TOKEN_PAYLOAD.format(username, password)

# Build the login payload as a python object.
data = json.loads(payload)
response = session.post(SESSION_TOKEN_URL, data=payload)

# Send the login command.
response = session.post(LOGIN_URL, data=data)
if response.status_code != 200:
raise Exception(f"An error occurred while logging in. Status code: {response.status_code} - {response.text}")

# Check login result.
loginData = response.json()
session_token = response.json().get("sessionToken")

response.raise_for_status()
Logger.debug("Session token: %s", session_token)

jar = http.cookiejar.CookieJar()

session = Session()
session.headers.update({"Content-Type": "application/json"})
session.headers.update({"X-Requested-With": "XMLHttpRequest"})

params = json.loads(AUTH_TOKEN_PARAMS.format(session_token))

response = session.get(AUTH_TOKEN_URL, params=params, allow_redirects=True, cookies=jar)

if "status" in loginData and "error" in loginData and loginData["status"] >= 400:
raise Exception(f"{loginData['error']} ({loginData['status']})")
if response.status_code != 200:
raise Exception(f"An error occurred while getting the auth token. Status code: {response.status_code} - {response.text}")

if "state" in loginData and loginData["state"] != "SUCCESS":
raise Exception(loginData["error"])
auth_token = session.cookies.get("auth_token", domain="monespace.grdf.fr")

return auth_token

@abstractmethod
def _loadFromSession(self, session: Session, pceIdentifier: str, startDate: date, endDate: date, frequencies: Optional[List[Frequency]] = None) -> MeterReadingsByFrequency:
def _loadFromSession(self, auth_token: str, pceIdentifier: str, startDate: date, endDate: date, frequencies: Optional[List[Frequency]] = None) -> MeterReadingsByFrequency:
pass


Expand Down Expand Up @@ -121,7 +133,7 @@ def __init__(self, username: str, password: str, tmpDirectory: str):
self.__tmpDirectory = tmpDirectory

# ------------------------------------------------------
def _loadFromSession(self, session: Session, pceIdentifier: str, startDate: date, endDate: date, frequencies: Optional[List[Frequency]] = None) -> MeterReadingsByFrequency:
def _loadFromSession(self, auth_token: str, pceIdentifier: str, startDate: date, endDate: date, frequencies: Optional[List[Frequency]] = None) -> MeterReadingsByFrequency:

res = {}

Expand All @@ -141,6 +153,12 @@ def _loadFromSession(self, session: Session, pceIdentifier: str, startDate: date
# Get unique values.
frequencyList = set(frequencies)

# Create a session.
session = Session()
session.headers.update({"domain": "grdf.fr"})
session.headers.update({"Cookie": f"auth_token={auth_token}"})
session.headers.update({"X-Requested-With": "XMLHttpRequest"})

for frequency in frequencyList:
# Inject parameters.
downloadUrl = ExcelWebDataSource.DATA_URL.format(startDate.strftime(ExcelWebDataSource.DATE_FORMAT), endDate.strftime(ExcelWebDataSource.DATE_FORMAT), pceIdentifier, ExcelWebDataSource.FREQUENCY_VALUES[frequency])
Expand Down Expand Up @@ -210,7 +228,7 @@ def load(self, pceIdentifier: str, startDate: date, endDate: date, frequencies:
# ------------------------------------------------------------------------------------------------------------
class JsonWebDataSource(WebDataSource):

DATA_URL = "https://monespace.grdf.fr/api/e-conso/pce/consommation/informatives?dateDebut={0}&dateFin={1}&pceList%5B%5D={2}"
DATA_URL = "https://monespace.grdf.fr/api/e-conso/pce/consommation/informatives?dateDebut={0}&dateFin={1}&pceList[]={2}"

TEMPERATURES_URL = "https://monespace.grdf.fr/api/e-conso/pce/{0}/meteo?dateFinPeriode={1}&nbJours={2}"

Expand All @@ -222,7 +240,7 @@ def __init__(self, username: str, password: str):

super().__init__(username, password)

def _loadFromSession(self, session: Session, pceIdentifier: str, startDate: date, endDate: date, frequencies: Optional[List[Frequency]] = None) -> MeterReadingsByFrequency:
def _loadFromSession(self, auth_token: str, pceIdentifier: str, startDate: date, endDate: date, frequencies: Optional[List[Frequency]] = None) -> MeterReadingsByFrequency:

res = {}

Expand All @@ -237,11 +255,38 @@ def _loadFromSession(self, session: Session, pceIdentifier: str, startDate: date
# Data URL: Inject parameters.
downloadUrl = JsonWebDataSource.DATA_URL.format(startDate.strftime(JsonWebDataSource.INPUT_DATE_FORMAT), endDate.strftime(JsonWebDataSource.INPUT_DATE_FORMAT), pceIdentifier)

# First request never returns data.
session.get(downloadUrl)
# Retry mechanism.
retry = 10
while retry > 0:

# Create a session.
session = Session()
session.headers.update({"Host": "monespace.grdf.fr"})
session.headers.update({"Domain": "grdf.fr"})
session.headers.update({"X-Requested-With": "XMLHttpRequest"})
session.headers.update({"Accept": "application/json"})
session.cookies.set("auth_token", auth_token, domain="monespace.grdf.fr")

try:
response = session.get(downloadUrl)

if "text/html" in response.headers.get("Content-Type"):
raise Exception("An error occurred while loading data. Please check your credentials.")

if response.status_code != 200:
raise Exception(f"An error occurred while loading data. Status code: {response.status_code} - {response.text}")

break
except Exception as e:

if retry == 1:
raise e

Logger.error("An error occurred while loading data. Retry in 3 seconds.")
time.sleep(3)
retry -= 1

# Get consumption data.
data = session.get(downloadUrl).text
data = response.text

# Temperatures URL: Inject parameters.
endDate = date.today() - timedelta(days=1) if endDate >= date.today() else endDate
Expand Down

0 comments on commit 85ad84a

Please sign in to comment.