105 lines
2.3 KiB
Python
105 lines
2.3 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(conf):
|
|
breakpoint()
|
|
if not logged:
|
|
api_login(conf)
|
|
apiurl = conf['api_status_url']
|
|
fetched = session.get("%s/status" % apiurl)
|
|
return fetched
|
|
|
|
|
|
def api_forward(config, data):
|
|
post_url = f'{config["forward_api_address"]}/sonde/conf{"identifiant_sonde"}'
|
|
breakpoint()
|
|
res = session.post(post_url, json=data)
|
|
print(res.ok)
|
|
print(res.json())
|
|
return res
|
|
|
|
|
|
def forward_api_error(conf,message=""):
|
|
breakpoint()
|
|
forward_urls = [
|
|
# "https://papi.silib.re/sonde/test/error/",
|
|
"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:
|
|
|
|
res = session.post(post_url, json={"msg": message})
|
|
print(post_url)
|
|
print(res.ok)
|
|
print(res.json())
|
|
|
|
pass
|
|
|
|
|
|
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(confconf,current)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main(maxloop=1)
|