|
|
|
import requests |
|
import pandas as pd |
|
from datetime import datetime |
|
from datetime import timezone |
|
from pages.Location import * |
|
|
|
class ApiCall: |
|
def getData(self): |
|
|
|
api_variablen = { |
|
"hochwasser": "/lhp/mapData", |
|
"polizei": "/police/mapData", |
|
"wetter": "/dwd/mapData", |
|
"katwarn": "/katwarn/mapData", |
|
"mowas": "/mowas/mapData" |
|
} |
|
|
|
|
|
columns = ['ID', 'Urgency','Area', 'Titel', 'Event', 'Datum'] |
|
|
|
|
|
ninaBaseUrl = "https://warnung.bund.de/api31" |
|
ninaWarningsUrl = "https://nina.api.proxy.bund.dev/api31/warnings/" |
|
|
|
|
|
def get_api_warning(meldung): |
|
response = requests.get(ninaBaseUrl+meldung+".json") |
|
return response.json() |
|
|
|
|
|
def get_api_details(warning): |
|
n = 0 |
|
response = get_api_warning(warning) |
|
df = pd.DataFrame(columns=columns) |
|
|
|
for responses in response: |
|
id = responses["id"] |
|
warningDetails = requests.get(ninaWarningsUrl+id+".json").json() |
|
meldungsText = warningDetails["info"][0]["headline"]+ ": "+warningDetails["info"][0]["description"] |
|
warnung = warningDetails["info"][0]["event"] |
|
start = warningDetails["sent"] |
|
urgency = warningDetails["info"][0]["urgency"] |
|
area = warningDetails["info"][0]["area"][0]["areaDesc"] |
|
d = datetime.fromisoformat(start).astimezone(timezone.utc) |
|
d.strftime('%Y-%m-%d %H:%M:') |
|
time = d.replace(tzinfo=None) |
|
df.loc[n] = [id, urgency, area, meldungsText, warnung, time] |
|
n = n+1 |
|
return df |
|
|
|
|
|
def df_to_csv(data, filename): |
|
data.to_csv(filename+'.csv') |
|
|
|
|
|
data = pd.DataFrame(columns=columns) |
|
|
|
|
|
for x in api_variablen: |
|
data_api = get_api_details(api_variablen[x]) |
|
data = pd.concat([data, data_api]) |
|
|
|
|
|
data['Area'] = data['Area'].str.split(', ') |
|
df2 = data.explode('Area') |
|
|
|
|
|
def berechnung_func(row): |
|
loca = row['Area'] |
|
ort = loca.replace("Gemeinde", "") |
|
ort = loca.replace("Stadt", "") |
|
ort = loca.replace("Mitgliedsgemeinde in Verwaltungsgemeinschaft", "") |
|
|
|
|
|
|
|
|
|
location = Location(ort.strip()).getPostalCode() |
|
plz = location.iloc[0]['name'] |
|
|
|
return plz |
|
|
|
df2['Plz'] = df2.apply(berechnung_func, axis=1) |
|
|
|
|
|
df_to_csv(df2, "Data") |
|
|
|
|
|
return df2 |
|
|
|
|
|
|
|
|
|
|
|
|
