<s> from git . util import ( <EOL> Actor , <EOL> LockedFD , <EOL> LockFile , <EOL> assure_directory_exists , <EOL> to_native_path , <EOL> ) <EOL> from gitdb . util import ( <EOL> bin_to_hex , <EOL> join , <EOL> file_contents_ro_filepath , <EOL> ) <EOL> from git . objects . util import ( <EOL> parse_date , <EOL> Serializable , <EOL> altz_to_utctz_str , <EOL> ) <EOL> from git . compat import ( <EOL> PY3 , <EOL> xrange , <EOL> string_types , <EOL> defenc <EOL> ) <EOL> import time <EOL> import re <EOL> __all__ = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> class RefLogEntry ( tuple ) : <EOL> """<STR_LIT>""" <EOL> _re_hexsha_only = re . compile ( '<STR_LIT>' ) <EOL> __slots__ = tuple ( ) <EOL> def __repr__ ( self ) : <EOL> """<STR_LIT>""" <EOL> res = self . format ( ) <EOL> if PY3 : <EOL> return res <EOL> else : <EOL> return res . encode ( defenc ) <EOL> def format ( self ) : <EOL> """<STR_LIT>""" <EOL> act = self . actor <EOL> time = self . time <EOL> return u"<STR_LIT>" . format ( self . oldhexsha , <EOL> self . newhexsha , <EOL> act . name , <EOL> act . email , <EOL> time [ <NUM_LIT:0> ] , <EOL> altz_to_utctz_str ( time [ <NUM_LIT:1> ] ) , <EOL> self . message ) <EOL> @ property <EOL> def oldhexsha ( self ) : <EOL> """<STR_LIT>""" <EOL> return self [ <NUM_LIT:0> ] <EOL> @ property <EOL> def newhexsha ( self ) : <EOL> """<STR_LIT>""" <EOL> return self [ <NUM_LIT:1> ] <EOL> @ property <EOL> def actor ( self ) : <EOL> """<STR_LIT>""" <EOL> return self [ <NUM_LIT:2> ] <EOL> @ property <EOL> def time ( self ) : <EOL> """<STR_LIT>""" <EOL> return self [ <NUM_LIT:3> ] <EOL> @ property <EOL> def message ( self ) : <EOL> """<STR_LIT>""" <EOL> return self [ <NUM_LIT:4> ] <EOL> @ classmethod <EOL> def new ( self , oldhexsha , newhexsha , actor , time , tz_offset , message ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( actor , Actor ) : <EOL> raise ValueError ( "<STR_LIT>" % actor ) <EOL> return RefLogEntry ( ( oldhexsha , newhexsha , actor , ( time , tz_offset ) , message ) ) <EOL> @ classmethod <EOL> def from_line ( cls , line ) : <EOL> """<STR_LIT>""" <EOL> line = line . decode ( defenc ) <EOL> fields = line . split ( '<STR_LIT:\t>' , <NUM_LIT:1> ) <EOL> if len ( fields ) == <NUM_LIT:1> : <EOL> info , msg = fields [ <NUM_LIT:0> ] , None <EOL> elif len ( fields ) == <NUM_LIT:2> : <EOL> info , msg = fields <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" % repr ( line ) ) <EOL> oldhexsha = info [ : <NUM_LIT> ] <EOL> newhexsha = info [ <NUM_LIT> : <NUM_LIT> ] <EOL> for hexsha in ( oldhexsha , newhexsha ) : <EOL> if not cls . _re_hexsha_only . match ( hexsha ) : <EOL> raise ValueError ( "<STR_LIT>" % hexsha ) <EOL> email_end = info . find ( '<STR_LIT:>>' , <NUM_LIT> ) <EOL> if email_end == - <NUM_LIT:1> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> actor = Actor . _from_string ( info [ <NUM_LIT> : email_end + <NUM_LIT:1> ] ) <EOL> time , tz_offset = parse_date ( info [ email_end + <NUM_LIT:2> : ] ) <EOL> return RefLogEntry ( ( oldhexsha , newhexsha , actor , ( time , tz_offset ) , msg ) ) <EOL> class RefLog ( list , Serializable ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( '<STR_LIT>' , ) <EOL> def __new__ ( cls , filepath = None ) : <EOL> inst = super ( RefLog , cls ) . __new__ ( cls ) <EOL> return inst <EOL> def __init__ ( self , filepath = None ) : <EOL> """<STR_LIT>""" <EOL> self . _path = filepath <EOL> if filepath is not None : <EOL> self . _read_from_file ( ) <EOL> def _read_from_file ( self ) : <EOL> try : <EOL> fmap = file_contents_ro_filepath ( self . _path , stream = True , allow_mmap = True ) <EOL> except OSError : <EOL> return <EOL> try : <EOL> self . _deserialize ( fmap ) <EOL> finally : <EOL> fmap . close ( ) <EOL> @ classmethod <EOL> def from_file ( cls , filepath ) : <EOL> """<STR_LIT>""" <EOL> return cls ( filepath ) <EOL> @ classmethod <EOL> def path ( cls , ref ) : <EOL> """<STR_LIT>""" <EOL> return join ( ref . repo . git_dir , "<STR_LIT>" , to_native_path ( ref . path ) ) <EOL> @ classmethod <EOL> def iter_entries ( cls , stream ) : <EOL> """<STR_LIT>""" <EOL> new_entry = RefLogEntry . from_line <EOL> if isinstance ( stream , string_types ) : <EOL> stream = file_contents_ro_filepath ( stream ) <EOL> while True : <EOL> line = stream . readline ( ) <EOL> if not line : <EOL> return <EOL> yield new_entry ( line . strip ( ) ) <EOL> stream . close ( ) <EOL> @ classmethod <EOL> def entry_at ( cls , filepath , index ) : <EOL> """<STR_LIT>""" <EOL> fp = open ( filepath , '<STR_LIT:rb>' ) <EOL> if index < <NUM_LIT:0> : <EOL> return RefLogEntry . from_line ( fp . readlines ( ) [ index ] . strip ( ) ) <EOL> else : <EOL> for i in xrange ( index + <NUM_LIT:1> ) : <EOL> line = fp . readline ( ) <EOL> if not line : <EOL> break <EOL> if i != index or not line : <EOL> raise IndexError <EOL> return RefLogEntry . from_line ( line . strip ( ) ) <EOL> def to_file ( self , filepath ) : <EOL> """<STR_LIT>""" <EOL> lfd = LockedFD ( filepath ) <EOL> assure_directory_exists ( filepath , is_file = True ) <EOL> fp = lfd . open ( write = True , stream = True ) <EOL> try : <EOL> self . _serialize ( fp ) <EOL> lfd . commit ( ) <EOL> except : <EOL> lfd . rollback ( ) <EOL> raise <EOL> @ classmethod <EOL> def append_entry ( cls , config_reader , filepath , oldbinsha , newbinsha , message ) : <EOL> """<STR_LIT>""" <EOL> if len ( oldbinsha ) != <NUM_LIT:20> or len ( newbinsha ) != <NUM_LIT:20> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> assure_directory_exists ( filepath , is_file = True ) <EOL> committer = isinstance ( config_reader , Actor ) and config_reader or Actor . committer ( config_reader ) <EOL> entry = RefLogEntry ( ( <EOL> bin_to_hex ( oldbinsha ) . decode ( '<STR_LIT:ascii>' ) , <EOL> bin_to_hex ( newbinsha ) . decode ( '<STR_LIT:ascii>' ) , <EOL> committer , ( int ( time . time ( ) ) , time . altzone ) , message <EOL> ) ) <EOL> lf = LockFile ( filepath ) <EOL> lf . _obtain_lock_or_raise ( ) <EOL> fd = open ( filepath , '<STR_LIT>' ) <EOL> try : <EOL> fd . write ( entry . format ( ) . encode ( defenc ) ) <EOL> finally : <EOL> fd . close ( ) <EOL> lf . _release_lock ( ) <EOL> return entry <EOL> def write ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _path is None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . to_file ( self . _path ) <EOL> return self <EOL> def _serialize ( self , stream ) : <EOL> write = stream . write <EOL> for e in self : <EOL> write ( e . format ( ) . encode ( defenc ) ) <EOL> def _deserialize ( self , stream ) : <EOL> self . extend ( self . iter_entries ( stream ) ) </s>
