76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
import requests
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
import os
|
|
import logging
|
|
|
|
class WheatherAPI():
|
|
|
|
def __init__(self, bucket_name, folder_name):
|
|
self.current_url = 'http://api.weatherapi.com/v1/current.json?key=6b0654bd658842b085395550230406&q=location&aqi=no'
|
|
self.forecast_url = 'http://api.weatherapi.com/v1/forecast.json?key=6b0654bd658842b085395550230406&q=location&days=no_days&aqi=no&alerts=no'
|
|
self.bucket_name = bucket_name
|
|
self.folder_name = folder_name
|
|
|
|
def current_wheather(self, location):
|
|
|
|
url = self.current_url.replace('location', location)
|
|
response = requests.get(url)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
|
|
filename = 'current_' + location + '_' + datetime.now().strftime('%Y-%m-%d-%H-%M') + '.json'
|
|
|
|
with open(os.path.join('C:/Users/jacob/PycharmProjects/GCP-Weather-Forecast/current', filename) ,'w') as file: #/results/current/
|
|
json.dump(data, file)
|
|
|
|
|
|
|
|
def forecast_wheather(self, location, no_days):
|
|
|
|
url = self.forecast_url.replace('location', location)
|
|
url = url.replace('no_days', str(no_days))
|
|
response = requests.get(url)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
|
|
filename = 'forecasts_' + location + '_' + str(no_days) + '_' + datetime.now().strftime('%Y-%m-%d-%H-%M') + '.json'
|
|
|
|
with open(os.path.join('C:/Users/jacob/PycharmProjects/GCP-Weather-Forecast/forecasts', filename), 'w') as file: #/results/forecasts/
|
|
json.dump(data, file)
|
|
|
|
def call_current_wheater(self, last_datetime, location_list):
|
|
|
|
current_time = datetime.now().strftime('%Y-%m-%d-%H')
|
|
|
|
if last_datetime != current_time:
|
|
|
|
logging.info('Run of current wheather successful')
|
|
for location_name in location_list:
|
|
self.current_wheather(location_name)
|
|
else:
|
|
logging.info('Run of current wheather does not meet conditions.')
|
|
|
|
return current_time
|
|
|
|
def call_forecast_wheather(self, next_date, hour_of_forecast_run, location_list, max_forecast):
|
|
|
|
current_date = datetime.today().strftime('%Y-%m-%d')
|
|
|
|
if next_date == current_date and \
|
|
datetime.now().strftime('%H') == hour_of_forecast_run:
|
|
|
|
for location_name in location_list:
|
|
self.forecast_wheather(location_name, max_forecast)
|
|
|
|
logging.info('Run of forecast wheather successful')
|
|
|
|
return (datetime.today() + timedelta(days=1)).strftime('%Y-%m-%d')
|
|
|
|
else:
|
|
logging.info('Run of forecast wheather does not meet conditions.')
|
|
return next_date
|
|
|