original_docstring: 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
      
docstring: Helper method to write per operator stolen list violation reports.
docstring_tokens: Helper method to write per operator stolen list violation reports .
code: 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)
code_tokens: 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 )
short_docstring: Helper method to write per operator stolen list violation reports.
short_docstring_tokens: Helper method to write per operator stolen list violation reports .
comment: """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
      """$SEP$# Push files into exit stack so that they will all be closed.$SEP$# Create a map from operator name to csv writer$SEP$# Write the header to each csvwriter$SEP$# Run a query to find all the classified IMEIs seen on multiple operators$SEP$# noqa: Q449