From 05aee98c50ec8f362cc7141d41f45e842339c278 Mon Sep 17 00:00:00 2001 From: CMDR furrycat Date: Wed, 12 Apr 2017 11:47:06 +0100 Subject: [PATCH] Track faction influence from EDDN. --- app.py | 5 +- eddb.py | 6 ++ eddn.py | 13 +-- plugin/faction/faction.py | 245 +++++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 260 insertions(+), 9 deletions(-) diff --git a/app.py b/app.py index efa6048..0c89ff9 100644 --- a/app.py +++ b/app.py @@ -352,4 +352,7 @@ async def on_member_join(member): async def on_member_update(before, after): await plugins.on_member_update(before, after) -client.run(token) +try: + client.run(token) +except RunTimeError: + sys.exit(1) diff --git a/eddb.py b/eddb.py index efffc71..2999e9e 100644 --- a/eddb.py +++ b/eddb.py @@ -24,6 +24,12 @@ eddb_states = { def state_name(id): return eddb_states.get(str(id), 'Unknown state {}'.format(id)) +def get_state_id(name): + for state_id, state_name in eddb_states.items(): + if state_name.lower() == name.lower(): + return int(state_id) + return None + def eddb_query(root, params, expand = None): try: queries = [] diff --git a/eddn.py b/eddn.py index cab01c5..6e6f7a6 100644 --- a/eddn.py +++ b/eddn.py @@ -8,9 +8,9 @@ import zmq.asyncio log = logging.getLogger('eddn') -url = 'tcp://eddn-gateway.elite-markets.net:9500' +url = 'tcp://eddn-relay.elite-markets.net:9500' schema_prefix = 'http://schemas.elite-markets.net/eddn' -topics = ['{}/{}/{}'.format(schema_prefix, 'journal', 1)] +topics = [] timeout = 600000 async def listen(plugins): @@ -26,16 +26,17 @@ async def listen(plugins): log.info('Subscribing to {}'.format(topic)) subscriber.setsockopt(zmq.SUBSCRIBE, topic.encode('utf-8')) else: + log.info('Listening for any topic') subscriber.setsockopt(zmq.SUBSCRIBE, b'') - subscriber.setsockopt(zmq.RCVTIMEO, timeout) + subscriber.setsockopt(zmq.RCVTIMEO, int(timeout)) poller = zmq.asyncio.Poller() poller.register(subscriber, zmq.POLLOUT) while True: try: messages = await subscriber.recv_multipart() except zmq.error.Again: - log.exception('recv_multipart') - messages = False + log.debug('ZeroMQ timeout after {}s'.format(timeout / 1000)) + continue if messages == False: break for message in messages: @@ -61,5 +62,5 @@ async def decode_message(compressed): return None def is_schema(schema, namespace, *, version = None, test = False): - regex = '^{}/{}/{}{}$'.format(schema_prefix, namespace, version if version is not None else '\d+', '/test' if test else '').encode('string-escape') + regex = '^{}/{}/{}{}$'.format(schema_prefix, namespace, version if version is not None else '\d+', '/test' if test else '') return bool(re.match(regex, schema)) diff --git a/plugin/faction/faction.py b/plugin/faction/faction.py index 6052669..2bb360a 100644 --- a/plugin/faction/faction.py +++ b/plugin/faction/faction.py @@ -1,4 +1,5 @@ import asyncio +import hashlib import shlex from db import DBConnection @@ -7,6 +8,7 @@ from throwingargparse import ThrowingArgumentParser, ArgumentParserError import bot import cat import eddb +import eddn class TrackedFaction(object): def __init__(self, row): @@ -26,6 +28,8 @@ class Faction(DBConnection): def __init__(self, filename = None): super(Faction, self).__init__(filename) self.create_tables() + self.cache_stale = True + self.tracked_factions = [] def description(self): return 'Reports on faction state and recommended objectives.' @@ -35,6 +39,14 @@ class Faction(DBConnection): cursor = self.query('create table if not exists factions (id char(36) not null, bot_id varchar(32) not null, member_id varchar(32) not null, server_id varchar(32) not null, channel_id varchar(32) not null, name varchar(128) not null, state_id int not null, interval int not null default 86400, updated datetime)') cursor = self.query('create unique index if not exists factions_id on factions (id)') cursor = self.query('create index if not exists factions_bot_id on factions (bot_id)') + cursor = self.query('create table if not exists faction_names (faction_id int not null, name varchar(128) not null)') + cursor = self.query('create unique index if not exists faction_names_name on faction_names (name)') + cursor = self.query('create table if not exists faction_systems (eddb_id int not null, name text collate nocase not null)') + cursor = self.query('create unique index if not exists faction_system_name on faction_systems (name)') + cursor = self.query('create table if not exists faction_influence (checksum char(32) not null, eddb_id int not null, timestamp datetime, faction_id int not null, influence float not null, state_id int not null)') + cursor = self.query('create index if not exists faction_influence_system on faction_influence (eddb_id)') + cursor = self.query('create index if not exists faction_influence_faction on faction_influence (faction_id)') + cursor = self.query('create unique index if not exists faction_influence_checksum on faction_influence (checksum, eddb_id, faction_id)') self.dbh.commit() self.close_db() @@ -48,7 +60,7 @@ class Faction(DBConnection): def get_tracked_factions(self): now = self.now() - cursor = self.query("select id, server_id, member_id, channel_id, interval, name, state_id, updated from factions where updated is null or updated < ? - interval and bot_id=?", [now, client.user.id]) + cursor = self.query("select id, name, server_id, member_id, channel_id, interval, name, state_id, updated from factions where updated is null or updated < ? - interval and bot_id=?", [now, client.user.id]) for row in cursor.fetchall(): yield TrackedFaction(row) self.close_db() @@ -62,10 +74,82 @@ class Faction(DBConnection): def delete_tracked_faction(self, id): return self.delete_from_table(client, 'factions', id) + def cache_faction(self, *, id, name): + cursor = self.query('insert or ignore into faction_names values (?, ?)', [id, name]) + if cursor.rowcount: + self.commit() + ret = True + else: + log.debug('Not caching faction {} with ID {}'.format(name, id)) + ret = False + self.close_db() + return ret + + def get_faction_ids(self, names): + ids = {} + cursor = self.query('select name, faction_id from faction_names where name in ({})'.format(', '.join(['?'] * len(names))), names) + for row in cursor.fetchall(): + ids[row['name']] = row['faction_id'] + self.close_db() + if len(ids) == len(names): + return ids + missing = [name for name in names if name not in ids] + for name in names: + if name in ids: + continue + factions = eddb.find_faction(name) + if len(factions) >= 1: + if factions[0]['name'].lower() == name.lower(): + faction = factions[0] + id = faction['id'] + self.cache_faction(id = id, name = name) + ids[name] = id + continue + log.warning("Can't find faction with name {}".format(name)) + ids[name] = None + return ids + + def get_faction_id(self, name): + return self.get_faction_ids([name]).get(name) + + def cache_system(self, *, id, name): + cursor = self.query('insert or ignore into faction_systems values (?, ?)', [id, name]) + if cursor.rowcount: + self.commit() + ret = True + else: + log.debug('Not caching system {} with ID {}'.format(name, id)) + ret = False + self.close_db() + return ret + + def get_system_id(self, name): + cursor = self.query('select eddb_id from faction_systems where name=?', [name]) + row = cursor.fetchone() + if row: + id = row['eddb_id'] + else: + id = None + self.close_db() + if id is not None: + return id + systems = eddb.find_system(name) + if len(systems) >= 1: + if systems[0]['name'].lower() == name.lower(): + system = systems[0] + id = system['id'] + self.cache_system(id = id, name = name) + return id + log.warning("Can't find system with name {}".format(name)) + return None + def valid_commands(self): return ['faction'] async def on_ready(self): + if self.cache_stale: + self.tracked_factions = list(self.get_all_tracked_factions()) + self.cache_stale = False waittime = 60 while True: # Convert to list because we will be sharing the cursor. @@ -89,17 +173,32 @@ class Faction(DBConnection): 'Commands to track faction states are:', '```', 'faction [show] FACTION', + 'faction cache', 'faction edit ID', + 'faction influence FACTION [SYSTEM]', 'faction list', 'faction track FACTION', 'faction untrack FACTION', '```' ] + elif command == 'cache': + lines = [ + 'Manage internal faction cache.' + ] elif command == 'edit': lines = [ 'Update faction tracking', 'See notes on `faction track` for the paramters to use.' ] + elif command == 'influence': + lines = [ + 'Show influence of faction:', + '```', + 'faction influence FACTION', + 'faction influence FACTION SYSTEM', + '```', + 'If no `SYSTEM` is supplied the influence in all known systems will be reported.' + ] elif command == 'list': lines = [ 'List tracked factions:', @@ -141,12 +240,16 @@ class Faction(DBConnection): if command == 'help': await self.help_faction(message) return + elif command == 'cache': + await self.show_faction_cache(message) elif command == 'list': await self.list_tracked_factions(message) elif command == 'show': await self.show_faction(message, args[1]) elif command == 'edit': await self.edit_faction(message, args[1:]) + elif command == 'influence': + await self.show_influence_report(message, args[1:]) elif command == 'track': await self.track_faction(message, args[1:]) elif command == 'untrack': @@ -154,6 +257,18 @@ class Faction(DBConnection): else: await self.show_faction(message, args[0]) + async def show_faction_cache(self, message): + cursor = self.query('select faction_id, name from faction_names') + lines = [] + for row in cursor.fetchall(): + lines.append('Faction **{}** #{}'.format(row['name'], row['faction_id'])) + cursor = self.query('select eddb_id, name from faction_systems') + for row in cursor.fetchall(): + lines.append('System **{}** #{}'.format(row['name'], row['eddb_id'])) + self.close_db() + if len(lines): + await bot.say_many(message.channel, lines) + async def list_tracked_factions(self, message): results = [] for faction in self.get_all_tracked_factions(self.get_message_servers(client, message, True)): @@ -222,6 +337,7 @@ class Faction(DBConnection): faction = await self.get_faction(message.channel, name) if faction is None: return + self.cache_faction(id = faction['id'], name = faction['name']) facts = [] if faction.get('allegianceName'): @@ -232,6 +348,7 @@ class Faction(DBConnection): text = '{} {} faction **{}** #{}'.format(', '.join(facts), 'player' if faction['is_player_faction'] else 'minor', faction['name'], faction['id']) system = eddb.get_system(faction['home_system_id']) if system: + self.cache_system(id = system['id'], name = system['name']) text += ' from *{}*'.format(system['name']) state = eddb.state_name(faction['state_id']) if state == 'None': @@ -304,6 +421,7 @@ class Faction(DBConnection): log.warning("Can't get channel for faction {}".format(tracked.id)) self.update_tracked_faction(tracked.id, **update) + await self.report_faction_influence(channel, name = tracked.name, yelp_if_no_data = False) async def get_faction(self, destination, name): factions = eddb.find_faction(name) @@ -314,6 +432,8 @@ class Faction(DBConnection): return None if len(factions) > 1: if factions[0]['name'].lower() == name.lower(): + if factions['state_id'] is None: + factions['state_id'] = eddb.get_state_id('None') return factions[0] log.info('Multiple factions matching {}.'.format(name)) if destination is not None: @@ -492,7 +612,7 @@ class Faction(DBConnection): if faction is None: return None parsed['name'] = faction['name'] - parsed['state_id'] = faction['state_id'] + parsed['state_id'] = faction['state_id'] if faction['state_id'] is not None else eddb.get_state_id('None') return parsed @@ -512,6 +632,8 @@ class Faction(DBConnection): await bot.say(message.channel, id) await cat.purr(None, bot.voice_channel_for_user(message.author), join = False) await self.report_tracked_faction(self.get_tracked_faction(id)) + self.tracked_factions = list(self.get_all_tracked_factions()) + self.cache_stale = False else: log.info('Failed to track faction: {}'.format(create)) await cat.yelp(message.channel) @@ -535,6 +657,8 @@ class Faction(DBConnection): log.info('Edited faction {}: {}'.format(id, update)) await bot.say(message.channel, ', '.join([self.faction_key(k) for k in update])) await cat.purr(None, bot.voice_channel_for_user(message.author), join = False) + self.tracked_factions = list(self.get_all_tracked_factions()) + self.cache_stale = False else: log.info('Failed to edit faction {}: {}'.format(id, update)) await cat.yelp(message.channel) @@ -553,5 +677,122 @@ class Faction(DBConnection): log.info('Deleting faction {}'.format(id)) if self.delete_tracked_faction(id): await cat.purr(message.channel) + self.tracked_factions = list(self.get_all_tracked_factions()) + self.cache_stale = False else: await cat.yelp(message.channel) + + async def eddn_message(self, schema, header, message): + if not eddn.is_schema(schema, 'journal'): + return + factions = message.get('Factions', []) + if not len(factions): + return + if self.cache_stale: + self.tracked_factions = list(self.get_all_tracked_factions()) + if not any(filter(lambda name: name in [tracked.name for tracked in self.tracked_factions], [faction['Name'] for faction in factions])): + log.debug('No tracked factions among {}; tracking {}'.format(', '.join([faction['Name'] for faction in factions]), ', '.join([tracked.name for tracked in self.tracked_factions]))) + return + await self.update_factions(message.get('StarSystem'), factions, message.get('timestamp')) + + async def update_factions(self, system, factions, iso8601): + influence = ', '.join(['{} {:.2f}%'.format(faction['Name'], faction['Influence'] * 100.0) for faction in reversed(sorted(factions, key = lambda faction: faction['Influence']))]) + log.debug('{} factions: {}'.format(system, influence)) + try: + timestamp = bot.parse_iso8601(iso8601) + except ValueError: + log.warning("Can't update influence in {} with invalid timestamp {}: {}".format(system, iso8601, influence)) + return False + + faction_ids = self.get_faction_ids([faction['Name'] for faction in factions]) + if not all(faction_ids.values()): + log.warning("Can't update influence in {} without mapping all factions: {}".format(system, influence)) + return False + total = sum([faction['Influence'] * 100.0 for faction in factions]) + if total < 99.99: + log.warning("Can't update influence with total {}% less than 100%: {}".format(total, influence)) + return False + + system_id = self.get_system_id(system) + if not system_id: + log.warning("Can't update influence for unknown system {}: {}".format(system, influence)) + return False + + cursor = self.query('select timestamp from faction_influence where eddb_id=? order by timestamp desc limit 1', [system_id]) + row = cursor.fetchone() + if row: + if timestamp < row['timestamp']: + log.debug("Ignoring old influence data.") + self.close_db() + return True + + # Use checksum to avoid writing duplicate data. + m = hashlib.md5() + m.update(str(system_id).encode('utf-8')) + m.update(influence.encode('utf-8')) + checksum = m.hexdigest() + + log.info('Updating influence in {}: {}'.format(system, influence)) + for faction in factions: + state_id = eddb.get_state_id(faction['FactionState']) + if state_id is None: + state_id = eddb.get_state_id('None') + name = faction['Name'] + faction_id = faction_ids[name] + cursor = self.query('insert or ignore into faction_influence values (?, ?, ?, ?, ?, ?)', [checksum, system_id, timestamp, faction_id, faction['Influence'], state_id]) + if not cursor.rowcount: + log.debug('No update for {} with checksum {}'.format(system, checksum)) + self.dbh.commit() + + self.close_db() + return True + + async def report_faction_influence(self, destination, *, id = None, name = None, system = None, yelp_if_no_data = True): + if id is None: + if name is None: + log.error("Can't report influence without a name or ID!") + return False + id = self.get_faction_id(name) + if id is None: + log.error("Can't report influence of unknown faction {}".format(name)) + return False + + sql = 'select s.name as system_name, i.timestamp as timestamp, i.faction_id as faction_id, n.name as faction_name, i.influence as influence, i.state_id as state_id from faction_influence i, faction_systems s, faction_names n, (select eddb_id, max(timestamp) as timestamp from faction_influence group by eddb_id) j where i.faction_id=n.faction_id and i.eddb_id=s.eddb_id and i.timestamp=j.timestamp and i.eddb_id in (select distinct eddb_id from faction_influence where faction_id=?)' + params = [id] + if system is not None: + sql += ' and s.name=?' + params.append(system) + sql += ' order by system_name, influence desc' + cursor = self.query(sql, params) + last_system = '' + lines = [] + for row in cursor.fetchall(): + if row['system_name'] != last_system: + if len(lines): + await bot.say_many(destination, lines) + lines = [] + lines.append('Influence in **{}** at {}:'.format(row['system_name'], bot.iso8601(row['timestamp']))) + lines.append('') + last_system = row['system_name'] + line = '{} {:.2f}% ({})'.format(row['faction_name'], row['influence'] * 100.0, eddb.state_name(row['state_id'])) + if row['faction_id'] == id: + line = '**{}**'.format(line) + lines.append(line) + if len(lines): + await bot.say_many(destination, lines) + ret = True + else: + if yelp_if_no_data: + await cat.yelp(destination) + ret = False + self.close_db() + return ret + + async def show_influence_report(self, message, args): + name = args[0] + if len(args) > 1: + system = args[1] + else: + system = None + result = await self.report_faction_influence(message.channel, name = name, system = system) + return result -- 2.7.4