Source code for mwreverts.detector

from . import defaults
from .historical_dict import HistoricalDict
from .revert import Revert


[docs]class Detector(HistoricalDict): """ Detects revert events in a stream of revisions (to the same page) based on matching checksums. To detect reverts, construct an instance of this class and call :func:`~mwreverts.Detector.process` in chronological order. See https://meta.wikimedia.org/wiki/R:Identity_revert :Parameters: radius : int a positive integer indicating the maximum revision distance that a revert can span. :Example: >>> import mwreverts >>> detector = mwreverts.Detector() >>> >>> detector.process("aaa", {'rev_id': 1}) >>> detector.process("bbb", {'rev_id': 2}) >>> detector.process("aaa", {'rev_id': 3}) Revert(reverting={'rev_id': 3}, reverteds=[{'rev_id': 2}], reverted_to={'rev_id': 1}) >>> detector.process("ccc", {'rev_id': 4}) """ def initialize(self, radius=defaults.RADIUS): if radius < 1: raise TypeError("invalid radius. Expected a positive integer.") super().initialize(maxsize=radius + 1)
[docs] def process(self, checksum, revision=None): """ Process a new revision and detect a revert if it occurred. Note that you can pass whatever you like as `revision` and it will be returned in the case that a revert occurs. :Parameters: checksum : str Any identity-machable string-based hash of revision content revision : `mixed` Revision metadata. Note that any data will just be returned in the case of a revert. :Returns: a :class:`~mwreverts.Revert` if one occured or `None` """ revert = None if checksum in self: # potential revert reverteds = list(self.up_to(checksum)) if len(reverteds) > 0: # If no reverted revisions, this is a noop revert = Revert(revision, reverteds, self[checksum]) self.insert(checksum, revision) return revert