PyPI 官网下载 | pyhamtools-0.5.6.tar.gz

import os import logging import logging.config import re import random, string from datetime import datetime import xml.etree.ElementTree as ET import urllib import json import copy import sys import unicodedata import requests from requests.exceptions import ConnectionError, HTTPError, Timeout from bs4 import BeautifulSoup import pytz import version from consts import LookupConventions as const from exceptions import APIKeyMissingError UTC = pytz.UTC timestamp_now = datetime.utcnow().replace(tzinfo=UTC) if sys.version_info < (2, 7,): class NullHandler(logging.Handler): def emit(self, record): pass class LookupLib(object): """ This class is a wrapper for the following three Amateur Radio databases: 1. Clublog.org (daily updated XML File) 2. Clublog.org (HTTPS lookup) 3. Country-files.com (infrequently updated PLIST File) 4. QRZ.com (HTTP / XML Lookup) It's aim is to provide a homogeneous interface to different databases. Typically an instance of this class is injected as a dependency in the :py:class:`Callinfo` class, but it can also be used directly. Even the interface is the same for all lookup sources, the returning data can be different. The documentation of the various methods provide more detail. By default, LookupLib requires an Internet connection to download the libraries or perform the lookup against the Clublog API or QRZ.com. The entire lookup data (where database files are downloaded) can also be copied into Redis, which an extremely fast in-memory Key/Value store. A LookupLib object can be instanciated to perform then all lookups in Redis, instead processing and loading the data from Internet / File. This saves some time and allows several instances of :py:class:`LookupLib` to query the same data concurrently. Args: lookuptype (str) : "clublogxml" or "clublogapi" or "countryfile" or "redis" or "qrz" apikey (str): Clublog API Key username (str): QRZ.com username pwd (str): QRZ.com password apiv (str, optional): QRZ.com API Version filename (str, optional): Filename for Clublog XML or Country-files.com cty.plist file. When a local file is used, no Internet connection not API Key is necessary. logger (logging.getLogger(__name__), optional): Python logger redis_instance (redis.Redis(), optional): Instance of Redis redis_prefix (str, optional): Prefix to identify the lookup data set in Redis """ def __init__(self, lookuptype = "countryfile", apikey=None, apiv="1.3.3", filename=None, logger=None, username=None, pwd=None, redis_instance=None, redis_prefix=None): self._logger = None if logger: self._logger = logger else: self._logger = logging.getLogger(__name__) if sys.version_info[:2] == (2, 6): self._logger.addHandler(NullHandler()) else: self._logger.addHandler(logging.NullHandler()) self._apikey = apikey self._apiv = apiv self._download = True self._lib_filename = filename self._redis = redis_instance self._redis_prefix = redis_prefix self._username = username self._pwd = pwd if self._lib_filename: self._download = False self._callsign_exceptions_index = {} self._invalid_operations_index = {} self._zone_exceptions_index = {} self._entities = {} self._callsign_exceptions = {} self._invalid_operations = {} self._zone_exceptions = {} self._lookuptype = lookuptype if self._lookuptype == "clublogxml": self._load_clublogXML(apikey=self._apikey, cty_file=self._lib_filename) elif self._lookuptype == "countryfile": self._load_countryfile(cty_file=self._lib_filename) elif self._lookuptype == "clublogapi": pass elif self._lookuptype == "redis": import redis elif self._lookuptype == "qrz": self._apikey = self._get_qrz_session_key(self._username, self._pwd) else: raise AttributeError("Lookup type missing") def _get_qrz_session_key(self, username, pwd): qrz_api_version = "1.3.3" url = "https://xmldata.qrz.com/xml/" + qrz_api_version + "/" agent = "PyHamTools"+version.__version__ params = {"username" : username, "password" : pwd, "agent" : agent } encodeurl = url + "?" + urllib.urlencode(params) response = requests.get(encodeurl, timeout=10) doc = BeautifulSoup(response.text) session_key = None if doc.session.key: session_key = doc.session.key.text else: if doc.session.error: raise ValueError(doc.session.error.text) else: raise ValueError("Could not retrieve Session Key from QRZ.com") return session_key def copy_data_in_redis(self, redis_prefix, redis_instance): """ Copy the complete lookup data into redis. Old data will be overwritten. Args: redis_prefix (str): Prefix to distinguish the data in redis for the different looktypes redis_instance (str): an Instance of Redis Returns: bool: returns True when the data has been copied successfully into Redis Example: Copy the entire lookup data from the Country-files.com PLIST File into Redis. This example requires a running instance of Redis, as well the python Redis connector (pip install redis-py). >>> from pyhamtools import LookupLib >>> import redis >>> r = redis.Redis() >>> my_lookuplib = LookupLib(lookuptype="countryfile") >>> print my_lookuplib.copy_data_in_redis(redis_prefix="CF", redis_instance=r) True Now let's create an instance of LookupLib, using Redis to query the data >>> from pyhamtools import LookupLib >>> import redis >>> r = redis.Redis() >>> my_lookuplib = LookupLib(lookuptype="countryfile", redis_instance=r, redis_prefix="CF") >>> my_lookuplib.lookup_callsign("3D2RI") { u'adif': 460, u'continent': u'OC', u'country': u'Rotuma Island', u'cqz': 32, u'ituz': 56, u'latitude': -12.48, u'longitude': -177.08 } Note: This method is available for the following lookup type - clublogxml - countryfile """ if redis_instance is not None: self._redis = redis_instance if self._redis is None: raise AttributeError("redis_instance is missing") if redis_prefix is None: raise KeyError("redis_prefix is missing") if self._lookuptype == "clublogxml" or self._lookuptype == "countryfile": self._push_dict_to_redis(self._entities, redis_prefix, "_entity_") self._push_dict_index_to_redis(self._callsign_exceptions_index, redis_prefix, "_call_ex_index_") self._push_dict_to_redis(self._callsign_exceptions, redis_prefix, "_call_ex_") self._push_dict_index_to_redis(self._prefixes_index, redis_prefix, "_prefix_index_") self._push_dict_to_redis(self._prefixes, redis_prefix, "_prefix_") self._push_dict_index_to_redis(self._invalid_operations_index, redis_prefix, "_inv_op_index_") self._push_dict_to_redis(self._invalid_operations, redis_prefix, "_inv_op_") self._push_dict_index_to_redis(self._zone_exceptions_index, redis_prefix, "_zone_ex_index_") self._push_dict_to_redis(self._zone_exceptions, redis_prefix, "_zone_ex_") return True def _push_dict_to_redis(self, push_dic