<s> import csv <EOL> import datetime <EOL> import logging <EOL> import os <EOL> import re <EOL> class MissingMaxMindError ( Exception ) : <EOL> def __init__ ( self , db_location , io_error ) : <EOL> Exception . __init__ ( <EOL> self , '<STR_LIT>' % ( <EOL> db_location , io_error ) ) <EOL> class IPTranslationStrategySpec ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , strategy_name , params ) : <EOL> self . strategy_name = strategy_name <EOL> self . params = params <EOL> class IPTranslationStrategyFactory ( object ) : <EOL> def __init__ ( self , file_opener = open ) : <EOL> self . _file_opener = file_opener <EOL> self . _cache = { } <EOL> def create ( self , ip_translation_spec ) : <EOL> if ip_translation_spec in self . _cache : <EOL> return self . _cache [ ip_translation_spec ] <EOL> if ip_translation_spec . strategy_name == '<STR_LIT>' : <EOL> ip_translator = self . _create_maxmind_strategy ( <EOL> ip_translation_spec . params ) <EOL> else : <EOL> ValueError ( '<STR_LIT>' ) <EOL> self . _cache [ ip_translation_spec ] = ip_translator <EOL> return ip_translator <EOL> def _create_maxmind_strategy ( self , maxmind_params ) : <EOL> db_snapshot_strings = maxmind_params [ '<STR_LIT>' ] <EOL> if len ( db_snapshot_strings ) == <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> maxmind_dir = maxmind_params [ '<STR_LIT>' ] <EOL> snapshots = [ ] <EOL> for db_snapshot_string in db_snapshot_strings : <EOL> snapshot_datetime = datetime . datetime . strptime ( db_snapshot_string , <EOL> '<STR_LIT>' ) <EOL> snapshot_path = IPTranslationStrategyMaxMind . get_maxmind_snapshot_path ( <EOL> snapshot_datetime , maxmind_dir ) <EOL> try : <EOL> snapshot_file = self . _file_opener ( snapshot_path , '<STR_LIT:rb>' ) <EOL> snapshot = ( snapshot_datetime , snapshot_file ) <EOL> snapshots . append ( snapshot ) <EOL> except IOError as io_error : <EOL> raise MissingMaxMindError ( snapshot_path , io_error ) <EOL> return IPTranslationStrategyMaxMind ( snapshots ) <EOL> class IPTranslationStrategy ( object ) : <EOL> def find_ip_blocks ( self , asn_search_name ) : <EOL> raise NotImplementedError ( ) <EOL> class IPTranslationStrategyMaxMind ( IPTranslationStrategy ) : <EOL> def __init__ ( self , snapshots ) : <EOL> """<STR_LIT>""" <EOL> self . logger = logging . getLogger ( '<STR_LIT>' ) <EOL> if len ( snapshots ) > <NUM_LIT:1> : <EOL> raise NotImplementedError ( <EOL> '<STR_LIT>' ) <EOL> snapshot_file = snapshots [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> self . _network_map = self . _parse_maxmind_snapshot ( snapshot_file ) <EOL> self . _cache = { } <EOL> def find_ip_blocks ( self , asn_search_name ) : <EOL> """<STR_LIT>""" <EOL> if asn_search_name in self . _cache : <EOL> return self . _cache [ asn_search_name ] <EOL> notification_cache = set ( ) <EOL> blocks_to_return = [ ] <EOL> asn_search_terms = self . _translate_short_name ( asn_search_name ) <EOL> asn_name_re = re . compile ( asn_search_terms , re . IGNORECASE ) <EOL> for block_dict in self . _network_map : <EOL> if asn_name_re . search ( block_dict [ '<STR_LIT>' ] ) is not None : <EOL> if not ( block_dict [ '<STR_LIT>' ] in notification_cache ) : <EOL> self . logger . debug ( ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( <EOL> asn_name = block_dict [ '<STR_LIT>' ] , <EOL> asn_search_name = asn_search_name ) ) <EOL> notification_cache . add ( block_dict [ '<STR_LIT>' ] ) <EOL> block_start = int ( block_dict [ '<STR_LIT>' ] ) <EOL> block_end = int ( block_dict [ '<STR_LIT>' ] ) <EOL> blocks_to_return . append ( ( block_start , block_end ) ) <EOL> self . _cache [ asn_search_name ] = blocks_to_return <EOL> return blocks_to_return <EOL> @ staticmethod <EOL> def get_maxmind_snapshot_path ( snapshot_datetime , maxmind_dir ) : <EOL> """<STR_LIT>""" <EOL> date_string = snapshot_datetime . strftime ( '<STR_LIT>' ) <EOL> snapshot_filename = '<STR_LIT>' % date_string <EOL> return os . path . join ( os . path . dirname ( __file__ ) , maxmind_dir , <EOL> snapshot_filename ) <EOL> def _parse_maxmind_snapshot ( self , snapshot_file ) : <EOL> """<STR_LIT>""" <EOL> block_rows = [ ] <EOL> csvReader = csv . DictReader ( <EOL> snapshot_file , <EOL> fieldnames = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> for block_row in csvReader : <EOL> block_rows . append ( block_row ) <EOL> self . logger . debug ( '<STR_LIT>' , <EOL> len ( block_rows ) ) <EOL> return block_rows <EOL> def _translate_short_name ( self , short_name ) : <EOL> """<STR_LIT>""" <EOL> short_name_map = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> } <EOL> if short_name in short_name_map : <EOL> long_names = short_name_map [ short_name ] <EOL> return self . _regex_xor_names ( long_names ) <EOL> return re . escape ( short_name ) <EOL> def _regex_xor_names ( self , names ) : <EOL> """<STR_LIT>""" <EOL> escaped_names = [ ] <EOL> for name in names : <EOL> escaped_names . append ( re . escape ( name ) ) <EOL> return '<STR_LIT:(>' + '<STR_LIT>' . join ( escaped_names ) + '<STR_LIT:)>' </s>
