qrm2/exts/qrz.py

183 lines
7.3 KiB
Python
Raw Normal View History

"""
QRZ extension for qrm
---
Copyright (C) 2019 Abigail Gold, 0x5c
This file is part of discord-qrm2 and is released under the terms of the GNU
General Public License, version 2.
"""
from collections import OrderedDict
from io import BytesIO
2019-10-18 11:13:24 -04:00
from discord.ext import commands, tasks
import aiohttp
from lxml import etree
import common as cmn
2019-12-07 17:13:06 -05:00
import data.keys as keys
class QRZCog(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
self.session = bot.qrm.session
2019-10-18 11:13:24 -04:00
self._qrz_session_init.start()
@commands.command(name="call", aliases=["qrz"], category=cmn.cat.lookup)
async def _qrz_lookup(self, ctx: commands.Context, callsign: str, *flags):
'''Look up a callsign on [QRZ.com](https://www.qrz.com/). Add `--link` to only link the QRZ page.'''
flags = [f.lower() for f in flags]
if keys.qrz_user == '' or keys.qrz_pass == '' or '--link' in flags:
await ctx.send(f'http://qrz.com/db/{callsign}')
return
2019-10-16 17:23:56 -04:00
try:
2019-10-18 15:27:35 -04:00
await qrz_test_session(self.key, self.session)
2019-10-16 17:23:56 -04:00
except ConnectionError:
2019-10-18 11:13:24 -04:00
await self.get_session()
2019-10-16 17:23:56 -04:00
url = f'http://xmldata.qrz.com/xml/current/?s={self.key};callsign={callsign}'
2019-10-18 15:27:35 -04:00
async with self.session.get(url) as resp:
if resp.status != 200:
raise ConnectionError(f'Unable to connect to QRZ (HTTP Error {resp.status})')
with BytesIO(await resp.read()) as resp_file:
resp_xml = etree.parse(resp_file).getroot()
resp_xml_session = resp_xml.xpath('/x:QRZDatabase/x:Session', namespaces={'x': 'http://xmldata.qrz.com'})
resp_session = {el.tag.split('}')[1]: el.text for el in resp_xml_session[0].getiterator()}
if 'Error' in resp_session:
2019-10-16 17:45:11 -04:00
if 'Session Timeout' in resp_session['Error']:
2019-10-18 11:13:24 -04:00
await self.get_session()
await self._qrz_lookup(ctx, callsign)
return
if 'Not found' in resp_session['Error']:
embed = cmn.embed_factory(ctx)
embed.title = f"QRZ Data for {callsign.upper()}"
embed.colour = cmn.colours.bad
embed.description = 'No data found!'
await ctx.send(embed=embed)
return
raise ValueError(resp_session['Error'])
resp_xml_data = resp_xml.xpath('/x:QRZDatabase/x:Callsign', namespaces={'x': 'http://xmldata.qrz.com'})
resp_data = {el.tag.split('}')[1]: el.text for el in resp_xml_data[0].getiterator()}
embed = cmn.embed_factory(ctx)
embed.title = f"QRZ Data for {resp_data['call']}"
embed.colour = cmn.colours.good
embed.url = f'http://www.qrz.com/db/{resp_data["call"]}'
if 'image' in resp_data:
2019-11-11 21:20:25 -05:00
embed.set_thumbnail(url=resp_data['image'])
2019-10-17 22:29:53 -04:00
data = qrz_process_info(resp_data)
2019-10-16 17:23:56 -04:00
for title, val in data.items():
if val is not None:
embed.add_field(name=title, value=val, inline=True)
await ctx.send(embed=embed)
2019-10-18 11:13:24 -04:00
async def get_session(self):
2019-10-18 11:26:06 -04:00
"""Session creation and caching."""
self.key = await qrz_login(keys.qrz_user, keys.qrz_pass, self.session)
2019-10-18 11:26:06 -04:00
with open('data/qrz_session', 'w') as qrz_file:
qrz_file.write(self.key)
2019-10-18 11:37:24 -04:00
@tasks.loop(count=1)
2019-10-18 11:26:06 -04:00
async def _qrz_session_init(self):
"""Helper task to allow obtaining a session at cog instantiation."""
2019-10-18 11:13:24 -04:00
try:
with open('data/qrz_session') as qrz_file:
self.key = qrz_file.readline().strip()
2019-10-18 15:27:35 -04:00
await qrz_test_session(self.key, self.session)
2019-10-18 11:26:06 -04:00
except (FileNotFoundError, ConnectionError):
2019-10-18 12:45:26 -04:00
await self.get_session()
2019-10-18 11:13:24 -04:00
2019-10-18 15:27:35 -04:00
async def qrz_login(user: str, passwd: str, session: aiohttp.ClientSession):
url = f'http://xmldata.qrz.com/xml/current/?username={user};password={passwd};agent=discord-qrm2'
2019-10-18 15:27:35 -04:00
async with session.get(url) as resp:
if resp.status != 200:
raise ConnectionError(f'Unable to connect to QRZ (HTTP Error {resp.status})')
with BytesIO(await resp.read()) as resp_file:
resp_xml = etree.parse(resp_file).getroot()
resp_xml_session = resp_xml.xpath('/x:QRZDatabase/x:Session', namespaces={'x': 'http://xmldata.qrz.com'})
resp_session = {el.tag.split('}')[1]: el.text for el in resp_xml_session[0].getiterator()}
if 'Error' in resp_session:
raise ConnectionError(resp_session['Error'])
if resp_session['SubExp'] == 'non-subscriber':
raise ConnectionError('Invalid QRZ Subscription')
return resp_session['Key']
2019-10-18 15:27:35 -04:00
async def qrz_test_session(key: str, session: aiohttp.ClientSession):
2019-10-16 17:23:56 -04:00
url = f'http://xmldata.qrz.com/xml/current/?s={key}'
2019-10-18 15:27:35 -04:00
async with session.get(url) as resp:
if resp.status != 200:
raise ConnectionError(f'Unable to connect to QRZ (HTTP Error {resp.status})')
with BytesIO(await resp.read()) as resp_file:
resp_xml = etree.parse(resp_file).getroot()
2019-10-16 17:23:56 -04:00
resp_xml_session = resp_xml.xpath('/x:QRZDatabase/x:Session', namespaces={'x': 'http://xmldata.qrz.com'})
resp_session = {el.tag.split('}')[1]: el.text for el in resp_xml_session[0].getiterator()}
2019-10-16 17:23:56 -04:00
if 'Error' in resp_session:
raise ConnectionError(resp_session['Error'])
2019-10-17 22:29:53 -04:00
def qrz_process_info(data: dict):
2019-10-16 17:23:56 -04:00
if 'name' in data:
if 'fname' in data:
name = data['fname'] + ' ' + data['name']
else:
name = data['name']
else:
name = None
if 'state' in data:
state = f', {data["state"]}'
else:
state = ''
2019-12-25 02:48:59 -05:00
address = data.get('addr1', '') + '\n' + data.get('addr2', '') + state + ' ' + data.get('zip', '')
address = address.strip()
if address == '':
address = None
2019-10-16 17:23:56 -04:00
if 'eqsl' in data:
eqsl = 'Yes' if data['eqsl'] == 1 else 'No'
else:
eqsl = 'Unknown'
if 'mqsl' in data:
mqsl = 'Yes' if data['mqsl'] == 1 else 'No'
else:
mqsl = 'Unknown'
if 'lotw' in data:
lotw = 'Yes' if data['lotw'] == 1 else 'No'
else:
lotw = 'Unknown'
return OrderedDict([('Name', name),
('Country', data.get('country', None)),
('Address', address),
('Grid Square', data.get('grid', None)),
('County', data.get('county', None)),
('CQ Zone', data.get('cqzone', None)),
('ITU Zone', data.get('ituzone', None)),
('IOTA Designator', data.get('iota', None)),
('Expires', data.get('expdate', None)),
('Aliases', data.get('aliases', None)),
('Previous Callsign', data.get('p_call', None)),
('License Class', data.get('class', None)),
2020-01-04 17:14:41 -05:00
('Trustee', data.get('trustee', None)),
2019-10-16 17:23:56 -04:00
('eQSL?', eqsl),
('Paper QSL?', mqsl),
('LotW?', lotw),
('QSL Info', data.get('qslmgr', None)),
('CQ Zone', data.get('cqzone', None)),
('ITU Zone', data.get('ituzone', None)),
('IOTA Designator', data.get('iota', None)),
('Born', data.get('born', None))])
2019-10-16 17:23:56 -04:00
def setup(bot):
bot.add_cog(QRZCog(bot))