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

   1 from collections import namedtuple
   2
   3 from ...util import ordered
   4 from . import defaults
   5
   6 Revert = namedtuple("Revert", ['reverting', 'reverteds', 'reverted_to'])
   7 """
   8 Represents a revert event.  This class behaves like
   9 :class:`collections.namedtuple`.  Note that the datatypes of `reverting`,
  10 `reverteds` and `reverted_to` is not specified since those types will depend
  11 on the revision data provided during revert detection.
  12
  13 :Members:
  14     **reverting**
  15         The reverting revision data : `mixed`
  16     **reverteds**
  17         The reverted revision data (ordered chronologically) : list( `mixed` )
  18     **reverted_to**
  19         The reverted-to revision data : `mixed`
  20 """
  21
  22
  23 class Detector(ordered.HistoricalMap):
  24     """
  25     Detects revert events in a stream of revisions (to the same page) based on
  26     matching checksums.  To detect reverts, construct an instance of this class and call
  27     :meth:`process` in chronological order (``direction == "newer"``).
  28
  29     See `<https://meta.wikimedia.org/wiki/R:Identity_revert>`_
  30
  31     :Parameters:
  32         radius : int
  33             a positive integer indicating the maximum revision distance that a revert can span.
  34
  35     :Example:
  36         >>> from mw.lib import reverts
  37         >>> detector = reverts.Detector()
  38         >>>
  39         >>> detector.process("aaa", {'rev_id': 1})
  40         >>> detector.process("bbb", {'rev_id': 2})
  41         >>> detector.process("aaa", {'rev_id': 3})
  42         Revert(reverting={'rev_id': 3}, reverteds=[{'rev_id': 2}], reverted_to={'rev_id': 1})
  43         >>> detector.process("ccc", {'rev_id': 4})
  44
  45     """
  46
  47     def __init__(self, radius=defaults.RADIUS):
  48         """
  49         :Parameters:
  50             radius : int
  51                 a positive integer indicating the maximum revision distance that a revert can span.
  52         """
  53         if radius < 1:
  54             raise TypeError("invalid radius. Expected a positive integer.")
  55         super().__init__(maxlen=radius + 1)
  56
  57     def process(self, checksum, revision=None):
  58         """
  59         Process a new revision and detect a revert if it occurred.  Note that
  60         you can pass whatever you like as `revision` and it will be returned in
  61         the case that a revert occurs.
  62
  63         :Parameters:
  64             checksum : str
  65                 Any identity-machable string-based hash of revision content
  66             revision : `mixed`
  67                 Revision meta data.  Note that any data will just be returned in the
  68                 case of a revert.
  69
  70         :Returns:
  71             a :class:`~mw.lib.reverts.Revert` if one occured or `None`
  72         """
  73         revert = None
  74
  75         if checksum in self:  # potential revert
  76
  77             reverteds = list(self.up_to(checksum))
  78
  79             if len(reverteds) > 0:  # If no reverted revisions, this is a noop
  80                 revert = Revert(revision, reverteds, self[checksum])
  81
  82         self.insert(checksum, revision)
  83         return revert