230 lines
9.5 KiB
Python
230 lines
9.5 KiB
Python
from typing import Type
|
|
from mautrix.util.config import BaseProxyConfig, ConfigUpdateHelper
|
|
from maubot import Plugin, MessageEvent
|
|
from maubot.handlers import command
|
|
from datetime import datetime
|
|
import aiohttp
|
|
import asyncio
|
|
import json
|
|
import base64
|
|
import difflib
|
|
|
|
|
|
class Terminus:
|
|
def __init__(self, name):
|
|
self.name = name
|
|
self.horaires = []
|
|
|
|
def add_horaire(self, horaire):
|
|
self.horaires.append(horaire)
|
|
|
|
def get_horaires(self):
|
|
return self.sort_horaires()
|
|
|
|
def get_name(self):
|
|
return self.name
|
|
|
|
def sort_horaires(self):
|
|
inth = []
|
|
for h in self.horaires:
|
|
if h == "Proche":
|
|
inth.append(0)
|
|
else :
|
|
inth.append(int(h[:-4]))
|
|
inth.sort()
|
|
return inth
|
|
|
|
class Ligne:
|
|
def __init__(self, name):
|
|
self.name = name
|
|
self.terminus = []
|
|
|
|
def add_terminus(self, newterminus):
|
|
if newterminus in self.terminus:
|
|
return
|
|
self.terminus.append(newterminus)
|
|
|
|
def get_terminus(self, term):
|
|
for t in self.terminus:
|
|
if t.get_name() == term:
|
|
return t
|
|
return None
|
|
|
|
def get_all_terminus(self):
|
|
return self.terminus
|
|
|
|
def get_name(self):
|
|
return self.name
|
|
|
|
|
|
|
|
class Config(BaseProxyConfig):
|
|
def do_update(self, helper: ConfigUpdateHelper) -> None:
|
|
helper.copy("mail")
|
|
helper.copy("password")
|
|
|
|
|
|
class Tcl(Plugin):
|
|
async def start(self) -> None:
|
|
self.config.load_and_update()
|
|
|
|
|
|
@command.new(name="tcl", help="")
|
|
@command.argument("pattern", pass_raw=True, required=True)
|
|
async def tclInfos(self, evt: MessageEvent, pattern: str) -> None:
|
|
await evt.mark_read()
|
|
strToEncode = self.config["mail"] + ':' + self.config["password"]
|
|
base64string = base64.b64encode(strToEncode.encode('utf-8'))
|
|
if not pattern:
|
|
await evt.respond(
|
|
"""
|
|
!tcl alerte ligne -> Retourne les alertes TCL sur la ligne donnée
|
|
"""
|
|
)
|
|
else:
|
|
self.log.info("Commande : " + pattern)
|
|
if pattern.__contains__("alerte"):
|
|
chunks = pattern.split(None, 1)
|
|
if len(chunks) != 2:
|
|
await evt.respond(
|
|
"""
|
|
!tcl alerte ligne -> Retourne les alertes TCL sur la ligne donnée
|
|
"""
|
|
)
|
|
url = 'https://download.data.grandlyon.com/ws/rdata/tcl_sytral.tclalertetrafic_2/all.json?maxfeatures=200&start=1'
|
|
resp = await self.http.get(url, headers={'Authorization' : 'Basic ' + base64string.decode('utf-8')})
|
|
ans = await resp.text()
|
|
try:
|
|
resp.raise_for_status()
|
|
except aiohttp.ClientError as Error:
|
|
await evt.respond(str(Error))
|
|
return
|
|
objet = json.loads(ans)
|
|
try:
|
|
values = objet["values"]
|
|
alerte=False
|
|
for value in values:
|
|
if(value["ligne_cli"] == chunks[1]):
|
|
alerte=True
|
|
debut = datetime.fromisoformat(value["debut"])
|
|
fin = datetime.fromisoformat(value["fin"])
|
|
respText = "<h3>" + value["type"] + "</h3><p><strong>" + value["titre"] + "</strong></p><p>" \
|
|
+ value["message"] + "</p><strong>" + debut.strftime("%d/%m/%Y à %H:%M") + " au " \
|
|
+ fin.strftime("%d/%m/%Y à %H:%M") + "</strong></p>"
|
|
await evt.respond(respText, allow_html=True)
|
|
if(alerte==False):
|
|
await evt.respond("<h3>Pas d'alerte sur cette ligne</h3>", allow_html=True)
|
|
except KeyError:
|
|
await evt.respond("Erreur d'analyse JSON ")
|
|
elif pattern.__contains__("horaires"):
|
|
chunks = pattern.split(None, 1)
|
|
if len(chunks) != 2:
|
|
await evt.respond(
|
|
"""
|
|
!tcl horaires arrêt -> Retourne les horaires à l'arrêt
|
|
"""
|
|
)
|
|
return
|
|
with open("/home/tfa/maubot/tcl/tcl_sytral.tclarret.json") as file:
|
|
objet = json.load(file)
|
|
values = objet["values"]
|
|
arrets = []
|
|
for value in values:
|
|
arrets.append(value["nom"].lower())
|
|
arret = difflib.get_close_matches(chunks[1].lower(),arrets,n=1,cutoff=0.8)
|
|
if len(arret) == 0:
|
|
await evt.respond("Arrêt non trouvé")
|
|
return
|
|
Ids =[]
|
|
nomArret = ""
|
|
for value in values:
|
|
if value["nom"].lower() == arret[0]:
|
|
nomArret = value["nom"]
|
|
Ids.append(value["id"])
|
|
|
|
|
|
url = 'https://download.data.grandlyon.com/ws/rdata/tcl_sytral.tclpassagearret/all.json?maxfeatures=2000&start=1'
|
|
respText = "<h3>Prochains départs " + nomArret + " :</h3><ul>"
|
|
self.log.info(respText)
|
|
lines = []
|
|
while len(url) > 0:
|
|
resp = await self.http.get(url, headers={'Authorization' : 'Basic ' + base64string.decode('utf-8')})
|
|
ans = await resp.text()
|
|
try:
|
|
resp.raise_for_status()
|
|
except aiohttp.ClientError as Error:
|
|
await evt.respond(str(Error))
|
|
return
|
|
objet = json.loads(ans)
|
|
try:
|
|
url = objet["next"]
|
|
except KeyError:
|
|
url = ""
|
|
values = objet["values"]
|
|
|
|
for value in values:
|
|
if value["id"] in Ids:
|
|
if value["type"] == "E": #Filtré sur les estimés et non théoriques
|
|
# Les metros ont leur propre code
|
|
if value["coursetheorique"].split('-')[0] == "301A" :
|
|
transport = "A"
|
|
elif value["coursetheorique"].split('-')[0] == "303" :
|
|
transport = "C"
|
|
elif value["coursetheorique"].split('-')[0] == "325" :
|
|
transport = "F1"
|
|
elif value["coursetheorique"].split('-')[0] == "326" :
|
|
transport = "F2"
|
|
elif value["coursetheorique"].split('-')[0] == "304" :
|
|
transport = "D"
|
|
elif value["coursetheorique"].split('-')[0] == "302A" :
|
|
transport = "B"
|
|
else:
|
|
transport = value["coursetheorique"].split('-')[0][:-1]
|
|
|
|
found = False
|
|
for line in lines:
|
|
if line.get_name() == transport:
|
|
term = line.get_terminus(value["direction"])
|
|
if term is None:
|
|
newterm = Terminus(value["direction"])
|
|
newterm.add_horaire(value["delaipassage"])
|
|
line.add_terminus(newterm)
|
|
else:
|
|
term.add_horaire(value["delaipassage"])
|
|
found = True
|
|
if not found :
|
|
newLine = Ligne(transport)
|
|
newterm = Terminus(value["direction"])
|
|
newterm.add_horaire(value["delaipassage"])
|
|
newLine.add_terminus(newterm)
|
|
lines.append(newLine)
|
|
respText += "<ul>"
|
|
for line in lines:
|
|
self.log.info("Ligne " + line.get_name())
|
|
respText += "<li><h5>Ligne " + line.get_name() + "</h5></li>"
|
|
terms = line.get_all_terminus()
|
|
for t in terms:
|
|
self.log.info(t.get_name())
|
|
respText += "<b>Direction " + t.get_name() + "</b><ul>"
|
|
hs = t.get_horaires()
|
|
hs.sort()
|
|
for idx, h in enumerate(hs):
|
|
self.log.info(h)
|
|
if h == 0:
|
|
respText += "<li>Proche</li>"
|
|
else:
|
|
respText += "<li>" + str(h) + " min</li>"
|
|
if idx == 2: ##On limite à 3 items
|
|
break
|
|
respText += "</ul>"
|
|
respText += "</ul>"
|
|
await evt.respond(respText, allow_html=True)
|
|
|
|
|
|
@classmethod
|
|
def get_config_class(cls) -> Type[BaseProxyConfig]:
|
|
return Config
|
|
|
|
|
|
|