forked from RocketMap/RocketMap
/
runserver.py
165 lines (132 loc) 路 5.54 KB
/
runserver.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
#!/usr/bin/python
# -*- coding: utf-8 -*-
import os
import sys
import shutil
import logging
import time
import re
# Currently supported pgoapi
pgoapi_version = "1.1.6"
# Moved here so logger is configured at load time
logging.basicConfig(format='%(asctime)s [%(threadName)16s][%(module)14s][%(levelname)8s] %(message)s')
log = logging.getLogger()
# Make sure pogom/pgoapi is actually removed if it is an empty directory
# This is a leftover directory from the time pgoapi was embedded in PokemonGo-Map
# The empty directory will cause problems with `import pgoapi` so it needs to go
oldpgoapiPath = os.path.join(os.path.dirname(__file__), "pogom/pgoapi")
if os.path.isdir(oldpgoapiPath):
log.info("I found %s, but its no longer used. Going to remove it...", oldpgoapiPath)
shutil.rmtree(oldpgoapiPath)
log.info("Done!")
# Assert pgoapi is installed
try:
import pgoapi
except ImportError:
log.critical("It seems `pgoapi` is not installed. You must run pip install -r requirements.txt again")
sys.exit(1)
# Assert pgoapi >= 1.1.6 is installed
from distutils.version import StrictVersion
if not hasattr(pgoapi, "__version__") or StrictVersion(pgoapi.__version__) < StrictVersion(pgoapi_version):
log.critical("It seems `pgoapi` is not up-to-date. You must run pip install -r requirements.txt again")
sys.exit(1)
from threading import Thread, Event
from queue import Queue
from flask_cors import CORS
from pogom import config
from pogom.app import Pogom
from pogom.utils import get_args, insert_mock_data, get_encryption_lib_path
from pogom.search import search_overseer_thread, fake_search_loop
from pogom.models import init_database, create_tables, drop_tables, Pokemon, Pokestop, Gym
from pgoapi import utilities as util
if __name__ == '__main__':
# Check if we have the proper encryption library file and get its path
encryption_lib_path = get_encryption_lib_path()
if encryption_lib_path is "":
sys.exit(1)
args = get_args()
if args.debug:
log.setLevel(logging.DEBUG);
else:
log.setLevel(logging.INFO);
# Let's not forget to run Grunt / Only needed when running with webserver
if not args.no_server:
if not os.path.exists(os.path.join(os.path.dirname(__file__), 'static/dist')):
log.critical('Please run "grunt build" before starting the server');
sys.exit();
# These are very noisey, let's shush them up a bit
logging.getLogger('peewee').setLevel(logging.INFO)
logging.getLogger('requests').setLevel(logging.WARNING)
logging.getLogger('pgoapi.pgoapi').setLevel(logging.WARNING)
logging.getLogger('pgoapi.rpc_api').setLevel(logging.INFO)
logging.getLogger('werkzeug').setLevel(logging.ERROR)
config['parse_pokemon'] = not args.no_pokemon
config['parse_pokestops'] = not args.no_pokestops
config['parse_gyms'] = not args.no_gyms
# Turn these back up if debugging
if args.debug:
logging.getLogger('requests').setLevel(logging.DEBUG)
logging.getLogger('pgoapi').setLevel(logging.DEBUG)
logging.getLogger('rpc_api').setLevel(logging.DEBUG)
# use lat/lng directly if matches such a pattern
prog = re.compile("^(\-?\d+\.\d+),?\s?(\-?\d+\.\d+)$")
res = prog.match(args.location)
if res:
log.debug('Using coords from CLI directly')
position = (float(res.group(1)), float(res.group(2)), 0)
else:
log.debug('Lookig up coords in API')
position = util.get_pos_by_name(args.location)
if not any(position):
log.error('Could not get a position by name, aborting')
sys.exit()
log.info('Parsed location is: %.4f/%.4f/%.4f (lat/lng/alt)',
position[0], position[1], position[2])
if args.no_pokemon:
log.info('Parsing of Pokemon disabled')
if args.no_pokestops:
log.info('Parsing of Pokestops disabled')
if args.no_gyms:
log.info('Parsing of Gyms disabled')
config['LOCALE'] = args.locale
config['CHINA'] = args.china
app = Pogom(__name__)
db = init_database(app)
if args.clear_db:
log.info('Clearing database')
if args.db_type == 'mysql':
drop_tables(db)
elif os.path.isfile(args.db):
os.remove(args.db)
create_tables(db)
app.set_current_location(position);
# Control the search status (running or not) across threads
pause_bit = Event()
pause_bit.clear()
# Setup the location tracking queue and push the first location on
new_location_queue = Queue()
new_location_queue.put(position)
if not args.only_server:
# Gather the pokemons!
if not args.mock:
log.debug('Starting a real search thread')
search_thread = Thread(target=search_overseer_thread, args=(args, new_location_queue, pause_bit, encryption_lib_path))
else:
log.debug('Starting a fake search thread')
insert_mock_data(position)
search_thread = Thread(target=fake_search_loop)
search_thread.daemon = True
search_thread.name = 'search_thread'
search_thread.start()
if args.cors:
CORS(app);
app.set_search_control(pause_bit)
app.set_location_queue(new_location_queue)
config['ROOT_PATH'] = app.root_path
config['GMAPS_KEY'] = args.gmaps_key
if args.no_server:
# This loop allows for ctrl-c interupts to work since flask won't be holding the program open
while search_thread.is_alive():
time.sleep(60)
else:
app.run(threaded=True, use_reloader=False, debug=args.debug, host=args.host, port=args.port)