from datetime import datetime, timedelta, timezone import json from urllib import parse import logging import re import typing as t import requests import models from config import env_OWM_KEY, env_OWM_UNITS logger = logging.getLogger("gunicorn.error") weather_template = "The current temperature is {0} and feels like {1}. The high today is {2} with a low of {3}. The current humidity is {4} percent. The summary for today is: {5}." def str_none(x): if x is None: return "" else: return str(x) def string_validator(input_str: str): # Decode the input string decoded_str = parse.unquote(input_str) # Sanitize the string sanitized = re.sub(r"[\s]", "", decoded_str) sanitized = re.sub(r'[<>"\'%;]', "", sanitized) # Check length of the string if len(sanitized) < 1: return None return sanitized def validate_data_presence(data: t.Dict[str, t.Any], keys: list[str]) -> bool: """ Validate that all given keys are present in the data. Args: data (Dict[str, Any]): The JSON data to be validated. keys (list[str]): A list of keys to look for in the data. Returns: bool: If any key is missing, returns False. Otherwise, returns True. """ for key in keys: if key not in data: return False return True def strq(str: str) -> str: return "\"" + str + "\"" def _get_weather(lat, long): try: if lat is None or long is None: return "An error has occured and the provided zipcode could not be understood." weather_json = _get_weather_json(lat, long) if weather_json is None: return "An error has occured and the weather could not be retrieved." weather = weather_template.format( weather_json.current.temp, weather_json.current.feels_like, weather_json.daily[0].temp.max, weather_json.daily[0].temp.min, weather_json.current.humidity, weather_json.daily[0].summary, ) return weather except Exception as e: logger.error("Error in _get_weather: " + e) return "An error has occured and the weather could not be retrieved." def _get_weather_json(lat, long): url = "https://api.openweathermap.org/data/3.0/onecall?lat={0}&lon={1}&exclude=alerts,minutely,hourly&units={2}&appid={3}".format( lat, long, env_OWM_UNITS, env_OWM_KEY ) try: response = requests.get(url) if response.status_code == 200: weather = response.json() return weather else: logger.error("Error in _get_weather_json: " + str(response.status_code)) return None except requests.exceptions.RequestException as e: logger.error("Error in _get_weather_json: " + e) return None def _get_cords(zipcode): url = "http://api.openweathermap.org/geo/1.0/zip?zip={0},US&appid={1}".format( zipcode, env_OWM_KEY ) try: response = requests.get(url) if response.status_code == 200: zipcode = response.json() return zipcode.lat, zipcode.long else: logger.error("Error in _get_cords: " + str(response.status_code)) return None, None except requests.exceptions.RequestException as e: logger.error("Error in _get_cords: " + e) return None, None