Nemesis/asmodee_matrix
1
0
Fork 0
generated from Nemesis/Exemple
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

correction token RTE

Browse source
This commit is contained in:
nemesis 2022-10-25 16:31:19 +02:00
parent c228b3feff
commit e665b335d8
1 changed files with 17 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
asmodee.py
View file

@ -25,7 +25,9 @@ import json
bot = None bot = None
PREFIX = '!' PREFIX = '!'
USERNAME = "" USERNAME = ""
oauth = None #Appels RTE RTE_ID = None # Bot's RTE ID
RTE_SECRET = None # Bot's RTE SECRET
#ecowatt_url = 'https://digital.iservices.rte-france.com/open_api/ecowatt/v4/sandbox/signals' #ecowatt_url = 'https://digital.iservices.rte-france.com/open_api/ecowatt/v4/sandbox/signals'
ecowatt_url = 'https://digital.iservices.rte-france.com/open_api/ecowatt/v4/signals' ecowatt_url = 'https://digital.iservices.rte-france.com/open_api/ecowatt/v4/signals'
commune_url = 'https://geo.api.gouv.fr/communes?<TYPE>=<NOM>&fields=nom,code,codesPostaux,population&format=json' commune_url = 'https://geo.api.gouv.fr/communes?<TYPE>=<NOM>&fields=nom,code,codesPostaux,population&format=json'
@ -1634,6 +1636,7 @@ async def ecowatt(room, message):
global wattobj global wattobj
if (wattobj.jour < jn or (wattobj.jour == jn and wattobj.heure + 1 < hn)): if (wattobj.jour < jn or (wattobj.jour == jn and wattobj.heure + 1 < hn)):
try: try:
oauth = rte_token()
result = oauth.get(ecowatt_url) result = oauth.get(ecowatt_url)
watt = json.loads(result.text) watt = json.loads(result.text)
except Exception as e: except Exception as e:
@ -1694,6 +1697,7 @@ async def commune(room, message):
cette_commune = commune_url.replace("<NOM>", ville).replace("<TYPE>", "codePostal") cette_commune = commune_url.replace("<NOM>", ville).replace("<TYPE>", "codePostal")
else: else:
cette_commune = commune_url.replace("<NOM>", ville).replace("<TYPE>", "nom") cette_commune = commune_url.replace("<NOM>", ville).replace("<TYPE>", "nom")
oauth = rte_token()
result = oauth.get(cette_commune) result = oauth.get(cette_commune)
villes = json.loads(result.text) villes = json.loads(result.text)
if (len(villes) > 0): if (len(villes) > 0):
@ -1730,6 +1734,7 @@ async def eau(room, message):
cette_commune = eau_url.replace("<NOM>", ville).replace("<TYPE>", "code_commune") cette_commune = eau_url.replace("<NOM>", ville).replace("<TYPE>", "code_commune")
else: else:
cette_commune = eau_url.replace("<NOM>", ville).replace("<TYPE>", "nom_commune") cette_commune = eau_url.replace("<NOM>", ville).replace("<TYPE>", "nom_commune")
oauth = rte_token()
result = oauth.get(cette_commune) result = oauth.get(cette_commune)
eaux = json.loads(result.text) eaux = json.loads(result.text)
if (len(eaux["data"]) > 0): if (len(eaux["data"]) > 0):
@ -2009,6 +2014,13 @@ async def callStats(room, message):
if match.is_not_from_this_bot(): if match.is_not_from_this_bot():
statistiques(room, message) statistiques(room, message)
def rte_token():
auth = HTTPBasicAuth(RTE_ID, RTE_SECRET)
client = BackendApplicationClient(client_id=RTE_ID)
oauth = OAuth2Session(client=client)
token = oauth.fetch_token(token_url='https://digital.iservices.rte-france.com/token/oauth/', auth=auth)
return(oauth)
def main(): def main():
config = configparser.ConfigParser() config = configparser.ConfigParser()
config.read('asmodee.ini') config.read('asmodee.ini')
@ -2025,20 +2037,16 @@ def main():
else: else:
print("Probleme de lecture de configuration asmodee.ini (AUTH)") print("Probleme de lecture de configuration asmodee.ini (AUTH)")
RTE_ID = None # Bot's RTE ID
RTE_SECRET = None # Bot's RTE SECRET
if ('RTE' in config): if ('RTE' in config):
global oauth global RTE_ID
global RTE_SECRET
global wattobj global wattobj
RTE_ID = config['RTE']['client_id'] # Bot's username RTE_ID = config['RTE']['client_id'] # Bot's username
RTE_SECRET = config['RTE']['client_secret'] # Bot's password RTE_SECRET = config['RTE']['client_secret'] # Bot's password
auth = HTTPBasicAuth(RTE_ID, RTE_SECRET)
client = BackendApplicationClient(client_id=RTE_ID)
oauth = OAuth2Session(client=client)
token = oauth.fetch_token(token_url='https://digital.iservices.rte-france.com/token/oauth/', auth=auth)
wattobj = Wattobject() wattobj = Wattobject()
else: else:
RTE_ID = None
RTE_SECRET = None
print("Probleme de lecture de configuration asmodee.ini (RTE)") print("Probleme de lecture de configuration asmodee.ini (RTE)")
global liste_mod global liste_mod