theraPy/arztapi/APIHandler.py
2024-09-10 21:02:23 +02:00

412 lines
18 KiB
Python

from typing import List
import requests
from requests import JSONDecodeError
import base64
import shelve
from datetime import datetime, timedelta
from arztapi.ArztPraxisDatas import ArztPraxisDatas
from arztapi.DoctorInformation import DoctorInformation, PhoneTime
from arztapi.DoctorPhoneTime import DoctorPhoneTime
class APIHandler:
"""
Class for accessing and handling the Arztsuche API -> Get necessary data and filter it for theraPy
"""
def __init__(self):
# Base URL as given by the base website
self.base_api_url = "https://arztsuche.116117.de/api/"
self.json_data = {}
# Containers for phone times, general doctor information and processed phone times of doctors
self.phone_times = []
self.general_information = []
self.processed_doctor_phone_times = []
self._cache = shelve.open("_therapydb.cache")
def get_lat_lon_location_list(self, location):
"""
Use a given location input string and search for the location with the Arztsuche API -> input string can contain
more or less everything to search for, but should not contain spaces (since the original API is not able to
handle spaces properly)
:param location: given as input string/plz with more or less validation, directly given to the Arztsuche API
:return: location matches in JSON format or catching raised JSONDecodeError (resulting in None)
"""
# API path as given by original website
api_path = self.base_api_url + "location"
# Headers copied by manual cURL
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": "de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7",
# Authorization header gathered by initial cURLing
"Authorization": "Basic YmRwczpma3I0OTNtdmdfZg==",
"Connection": "keep-alive",
}
params = {
"loc": location,
}
response = requests.get(api_path, params=params, headers=headers)
# Try to return the response in JSON
try:
# Possibly multiple results, further processing by caller
return response.json()
# JSONDecodeError for empty response object -> None, no location match
except JSONDecodeError:
return None
def get_list_of_doctors(self, lat, lon, req_val_base64, therapy_types, therapy_age, therapy_setting,
amount_of_weeks) -> ArztPraxisDatas:
# Data object as built by the original website, some fields might not be plausible or known (since there is no
# API documentation itself available
self.json_data = {
# TODO: Find out what r means
"r": 900,
"lat": lat,
"lon": lon,
"filterSelections": [
{
"title": "Fachgebiet Kategorie",
"fieldName": "fgg",
"selectedCodes": [
"12",
],
},
{
"title": "Psychotherapie: Verfahren",
"fieldName": "ptv",
"selectedCodes": therapy_types,
},
{
"title": "Psychotherapie: Altersgruppe",
"fieldName": "pta",
"selectedCodes": [
therapy_age
],
},
{
"title": "Psychotherapie: Setting",
"fieldName": "pts",
"selectedCodes": [
therapy_setting
],
},
],
"locOrigin": "USER_INPUT",
"initialSearch": False,
"viaDeeplink": False,
}
amount_of_days = amount_of_weeks * 7
cache_data = self.get_current_doctor_information_data_in_cache_with_time_check(amount_of_days)
if cache_data:
print("using cached data")
self.phone_times = cache_data
else:
print("call api")
self.get_list_of_doctors_from_api(lat, lon, req_val_base64, therapy_types, therapy_age, therapy_setting)
self.set_current_doctor_information_data_in_cache()
def get_list_of_doctors_from_api(self, lat, lon, req_val_base64, therapy_types, therapy_age,
therapy_setting) -> ArztPraxisDatas:
"""
:param lat: Latitude as given by location API
:param lon: Longitude as given by location API
:param req_val_base64: base64 value required for API access (is this a token?)
:param therapy_types: Therapy types of interest
:param therapy_age: Therapy age range of interest
:param therapy_setting: Therapy setting of interest
:return: Relevant doctor/therapist data
"""
# API path for doctor data
api_path = self.base_api_url + "data"
headers = {
"Accept": "application/json",
"Accept-Language": "de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7",
# Authorization header gathered by initial cURLing
"Authorization": "Basic YmRwczpma3I0OTNtdmdfZg==",
"Connection": "keep-alive",
# Calculated base64 value based on latitude and longitude
"req-val": req_val_base64,
}
response = requests.post(api_path, headers=headers, json=self.json_data)
# Check for HTTP errors
response.raise_for_status()
# Convert phone times to data format as input validation, save as class variable for further processing
self.phone_times = ArztPraxisDatas(**response.json())
# Return result for processing by caller
return self.phone_times
def get_current_doctor_information_data_in_cache_with_time_check(self, amount_of_days):
cached_data = self._cache.get(str(self.json_data))
if cached_data:
cache_timestamp = cached_data["timestamp"]
current_date = datetime.now()
time_difference = current_date - cache_timestamp
if time_difference.days <= amount_of_days:
return cached_data["data"]
def set_current_doctor_information_data_in_cache(self):
current_date = datetime.now()
self._cache[str(self.json_data)] = {"timestamp": current_date, "data": self.phone_times}
def get_general_doctor_information(self) -> List[DoctorInformation]:
"""
Transform and filter data to more usable format: Check for phone times and collect general doctor information
data of interest
Function should be called after initial API call, but doesn't create an error if not, just returning empty list
in case of
:return: General doctor information
"""
# Create container for saving information
general_doctor_information = []
# Iterate over phone times -> API should have been accessed before
for data in self.phone_times.arztPraxisDatas:
# Remove empty phone number fields
if data.tel == "":
continue
# Get times of doctors in a day (maybe tsz = tageszeit?)
doctor_day_times = data.tsz
phone_times = []
# Get phone times on every given day
for day in doctor_day_times:
# If available -> phone time(s) detected
if day.tszDesTyps:
# Collect every phone time
for contact_times in day.tszDesTyps:
# Check if phone time is actually phone time and not only opening time
if contact_times.typ == "Telefonische Erreichbarkeit":
# Get as call time available for speaking with therapist
phone_times_day = contact_times.sprechzeiten
# Process phone time properly
for phone_time_day in phone_times_day:
# String magic since the actual phone time is given as string such as 9:00-10:00
start_time_str, end_time_str = phone_time_day.zeit.split("-")
# Parse both to datetime object
start_date_time = self.parse_date_string(f"{day.d} {start_time_str}")
end_date_time = self.parse_date_string(f"{day.d} {end_time_str}")
# Create a dict out of it
current_phone_time_dict = {
"start": start_date_time,
"end": end_date_time
}
# Dict to actual PhoneTime object -> input validation
current_phone_time = PhoneTime(**current_phone_time_dict)
# Add result to overall list
phone_times.append(current_phone_time)
# Collect relevant information of doctor/therapist
doctor_information_dict = {
"name": data.name,
"tel": data.tel,
"fax": data.fax,
"anrede": data.anrede,
"email": data.email,
"distance": data.distance,
"strasse": data.strasse,
"hausnummer": data.hausnummer,
"plz": data.plz,
"ort": data.ort,
"telefonzeiten": phone_times
}
# Convert to actual DoctorInformation object -> input validation
doctor_information = DoctorInformation(**doctor_information_dict)
general_doctor_information.append(doctor_information)
# Save result to class for further processing and return for caller
self.general_information = general_doctor_information
return self.general_information
def filter_doctor_information_for_distance(self, distance):
"""
Filter the given list of doctors based on the distance to them calculated by the given location as a result of
the original API call
:param distance: Distance to location given in meters for doctors within given radius
:return: No return, only filter class variable
"""
# Keep every doctor information within the given distance/radius
self.general_information = [doctor_information for doctor_information in self.general_information
if doctor_information.distance <= distance]
def get_doctor_phone_times_sorted(self, therapy_phone_weeks):
"""
Sort the current list of doctor phone times by start date and return the result
:param therapy_phone_weeks: Amount of weeks for showing phone times
:return:
"""
# Process doctor information to desired format with relevant information for phone times
for doctor_information in self.general_information:
# Perspective: phone time as data object of choice
for phone_time in doctor_information.telefonzeiten:
# Relevant information
doctor_phone_time_dict = {
"phone_time": phone_time,
# workaround until properly assigned in sort
"doctor_nr": 0,
"doctor_name": doctor_information.name,
# Address in readable format
"doctor_address": f"{doctor_information.plz} {doctor_information.ort} "
f"{doctor_information.strasse} {doctor_information.hausnummer}",
"doctor_phone_number": doctor_information.tel
}
# Convert to actual DoctorPhoneTime object -> input validation
doctor_phone_time = DoctorPhoneTime(**doctor_phone_time_dict)
self.processed_doctor_phone_times.append(doctor_phone_time)
# Apply filter for desired amount of weeks for phone times
self.filter_for_relevant_weeks(therapy_phone_weeks)
# Sort the times by the starting time
self.processed_doctor_phone_times.sort(key=lambda dpt: dpt.phone_time.start)
# Throw out already filtered times
self.filter_for_already_passed_times_today()
# Assign the numbers for showing them properly in the web interface
self.assign_numbers_to_doctor_phone_times()
return self.processed_doctor_phone_times
def filter_for_relevant_weeks(self, therapy_phone_weeks):
"""
Get the desired amount of weeks and filter for them
:param therapy_phone_weeks: Desired amount of weeks for showing phone times
:return: None, class variable affected
"""
# Get the current date for calculation and determining the next weeks
current_date = datetime.now()
# Calculate the end date by a timedelta based on the given amount of weeks
end_date = current_date + timedelta(weeks=therapy_phone_weeks)
# Filter for the relevant weeks
self.processed_doctor_phone_times = [dpt for dpt in self.processed_doctor_phone_times
if current_date <= dpt.phone_time.start <= end_date]
def filter_for_already_passed_times_today(self):
"""
Get the phone times of the current day and filter already passed phone times since they are considered not
relevant anymore
:return: None, class variable affected
"""
# Get the current date with time for calculations
current_datetime = datetime.now()
# Check if a phone time on the same day has already passed based on the end time or if the phone time is on
# another date (and keep them)
self.processed_doctor_phone_times = [dpt for dpt in self.processed_doctor_phone_times
if (dpt.phone_time.start.date() == current_datetime.date() and
dpt.phone_time.end > current_datetime)
or dpt.phone_time.start.date() != current_datetime.date()]
def assign_numbers_to_doctor_phone_times(self):
"""
Assign numbers to the doctors in phone time objects to distinct them from each other visually, since a doctor
can have multiple phone times in the next week(s)
:return:
"""
# Store known doctors and their number
known_doctor_names_with_nr = {}
# Users prefer starting with 1 instead of 0
doctor_count = 1
# For every phone time, process the doctor
for doctor_phone_time in self.processed_doctor_phone_times:
# New count for new doctor
if doctor_phone_time.doctor_name not in known_doctor_names_with_nr:
doctor_phone_time.doctor_nr = doctor_count
known_doctor_names_with_nr[doctor_phone_time.doctor_name] = doctor_count
doctor_count += 1
# Add known count to phone time with known doctor
else:
known_doctor_count = known_doctor_names_with_nr[doctor_phone_time.doctor_name]
doctor_phone_time.doctor_nr = known_doctor_count
@staticmethod
def calculate_req_value_base64(lat, lon):
"""
This function is based on the initial Javascript code found in app.js.
It is rewritten in Python to calculate the HTTP header req_val for proper requests with the correct location.
:param lat: Latitude given by the Arztsuche API
:param lon: Longitude given by the Arztsuche API
:return:
"""
# Adjust lat and lon values slightly
adjusted_lat = lat + 1.1
adjusted_lon = lon + 2.3
# Get the current time in milliseconds since epoch
current_time = datetime.now()
timestamp_str = str(int(current_time.timestamp() * 1000)) # Convert to milliseconds
# Extract digits from latitude
lat_integer_part = str(adjusted_lat).split(".")[0]
lat_last_digit = lat_integer_part[-1]
lat_first_fraction_digit = str(adjusted_lat).split(".")[1][0] if len(str(adjusted_lat).split(".")) > 1 else "0"
# Extract digits from longitude
lon_integer_part = str(adjusted_lon).split(".")[0]
lon_last_digit = lon_integer_part[-1]
lon_first_fraction_digit = str(adjusted_lon).split(".")[1][0] if len(str(adjusted_lon).split(".")) > 1 else "0"
# Create the final string by combining digits
combined_string = (
lat_last_digit +
timestamp_str[-1] +
lon_last_digit +
timestamp_str[-2] +
lat_first_fraction_digit +
timestamp_str[-3] +
lon_first_fraction_digit
)
# Encode the combined string in Base64
encoded_value = base64.b64encode(combined_string.encode()).decode()
return encoded_value
@staticmethod
def parse_date_string(date_string):
"""
Parse a date string given by the Arztsuche API and return an actual datetime object
:param date_string: Date string to parse
:return: parsed datetime as actual object or exception for failed parsing (we like input validation)
"""
# String as given as result by the API (based on known values)
format_string = "%d.%m. %H:%M"
# Get the current year since the year is not given as part of the date string to prevent the use of a wrong year
current_year = datetime.now().year
# Try block to prevent invalid dates
try:
# 24:00 can be part of a date string returned by the API - even though it doesn't make lots of sense
if "24:00" in date_string:
# Sometimes the API returns 24:00 as time, so filtering for those cases and replacing it with a minute
# less to work with proper input
date_string = date_string.replace("24:00", "23:59")
# Add the current year since it is not part of the original date sent by the API
parsed_date = datetime.strptime(date_string, format_string).replace(year=datetime.now().year)
# Handle turn of the year: if date is in the past relative to today, consider it as part of the next year
if parsed_date < datetime.now():
parsed_date = parsed_date.replace(year=current_year + 1)
return parsed_date
except ValueError as e:
raise ValueError(f"Error parsing date string: '{date_string}'. Details: {e}")