mgokg commited on
Commit
70394eb
·
verified ·
1 Parent(s): 3ebcec9

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +4 -131
app.py CHANGED
@@ -4,136 +4,8 @@ from datetime import datetime, timedelta
4
  import xml.etree.ElementTree as ET
5
  import json
6
 
7
- api_key=os.environ.get("api_key")
8
- client_id=os.environ.get("client_id")
9
-
10
- def
11
- url = f"https://apis.deutschebahn.com/db-api-marketplace/apis/timetables/v1/station/{pattern}"
12
- headers = {
13
- 'DB-Client-Id': client_id,
14
- 'DB-Api-Key': api_key,
15
- 'accept': 'application/xml'
16
- }
17
- try:
18
- response = requests.get(url, headers=headers, timeout=10)
19
- response.raise_for_status()
20
- root = ET.fromstring(response.content)
21
- station = root.find('station')
22
- if station is not None:
23
- return {'name': station.get('name'), 'eva': station.get('eva')}
24
- except Exception:
25
- return None
26
- return None
27
-
28
- def fetch_timetable_hour(eva, date_str, hour_str, client_id, api_key):
29
-
30
- url = f"https://apis.deutschebahn.com/db-api-marketplace/apis/timetables/v1/plan/{eva}/{date_str}/{hour_str}"
31
- headers = {
32
- 'DB-Client-Id': client_id,
33
- 'DB-Api-Key': api_key,
34
- 'accept': 'application/xml'
35
- }
36
- try:
37
- response = requests.get(url, headers=headers, timeout=10)
38
- if response.status_code == 200:
39
- return ET.fromstring(response.content)
40
- except Exception:
41
- pass
42
- return None
43
-
44
- def search_next_3_connections(departure, destination, client_id, api_key):
45
-
46
- # 1. IBNRs ermitteln
47
- dep_info = get_station_info(departure, client_id, api_key)
48
- dest_info = get_station_info(destination, client_id, api_key)
49
-
50
- if not dep_info: return {"error": f"Startort '{departure}' nicht gefunden."}
51
- if not dest_info: return {"error": f"Zielort '{destination}' nicht gefunden."}
52
-
53
- now = datetime.now()
54
- found_connections = []
55
-
56
- # 2. Suche in der aktuellen und der nächsten Stunde (um 3 Verbindungen zu garantieren)
57
- for hour_offset in range(3): # Wir schauen bis zu 2 Stunden voraus
58
- search_time = now + timedelta(hours=hour_offset)
59
- datum = search_time.strftime("%y%m%d")
60
- stunde = search_time.strftime("%H")
61
-
62
- root = fetch_timetable_hour(dep_info['eva'], datum, stunde, client_id, api_key)
63
- if root is None: continue
64
-
65
- for train in root.findall('.//s'):
66
- dp = train.find('dp')
67
- if dp is None: continue
68
-
69
- # Prüfen ob Ziel im geplanten Pfad (ppth)
70
- path = dp.get('ppth', '')
71
- if dest_info['name'].lower() in path.lower() or destination.lower() in path.lower():
72
-
73
- pt = dp.get('pt', '') # Geplante Zeit: YYMMDDHHMM
74
- if not pt: continue
75
-
76
- dep_dt = datetime.strptime(pt, "%y%m%d%H%M")
77
-
78
- # Nur Verbindungen in der Zukunft (ab Zeitstempel jetzt)
79
- if dep_dt >= now:
80
- tl = train.find('tl')
81
- found_connections.append({
82
- "startort": dep_info['name'],
83
- "zielort": dest_info['name'],
84
- "abfahrtszeit": dep_dt.strftime("%d.%m.%Y %H:%M"),
85
- "gleis": dp.get('pp', 'n/a'),
86
- "zug": f"{tl.get('c', '')} {tl.get('n', '')}",
87
- "_dt": dep_dt # Hilfsfeld zum Sortieren
88
- })
89
-
90
- # 3. Sortieren und Top 3 auswählen
91
- found_connections.sort(key=lambda x: x['_dt'])
92
-
93
- # Bereinigen der Ergebnisse für die JSON-Ausgabe
94
- final_results = []
95
- for c in found_connections[:3]:
96
- # Entferne das interne Datetime-Objekt vor der JSON-Konvertierung
97
- res = {k: v for k, v in c.items() if k != "_dt"}
98
- final_results.append(res)
99
-
100
- return final_results
101
-
102
- # --- Gradio UI ---
103
-
104
- def ui_wrapper(dep, dest, cid, akey):
105
- results = search_next_3_connections(dep, dest, cid, akey)
106
- return json.dumps(results, indent=4, ensure_ascii=False)
107
-
108
- with gr.Blocks(title="DB JSON Fahrplan", theme=gr.themes.Soft()) as demo:
109
- gr.Markdown("# 🚆 DB Verbindungs-Suche (JSON)")
110
- gr.Markdown("Ermittelt die nächsten 3 Verbindungen ab dem aktuellen Zeitstempel.")
111
-
112
- with gr.Row():
113
- cid_input = gr.Textbox(label="DB Client ID", type="password")
114
- akey_input = gr.Textbox(label="DB API Key", type="password")
115
-
116
- with gr.Row():
117
- dep_input = gr.Textbox(label="Abfahrtsort", placeholder="z.B. Berlin")
118
- dest_input = gr.Textbox(label="Zielort", placeholder="z.B. Hamburg")
119
-
120
- btn = gr.Button("Suchen", variant="primary")
121
- output = gr.JSON(label="JSON Ergebnis") # Nutzt Gradio JSON Komponente für schöneres Format
122
-
123
- btn.click(fn=ui_wrapper, inputs=[dep_input, dest_input, cid_input, akey_input], outputs=output)
124
-
125
- if __name__ == "__main__":
126
- demo.launch()
127
-
128
-
129
- import gradio as gr
130
- import requests
131
- from datetime import datetime, timedelta
132
- import xml.etree.ElementTree as ET
133
- import json
134
-
135
  def get_station_info(pattern, client_id, api_key):
136
-
137
  url = f"https://apis.deutschebahn.com/db-api-marketplace/apis/timetables/v1/station/{pattern}"
138
  headers = {
139
  'DB-Client-Id': client_id,
@@ -152,7 +24,7 @@ def get_station_info(pattern, client_id, api_key):
152
  return None
153
 
154
  def fetch_timetable_hour(eva, date_str, hour_str, client_id, api_key):
155
-
156
  url = f"https://apis.deutschebahn.com/db-api-marketplace/apis/timetables/v1/plan/{eva}/{date_str}/{hour_str}"
157
  headers = {
158
  'DB-Client-Id': client_id,
@@ -168,7 +40,7 @@ def fetch_timetable_hour(eva, date_str, hour_str, client_id, api_key):
168
  return None
169
 
170
  def search_next_3_connections(departure, destination, client_id, api_key):
171
-
172
 
173
  if not all([departure, destination, client_id, api_key]):
174
  return {"error": "Bitte alle Felder ausfüllen."}
@@ -253,3 +125,4 @@ with gr.Blocks(title="DB JSON Fahrplan", theme=gr.themes.Soft()) as demo:
253
  btn.click(fn=ui_wrapper, inputs=[dep_input, dest_input, cid_input, akey_input], outputs=output)
254
 
255
  if __name__ == "__main__":
 
 
4
  import xml.etree.ElementTree as ET
5
  import json
6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7
  def get_station_info(pattern, client_id, api_key):
8
+ """Ermittelt EVA-Nummer und offiziellen Namen."""
9
  url = f"https://apis.deutschebahn.com/db-api-marketplace/apis/timetables/v1/station/{pattern}"
10
  headers = {
11
  'DB-Client-Id': client_id,
 
24
  return None
25
 
26
  def fetch_timetable_hour(eva, date_str, hour_str, client_id, api_key):
27
+ """Ruft den Fahrplan für eine bestimmte Stunde ab."""
28
  url = f"https://apis.deutschebahn.com/db-api-marketplace/apis/timetables/v1/plan/{eva}/{date_str}/{hour_str}"
29
  headers = {
30
  'DB-Client-Id': client_id,
 
40
  return None
41
 
42
  def search_next_3_connections(departure, destination, client_id, api_key):
43
+ """Hauptfunktion: Findet die nächsten 3 Verbindungen ab 'jetzt'."""
44
 
45
  if not all([departure, destination, client_id, api_key]):
46
  return {"error": "Bitte alle Felder ausfüllen."}
 
125
  btn.click(fn=ui_wrapper, inputs=[dep_input, dest_input, cid_input, akey_input], outputs=output)
126
 
127
  if __name__ == "__main__":
128
+ demo.launch()