web_sanic.py

#!/usr/bin/env python3

from pkg_resources import resource_filename
from time import time

from sanic import Sanic
from sanic.response import html, json
from jinja2 import Environment, PackageLoader, Markup
from asyncpg import create_pool

from monocle import sanitized as conf
from monocle.bounds import center
from monocle.names import DAMAGE, MOVES, POKEMON
from monocle.web_utils import get_scan_coords, get_worker_markers, Workers, get_args


env = Environment(loader=PackageLoader('monocle', 'templates'))
app = Sanic(__name__)
app.static('/static', resource_filename('monocle', 'static'))


def social_links():
    social_links = ''

    if conf.FB_PAGE_ID:
        social_links = '<a class="map_btn facebook-icon" target="_blank" href="https://www.facebook.com/' + conf.FB_PAGE_ID + '"></a>'
    if conf.TWITTER_SCREEN_NAME:
        social_links += '<a class="map_btn twitter-icon" target="_blank" href="https://www.twitter.com/' + conf.TWITTER_SCREEN_NAME + '"></a>'
    if conf.DISCORD_INVITE_ID:
        social_links += '<a class="map_btn discord-icon" target="_blank" href="https://discord.gg/' + conf.DISCORD_INVITE_ID + '"></a>'
    if conf.TELEGRAM_USERNAME:
        social_links += '<a class="map_btn telegram-icon" target="_blank" href="https://www.telegram.me/' + conf.TELEGRAM_USERNAME + '"></a>'

    return Markup(social_links)


def render_map():
    css_js = ''

    if conf.LOAD_CUSTOM_CSS_FILE:
        css_js = '<link rel="stylesheet" href="static/css/custom.css">'
    if conf.LOAD_CUSTOM_JS_FILE:
        css_js += '<script type="text/javascript" src="static/js/custom.js"></script>'

    js_vars = Markup(
        "_defaultSettings['FIXED_OPACITY'] = '{:d}'; "
        "_defaultSettings['SHOW_TIMER'] = '{:d}'; "
        "_defaultSettings['TRASH_IDS'] = [{}]; ".format(conf.FIXED_OPACITY, conf.SHOW_TIMER, ', '.join(str(p_id) for p_id in conf.TRASH_IDS)))

    template = env.get_template('custom.html' if conf.LOAD_CUSTOM_HTML_FILE else 'newmap.html')
    return html(template.render(
        area_name=conf.AREA_NAME,
        map_center=center,
        map_provider_url=conf.MAP_PROVIDER_URL,
        map_provider_attribution=conf.MAP_PROVIDER_ATTRIBUTION,
        social_links=social_links(),
        init_js_vars=js_vars,
        extra_css_js=Markup(css_js)
    ))


def render_worker_map():
    template = env.get_template('workersmap.html')
    return html(template.render(
        area_name=conf.AREA_NAME,
        map_center=center,
        map_provider_url=conf.MAP_PROVIDER_URL,
        map_provider_attribution=conf.MAP_PROVIDER_ATTRIBUTION,
        social_links=social_links()
    ))


@app.get('/')
async def fullmap(request, html_map=render_map()):
    return html_map


if conf.MAP_WORKERS:
    workers = Workers()


    @app.get('/workers_data')
    async def workers_data(request):
        return json(get_worker_markers(workers))


    @app.get('/workers')
    async def workers_map(request, html_map=render_worker_map()):
        return html_map


del env


@app.get('/data')
async def pokemon_data(request, _time=time):
    last_id = request.args.get('last_id', 0)
    async with app.pool.acquire() as conn:
        results = await conn.fetch('''
            SELECT id, pokemon_id, expire_timestamp, lat, lon, atk_iv, def_iv, sta_iv, move_1, move_2
            FROM sightings
            WHERE expire_timestamp > {} AND id > {}
        '''.format(_time(), last_id))
    return json(list(map(sighting_to_marker, results)))


@app.get('/gym_data')
async def gym_data(request, names=POKEMON, _str=str):
    async with app.pool.acquire() as conn:
        results = await conn.fetch('''
            SELECT
                fs.fort_id,
                fs.id,
                fs.team,
                fs.prestige,
                fs.guard_pokemon_id,
                fs.last_modified,
                f.lat,
                f.lon
            FROM fort_sightings fs
            JOIN forts f ON f.id=fs.fort_id
            WHERE (fs.fort_id, fs.last_modified) IN (
                SELECT fort_id, MAX(last_modified)
                FROM fort_sightings
                GROUP BY fort_id
            )
        ''')
    return json([{
            'id': 'fort-' + _str(fort['fort_id']),
            'sighting_id': fort['id'],
            'prestige': fort['prestige'],
            'pokemon_id': fort['guard_pokemon_id'],
            'pokemon_name': names[fort['guard_pokemon_id']],
            'team': fort['team'],
            'lat': fort['lat'],
            'lon': fort['lon']
    } for fort in results])


@app.get('/spawnpoints')
async def spawn_points(request, _dict=dict):
    async with app.pool.acquire() as conn:
         results = await conn.fetch('SELECT spawn_id, despawn_time, lat, lon, duration FROM spawnpoints')
    return json([_dict(x) for x in results])


@app.get('/pokestops')
async def get_pokestops(request, _dict=dict):
    async with app.pool.acquire() as conn:
        results = await conn.fetch('SELECT external_id, lat, lon FROM pokestops')
    return json([_dict(x) for x in results])


@app.get('/scan_coords')
async def scan_coords(request):
    return json(get_scan_coords())


def sighting_to_marker(pokemon, names=POKEMON, moves=MOVES, damage=DAMAGE, trash=conf.TRASH_IDS, _str=str):
    pokemon_id = pokemon['pokemon_id']
    marker = {
        'id': 'pokemon-' + _str(pokemon['id']),
        'trash': pokemon_id in trash,
        'name': names[pokemon_id],
        'pokemon_id': pokemon_id,
        'lat': pokemon['lat'],
        'lon': pokemon['lon'],
        'expires_at': pokemon['expire_timestamp'],
    }
    move1 = pokemon['move_1']
    if move1:
        move2 = pokemon['move_2']
        marker['atk'] = pokemon['atk_iv']
        marker['def'] = pokemon['def_iv']
        marker['sta'] = pokemon['sta_iv']
        marker['move1'] = moves[move1]
        marker['move2'] = moves[move2]
        marker['damage1'] = damage[move1]
        marker['damage2'] = damage[move2]
    return marker


@app.listener('before_server_start')
async def register_db(app, loop):
    app.pool = await create_pool(**conf.DB, loop=loop)


def main():
    args = get_args()
    app.run(debug=args.debug, host=args.host, port=args.port)


if __name__ == '__main__':
    main()