src/dirbs/reports/csv_reports.py
def write_association_list_violations(logger: callable, config: callable, report_dir: str,
                                      conn: callable, month: int, year: int) -> callable:
    """Helper method to write association list violations reports.

     Arguments:
        logger: DIRBS logger object
        config: DIRBS config object
        report_dir: reporting directory
        conn: DIRBS PostgreSQL connection object
        month: reporting month
        year: reporting year
    Returns:
        Report Metadata
    """
    logger.info('Generating per-MNO association list violations...')
    with contextlib.ExitStack() as stack:
        operator_ids = [o.id for o in config.region_config.operators]
        filename_op_map = {'association_violations_{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(['imei', 'imsi', 'msisdn', 'first_seen', 'last_seen'])

        with conn.cursor() as cursor:
            cursor.execute("""SELECT imei_norm imei, imsi, msisdn, first_seen, last_seen, operator_id
                                FROM monthly_network_triplets_per_mno_no_null_imeis mno
                               WHERE NOT EXISTS(SELECT 1
                                                  FROM (SELECT imei_norm, imsi
                                                          FROM device_association_list dal
                                                    INNER JOIN subscribers_registration_list srl
                                                               ON dal.uid = srl.uid) association
                                                 WHERE mno.imei_norm = association.imei_norm
                                                   AND mno.imsi = association.imsi)
                                 AND mno.triplet_month = %s
                                 AND mno.triplet_year = %s""", [month, year])  # noqa: Q449
            for res in cursor:
                opname_csvwriter_map[res.operator_id].writerow([res.imei,
                                                                res.imsi,
                                                                res.msisdn,
                                                                res.first_seen,
                                                                res.last_seen])
        logger.info('Per-MNO association list violations list generated successfully.')
    return _gen_metadata_for_reports(list(filename_op_map.keys()), report_dir)
Helper method to write association list violations reports.

     Arguments:
        logger: DIRBS logger object
        config: DIRBS config object
        report_dir: reporting directory
        conn: DIRBS PostgreSQL connection object
        month: reporting month
        year: reporting year
    Returns:
        Report Metadata
    
Helper method to write association list violations reports.
Helper method to write association list violations reports .
def write_association_list_violations(logger: callable, config: callable, report_dir: str,
                                      conn: callable, month: int, year: int) -> callable:
    logger.info('Generating per-MNO association list violations...')
    with contextlib.ExitStack() as stack:
        operator_ids = [o.id for o in config.region_config.operators]
        filename_op_map = {'association_violations_{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(['imei', 'imsi', 'msisdn', 'first_seen', 'last_seen'])
        with conn.cursor() as cursor:
            cursor.execute("""SELECT imei_norm imei, imsi, msisdn, first_seen, last_seen, operator_id
                                FROM monthly_network_triplets_per_mno_no_null_imeis mno
                               WHERE NOT EXISTS(SELECT 1
                                                  FROM (SELECT imei_norm, imsi
                                                          FROM device_association_list dal
                                                    INNER JOIN subscribers_registration_list srl
                                                               ON dal.uid = srl.uid) association
                                                 WHERE mno.imei_norm = association.imei_norm
                                                   AND mno.imsi = association.imsi)
                                 AND mno.triplet_month = %s
                                 AND mno.triplet_year = %s""", [month, year])  
            for res in cursor:
                opname_csvwriter_map[res.operator_id].writerow([res.imei,
                                                                res.imsi,
                                                                res.msisdn,
                                                                res.first_seen,
                                                                res.last_seen])
        logger.info('Per-MNO association list violations list generated successfully.')
    return _gen_metadata_for_reports(list(filename_op_map.keys()), report_dir)
def write_association_list_violations ( logger : callable , config : callable , report_dir : str , conn : callable , month : int , year : int ) -> callable : logger . info ( 'Generating per-MNO association list violations...' ) with contextlib . ExitStack ( ) as stack : operator_ids = [ o . id for o in config . region_config . operators ] filename_op_map = { 'association_violations_{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 ( [ 'imei' , 'imsi' , 'msisdn' , 'first_seen' , 'last_seen' ] ) with conn . cursor ( ) as cursor : cursor . execute ( """SELECT imei_norm imei, imsi, msisdn, first_seen, last_seen, operator_id
                                FROM monthly_network_triplets_per_mno_no_null_imeis mno
                               WHERE NOT EXISTS(SELECT 1
                                                  FROM (SELECT imei_norm, imsi
                                                          FROM device_association_list dal
                                                    INNER JOIN subscribers_registration_list srl
                                                               ON dal.uid = srl.uid) association
                                                 WHERE mno.imei_norm = association.imei_norm
                                                   AND mno.imsi = association.imsi)
                                 AND mno.triplet_month = %s
                                 AND mno.triplet_year = %s""" , [ month , year ] ) for res in cursor : opname_csvwriter_map [ res . operator_id ] . writerow ( [ res . imei , res . imsi , res . msisdn , res . first_seen , res . last_seen ] ) logger . info ( 'Per-MNO association list violations list generated successfully.' ) return _gen_metadata_for_reports ( list ( filename_op_map . keys ( ) ) , report_dir )
Helper method to write association list violations reports.
Helper method to write association list violations reports .
"""Helper method to write association list violations reports.

     Arguments:
        logger: DIRBS logger object
        config: DIRBS config object
        report_dir: reporting directory
        conn: DIRBS PostgreSQL connection object
        month: reporting month
        year: reporting year
    Returns:
        Report Metadata
    """ # noqa: Q449
None
logger
callable
config
callable
report_dir
str
conn
callable
month
int
year
int