<s> r"""<STR_LIT>""" <EOL> import logging <EOL> import sys <EOL> from collections import deque , namedtuple <EOL> from functools import lru_cache <EOL> import docopt <EOL> import mwapi <EOL> import mwreverts <EOL> import mwtypes <EOL> import mwxml <EOL> from mwtypes import Timestamp <EOL> import mysqltsv <EOL> Revision = namedtuple ( "<STR_LIT>" , [ '<STR_LIT:id>' , '<STR_LIT:status>' , '<STR_LIT>' ] ) <EOL> User = namedtuple ( "<STR_LIT>" , [ '<STR_LIT:id>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> logger = logging . getLogger ( __name__ ) <EOL> def main ( argv = None ) : <EOL> args = docopt . docopt ( __doc__ , argv = argv ) <EOL> logging . basicConfig ( <EOL> level = logging . INFO , <EOL> format = '<STR_LIT>' <EOL> ) <EOL> revert_radius = int ( args [ '<STR_LIT>' ] ) <EOL> revert_window = int ( args [ '<STR_LIT>' ] ) * ( <NUM_LIT> * <NUM_LIT> ) <EOL> if args [ '<STR_LIT>' ] : <EOL> session = mwapi . Session ( args [ '<STR_LIT>' ] , <EOL> user_agent = "<STR_LIT>" ) <EOL> else : <EOL> session = None <EOL> dumps = args [ '<STR_LIT>' ] <EOL> verbose = args [ '<STR_LIT>' ] <EOL> start = args [ '<STR_LIT>' ] <EOL> if start : <EOL> start = Timestamp ( start ) <EOL> end = args [ '<STR_LIT>' ] <EOL> if end : <EOL> end = Timestamp ( end ) <EOL> reverted_only = args [ '<STR_LIT>' ] <EOL> trusted_groups = args [ '<STR_LIT>' ] <EOL> if trusted_groups : <EOL> trusted_groups = trusted_groups . split ( '<STR_LIT:U+002C>' ) <EOL> trusted_users = load_user_group_members ( trusted_groups , session ) <EOL> else : <EOL> trusted_users = None <EOL> trusted_edits = args [ '<STR_LIT>' ] <EOL> if trusted_edits : <EOL> trusted_edits = int ( trusted_edits ) <EOL> if args [ '<STR_LIT>' ] == "<STR_LIT>" : <EOL> rev_reverteds = mysqltsv . Writer ( sys . stdout ) <EOL> else : <EOL> rev_reverteds = mysqltsv . Writer ( open ( args [ '<STR_LIT>' ] , "<STR_LIT:w>" ) ) <EOL> check_blocked = args [ '<STR_LIT>' ] <EOL> run ( dumps , session , start , end , revert_radius , revert_window , <EOL> reverted_only , trusted_users , trusted_edits , rev_reverteds , <EOL> check_blocked , verbose = verbose ) <EOL> def load_user_group_members ( groups , session ) : <EOL> users = [ ] <EOL> aufrom = None <EOL> while True : <EOL> res = session . get ( action = '<STR_LIT>' , list = '<STR_LIT>' , auprop = '<STR_LIT>' , <EOL> augroup = '<STR_LIT:|>' . join ( groups ) , aulimit = <NUM_LIT> , <EOL> aufrom = aufrom ) <EOL> user_list = res [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> for user in user_list : <EOL> users . append ( User ( user [ '<STR_LIT>' ] , <NUM_LIT:0> , user [ '<STR_LIT>' ] ) ) <EOL> if not res . get ( '<STR_LIT>' ) : <EOL> break <EOL> aufrom = res [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> return users <EOL> @ lru_cache ( maxsize = <NUM_LIT> ) <EOL> def load_user_info ( user_name , session ) : <EOL> logger . debug ( "<STR_LIT>" . format ( user_name ) ) <EOL> res = session . get ( action = '<STR_LIT>' , list = '<STR_LIT>' , ususers = user_name , <EOL> usprop = '<STR_LIT>' ) <EOL> user_data = res [ "<STR_LIT>" ] [ "<STR_LIT>" ] [ <NUM_LIT:0> ] <EOL> if '<STR_LIT>' in user_data : <EOL> return User ( user_name , <NUM_LIT:0> , set ( ) ) <EOL> return User ( user_data [ "<STR_LIT>" ] , user_data [ '<STR_LIT>' ] , <EOL> set ( user_data [ "<STR_LIT>" ] ) ) <EOL> @ lru_cache ( maxsize = <NUM_LIT> ) <EOL> def user_blocks ( user_text , session ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" . format ( user_text ) ) <EOL> doc = session . get ( action = '<STR_LIT>' , list = '<STR_LIT>' , bkusers = user_text , <EOL> bkprop = [ '<STR_LIT:id>' , '<STR_LIT>' ] ) <EOL> return [ mwtypes . Timestamp ( b [ '<STR_LIT>' ] ) for b in doc [ '<STR_LIT>' ] [ '<STR_LIT>' ] ] <EOL> def user_recently_blocked ( user_text , session , before ) : <EOL> blocks = user_blocks ( user_text , session ) <EOL> return sum ( <NUM_LIT:1> for b in blocks if b <= before ) > <NUM_LIT:0> <EOL> def run ( paths , session , start , end , revert_radius , revert_window , <EOL> reverted_only , trusted_users , trusted_edits , rev_reverteds , <EOL> check_blocked , verbose = False ) : <EOL> def process_dump ( dump , path ) : <EOL> for page in dump : <EOL> detector = mwreverts . Detector ( radius = revert_radius ) <EOL> window = deque ( maxlen = revert_radius ) <EOL> for revision in page : <EOL> revision . maybe_damaging = None <EOL> revision . reason = None <EOL> revert = detector . process ( revision . sha1 , revision ) <EOL> if start and revision . timestamp < start : <EOL> continue <EOL> if end and revision . timestamp > end : <EOL> continue <EOL> window . append ( revision ) <EOL> if revert is not None : <EOL> for reverted in revert . reverteds : <EOL> if ( revision . timestamp - reverted . timestamp ) <= revert_window and reverted . user is not None and revision . user is not None and reverted . user . text != revision . user . text and reverted . maybe_damaging is not False : <EOL> reverted . maybe_damaging = True <EOL> reverted . reason = "<STR_LIT>" <EOL> if revert . reverted_to . maybe_damaging and revert . reverted_to . user . text != revision . user . text : <EOL> revert . reverted_to . maybe_damaging = False <EOL> revert . reverted_to . reason = "<STR_LIT>" + "<STR_LIT>" <EOL> load_user_data = trusted_edits or check_blocked <EOL> if revision . user . id is not None and revision . user . id > <NUM_LIT:0> and load_user_data : <EOL> info = load_user_info ( revision . user . text , session ) <EOL> else : <EOL> info = User ( revision . user . id , <NUM_LIT:0> , set ( ) ) <EOL> two_days_later = revision . timestamp + ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT:2> ) <EOL> if trusted_users and info . id in trusted_users : <EOL> revision . maybe_damaging = False <EOL> revision . reason = "<STR_LIT>" <EOL> elif check_blocked and user_recently_blocked ( <EOL> revision . user . text , session , two_days_later ) : <EOL> revision . maybe_damaging = True <EOL> revision . reason = "<STR_LIT>" <EOL> elif trusted_edits and info . editcount >= trusted_edits : <EOL> revision . maybe_damaging = False <EOL> revision . reason = "<STR_LIT>" <EOL> else : <EOL> revision . reason = "<STR_LIT>" <EOL> if len ( window ) == revert_radius : <EOL> old_revision = window . popleft ( ) <EOL> yield ( old_revision . id , old_revision . maybe_damaging , <EOL> old_revision . reason ) <EOL> for old_revision in window : <EOL> yield ( old_revision . id , old_revision . maybe_damaging , <EOL> old_revision . reason ) <EOL> for rev_id , maybe_damaging , reason in mwxml . map ( process_dump , paths ) : <EOL> rev_reverteds . write ( [ rev_id , bool ( maybe_damaging ) , reason ] ) <EOL> if maybe_damaging : <EOL> if reason and "<STR_LIT>" in reason : <EOL> if verbose : <EOL> sys . stderr . write ( "<STR_LIT:r>" ) <EOL> sys . stderr . flush ( ) <EOL> elif reason and "<STR_LIT>" in reason : <EOL> if verbose : <EOL> sys . stderr . write ( "<STR_LIT:b>" ) <EOL> sys . stderr . flush ( ) <EOL> else : <EOL> if verbose : <EOL> sys . stderr . write ( "<STR_LIT:.>" ) <EOL> sys . stderr . flush ( ) <EOL> else : <EOL> if reason and "<STR_LIT>" in reason : <EOL> if verbose : <EOL> sys . stderr . write ( "<STR_LIT:e>" ) <EOL> sys . stderr . flush ( ) <EOL> elif reason and "<STR_LIT>" in reason : <EOL> if verbose : <EOL> sys . stderr . write ( "<STR_LIT:.>" ) <EOL> sys . stderr . flush ( ) <EOL> else : <EOL> if verbose : <EOL> sys . stderr . write ( "<STR_LIT:g>" ) <EOL> sys . stderr . flush ( ) <EOL> if verbose : <EOL> sys . stderr . write ( "<STR_LIT:\n>" ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> try : <EOL> main ( ) <EOL> except KeyboardInterrupt : <EOL> sys . stderr . write ( "<STR_LIT>" ) </s>
