mirror of https://github.com/miaowware/qrm2.git
(NOT FUNCTIONAL) progress on qrz cmd
This commit is contained in:
Abigail Gold
parent
10c43e83bd
commit
907065e718
145
cogs/qrzcog.py
145
cogs/qrzcog.py
|
|
@ -7,11 +7,11 @@ This file is part of discord-qrmbot and is released under the terms of the GNU
|
|||
General Public License, version 2.
|
||||
"""
|
||||
from collections import OrderedDict
|
||||
from datetime import datetime
|
||||
|
||||
import discord
|
||||
import discord.ext.commands as commands
|
||||
|
||||
from datetime import datetime
|
||||
import aiohttp
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
|
|
@ -20,6 +20,18 @@ class QRZCog(commands.Cog):
|
|||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
self.gs = bot.get_cog("GlobalSettings")
|
||||
try:
|
||||
with open('data/qrz_session') as qrz_file:
|
||||
self.key = qrz_file.readline().strip()
|
||||
qrz_test_session(self.key)
|
||||
except FileNotFoundError:
|
||||
self.key = qrz_login(self.gs.keys.qrz_user, self.gs.keys.qrz_pass)
|
||||
with open('data/qrz_session', 'w') as qrz_file:
|
||||
qrz_file.write(self.key)
|
||||
except ConnectionError:
|
||||
self.key = qrz_login(self.gs.keys.qrz_user, self.gs.keys.qrz_pass)
|
||||
with open('data/qrz_session', 'w') as qrz_file:
|
||||
qrz_file.write(self.key)
|
||||
|
||||
@commands.command(name="qrz", aliases=["call"])
|
||||
async def _qrz_lookup(self, ctx: commands.Context, call: str):
|
||||
|
|
@ -27,13 +39,15 @@ class QRZCog(commands.Cog):
|
|||
if self.gs.keys.qrz_user == '' or self.gs.keys.qrz_pass == '':
|
||||
await ctx.send(f'http://qrz.com/db/{call}')
|
||||
return
|
||||
|
||||
try:
|
||||
# TODO: see if there's a key first (i.e. don't log in every time)
|
||||
# TODO: maybe make it a task to generate a key?
|
||||
key = await _qrz_login(self.gs.keys.qrz_user, self.gs.keys.qrz_pass)
|
||||
except ConnectionError as err:
|
||||
print(err)
|
||||
url = f'http://xmldata.qrz.com/xml/current/?s={key};callsign={call}'
|
||||
await qrz_test_session(self.key)
|
||||
except ConnectionError:
|
||||
self.key = await qrz_login(self.gs.keys.qrz_user, self.gs.keys.qrz_pass)
|
||||
with open('data/qrz_session', 'w') as qrz_file:
|
||||
qrz_file.write(self.key)
|
||||
|
||||
url = f'http://xmldata.qrz.com/xml/current/?s={self.key};callsign={call}'
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
if resp.status != 200:
|
||||
|
|
@ -41,6 +55,8 @@ class QRZCog(commands.Cog):
|
|||
resp_xml = await resp.text()
|
||||
|
||||
xml_soup = BeautifulSoup(resp_xml, "xml")
|
||||
print(xml_soup)
|
||||
|
||||
resp_data = {tag.name: tag.contents[0] for tag in xml_soup.select('QRZDatabase Callsign *')}
|
||||
resp_session = {tag.name: tag.contents[0] for tag in xml_soup.select('QRZDatabase Session *')}
|
||||
if 'Error' in resp_session:
|
||||
|
|
@ -55,60 +71,15 @@ class QRZCog(commands.Cog):
|
|||
if 'image' in resp_data:
|
||||
embed.set_image(url=resp_data['image'])
|
||||
|
||||
if 'name' in resp_data:
|
||||
if 'fname' in resp_data:
|
||||
name = resp_data['fname'] + ' ' + resp_data['name']
|
||||
else:
|
||||
name = resp_data['name']
|
||||
else:
|
||||
name = None
|
||||
if 'state' in resp_data:
|
||||
state = f', {resp_data["state"]}'
|
||||
else:
|
||||
state = ''
|
||||
address = resp_data.get('addr1', '') + '\n' + resp_data.get('addr2', '') + \
|
||||
state + ' ' + resp_data.get('zip', '')
|
||||
if 'eqsl' in resp_data:
|
||||
eqsl = 'Yes' if resp_data['eqsl'] == 1 else 'No'
|
||||
else:
|
||||
eqsl = 'Unknown'
|
||||
if 'mqsl' in resp_data:
|
||||
mqsl = 'Yes' if resp_data['mqsl'] == 1 else 'No'
|
||||
else:
|
||||
mqsl = 'Unknown'
|
||||
if 'lotw' in resp_data:
|
||||
lotw = 'Yes' if resp_data['lotw'] == 1 else 'No'
|
||||
else:
|
||||
lotw = 'Unknown'
|
||||
data = await qrz_process_info(resp_data)
|
||||
|
||||
data = OrderedDict([('Name', name),
|
||||
('Country', resp_data.get('country', None)),
|
||||
('Address', address),
|
||||
('Grid Square', resp_data.get('grid', None)),
|
||||
('County', resp_data.get('county', None)),
|
||||
('CQ Zone', resp_data.get('cqzone', None)),
|
||||
('ITU Zone', resp_data.get('ituzone', None)),
|
||||
('IOTA Designator', resp_data.get('iota', None)),
|
||||
('Expires', resp_data.get('expdate', None)),
|
||||
('Aliases', resp_data.get('aliases', None)),
|
||||
('Previous Callsign', resp_data.get('p_call', None)),
|
||||
('License Class', resp_data.get('class', None)),
|
||||
('eQSL?', eqsl),
|
||||
('Paper QSL?', mqsl),
|
||||
('LotW?', lotw),
|
||||
('QSL Info', resp_data.get('qslmgr', None)),
|
||||
('CQ Zone', resp_data.get('cqzone', None)),
|
||||
('ITU Zone', resp_data.get('ituzone', None)),
|
||||
('IOTA Designator', resp_data.get('iota', None)),
|
||||
('Born', resp_data.get('born', None)),
|
||||
])
|
||||
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)
|
||||
|
||||
|
||||
async def _qrz_login(user: str, passwd: str):
|
||||
async def qrz_login(user: str, passwd: str):
|
||||
url = f'http://xmldata.qrz.com/xml/current/?username={user};password={passwd};agent=qrmbot'
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
|
|
@ -125,5 +96,71 @@ async def _qrz_login(user: str, passwd: str):
|
|||
return resp_data['Key']
|
||||
|
||||
|
||||
async def qrz_test_session(key: str):
|
||||
url = f'http://xmldata.qrz.com/xml/current/?s={key}'
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
if resp.status != 200:
|
||||
raise ConnectionError(f'Unable to connect to QRZ (HTTP Error {resp.status})')
|
||||
resp_xml = await resp.text()
|
||||
|
||||
xml_soup = BeautifulSoup(resp_xml, "xml")
|
||||
print(xml_soup)
|
||||
|
||||
resp_session = {tag.name: tag.contents[0] for tag in xml_soup.select('QRZDatabase Session *')}
|
||||
if 'Error' in resp_session:
|
||||
raise ConnectionError(resp_session['Error'])
|
||||
|
||||
|
||||
async def qrz_process_info(data: dict):
|
||||
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 = ''
|
||||
address = data.get('addr1', '') + '\n' + data.get('addr2', '') + \
|
||||
state + ' ' + data.get('zip', '')
|
||||
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)),
|
||||
('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)),
|
||||
])
|
||||
|
||||
|
||||
def setup(bot):
|
||||
bot.add_cog(QRZCog(bot))
|
||||
|
|
|
|||
Loading…
Reference in New Issue