Common Utilities API

Database

telecom_common.db_connect

db_connect(path=DB_PATH)

Source code in telecom_common.py

def db_connect(path=DB_PATH):
    con = sqlite3.connect(path)
    con.execute("PRAGMA journal_mode=WAL")
    return con

telecom_common.db_df

db_df(sql, params=(), path=DB_PATH)

Source code in telecom_common.py

def db_df(sql, params=(), path=DB_PATH):
    con = db_connect(path)
    try:
        return pd.read_sql(sql, con, params=params)
    finally:
        con.close()

telecom_common.db_exec

db_exec(query, params=(), path=DB_PATH)

Source code in telecom_common.py

def db_exec(query, params=(), path=DB_PATH):
    con = db_connect(path)
    try:
        cur = con.cursor()
        cur.execute(query, params)
        con.commit()
    finally:
        con.close()

telecom_common.db_one

db_one(sql, params=(), path=DB_PATH)

Source code in telecom_common.py

def db_one(sql, params=(), path=DB_PATH):
    con = db_connect(path)
    try:
        cur = con.cursor()
        cur.execute(sql, params)
        return cur.fetchone()
    finally:
        con.close()

telecom_common.db_count

db_count(table, path=DB_PATH)

Source code in telecom_common.py

def db_count(table, path=DB_PATH):
    try:
        r = db_one(f"SELECT COUNT(*) FROM {table}", path=path)
        return r[0] if r else 0
    except Exception:
        return 0

telecom_common.db_chunked_in_update

db_chunked_in_update(table, column, set_clause, values, where_values, chunk_size=800, path=DB_PATH)

Source code in telecom_common.py

def db_chunked_in_update(
    table, column, set_clause, values, where_values, chunk_size=800, path=DB_PATH
):
    total = 0
    con = db_connect(path)
    try:
        cur = con.cursor()
        for i in range(0, len(where_values), chunk_size):
            chunk = where_values[i : i + chunk_size]
            ph = ",".join("?" * len(chunk))
            cur.execute(
                f"UPDATE {table} SET {set_clause} " f"WHERE {column} IN ({ph})",
                tuple(values + chunk),
            )
            total += cur.rowcount
        con.commit()
    finally:
        con.close()
    return total

Password

telecom_common.has_password

has_password()

Source code in telecom_common.py

def has_password():
    return os.path.exists(_AUTH_FILE) and os.path.exists(_SALT_FILE)

telecom_common.set_password

set_password(pw)

Source code in telecom_common.py

def set_password(pw):
    os.makedirs(OUT_DIR, exist_ok=True)
    salt = secrets.token_bytes(16)
    digest = _scrypt(pw, salt)
    with open(_SALT_FILE, "wb") as f:
        f.write(salt)
    with open(_AUTH_FILE, "wb") as f:
        f.write(digest)

telecom_common.verify_password

verify_password(pw)

Source code in telecom_common.py

def verify_password(pw):
    if not has_password():
        return False
    try:
        with open(_SALT_FILE, "rb") as f:
            salt = f.read()
        with open(_AUTH_FILE, "rb") as f:
            expected = f.read()
        return hmac.compare_digest(_scrypt(pw, salt), expected)
    except Exception:
        return False

Identity generators

telecom_common.luhn

luhn(number)

Source code in telecom_common.py

def luhn(number):
    d = [int(x) for x in number]
    for i in range(len(d) - 1, -1, -2):
        d[i] *= 2
        if d[i] > 9:
            d[i] -= 9
    return str((10 - sum(d) % 10) % 10)

telecom_common.gen_imsi

gen_imsi()

Source code in telecom_common.py

def gen_imsi():
    return f"{MCC}{MNC}" + "".join(random.choices("0123456789", k=10))

telecom_common.gen_msisdn

gen_msisdn()

Source code in telecom_common.py

def gen_msisdn():
    return "98" + random.choice(MSISDN_PREFIXES) + "".join(random.choices("0123456789", k=7))

telecom_common.gen_imei

gen_imei()

Source code in telecom_common.py

def gen_imei():
    p = random.choice(IMEI_TACS) + "".join(random.choices("0123456789", k=6))
    return p + luhn(p)

telecom_common.gen_key_id

gen_key_id(seed='')

Source code in telecom_common.py

def gen_key_id(seed=""):
    src = seed or f"{random.random()}{datetime.now()}"
    return "KEY-" + hashlib.sha1(src.encode()).hexdigest()[:16].upper()