<s> import re <EOL> ISO8601_REGEX = re . compile ( r"<STR_LIT>" <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" ) <EOL> URL_REGEX = re . compile ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' , re . IGNORECASE ) <EOL> ACCRUAL_PERIODICITY_VALUES = ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> LANGUAGE_REGEX = re . compile ( "<STR_LIT>" ) <EOL> COMMON_MIMETYPES = ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:application/json>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> MIMETYPE_REGEX = re . compile ( "<STR_LIT>" ) <EOL> import lepl . apps . rfc3696 <EOL> email_validator = lepl . apps . rfc3696 . Email ( ) <EOL> import urllib , csv <EOL> omb_burueau_codes = set ( ) <EOL> for row in csv . DictReader ( urllib . urlopen ( "<STR_LIT>" ) ) : <EOL> omb_burueau_codes . add ( row [ "<STR_LIT>" ] + "<STR_LIT::>" + row [ "<STR_LIT>" ] ) <EOL> def do_validation ( doc , src_url , errors_array ) : <EOL> errs = { } <EOL> if type ( doc ) != list : <EOL> add_error ( errs , <NUM_LIT:0> , "<STR_LIT>" , "<STR_LIT>" ) <EOL> elif len ( doc ) == <NUM_LIT:0> : <EOL> add_error ( errs , <NUM_LIT:0> , "<STR_LIT>" , "<STR_LIT>" ) <EOL> else : <EOL> seen_identifiers = set ( ) <EOL> for i , item in enumerate ( doc ) : <EOL> dataset_name = "<STR_LIT>" % ( i + <NUM_LIT:1> ) <EOL> if check_string_field ( item , "<STR_LIT:title>" , <NUM_LIT:5> , dataset_name , errs ) : <EOL> dataset_name = '<STR_LIT>' % item . get ( "<STR_LIT:title>" , "<STR_LIT>" ) . strip ( ) <EOL> for k , v in item . items ( ) : <EOL> if v is None : <EOL> add_error ( errs , <NUM_LIT:2> , "<STR_LIT>" , "<STR_LIT>" % k , dataset_name ) <EOL> if isinstance ( v , list ) and len ( v ) == <NUM_LIT:0> : <EOL> add_error ( errs , <NUM_LIT:2> , "<STR_LIT>" , "<STR_LIT>" % k , dataset_name ) <EOL> check_string_field ( item , "<STR_LIT:description>" , <NUM_LIT:30> , dataset_name , errs ) <EOL> if isinstance ( item . get ( "<STR_LIT>" ) , ( str , unicode ) ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif check_required_field ( item , "<STR_LIT>" , list , dataset_name , errs ) : <EOL> for kw in item [ "<STR_LIT>" ] : <EOL> if not isinstance ( kw , ( str , unicode ) ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif len ( kw . strip ( ) ) == <NUM_LIT:0> : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if item . get ( "<STR_LIT>" , True ) == True and check_required_field ( item , "<STR_LIT>" , list , dataset_name , errs ) : <EOL> for bc in item [ "<STR_LIT>" ] : <EOL> if not isinstance ( bc , ( str , unicode ) ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif "<STR_LIT::>" not in bc : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" % bc , dataset_name ) <EOL> elif bc not in omb_burueau_codes : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" % bc , dataset_name ) <EOL> check_date_field ( item , "<STR_LIT>" , dataset_name , errs ) <EOL> check_string_field ( item , "<STR_LIT>" , <NUM_LIT:1> , dataset_name , errs ) <EOL> check_string_field ( item , "<STR_LIT>" , <NUM_LIT:3> , dataset_name , errs ) <EOL> if check_string_field ( item , "<STR_LIT>" , <NUM_LIT:3> , dataset_name , errs ) : <EOL> if not email_validator ( item [ "<STR_LIT>" ] ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" % item [ "<STR_LIT>" ] , dataset_name ) <EOL> if check_string_field ( item , "<STR_LIT>" , <NUM_LIT:1> , dataset_name , errs ) : <EOL> if item [ "<STR_LIT>" ] in seen_identifiers : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" % item [ "<STR_LIT>" ] , dataset_name ) <EOL> seen_identifiers . add ( item [ "<STR_LIT>" ] ) <EOL> if isinstance ( item . get ( "<STR_LIT>" ) , list ) and check_required_field ( item , "<STR_LIT>" , list , dataset_name , errs ) : <EOL> for s in item [ "<STR_LIT>" ] : <EOL> if not isinstance ( s , ( str , unicode ) ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif len ( s . strip ( ) ) == <NUM_LIT:0> : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if check_string_field ( item , "<STR_LIT>" , <NUM_LIT:0> , dataset_name , errs ) : <EOL> if item [ "<STR_LIT>" ] not in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" % item [ "<STR_LIT>" ] , dataset_name ) <EOL> elif item [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> add_error ( errs , <NUM_LIT:1> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) != "<STR_LIT>" : <EOL> check_string_field ( item , "<STR_LIT>" , <NUM_LIT:10> , dataset_name , errs ) <EOL> if check_url_field ( False , item , "<STR_LIT>" , dataset_name , errs ) : <EOL> if item . get ( "<STR_LIT>" ) and isinstance ( item . get ( "<STR_LIT>" ) , list ) : <EOL> for d in item [ "<STR_LIT>" ] : <EOL> if d . get ( "<STR_LIT>" ) == item . get ( "<STR_LIT>" ) : <EOL> break <EOL> else : <EOL> add_error ( errs , <NUM_LIT:20> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> check_url_field ( False , item , "<STR_LIT>" , dataset_name , errs ) <EOL> if item . get ( "<STR_LIT>" ) == "<STR_LIT>" and item . get ( "<STR_LIT>" ) is None and item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT:20> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if i == <NUM_LIT:0> and item . get ( "<STR_LIT>" ) != src_url : <EOL> add_error ( errs , <NUM_LIT:2> , "<STR_LIT>" , "<STR_LIT>" % src_url , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> if item . get ( "<STR_LIT>" ) != None : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif check_string_field ( item , "<STR_LIT>" , - <NUM_LIT:1> , dataset_name , errs ) : <EOL> check_mime_type ( item . get ( "<STR_LIT>" ) , "<STR_LIT>" , dataset_name , errs ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif not isinstance ( item . get ( "<STR_LIT>" ) , ( str , unicode ) ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif not isinstance ( item . get ( "<STR_LIT>" ) , ( str , unicode ) ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif not isinstance ( item [ "<STR_LIT>" ] , ( str , unicode ) ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif "<STR_LIT:/>" not in item [ "<STR_LIT>" ] : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> else : <EOL> d1 , d2 = item [ "<STR_LIT>" ] . split ( "<STR_LIT:/>" , <NUM_LIT:1> ) <EOL> if not ISO8601_REGEX . match ( d1 ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" % d1 , dataset_name ) <EOL> if not ISO8601_REGEX . match ( d2 ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" % d2 , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif not isinstance ( item [ "<STR_LIT>" ] , list ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> else : <EOL> for s in item [ "<STR_LIT>" ] : <EOL> if not isinstance ( s , ( str , unicode ) ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif len ( s . strip ( ) ) == <NUM_LIT:0> : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if check_url_field ( False , item , "<STR_LIT>" , dataset_name , errs ) : <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif not isinstance ( item [ "<STR_LIT>" ] , bool ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> pass <EOL> elif not isinstance ( item [ "<STR_LIT>" ] , list ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> else : <EOL> if len ( item [ "<STR_LIT>" ] ) > <NUM_LIT:0> and item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT:10> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> for j , d in enumerate ( item [ "<STR_LIT>" ] ) : <EOL> resource_name = dataset_name + ( "<STR_LIT>" % ( j + <NUM_LIT:1> ) ) <EOL> check_url_field ( True , d , "<STR_LIT>" , resource_name , errs ) <EOL> if check_string_field ( d , "<STR_LIT>" , <NUM_LIT:1> , resource_name , errs ) : <EOL> check_mime_type ( d [ "<STR_LIT>" ] , "<STR_LIT>" , resource_name , errs ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif item . get ( "<STR_LIT>" ) not in ACCRUAL_PERIODICITY_VALUES : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if check_url_field ( False , item , "<STR_LIT>" , dataset_name , errs ) : <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif not isinstance ( item [ "<STR_LIT>" ] , list ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> else : <EOL> for s in item [ "<STR_LIT>" ] : <EOL> if not LANGUAGE_REGEX . match ( s ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" % s , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif not isinstance ( item [ "<STR_LIT>" ] , ( str , unicode ) ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> elif not isinstance ( item [ "<STR_LIT>" ] , list ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> else : <EOL> for s in item [ "<STR_LIT>" ] : <EOL> if not URL_REGEX . match ( s ) : <EOL> add_error ( errs , <NUM_LIT:50> , "<STR_LIT>" , "<STR_LIT>" % s , dataset_name ) <EOL> if item . get ( "<STR_LIT>" ) is None : <EOL> add_error ( errs , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , dataset_name ) <EOL> else : <EOL> check_date_field ( item , "<STR_LIT>" , dataset_name , errs ) <EOL> for err_type in sorted ( errs ) : <EOL> errors_array . append ( ( <EOL> err_type [ <NUM_LIT:1> ] , <EOL> [ err_item + ( "<STR_LIT>" % len ( errs [ err_type ] [ err_item ] ) if len ( errs [ err_type ] [ err_item ] ) else "<STR_LIT>" ) <EOL> for err_item in sorted ( errs [ err_type ] , key = lambda x : ( - len ( errs [ err_type ] [ x ] ) , x ) ) <EOL> ] ) ) <EOL> def add_error ( errs , severity , heading , description , context = None ) : <EOL> s = errs . setdefault ( ( severity , heading ) , { } ) . setdefault ( description , set ( ) ) <EOL> if context : s . add ( context ) <EOL> def nice_type_name ( data_type ) : <EOL> if data_type == ( str , unicode ) or data_type in ( str , unicode ) : <EOL> return "<STR_LIT>" <EOL> elif data_type == list : <EOL> return "<STR_LIT>" <EOL> else : <EOL> return "<STR_LIT>" + str ( data_type ) <EOL> def check_required_field ( obj , field_name , data_type , dataset_name , errs , display_field_name = None ) : <EOL> if not display_field_name : display_field_name = field_name <EOL> if field_name not in obj : <EOL> add_error ( errs , <NUM_LIT:10> , "<STR_LIT>" , "<STR_LIT>" % display_field_name , dataset_name ) <EOL> return False <EOL> elif obj [ field_name ] is None : <EOL> add_error ( errs , <NUM_LIT:10> , "<STR_LIT>" , "<STR_LIT>" % display_field_name , dataset_name ) <EOL> return False <EOL> elif not isinstance ( obj [ field_name ] , data_type ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" % ( display_field_name , nice_type_name ( data_type ) , nice_type_name ( type ( obj [ field_name ] ) ) ) , dataset_name ) <EOL> return False <EOL> elif isinstance ( obj [ field_name ] , list ) and len ( obj [ field_name ] ) == <NUM_LIT:0> : <EOL> add_error ( errs , <NUM_LIT:10> , "<STR_LIT>" , "<STR_LIT>" % display_field_name , dataset_name ) <EOL> return False <EOL> return True <EOL> def check_string_field ( obj , field_name , min_length , dataset_name , errs ) : <EOL> if not check_required_field ( obj , field_name , ( str , unicode ) , dataset_name , errs ) : <EOL> return False <EOL> elif len ( obj [ field_name ] . strip ( ) ) == <NUM_LIT:0> : <EOL> add_error ( errs , <NUM_LIT:10> , "<STR_LIT>" , "<STR_LIT>" % field_name , dataset_name ) <EOL> return False <EOL> elif len ( obj [ field_name ] . strip ( ) ) <= min_length : <EOL> add_error ( errs , <NUM_LIT:100> , "<STR_LIT>" , "<STR_LIT>" % ( field_name , obj [ field_name ] ) , dataset_name ) <EOL> return False <EOL> return True <EOL> def check_date_field ( obj , field_name , dataset_name , errs ) : <EOL> if not check_required_field ( obj , field_name , ( str , unicode ) , dataset_name , errs ) : <EOL> return False <EOL> elif len ( obj [ field_name ] . strip ( ) ) == <NUM_LIT:0> : <EOL> add_error ( errs , <NUM_LIT:10> , "<STR_LIT>" , "<STR_LIT>" % field_name , dataset_name ) <EOL> return False <EOL> else : <EOL> if not ISO8601_REGEX . match ( obj [ field_name ] ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" % ( field_name , obj [ field_name ] ) , dataset_name ) <EOL> return False <EOL> return True <EOL> def check_url_field ( required , obj , field_name , dataset_name , errs ) : <EOL> display_field_name = field_name <EOL> field_name = field_name . split ( "<STR_LIT:U+0020>" ) [ - <NUM_LIT:1> ] <EOL> if not required and ( field_name not in obj or obj [ field_name ] is None ) : return True <EOL> if not check_required_field ( obj , field_name , ( str , unicode ) , dataset_name , errs , display_field_name = display_field_name ) : return False <EOL> if not URL_REGEX . match ( obj [ field_name ] ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" % ( display_field_name , obj [ field_name ] ) , dataset_name ) <EOL> return False <EOL> return True <EOL> def check_mime_type ( format , field_name , dataset_name , errs ) : <EOL> if format . lower ( ) in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:text>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" % field_name , dataset_name ) <EOL> elif not MIMETYPE_REGEX . match ( format ) : <EOL> add_error ( errs , <NUM_LIT:5> , "<STR_LIT>" , "<STR_LIT>" % ( field_name , format ) , dataset_name ) <EOL> elif format . split ( "<STR_LIT:;>" ) [ <NUM_LIT:0> ] not in COMMON_MIMETYPES : <EOL> add_error ( errs , <NUM_LIT:100> , "<STR_LIT>" , "<STR_LIT>" % ( field_name , format ) , dataset_name ) </s>
