CSS7ID command line search tool

css7id.py

#!/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()