mediawiki_dump_tools/Mediawiki-Utilities/mw/lib/reverts/api.py

   1 from itertools import chain
   2
   3 from . import defaults
   4 from ...types import Timestamp
   5 from ...util import none_or
   6 from .dummy_checksum import DummyChecksum
   7 from .functions import detect
   8
   9
  10 def check_rev(session, rev, **kwargs):
  11     """
  12     Checks whether a revision (database row) was reverted (identity) and returns
  13     a named tuple of Revert(reverting, reverteds, reverted_to).
  14
  15     :Parameters:
  16         session : :class:`mw.api.Session`
  17             An API session to make use of
  18         rev : dict
  19             a revision dict containing 'revid' and 'page.id'
  20         radius : int
  21             a positive integer indicating the maximum number of revisions that can be reverted
  22         before : :class:`mw.Timestamp`
  23             if set, limits the search for *reverting* revisions to those which were saved before this timestamp
  24         properties : set( str )
  25             a set of properties to include in revisions (see :class:`mw.api.Revisions`)
  26     """
  27
  28     # extract rev_id, sha1, page_id
  29     if 'revid' in rev:
  30         rev_id = rev['revid']
  31     else:
  32         raise TypeError("rev must have 'rev_id'")
  33     if 'page' in rev:
  34         page_id = rev['page']['id']
  35     elif 'pageid' in rev:
  36         page_id = rev['pageid']
  37     else:
  38         raise TypeError("rev must have 'page' or 'pageid'")
  39
  40     # run the regular check
  41     return check(session, rev_id, page_id=page_id, **kwargs)
  42
  43
  44 def check(session, rev_id, page_id=None, radius=defaults.RADIUS,
  45           before=None, window=None, properties=None):
  46     """
  47     Checks whether a revision was reverted (identity) and returns a named tuple
  48     of Revert(reverting, reverteds, reverted_to).
  49
  50     :Parameters:
  51         session : :class:`mw.api.Session`
  52             An API session to make use of
  53         rev_id : int
  54             the ID of the revision to check
  55         page_id : int
  56             the ID of the page the revision occupies (slower if not provided)
  57         radius : int
  58             a positive integer indicating the maximum number of revisions
  59             that can be reverted
  60         before : :class:`mw.Timestamp`
  61             if set, limits the search for *reverting* revisions to those which
  62             were saved before this timestamp
  63         window : int
  64             if set, limits the search for *reverting* revisions to those which
  65             were saved within `window` seconds after the reverted edit
  66         properties : set( str )
  67             a set of properties to include in revisions (see :class:`mw.api.Revisions`)
  68     """
  69
  70     if not hasattr(session, "revisions"):
  71         raise TypeError("session wrong type.  Expected a mw.api.Session.")
  72
  73     rev_id = int(rev_id)
  74     radius = int(radius)
  75     if radius < 1:
  76         raise TypeError("invalid radius.  Expected a positive integer.")
  77
  78     page_id = none_or(page_id, int)
  79     before = none_or(before, Timestamp)
  80     properties = set(properties) if properties is not None else set()
  81
  82     # If we don't have the page_id, we're going to need to look them up
  83     if page_id is None:
  84         rev = session.revisions.get(rev_id, properties={'ids'})
  85         page_id = rev['page']['pageid']
  86
  87     # Load history and current rev
  88     current_and_past_revs = list(session.revisions.query(
  89         pageids={page_id},
  90         limit=radius + 1,
  91         start_id=rev_id,
  92         direction="older",
  93         properties={'ids', 'timestamp', 'sha1'} | properties
  94     ))
  95
  96     try:
  97         # Extract current rev and reorder history
  98         current_rev, past_revs = (
  99             current_and_past_revs[0],  # Current rev is the first one returned
 100             reversed(current_and_past_revs[1:])  # The rest are past revs, but they are in the wrong order
 101         )
 102     except IndexError:
 103         # Only way to get here is if there isn't enough history.  Couldn't be
 104         # reverted.  Just return None.
 105         return None
 106
 107     if window is not None and before is None:
 108         before = Timestamp(current_rev['timestamp']) + window
 109
 110     # Load future revisions
 111     future_revs = session.revisions.query(
 112         pageids={page_id},
 113         limit=radius,
 114         start_id=rev_id + 1, # Ensures that we skip the current revision
 115         end=before,
 116         direction="newer",
 117         properties={'ids', 'timestamp', 'sha1'} | properties
 118     )
 119
 120     # Convert to an iterable of (checksum, rev) pairs for detect() to consume
 121     checksum_revisions = chain(
 122         ((rev['sha1'] if 'sha1' in rev else DummyChecksum(), rev)
 123          for rev in past_revs),
 124         [(current_rev.get('sha1', DummyChecksum()), current_rev)],
 125         ((rev['sha1'] if 'sha1' in rev else DummyChecksum(), rev)
 126          for rev in future_revs),
 127     )
 128
 129     for revert in detect(checksum_revisions, radius=radius):
 130         # Check that this is a relevant revert
 131         if rev_id in [rev['revid'] for rev in revert.reverteds]:
 132             return revert
 133
 134     return None