src/dirbs/reports/csv_reports.py
def write_un_registered_subscribers(logger: callable, config: callable, report_dir: str, conn: callable,
                                    newer_than: str) -> callable:
    """Helper method to write per operator unregistered subscribers reports.

    Arguments:
        logger: DIRBS logger object
        config: DIRBS config object
        report_dir: reporting dir path
        conn: DIRBS PostgreSQL connection object
        newer_than: newer than this date
    Returns:
        metadata
    """
    logger.info('Generating per-MNO unregistered subscribers list...')
    with contextlib.ExitStack() as stack:
        # push files to the exit stack so that they will all be closed properly.
        operator_ids = [o.id for o in config.region_config.operators]
        filename_op_map = {'unregistered_subscribers_{0}.csv'.format(o): o for o in operator_ids}
        opname_file_map = {o: stack.enter_context(open(os.path.join(report_dir, fn), 'w', encoding='utf-8'))
                           for fn, o in filename_op_map.items()}

        # create a map from operator name to csv writer
        opname_csvwriter_map = {o: csv.writer(opname_file_map[o]) for o in operator_ids}

        # write the header to each file
        for _, writer in opname_csvwriter_map.items():
            writer.writerow(['imsi', 'first_seen', 'last_seen'])

        # query to find all the unregistered imsis across the operators
        with conn.cursor() as cursor:
            query = sql.SQL("""SELECT imsi, first_seen, last_seen, operator_id
                                 FROM monthly_network_triplets_per_mno_no_null_imeis AS mno
                                WHERE NOT EXISTS (SELECT 1
                                                    FROM subscribers_registration_list
                                                   WHERE imsi = mno.imsi) {0}""")

            if newer_than:
                newer_than_query = 'AND last_seen > %s'
                sql_bytes = cursor.mogrify(newer_than_query, [newer_than])
                date_filter_sql = sql.SQL(str(sql_bytes, conn.encoding))
            else:
                date_filter_sql = sql.SQL('')

            cursor.execute(query.format(date_filter_sql))
            for res in cursor:
                opname_csvwriter_map[res.operator_id].writerow([res.imsi,
                                                                res.first_seen.strftime('%Y%m%d'),
                                                                res.last_seen.strftime('%Y%m%d')])

    logger.info('per-MNO unregistered subscribers list generated successfully')
    return _gen_metadata_for_reports(list(filename_op_map.keys()), report_dir)
Helper method to write per operator unregistered subscribers reports.

    Arguments:
        logger: DIRBS logger object
        config: DIRBS config object
        report_dir: reporting dir path
        conn: DIRBS PostgreSQL connection object
        newer_than: newer than this date
    Returns:
        metadata
    
Helper method to write per operator unregistered subscribers reports.
Helper method to write per operator unregistered subscribers reports .
def write_un_registered_subscribers(logger: callable, config: callable, report_dir: str, conn: callable,
                                    newer_than: str) -> callable:
    logger.info('Generating per-MNO unregistered subscribers list...')
    with contextlib.ExitStack() as stack:
        operator_ids = [o.id for o in config.region_config.operators]
        filename_op_map = {'unregistered_subscribers_{0}.csv'.format(o): o for o in operator_ids}
        opname_file_map = {o: stack.enter_context(open(os.path.join(report_dir, fn), 'w', encoding='utf-8'))
                           for fn, o in filename_op_map.items()}
        opname_csvwriter_map = {o: csv.writer(opname_file_map[o]) for o in operator_ids}
        for _, writer in opname_csvwriter_map.items():
            writer.writerow(['imsi', 'first_seen', 'last_seen'])
        with conn.cursor() as cursor:
            query = sql.SQL("""SELECT imsi, first_seen, last_seen, operator_id
                                 FROM monthly_network_triplets_per_mno_no_null_imeis AS mno
                                WHERE NOT EXISTS (SELECT 1
                                                    FROM subscribers_registration_list
                                                   WHERE imsi = mno.imsi) {0}""")
            if newer_than:
                newer_than_query = 'AND last_seen > %s'
                sql_bytes = cursor.mogrify(newer_than_query, [newer_than])
                date_filter_sql = sql.SQL(str(sql_bytes, conn.encoding))
            else:
                date_filter_sql = sql.SQL('')
            cursor.execute(query.format(date_filter_sql))
            for res in cursor:
                opname_csvwriter_map[res.operator_id].writerow([res.imsi,
                                                                res.first_seen.strftime('%Y%m%d'),
                                                                res.last_seen.strftime('%Y%m%d')])
    logger.info('per-MNO unregistered subscribers list generated successfully')
    return _gen_metadata_for_reports(list(filename_op_map.keys()), report_dir)
def write_un_registered_subscribers ( logger : callable , config : callable , report_dir : str , conn : callable , newer_than : str ) -> callable : logger . info ( 'Generating per-MNO unregistered subscribers list...' ) with contextlib . ExitStack ( ) as stack : operator_ids = [ o . id for o in config . region_config . operators ] filename_op_map = { 'unregistered_subscribers_{0}.csv' . format ( o ) : o for o in operator_ids } opname_file_map = { o : stack . enter_context ( open ( os . path . join ( report_dir , fn ) , 'w' , encoding = 'utf-8' ) ) for fn , o in filename_op_map . items ( ) } opname_csvwriter_map = { o : csv . writer ( opname_file_map [ o ] ) for o in operator_ids } for _ , writer in opname_csvwriter_map . items ( ) : writer . writerow ( [ 'imsi' , 'first_seen' , 'last_seen' ] ) with conn . cursor ( ) as cursor : query = sql . SQL ( """SELECT imsi, first_seen, last_seen, operator_id
                                 FROM monthly_network_triplets_per_mno_no_null_imeis AS mno
                                WHERE NOT EXISTS (SELECT 1
                                                    FROM subscribers_registration_list
                                                   WHERE imsi = mno.imsi) {0}""" ) if newer_than : newer_than_query = 'AND last_seen > %s' sql_bytes = cursor . mogrify ( newer_than_query , [ newer_than ] ) date_filter_sql = sql . SQL ( str ( sql_bytes , conn . encoding ) ) else : date_filter_sql = sql . SQL ( '' ) cursor . execute ( query . format ( date_filter_sql ) ) for res in cursor : opname_csvwriter_map [ res . operator_id ] . writerow ( [ res . imsi , res . first_seen . strftime ( '%Y%m%d' ) , res . last_seen . strftime ( '%Y%m%d' ) ] ) logger . info ( 'per-MNO unregistered subscribers list generated successfully' ) return _gen_metadata_for_reports ( list ( filename_op_map . keys ( ) ) , report_dir )
Helper method to write per operator unregistered subscribers reports.
Helper method to write per operator unregistered subscribers reports .
"""Helper method to write per operator unregistered subscribers reports.

    Arguments:
        logger: DIRBS logger object
        config: DIRBS config object
        report_dir: reporting dir path
        conn: DIRBS PostgreSQL connection object
        newer_than: newer than this date
    Returns:
        metadata
    """ # push files to the exit stack so that they will all be closed properly. # create a map from operator name to csv writer # write the header to each file # query to find all the unregistered imsis across the operators
None
logger
callable
DIRBS logger object
DIRBS logger object
config
callable
DIRBS config object
DIRBS config object
report_dir
str
reporting dir path
reporting dir path
conn
callable
DIRBS PostgreSQL connection object
DIRBS PostgreSQL connection object
newer_than
str
newer than this date
newer than this date