miaowware/qrm2
1
0
Fork 0
mirror of https://github.com/miaowware/qrm2.git synced 2024-11-16 13:11:47 -05:00
Code Issues Actions Projects Releases Wiki Activity

convert qrz lookup to use pure lxml instead of bs4

Browse Source
This commit is contained in:
Abigail Gold 2019-10-19 22:55:53 -04:00
parent 93b0bae2c1
commit f5ec8f0d8b
No known key found for this signature in database
GPG Key ID: 80A676456AB6B045
1 changed files with 21 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
cogs/qrzcog.py
View File

@ -8,12 +8,13 @@ General Public License, version 2.
""" """
from collections import OrderedDict from collections import OrderedDict
from datetime import datetime from datetime import datetime
from io import BytesIO
import discord import discord
from discord.ext import commands, tasks from discord.ext import commands, tasks
import aiohttp import aiohttp
from bs4 import BeautifulSoup from lxml import etree
class QRZCog(commands.Cog): class QRZCog(commands.Cog):
@ -38,12 +39,14 @@ class QRZCog(commands.Cog):
async with self.session.get(url) as resp: async with self.session.get(url) as resp:
if resp.status != 200: if resp.status != 200:
raise ConnectionError(f'Unable to connect to QRZ (HTTP Error {resp.status})') raise ConnectionError(f'Unable to connect to QRZ (HTTP Error {resp.status})')
resp_xml = await resp.text() resp_xml = etree.parse(BytesIO(await resp.read())).getroot()
xml_soup = BeautifulSoup(resp_xml, "xml") resp_xml_data = resp_xml.xpath('/x:QRZDatabase/x:Callsign',
namespaces={'x':'http://xmldata.qrz.com'})
resp_data = {tag.name: tag.contents[0] for tag in xml_soup.select('QRZDatabase Callsign *')} resp_data = {el.tag.split('}')[1]: el.text for el in resp_xml_data[0].getiterator()}
resp_session = {tag.name: tag.contents[0] for tag in xml_soup.select('QRZDatabase Session *')} 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: if 'Error' in resp_session:
if 'Session Timeout' in resp_session['Error']: if 'Session Timeout' in resp_session['Error']:
await self.get_session() await self.get_session()
@ -87,15 +90,16 @@ async def qrz_login(user: str, passwd: str, session: aiohttp.ClientSession):
async with session.get(url) as resp: async with session.get(url) as resp:
if resp.status != 200: if resp.status != 200:
raise ConnectionError(f'Unable to connect to QRZ (HTTP Error {resp.status})') raise ConnectionError(f'Unable to connect to QRZ (HTTP Error {resp.status})')
resp_xml = await resp.text() resp_xml = etree.parse(BytesIO(await resp.read())).getroot()
xml_soup = BeautifulSoup(resp_xml, "xml") resp_xml_session = resp_xml.xpath('/x:QRZDatabase/x:Session',
resp_data = {tag.name: tag.contents[0] for tag in xml_soup.select('QRZDatabase Session *')} namespaces={'x':'http://xmldata.qrz.com'})
if 'Error' in resp_data: resp_session = {el.tag.split('}')[1]: el.text for el in resp_xml_session[0].getiterator()}
raise ConnectionError(resp_data['Error']) if 'Error' in resp_session:
if resp_data['SubExp'] == 'non-subscriber': raise ConnectionError(resp_session['Error'])
if resp_session['SubExp'] == 'non-subscriber':
raise ConnectionError('Invalid QRZ Subscription') raise ConnectionError('Invalid QRZ Subscription')
return resp_data['Key'] return resp_session['Key']
async def qrz_test_session(key: str, session: aiohttp.ClientSession): async def qrz_test_session(key: str, session: aiohttp.ClientSession):
@ -103,11 +107,11 @@ async def qrz_test_session(key: str, session: aiohttp.ClientSession):
async with session.get(url) as resp: async with session.get(url) as resp:
if resp.status != 200: if resp.status != 200:
raise ConnectionError(f'Unable to connect to QRZ (HTTP Error {resp.status})') raise ConnectionError(f'Unable to connect to QRZ (HTTP Error {resp.status})')
resp_xml = await resp.text() resp_xml = etree.parse(BytesIO(await resp.read())).getroot()
xml_soup = BeautifulSoup(resp_xml, "xml") resp_xml_session = resp_xml.xpath('/x:QRZDatabase/x:Session',
namespaces={'x':'http://xmldata.qrz.com'})
resp_session = {tag.name: tag.contents[0] for tag in xml_soup.select('QRZDatabase Session *')} resp_session = {el.tag.split('}')[1]: el.text for el in resp_xml_session[0].getiterator()}
if 'Error' in resp_session: if 'Error' in resp_session:
raise ConnectionError(resp_session['Error']) raise ConnectionError(resp_session['Error'])