forked from RocketMap/RocketMap
-
Notifications
You must be signed in to change notification settings - Fork 0
/
pokemonhandler.py
82 lines (62 loc) · 2.34 KB
/
pokemonhandler.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
import calendar
from threading import Thread
import requests
import json
from queue import Queue
from datetime import datetime as dt
from accounts import *
from geography import *
from pogom.fnord_webhook import wh_updater
args = None
queue = []
threads = []
pokemon = os.path.dirname(os.path.abspath(os.path.realpath(__file__))) + "/docs/pokemon.min.json"
with open(pokemon) as data_file:
pokemon_data = json.load(data_file)
wh_key_cache = {}
wh_updates_queue = Queue()
headers = {
'User-Agent': 'discord-simple-webhook (0.0.1)',
'Content-Type': 'application/json'
}
def pms2(pokemon_id, cell_id):
name = pokemon_data[str(pokemon_id)]["name"]
uri = "http://s2map.com/#order=latlng&mode=polygon&s2=false&points="
s2msg("{} funnet kl {} i sone {}{}".format(name, dt.now().strftime('%H:%M'), uri, cell_id))
def s2msg(msg_to_send):
if args.s2_hook:
d = json.dumps({'content': msg_to_send})
requests.post(args.s2_hook, headers=headers, data=d)
def send_to_webhook(pkmn):
wh_poke = pkmn
if "latitude" not in pkmn:
return
if "disappear_time" not in pkmn:
now = datetime.utcnow()
hardcoded_disappear = now + timedelta(hours=4, minutes=20, seconds=1)
pkmn["disappear_time"] = hardcoded_disappear.timetuple()
# pkmn["atitude"] = 42.0
if "longitude" not in pkmn:
pkmn["longitude"] = 43.0
wh_poke.update({
'disappear_time': calendar.timegm(pkmn["disappear_time"]),
# 'disappear_time': int(time.mktime(now_plus_10.timetuple()) * 1000),
'last_modified_time': "", # pkmn['last_modified_timestamp_ms'],
'time_until_hidden_ms': "", # p['time_till_hidden_ms'],
'verified': "", # SpawnPoint.tth_found(sp),
'seconds_until_despawn': "", # seconds_until_despawn,
'spawn_start': "", # start_end[0],
'spawn_end': "", # start_end[1],
'player_level': "", # encounter_level
})
wh_updates_queue.put(('pokemon', wh_poke))
def set_args(args_in):
global args
args = args_in
# Thread to process webhook updates.
for i in range(args.wh_threads):
log.debug('Starting wh-updater worker thread %d', i)
t = Thread(target=wh_updater, name='wh-updater-{}'.format(i),
args=(args, wh_updates_queue, wh_key_cache))
t.daemon = True
t.start()