ZVhqMLA6

1import requests
2
3from sqlalchemy import Table, Column, PrimaryKeyConstraint, String
4from sqlalchemy.sql import select
5from cloudbot import hook
6from cloudbot.util import web, database
7
8
9class APIError(Exception):
10    pass
11
12# Define database table
13
14table = Table(
15    "weather",
16    database.metadata,
17    Column('nick', String),
18    Column('loc', String),
19    PrimaryKeyConstraint('nick')
20)
21
22# Define some constants
23google_base = 'https://maps.googleapis.com/maps/api/'
24geocode_api = google_base + 'geocode/json'
25
26wunder_api = "http://api.wunderground.com/api/{}/forecast/geolookup/conditions/q/{}.json"
27
28# Change this to a ccTLD code (eg. uk, nz) to make results more targeted towards that specific country.
29# <https://developers.google.com/maps/documentation/geocoding/#RegionCodes>
30bias = None
31
32
33def check_status(status):
34    """
35    A little helper function that checks an API error code and returns a nice message.
36    Returns None if no errors found
37    """
38    if status == 'REQUEST_DENIED':
39        return 'The geocode API is off in the Google Developers Console.'
40    elif status == 'ZERO_RESULTS':
41        return 'No results found.'
42    elif status == 'OVER_QUERY_LIMIT':
43        return 'The geocode API quota has run out.'
44    elif status == 'UNKNOWN_ERROR':
45        return 'Unknown Error.'
46    elif status == 'INVALID_REQUEST':
47        return 'Invalid Request.'
48    elif status == 'OK':
49        return None
50
51
52def find_location(location):
53    """
54    Takes a location as a string, and returns a dict of data
55    :param location: string
56    :return: dict
57    """
58    params = {"address": location, "key": dev_key}
59    if bias:
60        params['region'] = bias
61
62    json = requests.get(geocode_api, params=params).json()
63
64    error = check_status(json['status'])
65    if error:
66        raise APIError(error)
67
68    return json['results'][0]['geometry']['location']
69
70def load_cache(db):
71    global location_cache
72    location_cache = []
73    for row in db.execute(table.select()):
74        nick = row["nick"]
75        location = row["loc"]
76        location_cache.append((nick,location))
77
78def add_location(nick, location, db):
79    test = dict(location_cache)
80    location = str(location)
81    if nick.lower() in test:
82        db.execute(table.update().values(loc=location.lower()).where(table.c.nick == nick.lower()))
83        db.commit()
84        load_cache(db)
85    else:
86        db.execute(table.insert().values(nick=nick.lower(), loc=location.lower()))
87        db.commit()
88        load_cache(db)
89
90@hook.on_start
91def on_start(bot, db):
92    """ Loads API keys """
93    global dev_key, wunder_key
94    dev_key = bot.config.get("api_keys", {}).get("google_dev_key", None)
95    wunder_key = bot.config.get("api_keys", {}).get("wunderground", None)
96    load_cache(db)
97
98
99def get_location(nick):
100    """looks in location_cache for a saved location"""
101    location = [row[1] for row in location_cache if nick.lower() == row[0]]
102    if not location:
103        return
104    else:
105        location = location[0]
106    return location
107
108@hook.command("weather", "we", autohelp=False)
109def weather(text, reply, db, nick, notice):
110    """weather <location> -- Gets weather data for <location>."""
111    if not wunder_key:
112        return "This command requires a Weather Underground API key."
113    if not dev_key:
114        return "This command requires a Google Developers Console API key."
115
116    location = ""
117    # If no input try the db
118    if not text:
119        location = get_location(nick)
120        if not location:
121            notice(weather.__doc__)
122            return
123    else:
124        location = text
125    # use find_location to get location data from the user input
126    try:
127        location_data = find_location(location)
128    except APIError as e:
129        return e
130
131    formatted_location = "{lat},{lng}".format(**location_data)
132
133    url = wunder_api.format(wunder_key, formatted_location)
134    response = requests.get(url).json()
135
136    if response['response'].get('error'):
137        return "{}".format(response['response']['error']['description'])
138
139    forecast_today = response["forecast"]["simpleforecast"]["forecastday"][0]
140    forecast_tomorrow = response["forecast"]["simpleforecast"]["forecastday"][1]
141
142    # put all the stuff we want to use in a dictionary for easy formatting of the output
143    weather_data = {
144        "place": response['current_observation']['display_location']['full'],
145        "conditions": response['current_observation']['weather'],
146        "temp_f": response['current_observation']['temp_f'],
147        "temp_c": response['current_observation']['temp_c'],
148        "humidity": response['current_observation']['relative_humidity'],
149        "wind_kph": response['current_observation']['wind_kph'],
150        "wind_mph": response['current_observation']['wind_mph'],
151        "wind_direction": response['current_observation']['wind_dir'],
152        "today_conditions": forecast_today['conditions'],
153        "today_high_f": forecast_today['high']['fahrenheit'],
154        "today_high_c": forecast_today['high']['celsius'],
155        "today_low_f": forecast_today['low']['fahrenheit'],
156        "today_low_c": forecast_today['low']['celsius'],
157        "tomorrow_conditions": forecast_tomorrow['conditions'],
158        "tomorrow_high_f": forecast_tomorrow['high']['fahrenheit'],
159        "tomorrow_high_c": forecast_tomorrow['high']['celsius'],
160        "tomorrow_low_f": forecast_tomorrow['low']['fahrenheit'],
161        "tomorrow_low_c": forecast_tomorrow['low']['celsius']
162    }
163
164    # Get the more accurate URL if available, if not, get the generic one.
165    if "?query=," in response["current_observation"]['ob_url']:
166        try:
167            weather_data['url'] = web.try_shorten(response["current_observation"]['forecast_url'])
168        except:
169            weather_data['url'] = response["current_observation"]["forcast_url"]
170            pass
171    else:
172        try:
173            weather_data['url'] = web.try_shorten(response["current_observation"]['ob_url'])
174        except:
175            weather_data['url'] = response["current_observation"]["ob_url"]
176            pass
177
178    reply("{place} - \x02Current:\x02 {conditions}, {temp_f}F/{temp_c}C, {humidity}, "
179          "Wind: {wind_mph}MPH/{wind_kph}KPH {wind_direction}, \x02Today:\x02 {today_conditions}, "
180          "High: {today_high_f}F/{today_high_c}C, Low: {today_low_f}F/{today_low_c}C. "
181          "\x02Tomorrow:\x02 {tomorrow_conditions}, High: {tomorrow_high_f}F/{tomorrow_high_c}C, "
182          "Low: {tomorrow_low_f}F/{tomorrow_low_c}C - {url}".format(**weather_data))
183    if text:
184        add_location(nick, location, db)