<s> """<STR_LIT>""" <EOL> import time <EOL> import codecs <EOL> from cvs2svn_lib . log import logger <EOL> DB_OPEN_READ = '<STR_LIT:r>' <EOL> DB_OPEN_WRITE = '<STR_LIT:w>' <EOL> DB_OPEN_NEW = '<STR_LIT:n>' <EOL> SVN_INVALID_REVNUM = - <NUM_LIT:1> <EOL> warning_prefix = "<STR_LIT>" <EOL> error_prefix = "<STR_LIT>" <EOL> class FatalException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class InternalError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class FatalError ( FatalException ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , msg ) : <EOL> """<STR_LIT>""" <EOL> FatalException . __init__ ( self , '<STR_LIT>' % ( error_prefix , msg , ) ) <EOL> class CommandError ( FatalError ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , command , exit_status , error_output = '<STR_LIT>' ) : <EOL> self . command = command <EOL> self . exit_status = exit_status <EOL> self . error_output = error_output <EOL> if error_output . rstrip ( ) : <EOL> FatalError . __init__ ( <EOL> self , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT:%s>' <EOL> % ( self . command , self . exit_status , self . error_output . rstrip ( ) ) ) <EOL> else : <EOL> FatalError . __init__ ( <EOL> self , <EOL> '<STR_LIT>' <EOL> % ( self . command , self . exit_status ) ) <EOL> def canonicalize_eol ( text , eol ) : <EOL> """<STR_LIT>""" <EOL> text = text . replace ( '<STR_LIT:\r\n>' , '<STR_LIT:\n>' ) <EOL> text = text . replace ( '<STR_LIT:\r>' , '<STR_LIT:\n>' ) <EOL> if eol != '<STR_LIT:\n>' : <EOL> text = text . replace ( '<STR_LIT:\n>' , eol ) <EOL> return text <EOL> def path_join ( * components ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT:/>' . join ( filter ( None , components ) ) <EOL> def path_split ( path ) : <EOL> """<STR_LIT>""" <EOL> pos = path . rfind ( '<STR_LIT:/>' ) <EOL> if pos == - <NUM_LIT:1> : <EOL> return ( '<STR_LIT>' , path , ) <EOL> else : <EOL> return ( path [ : pos ] , path [ pos + <NUM_LIT:1> : ] , ) <EOL> class IllegalSVNPathError ( FatalException ) : <EOL> pass <EOL> def normalize_svn_path ( path , allow_empty = False ) : <EOL> """<STR_LIT>""" <EOL> norm_path = path_join ( * path . split ( '<STR_LIT:/>' ) ) <EOL> if not allow_empty and not norm_path : <EOL> raise IllegalSVNPathError ( "<STR_LIT>" ) <EOL> return norm_path <EOL> class PathRepeatedException ( Exception ) : <EOL> def __init__ ( self , path , count ) : <EOL> self . path = path <EOL> self . count = count <EOL> Exception . __init__ ( <EOL> self , '<STR_LIT>' % ( self . path , self . count , ) <EOL> ) <EOL> class PathsNestedException ( Exception ) : <EOL> def __init__ ( self , nest , nestlings ) : <EOL> self . nest = nest <EOL> self . nestlings = nestlings <EOL> Exception . __init__ ( <EOL> self , <EOL> '<STR_LIT>' <EOL> % ( self . nest , '<STR_LIT:U+002CU+0020>' . join ( self . nestlings ) , ) <EOL> ) <EOL> class PathsNotDisjointException ( FatalException ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , problems ) : <EOL> self . problems = problems <EOL> Exception . __init__ ( <EOL> self , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( '<STR_LIT>' . join ( [ str ( problem ) for problem in self . problems ] ) , ) <EOL> ) <EOL> def verify_paths_disjoint ( * paths ) : <EOL> """<STR_LIT>""" <EOL> def split ( path ) : <EOL> if not path : <EOL> return [ ] <EOL> else : <EOL> return path . split ( '<STR_LIT:/>' ) <EOL> def contains ( split_path1 , split_path2 ) : <EOL> """<STR_LIT>""" <EOL> return ( <EOL> len ( split_path1 ) < len ( split_path2 ) <EOL> and split_path2 [ : len ( split_path1 ) ] == split_path1 <EOL> ) <EOL> paths = [ ( split ( path ) , path ) for path in paths ] <EOL> paths . sort ( ) <EOL> problems = [ ] <EOL> i = <NUM_LIT:0> <EOL> while i < len ( paths ) : <EOL> split_path , path = paths [ i ] <EOL> j = i + <NUM_LIT:1> <EOL> while j < len ( paths ) and split_path == paths [ j ] [ <NUM_LIT:0> ] : <EOL> j += <NUM_LIT:1> <EOL> if j - i > <NUM_LIT:1> : <EOL> problems . append ( PathRepeatedException ( path , j - i ) ) <EOL> del paths [ i + <NUM_LIT:1> : j ] <EOL> i += <NUM_LIT:1> <EOL> i = <NUM_LIT:0> <EOL> while i < len ( paths ) : <EOL> split_path , path = paths [ i ] <EOL> j = i + <NUM_LIT:1> <EOL> while j < len ( paths ) and contains ( split_path , paths [ j ] [ <NUM_LIT:0> ] ) : <EOL> j += <NUM_LIT:1> <EOL> if j - i > <NUM_LIT:1> : <EOL> problems . append ( PathsNestedException ( <EOL> path , [ path2 for ( split_path2 , path2 ) in paths [ i + <NUM_LIT:1> : j ] ] <EOL> ) ) <EOL> i += <NUM_LIT:1> <EOL> if problems : <EOL> raise PathsNotDisjointException ( problems ) <EOL> def is_trunk_revision ( rev ) : <EOL> """<STR_LIT>""" <EOL> return rev . count ( '<STR_LIT:.>' ) == <NUM_LIT:1> <EOL> def is_branch_revision_number ( rev ) : <EOL> """<STR_LIT>""" <EOL> return rev . count ( '<STR_LIT:.>' ) % <NUM_LIT:2> == <NUM_LIT:0> <EOL> def format_date ( date ) : <EOL> """<STR_LIT>""" <EOL> return time . strftime ( "<STR_LIT>" , time . gmtime ( date ) ) <EOL> class CVSTextDecoder : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , encodings , fallback_encoding = None , eol_fix = None ) : <EOL> """<STR_LIT>""" <EOL> self . decoders = [ ] <EOL> for encoding in encodings : <EOL> self . add_encoding ( encoding ) <EOL> self . set_fallback_encoding ( fallback_encoding ) <EOL> self . eol_fix = eol_fix <EOL> def add_encoding ( self , encoding ) : <EOL> """<STR_LIT>""" <EOL> for ( name , decoder ) in self . decoders : <EOL> if name == encoding : <EOL> return <EOL> else : <EOL> self . decoders . append ( ( encoding , codecs . lookup ( encoding ) [ <NUM_LIT:1> ] ) ) <EOL> def set_fallback_encoding ( self , encoding ) : <EOL> """<STR_LIT>""" <EOL> if encoding is None : <EOL> self . fallback_decoder = None <EOL> else : <EOL> self . fallback_decoder = ( encoding , codecs . lookup ( encoding ) [ <NUM_LIT:1> ] ) <EOL> def decode ( self , s ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( s , unicode ) : <EOL> return s <EOL> for ( name , decoder ) in self . decoders : <EOL> try : <EOL> return decoder ( s ) [ <NUM_LIT:0> ] <EOL> except ValueError : <EOL> logger . verbose ( "<STR_LIT>" % ( name , s ) ) <EOL> if self . fallback_decoder is not None : <EOL> ( name , decoder ) = self . fallback_decoder <EOL> return decoder ( s , '<STR_LIT:replace>' ) [ <NUM_LIT:0> ] <EOL> else : <EOL> raise UnicodeError ( ) <EOL> def __call__ ( self , s ) : <EOL> s = self . decode ( s ) <EOL> if self . eol_fix is not None : <EOL> s = canonicalize_eol ( s , self . eol_fix ) <EOL> return s <EOL> def decode_path ( self , path ) : <EOL> """<STR_LIT>""" <EOL> return u'<STR_LIT:/>' . join ( [ <EOL> self . decode ( piece ) <EOL> for piece in path . split ( '<STR_LIT:/>' ) <EOL> ] ) <EOL> class Timestamper : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . timestamp = <NUM_LIT:0.0> <EOL> self . max_timestamp = time . time ( ) + <NUM_LIT> * <NUM_LIT> * <NUM_LIT> <EOL> def get ( self , timestamp , change_expected ) : <EOL> """<STR_LIT>""" <EOL> if timestamp > self . max_timestamp : <EOL> self . timestamp = self . timestamp + <NUM_LIT:1.0> <EOL> if not change_expected : <EOL> logger . warn ( <EOL> '<STR_LIT>' <EOL> % ( time . asctime ( time . gmtime ( timestamp ) ) , <EOL> time . asctime ( time . gmtime ( self . timestamp ) ) , ) <EOL> ) <EOL> elif timestamp < self . timestamp + <NUM_LIT:1.0> : <EOL> self . timestamp = self . timestamp + <NUM_LIT:1.0> <EOL> if not change_expected and logger . is_on ( logger . VERBOSE ) : <EOL> logger . verbose ( <EOL> '<STR_LIT>' <EOL> % ( time . asctime ( time . gmtime ( timestamp ) ) , <EOL> time . asctime ( time . gmtime ( self . timestamp ) ) , ) <EOL> ) <EOL> else : <EOL> self . timestamp = timestamp <EOL> return self . timestamp </s>
