src/dirbs/reports/csv_reports.py
def write_stolen_violations(config: callable, logger: callable, report_dir: str, conn: callable,
                            filter_by_conditions: list, newer_than: str) -> callable:
    """Helper method to write per operator stolen list violation reports.

    Arguments:
        config: DIRBS config object
        logger: DIRBS logger object
        report_dir: path to report directory
        conn: DIRBS PostgreSQL connection object
        filter_by_conditions: list of condition to filter
        newer_than: violation newer then this date
    Return:
        Report metadata
    """
    logger.info('Generating per-MNO stolen list violations reports...')
    with contextlib.ExitStack() as stack:
        # Push files into exit stack so that they will all be closed.
        operator_ids = [o.id for o in config.region_config.operators]
        filename_op_map = {'stolen_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='utf8'))
                           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 csvwriter
        for _, writer in opname_csvwriter_map.items():
            writer.writerow(['imei_norm', 'last_seen', 'reporting_date'])

        # Run a query to find all the classified IMEIs seen on multiple operators
        blacklist_violations_grace_period_days = config.report_config.blacklist_violations_grace_period_days
        with conn.cursor() as cursor:
            query = sql.SQL("""SELECT imei_norm, last_seen, reporting_date, operator_id
                                 FROM (SELECT imei_norm, MIN(reporting_date) AS reporting_date
                                         FROM stolen_list
                                     GROUP BY imei_norm) AS stolen_imeis
                                 JOIN LATERAL (
                                       SELECT imei_norm, operator_id, MAX(last_seen) AS last_seen
                                         FROM monthly_network_triplets_per_mno_no_null_imeis nt
                                        WHERE imei_norm = stolen_imeis.imei_norm
                                          AND virt_imei_shard = calc_virt_imei_shard(stolen_imeis.imei_norm)
                                     GROUP BY imei_norm, operator_id) network_imeis
                                USING (imei_norm)
                                WHERE network_imeis.last_seen > stolen_imeis.reporting_date + %s
                                      {0}
                                      {1}""")

            if filter_by_conditions:
                cond_filter_query = """AND EXISTS(SELECT 1
                                                    FROM classification_state
                                                   WHERE imei_norm = stolen_imeis.imei_norm
                                                     AND virt_imei_shard =
                                                            calc_virt_imei_shard(stolen_imeis.imei_norm)
                                                     AND cond_name IN %s
                                                     AND end_date IS NULL)"""  # noqa: Q449
                sql_bytes = cursor.mogrify(cond_filter_query, [tuple([c.label for c in filter_by_conditions])])
                conditions_filter_sql = sql.SQL(str(sql_bytes, conn.encoding))
            else:
                conditions_filter_sql = sql.SQL('')

            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(conditions_filter_sql, date_filter_sql),
                           [blacklist_violations_grace_period_days])
            for res in cursor:
                opname_csvwriter_map[res.operator_id].writerow([res.imei_norm, res.last_seen.strftime('%Y%m%d'),
                                                                res.reporting_date.strftime('%Y%m%d')])

    return _gen_metadata_for_reports(list(filename_op_map.keys()), report_dir)
Helper method to write per operator stolen list violation reports.

    Arguments:
        config: DIRBS config object
        logger: DIRBS logger object
        report_dir: path to report directory
        conn: DIRBS PostgreSQL connection object
        filter_by_conditions: list of condition to filter
        newer_than: violation newer then this date
    Return:
        Report metadata
    
Helper method to write per operator stolen list violation reports.
Helper method to write per operator stolen list violation reports .
def write_stolen_violations(config: callable, logger: callable, report_dir: str, conn: callable,
                            filter_by_conditions: list, newer_than: str) -> callable:
    logger.info('Generating per-MNO stolen list violations reports...')
    with contextlib.ExitStack() as stack:
        operator_ids = [o.id for o in config.region_config.operators]
        filename_op_map = {'stolen_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='utf8'))
                           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_norm', 'last_seen', 'reporting_date'])
        blacklist_violations_grace_period_days = config.report_config.blacklist_violations_grace_period_days
        with conn.cursor() as cursor:
            query = sql.SQL("""SELECT imei_norm, last_seen, reporting_date, operator_id
                                 FROM (SELECT imei_norm, MIN(reporting_date) AS reporting_date
                                         FROM stolen_list
                                     GROUP BY imei_norm) AS stolen_imeis
                                 JOIN LATERAL (
                                       SELECT imei_norm, operator_id, MAX(last_seen) AS last_seen
                                         FROM monthly_network_triplets_per_mno_no_null_imeis nt
                                        WHERE imei_norm = stolen_imeis.imei_norm
                                          AND virt_imei_shard = calc_virt_imei_shard(stolen_imeis.imei_norm)
                                     GROUP BY imei_norm, operator_id) network_imeis
                                USING (imei_norm)
                                WHERE network_imeis.last_seen > stolen_imeis.reporting_date + %s
                                      {0}
                                      {1}""")
            if filter_by_conditions:
                cond_filter_query = """AND EXISTS(SELECT 1
                                                    FROM classification_state
                                                   WHERE imei_norm = stolen_imeis.imei_norm
                                                     AND virt_imei_shard =
                                                            calc_virt_imei_shard(stolen_imeis.imei_norm)
                                                     AND cond_name IN %s
                                                     AND end_date IS NULL)"""  
                sql_bytes = cursor.mogrify(cond_filter_query, [tuple([c.label for c in filter_by_conditions])])
                conditions_filter_sql = sql.SQL(str(sql_bytes, conn.encoding))
            else:
                conditions_filter_sql = sql.SQL('')
            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(conditions_filter_sql, date_filter_sql),
                           [blacklist_violations_grace_period_days])
            for res in cursor:
                opname_csvwriter_map[res.operator_id].writerow([res.imei_norm, res.last_seen.strftime('%Y%m%d'),
                                                                res.reporting_date.strftime('%Y%m%d')])
    return _gen_metadata_for_reports(list(filename_op_map.keys()), report_dir)
def write_stolen_violations ( config : callable , logger : callable , report_dir : str , conn : callable , filter_by_conditions : list , newer_than : str ) -> callable : logger . info ( 'Generating per-MNO stolen list violations reports...' ) with contextlib . ExitStack ( ) as stack : operator_ids = [ o . id for o in config . region_config . operators ] filename_op_map = { 'stolen_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 = 'utf8' ) ) 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_norm' , 'last_seen' , 'reporting_date' ] ) blacklist_violations_grace_period_days = config . report_config . blacklist_violations_grace_period_days with conn . cursor ( ) as cursor : query = sql . SQL ( """SELECT imei_norm, last_seen, reporting_date, operator_id
                                 FROM (SELECT imei_norm, MIN(reporting_date) AS reporting_date
                                         FROM stolen_list
                                     GROUP BY imei_norm) AS stolen_imeis
                                 JOIN LATERAL (
                                       SELECT imei_norm, operator_id, MAX(last_seen) AS last_seen
                                         FROM monthly_network_triplets_per_mno_no_null_imeis nt
                                        WHERE imei_norm = stolen_imeis.imei_norm
                                          AND virt_imei_shard = calc_virt_imei_shard(stolen_imeis.imei_norm)
                                     GROUP BY imei_norm, operator_id) network_imeis
                                USING (imei_norm)
                                WHERE network_imeis.last_seen > stolen_imeis.reporting_date + %s
                                      {0}
                                      {1}""" ) if filter_by_conditions : cond_filter_query = """AND EXISTS(SELECT 1
                                                    FROM classification_state
                                                   WHERE imei_norm = stolen_imeis.imei_norm
                                                     AND virt_imei_shard =
                                                            calc_virt_imei_shard(stolen_imeis.imei_norm)
                                                     AND cond_name IN %s
                                                     AND end_date IS NULL)""" sql_bytes = cursor . mogrify ( cond_filter_query , [ tuple ( [ c . label for c in filter_by_conditions ] ) ] ) conditions_filter_sql = sql . SQL ( str ( sql_bytes , conn . encoding ) ) else : conditions_filter_sql = sql . SQL ( '' ) 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 ( conditions_filter_sql , date_filter_sql ) , [ blacklist_violations_grace_period_days ] ) for res in cursor : opname_csvwriter_map [ res . operator_id ] . writerow ( [ res . imei_norm , res . last_seen . strftime ( '%Y%m%d' ) , res . reporting_date . strftime ( '%Y%m%d' ) ] ) return _gen_metadata_for_reports ( list ( filename_op_map . keys ( ) ) , report_dir )
Helper method to write per operator stolen list violation reports.
Helper method to write per operator stolen list violation reports .
"""Helper method to write per operator stolen list violation reports.

    Arguments:
        config: DIRBS config object
        logger: DIRBS logger object
        report_dir: path to report directory
        conn: DIRBS PostgreSQL connection object
        filter_by_conditions: list of condition to filter
        newer_than: violation newer then this date
    Return:
        Report metadata
    """ # Push files into exit stack so that they will all be closed. # Create a map from operator name to csv writer # Write the header to each csvwriter # Run a query to find all the classified IMEIs seen on multiple operators # noqa: Q449
config
callable
DIRBS config object
DIRBS config object
logger
callable
DIRBS logger object
DIRBS logger object
report_dir
str
path to report directory
path to report directory
conn
callable
DIRBS PostgreSQL connection object
DIRBS PostgreSQL connection object
filter_by_conditions
list
list of condition to filter
list of condition to filter
newer_than
str
violation newer then this date
violation newer then this date