]> code.communitydata.science - rises_declines_wikia_code.git/blob - mediawiki_dump_tools/Mediawiki-Utilities/mw/lib/reverts/api.py
add copy of the GPL
[rises_declines_wikia_code.git] / 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

Community Data Science Collective || Want to submit a patch?