Source code for mw.lib.reverts.api

from itertools import chain

from . import defaults
from ...types import Timestamp
from ...util import none_or
from .dummy_checksum import DummyChecksum
from .functions import detect


[docs]def check_rev(session, rev, **kwargs): """ Checks whether a revision (database row) was reverted (identity) and returns a named tuple of Revert(reverting, reverteds, reverted_to). :Parameters: session : :class:`mw.api.Session` An API session to make use of rev : dict a revision dict containing 'revid' and 'page.id' radius : int a positive integer indicating the maximum number of revisions that can be reverted before : :class:`mw.Timestamp` if set, limits the search for *reverting* revisions to those which were saved before this timestamp properties : set( str ) a set of properties to include in revisions (see :class:`mw.api.Revisions`) """ # extract rev_id, sha1, page_id if 'revid' in rev: rev_id = rev['revid'] else: raise TypeError("rev must have 'rev_id'") if 'page' in rev: page_id = rev['page']['id'] elif 'pageid' in rev: page_id = rev['pageid'] else: raise TypeError("rev must have 'page' or 'pageid'") # run the regular check return check(session, rev_id, page_id=page_id, **kwargs)
[docs]def check(session, rev_id, page_id=None, radius=defaults.RADIUS, before=None, window=None, properties=None): """ Checks whether a revision was reverted (identity) and returns a named tuple of Revert(reverting, reverteds, reverted_to). :Parameters: session : :class:`mw.api.Session` An API session to make use of rev_id : int the ID of the revision to check page_id : int the ID of the page the revision occupies (slower if not provided) radius : int a positive integer indicating the maximum number of revisions that can be reverted before : :class:`mw.Timestamp` if set, limits the search for *reverting* revisions to those which were saved before this timestamp window : int if set, limits the search for *reverting* revisions to those which were saved within `window` seconds after the reverted edit properties : set( str ) a set of properties to include in revisions (see :class:`mw.api.Revisions`) """ if not hasattr(session, "revisions"): raise TypeError("session wrong type. Expected a mw.api.Session.") rev_id = int(rev_id) radius = int(radius) if radius < 1: raise TypeError("invalid radius. Expected a positive integer.") page_id = none_or(page_id, int) before = none_or(before, Timestamp) properties = set(properties) if properties is not None else set() # If we don't have the page_id, we're going to need to look them up if page_id is None: rev = session.revisions.get(rev_id, properties={'ids'}) page_id = rev['page']['pageid'] # Load history and current rev current_and_past_revs = list(session.revisions.query( pageids={page_id}, limit=radius + 1, start_id=rev_id, direction="older", properties={'ids', 'timestamp', 'sha1'} | properties )) try: # Extract current rev and reorder history current_rev, past_revs = ( current_and_past_revs[0], # Current rev is the first one returned reversed(current_and_past_revs[1:]) # The rest are past revs, but they are in the wrong order ) except IndexError: # Only way to get here is if there isn't enough history. Couldn't be # reverted. Just return None. return None if window is not None and before is None: before = Timestamp(current_rev['timestamp']) + window # Load future revisions future_revs = session.revisions.query( pageids={page_id}, limit=radius, start_id=rev_id + 1, # Ensures that we skip the current revision end=before, direction="newer", properties={'ids', 'timestamp', 'sha1'} | properties ) # Convert to an iterable of (checksum, rev) pairs for detect() to consume checksum_revisions = chain( ((rev['sha1'] if 'sha1' in rev else DummyChecksum(), rev) for rev in past_revs), [(current_rev.get('sha1', DummyChecksum()), current_rev)], ((rev['sha1'] if 'sha1' in rev else DummyChecksum(), rev) for rev in future_revs), ) for revert in detect(checksum_revisions, radius=radius): # Check that this is a relevant revert if rev_id in [rev['revid'] for rev in revert.reverteds]: return revert return None