Spaces:
Sleeping
Sleeping
File size: 4,652 Bytes
1bebf37 0d0c101 1bebf37 0d0c101 1bebf37 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 |
# -*- coding: utf-8 -*-
from flask import Flask, render_template, jsonify
import requests
import json
import re
import os
from dotenv import load_dotenv
# Charger les variables d'environnement
load_dotenv()
app = Flask(__name__)
TOMTOM_API_KEY = os.getenv("TOMTOM_API_KEY")
API_KEY = os.getenv("IDFM_API_KEY")
BASE_URL_API = "https://prim.iledefrance-mobilites.fr/marketplace/v2/navitia/line_reports/physical_modes/physical_mode%3A"
CATEGORIES = {
"Métro": "Metro",
"RER": "RapidTransit",
"Bus": "Bus",
"Transilien": "LocalTrain",
"Tram": "Tramway"
}
# Fonction pour récupérer les perturbations
def fetch_disruptions(category, api_value):
url = f"{BASE_URL_API}{api_value}/line_reports?"
headers = {"accept": "application/json", "apiKey": API_KEY}
try:
response = requests.get(url, headers=headers, timeout=5)
response.raise_for_status()
data = response.json()
return data.get("disruptions", [])
except requests.exceptions.RequestException as e:
print(f"Erreur API pour {category}: {e}")
return []
# Normaliser le nom des lignes
def normalize_line_name(line_name):
line_name = line_name.lower()
line_name = re.sub(r"\b(ratp|sncf)\b", "", line_name)
line_name = re.sub(r"\s+", " ", line_name).strip()
return line_name.capitalize()
# Générer le HTML des perturbations
def generate_traffic_html():
all_disruptions = {}
for category, api_value in CATEGORIES.items():
disruptions = fetch_disruptions(category, api_value)
active_disruptions = [
{
"id": disruption.get("id"),
"cause": disruption.get("cause", "Inconnue"),
"severity": disruption.get("severity", {}).get("name", "N/A"),
"message": disruption["messages"][0]["text"] if disruption.get("messages") else "No message",
"impacted_lines": [
obj.get("pt_object", {}).get("name", "Unknown Line")
for obj in disruption.get("impacted_objects", [])
if obj.get("pt_object", {}).get("embedded_type") == "line"
]
}
for disruption in disruptions if disruption.get("status") == "active"
]
all_disruptions[category] = active_disruptions
# Récapitulatif des lignes impactées
summary_data = []
seen_lines = set()
for category, disruptions in all_disruptions.items():
for d in disruptions:
for line in d["impacted_lines"]:
norm_line = normalize_line_name(line)
if norm_line not in seen_lines:
summary_data.append({
"category": category,
"line": norm_line,
"cause": d["cause"],
"severity": d["severity"],
"message": d["message"]
})
seen_lines.add(norm_line)
return summary_data
@app.route('/')
def home():
return render_template('index.html')
@app.route('/api/incidents', methods=['GET'])
def get_incidents():
"""
Récupère les incidents depuis l'API TomTom Traffic.
"""
try:
# Zone géographique pour Paris
bounding_box = "1.79804455,48.52917947,2.88843762,49.24075760"
url = f"https://api.tomtom.com/traffic/services/5/incidentDetails?bbox={bounding_box}&fields=%7Bincidents%7Btype%2Cgeometry%7Btype%2Ccoordinates%7D%2Cproperties%7BiconCategory%7D%7D%7D&language=en-GB&categoryFilter=0%2C1%2C2%2C3%2C4%2C5%2C6%2C7%2C8%2C9%2C10%2C11%2C14&timeValidityFilter=present&key={TOMTOM_API_KEY}"
response = requests.get(url)
response.raise_for_status()
data = response.json()
incidents = data.get("incidents", [])
return jsonify({"status": "success", "data": incidents}), 200
except requests.exceptions.RequestException as e:
return jsonify({"status": "error", "message": str(e)}), 500
@app.route('/api/ratp_disruptions', methods=['GET'])
def ratp_disruptions():
data = generate_traffic_html()
return jsonify({"status": "success", "data": data})
@app.route('/api/traffic-disruptions', methods=['GET'])
def get_traffic_disruptions():
"""
Récupère les perturbations de trafic pour tous les modes de transport.
"""
try:
summary_data = generate_traffic_html()
return jsonify({"status": "success", "data": summary_data}), 200
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500
if __name__ == '__main__':
app.run(debug=True) |