116 lines
2.9 KiB
Python
116 lines
2.9 KiB
Python
import requests
|
|
import urllib3
|
|
import logging as logger
|
|
import math
|
|
import itertools
|
|
from . import config
|
|
|
|
https = False
|
|
|
|
|
|
urllib3.disable_warnings()
|
|
|
|
|
|
session = requests.Session()
|
|
session.verify = False
|
|
|
|
logged = []
|
|
|
|
|
|
def api_login(config):
|
|
post_url = config["api_login_url"]
|
|
login_r = session.post(
|
|
post_url,
|
|
json={
|
|
"login": config["api_user_login"],
|
|
"password": config["api_user_password"],
|
|
},
|
|
)
|
|
|
|
if login_r.ok:
|
|
logged.append(session)
|
|
logger.info("Logged")
|
|
else: # pragma: no cover
|
|
err = f'{login_r.reason}: {login_r.text} for "{config["api_user_login"]}" on "{config["api_login_url"]}"'
|
|
logger.info("Login error {err}")
|
|
raise ValueError(err)
|
|
|
|
return login_r
|
|
|
|
|
|
def api_fetch(config):
|
|
if not logged:
|
|
api_login(config)
|
|
apiurl = config["api_status_url"]
|
|
fetched = session.get(apiurl)
|
|
if fetched.ok:
|
|
logger.info("Fetched")
|
|
else: # pragma: no cover
|
|
err = f'{fetched.reason}: {fetched.text} on "{config["api_status_url"]}"'
|
|
logger.info("Login error {err}")
|
|
raise ValueError(err)
|
|
return fetched
|
|
|
|
|
|
def api_forward(config, data):
|
|
post_url = f'{config["forward_api_address"]}/sonde/{config["identifiant_sonde"]}'
|
|
res = session.post(post_url, json=data)
|
|
print(res.ok)
|
|
print(res.json())
|
|
return res
|
|
|
|
|
|
def forward_api_error(config, message=""):
|
|
forward_urls = [
|
|
# "https://papi.silib.re/sonde/test/error/",
|
|
f'{config["forward_api_address"]}/sonde/{config["identifiant_sonde"]}',
|
|
|
|
"http://localhost:8000/sonde/838266b2-fc3a-4430-95e8-f7f0d0fc9871/error/",
|
|
# "http://localhost:8000/sonde/test/error/",
|
|
]
|
|
message = message or "Erreur inconnue"
|
|
for post_url in forward_urls:
|
|
|
|
forwarded_error = session.post(post_url, json={"msg": message})
|
|
print(post_url)
|
|
print(forwarded_error.ok)
|
|
print(forwarded_error.json())
|
|
|
|
|
|
if forwarded_error.ok:
|
|
logger.info("Fetched")
|
|
else: # pragma: no cover
|
|
err = f'{forwarded_error.reason}: {fetched.text} on "{config["api_status_url"]}"'
|
|
logger.info("Login error {err}")
|
|
raise ValueError(err)
|
|
|
|
|
|
def main(
|
|
*,
|
|
maxloop=math.inf,
|
|
login=api_login,
|
|
fetch=api_fetch,
|
|
forward=api_forward,
|
|
forward_error=forward_api_error,
|
|
pattern="relais_*",
|
|
):
|
|
loopcount = itertools.count().__next__
|
|
conf = config.read_config(kind="relais", pattern=pattern)
|
|
assert len(conf) == 1
|
|
conf = conf[0]
|
|
try:
|
|
login(conf)
|
|
except Exception as E:
|
|
forward_error(conf, str(E))
|
|
while loopcount() < maxloop:
|
|
try:
|
|
current = fetch(conf).json()
|
|
except Exception as E:
|
|
forward_error(conf, str(E))
|
|
else:
|
|
forward(conf, current)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main(maxloop=1)
|