theraPy/arztapi/APIHandler.py

from typing import List

import requests
from requests import JSONDecodeError
import base64
from datetime import datetime
from arztapi.ArztPraxisDatas import ArztPraxisDatas
from arztapi.DoctorInformation import DoctorInformation, PhoneTime
from arztapi.DoctorPhoneTime import DoctorPhoneTime


class APIHandler:
    def __init__(self):
        self.base_api_url = "https://arztsuche.116117.de/api/"
        self.phone_times = []
        self.general_information = []

    def get_lat_lon_location_list(self, location):
        api_path = self.base_api_url + "location"
        headers = {
            "Accept": "application/json, text/plain, */*",
            "Accept-Language": "de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7",
            "Authorization": "Basic YmRwczpma3I0OTNtdmdfZg==",
            "Connection": "keep-alive",
        }

        params = {
            "loc": location,
        }

        response = requests.get(api_path, params=params, headers=headers)

        try:
            return response.json()

        except JSONDecodeError:
            print("foo")
            print(response.text)

    def get_list_of_doctors(self, lat, lon, req_val_base64, therapy_types, therapy_age, therapy_setting) -> ArztPraxisDatas:
        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": "Basic YmRwczpma3I0OTNtdmdfZg==",
            "Connection": "keep-alive",
            "req-val": req_val_base64,
        }

        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,
        }

        response = requests.post(api_path, headers=headers, json=json_data)
        response.raise_for_status()
        self.phone_times = ArztPraxisDatas(**response.json())
        return self.phone_times

    def get_general_doctor_information(self) -> List[DoctorInformation]:
        general_doctor_information = []
        for data in self.phone_times.arztPraxisDatas:
            doctor_day_times = data.tsz
            phone_times = []
            for day in doctor_day_times:
                if day.tszDesTyps:
                    for contact_times in day.tszDesTyps:
                        if contact_times.typ == "Telefonische Erreichbarkeit":
                            phone_times_day = contact_times.sprechzeiten
                            for phone_time_day in phone_times_day:
                                start_time_str, end_time_str = phone_time_day.zeit.split("-")
                                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}")
                                current_phone_time_dict = {
                                    "start": start_date_time,
                                    "end": end_date_time
                                }
                                current_phone_time = PhoneTime(**current_phone_time_dict)
                                phone_times.append(current_phone_time)

            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
            }
            doctor_information = DoctorInformation(**doctor_information_dict)
            general_doctor_information.append(doctor_information)

        self.general_information = general_doctor_information
        return self.general_information

    def filter_doctor_information_for_distance(self, distance):
        """
        :param distance: in meters
        :return:
        """
        self.general_information = [doctor_information for doctor_information in self.general_information
                                    if doctor_information.distance <= distance]

    def get_doctor_phone_times_sorted(self):
        doctor_phone_times = []
        for doctor_information in self.general_information:
            for phone_time in doctor_information.telefonzeiten:
                doctor_phone_time_dict = {
                    "phone_time": phone_time,
                    "doctor_name": doctor_information.name,
                    "doctor_address": f"{doctor_information.plz} {doctor_information.ort} "
                                      f"{doctor_information.strasse} {doctor_information.hausnummer}",
                    "doctor_phone_number": doctor_information.tel
                }
                doctor_phone_time = DoctorPhoneTime(**doctor_phone_time_dict)
                doctor_phone_times.append(doctor_phone_time)
        return sorted(doctor_phone_times, key=lambda dpt: dpt.phone_time.start)

    @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:
        :param lon:
        :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):
        format_string = "%d.%m. %H:%M"
        # TODO: handle turn of year
        parsed_date = datetime.strptime(date_string, format_string).replace(year=datetime.now().year)
        return parsed_date