src/dirbs/reports/csv_reports.py
def write_non_active_pairs(conn: callable, logger: callable, report_dir: str, last_seen_date: str) -> callable:
    """Helper method to write non active paris over the network for a specific period.

    Arguments:
        conn: DIRBS PostgreSQL connection object
        logger: DIRBS logger object
        report_dir: reporting directory
        last_seen_date: date on which pair is last_seen
    Returns:
        report metadata
    """
    logger.info('Generating Non-Active Pairs report...')
    with open(os.path.join(report_dir, 'non_active_pairs_{0}.csv'.format(last_seen_date)),
              'w', encoding='utf-8') as pairs_file, conn.cursor() as cursor:
        csv_writer = csv.DictWriter(pairs_file,
                                    fieldnames=['imei_norm', 'imsi', 'last_seen'],
                                    extrasaction='ignore')
        csv_writer.writeheader()
        cursor.execute(sql.SQL("""SELECT pl.imei_norm, pl.imsi, mnt.last_seen
                                    FROM pairing_list AS pl
                              INNER JOIN monthly_network_triplets_country AS mnt
                                         ON pl.imei_norm = mnt.imei_norm
                                     AND pl.imsi = mnt.imsi
                                   WHERE mnt.last_seen < %s"""), [last_seen_date])
        num_written_records = 0
        for row_data in cursor:
            csv_writer.writerow(row_data._asdict())
            num_written_records += 1

        cursor.execute('SELECT COUNT(*) FROM pairing_list')
        total_records = cursor.fetchone()[0]
        logger.info('total_records: {0}, written_records: {1}'.format(total_records, num_written_records))

    return _gen_metadata_for_reports(['non_active_pairs_{0}.csv'.format(last_seen_date)], report_dir)
Helper method to write non active paris over the network for a specific period.

    Arguments:
        conn: DIRBS PostgreSQL connection object
        logger: DIRBS logger object
        report_dir: reporting directory
        last_seen_date: date on which pair is last_seen
    Returns:
        report metadata
    
Helper method to write non active paris over the network for a specific period.
Helper method to write non active paris over the network for a specific period .
def write_non_active_pairs(conn: callable, logger: callable, report_dir: str, last_seen_date: str) -> callable:
    logger.info('Generating Non-Active Pairs report...')
    with open(os.path.join(report_dir, 'non_active_pairs_{0}.csv'.format(last_seen_date)),
              'w', encoding='utf-8') as pairs_file, conn.cursor() as cursor:
        csv_writer = csv.DictWriter(pairs_file,
                                    fieldnames=['imei_norm', 'imsi', 'last_seen'],
                                    extrasaction='ignore')
        csv_writer.writeheader()
        cursor.execute(sql.SQL("""SELECT pl.imei_norm, pl.imsi, mnt.last_seen
                                    FROM pairing_list AS pl
                              INNER JOIN monthly_network_triplets_country AS mnt
                                         ON pl.imei_norm = mnt.imei_norm
                                     AND pl.imsi = mnt.imsi
                                   WHERE mnt.last_seen < %s"""), [last_seen_date])
        num_written_records = 0
        for row_data in cursor:
            csv_writer.writerow(row_data._asdict())
            num_written_records += 1
        cursor.execute('SELECT COUNT(*) FROM pairing_list')
        total_records = cursor.fetchone()[0]
        logger.info('total_records: {0}, written_records: {1}'.format(total_records, num_written_records))
    return _gen_metadata_for_reports(['non_active_pairs_{0}.csv'.format(last_seen_date)], report_dir)
def write_non_active_pairs ( conn : callable , logger : callable , report_dir : str , last_seen_date : str ) -> callable : logger . info ( 'Generating Non-Active Pairs report...' ) with open ( os . path . join ( report_dir , 'non_active_pairs_{0}.csv' . format ( last_seen_date ) ) , 'w' , encoding = 'utf-8' ) as pairs_file , conn . cursor ( ) as cursor : csv_writer = csv . DictWriter ( pairs_file , fieldnames = [ 'imei_norm' , 'imsi' , 'last_seen' ] , extrasaction = 'ignore' ) csv_writer . writeheader ( ) cursor . execute ( sql . SQL ( """SELECT pl.imei_norm, pl.imsi, mnt.last_seen
                                    FROM pairing_list AS pl
                              INNER JOIN monthly_network_triplets_country AS mnt
                                         ON pl.imei_norm = mnt.imei_norm
                                     AND pl.imsi = mnt.imsi
                                   WHERE mnt.last_seen < %s""" ) , [ last_seen_date ] ) num_written_records = 0 for row_data in cursor : csv_writer . writerow ( row_data . _asdict ( ) ) num_written_records += 1 cursor . execute ( 'SELECT COUNT(*) FROM pairing_list' ) total_records = cursor . fetchone ( ) [ 0 ] logger . info ( 'total_records: {0}, written_records: {1}' . format ( total_records , num_written_records ) ) return _gen_metadata_for_reports ( [ 'non_active_pairs_{0}.csv' . format ( last_seen_date ) ] , report_dir )
Helper method to write non active paris over the network for a specific period.
Helper method to write non active paris over the network for a specific period .
"""Helper method to write non active paris over the network for a specific period.

    Arguments:
        conn: DIRBS PostgreSQL connection object
        logger: DIRBS logger object
        report_dir: reporting directory
        last_seen_date: date on which pair is last_seen
    Returns:
        report metadata
    """
None
conn
callable
DIRBS PostgreSQL connection object
DIRBS PostgreSQL connection object
logger
callable
DIRBS logger object
DIRBS logger object
report_dir
str
None
last_seen_date
str
date on which pair is last_seen
date on which pair is last_seen