Skip to content

Instantly share code, notes, and snippets.

@0x9900
Last active November 23, 2022 02:20
Show Gist options
  • Save 0x9900/30ed4e83a77f79393b99cee4447c2324 to your computer and use it in GitHub Desktop.
Save 0x9900/30ed4e83a77f79393b99cee4447c2324 to your computer and use it in GitHub Desktop.
CSS7ID command line search tool
#!/usr/bin/env python3
#
# BSD 3-Clause License
#
# Copyright (c) 2022 Fred W6BSD
# All rights reserved.
#
#
import argparse
import json
import logging
import os
import signal
import sqlite3
import sys
import time
from urllib.request import urlretrieve, URLopener
from tqdm import tqdm
RADIOID_URL = 'https://radioid.net/static/users.json'
RADIOID_DB = '/Users/fred/.local/dmrid.db'
EXPIRATION_TIME = 48 # 2 days.
SQL_TABLE = """
CREATE TABLE IF NOT EXISTS radioid
(
id INTEGER,
radio_id INTEGER,
fname TEXT,
name TEXT,
country TEXT,
callsign TEXT,
city TEXT,
surname TEXT,
state TEXT
);
CREATE INDEX IF NOT EXISTS idx_radio_id on radioid (radio_id);
CREATE INDEX IF NOT EXISTS idx_callsign on radioid (callsign);
"""
def sig_handler(_signum, _frame):
LOG.critical('Database creation has been interrupted')
try:
os.unlink(RADIOID_DB)
except FileNotFoundError:
pass
sys.exit(os.EX_IOERR)
def show_vars():
LOG.info('RADIOID_URL: %s', RADIOID_URL)
LOG.info('RADIOID_DB: %s', RADIOID_DB)
LOG.info('EXPIRATION_TIME: %s', EXPIRATION_TIME)
class DBError(FileNotFoundError):
pass
class RadioID:
def __init__(self, db_name=RADIOID_DB, url=RADIOID_URL):
self.url = url
self.db_name = db_name
self.conn = None
def connect(self):
if not os.path.exists(self.db_name):
raise DBError('Database not found')
self.conn = self._connect()
def download(self, force):
now = time.time()
try:
st_db = os.stat(self.db_name)
if force or st_db.st_mtime < now - (3600 * EXPIRATION_TIME):
os.unlink(self.db_name)
raise FileNotFoundError
LOG.warning('Database update canceled: the database is less than %d hours old.',
EXPIRATION_TIME)
return
except FileNotFoundError:
pass
LOG.info('Download the CCS7ID database')
URLopener.version = 'css7id query cache/1.1'
filename, _ = urlretrieve(self.url)
with open(filename, 'rb') as ufd:
json_content = json.load(ufd)
os.unlink(filename)
insert = "INSERT INTO radioid VALUES (?,?,?,?,?,?,?,?,?)"
LOG.info('Start rebuilding the database')
LOG.debug('%d users to import', len(json_content['users']))
if self.conn:
self.conn.close()
self.conn = None
start_time = time.time()
sig_default = signal.signal(signal.SIGINT, sig_handler)
with self._connect() as conn:
curs = conn.cursor()
curs.execute('BEGIN TRANSACTION')
for user in tqdm(json_content['users']):
curs.execute(insert, (user['id'], user['radio_id'], user['fname'], user['name'],
user['country'], user['callsign'], user['city'],
user['surname'], user['state']))
curs.execute('COMMIT')
conn.close()
signal.signal(signal.SIGINT, sig_default)
LOG.info('End database creation (%.2f seconds)', time.time() - start_time)
def lookup_id(self, radioid):
with self.conn:
cursor = self.conn.cursor()
reply = cursor.execute('SELECT * from radioid where radio_id=?', (radioid,)).fetchone()
reply = dict(reply) if reply else None
LOG.debug(reply)
return reply
def lookup_call(self, call):
with self.conn:
cursor = self.conn.cursor()
reply = cursor.execute('SELECT * from radioid where callsign=?', (call,)).fetchone()
reply = dict(reply) if reply else None
LOG.debug(reply)
return reply
def _connect(self):
try:
conn = sqlite3.connect(self.db_name, isolation_level=None)
conn.row_factory = sqlite3.Row
LOG.info("Using the %s database", self.db_name)
except sqlite3.OperationalError as err:
LOG.error("Database: %s - %s", self.db_name, err)
sys.exit(os.EX_IOERR)
with conn:
curs = conn.cursor()
curs.executescript(SQL_TABLE)
return conn
def main():
parser = argparse.ArgumentParser(description="CSS7ID database lookup")
parser.add_argument('-V', '--version', action='version', version='%(prog)s 1.0')
update_grp = parser.add_argument_group('update', 'CSS7ID database update')
update_grp.add_argument("-u", "--update", action='store_true', default=False,
help="Update the local CSS7ID database ")
update_grp.add_argument("-f", "--force", action='store_true', default=False,
help="Force CSS7ID database update")
parser.add_argument('-v', '--variables', action='store_true', default=False,
help="Show internal variables")
parser.add_argument('query', nargs='*', help='callsign|radio_id')
opts = parser.parse_args()
if opts.variables:
show_vars()
radioid = RadioID()
if opts.update:
radioid.download(opts.force)
if not opts.query:
if not (opts.update or opts.variables):
parser.error('Empty query')
sys.exit(os.EX_OK)
try:
radioid.connect()
except DBError as err:
LOG.error(err)
LOG.error('Try using the --update argument to download a new version of the database')
sys.exit(os.EX_IOERR)
query = opts.query.pop().upper()
if query.isdigit():
user = radioid.lookup_id(int(query))
else:
user = radioid.lookup_call(query.upper())
if user is None:
LOG.error('Call %s Not found', query)
sys.exit(os.EX_NOUSER)
for key in ("radio_id", "callsign", "name", "surname", "city", "country", "state"):
print(f'{key.upper():>10s}: {user[key]}')
print(f' QRZ: https://qrz.com/db/{user["callsign"]}')
if __name__ == "__main__":
logging.basicConfig(format='%(levelname)s - %(message)s',
datefmt='%H:%M:%S', level=logging.INFO)
LOG = logging.getLogger('css7id')
LOG.setLevel(os.getenv('LOG_LEVEL', 'INFO').upper())
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment