]> code.communitydata.science - rises_declines_wikia_code.git/blob - mediawiki_dump_tools/Mediawiki-Utilities/mw/lib/reverts/database.py
add copy of the GPL
[rises_declines_wikia_code.git] / mediawiki_dump_tools / Mediawiki-Utilities / mw / lib / reverts / database.py
1 import random
2 from itertools import chain
3
4 from . import defaults
5 from ...types import Timestamp
6 from ...util import none_or
7 from .dummy_checksum import DummyChecksum
8 from .functions import detect
9
10 HEX = "1234567890abcdef"
11
12 def random_sha1():
13     return ''.join(random.choice(HEX) for i in range(40))
14
15 """
16 Simple constant used in order to not do weird things with a dummy revision.
17 """
18
19
20 def check_row(db, rev_row, **kwargs):
21     """
22     Checks whether a revision (database row) was reverted (identity) and returns
23     a named tuple of Revert(reverting, reverteds, reverted_to).
24
25     :Parameters:
26         db : :class:`mw.database.DB`
27             A database connection to make use of.
28         rev_row : dict
29             a revision row containing 'rev_id' and 'rev_page' or 'page_id'
30         radius : int
31             a positive integer indicating the the maximum number of revisions that can be reverted
32         check_archive : bool
33             should the archive table be checked for reverting revisions?
34         before : `Timestamp`
35             if set, limits the search for *reverting* revisions to those which were saved before this timestamp
36     """
37
38     # extract rev_id, sha1, page_id
39     if 'rev_id' in rev_row:
40         rev_id = rev_row['rev_id']
41     else:
42         raise TypeError("rev_row must have 'rev_id'")
43     if 'page_id' in rev_row:
44         page_id = rev_row['page_id']
45     elif 'rev_page' in rev_row:
46         page_id = rev_row['rev_page']
47     else:
48         raise TypeError("rev_row must have 'page_id' or 'rev_page'")
49
50     # run the regular check
51     return check(db, rev_id, page_id=page_id, **kwargs)
52
53
54 def check(db, rev_id, page_id=None, radius=defaults.RADIUS, check_archive=False,
55           before=None, window=None):
56
57     """
58     Checks whether a revision was reverted (identity) and returns a named tuple
59     of Revert(reverting, reverteds, reverted_to).
60
61     :Parameters:
62         db : `mw.database.DB`
63             A database connection to make use of.
64         rev_id : int
65             the ID of the revision to check
66         page_id : int
67             the ID of the page the revision occupies (slower if not provided)
68         radius : int
69             a positive integer indicating the maximum number of revisions that can be reverted
70         check_archive : bool
71             should the archive table be checked for reverting revisions?
72         before : `Timestamp`
73             if set, limits the search for *reverting* revisions to those which were saved before this timestamp
74         window : int
75             if set, limits the search for *reverting* revisions to those which
76             were saved within `window` seconds after the reverted edit
77     """
78
79     if not hasattr(db, "revisions") and hasattr(db, "all_revisions"):
80         raise TypeError("db wrong type.  Expected a mw.database.DB.")
81
82     rev_id = int(rev_id)
83     radius = int(radius)
84     if radius < 1:
85         raise TypeError("invalid radius.  Expected a positive integer.")
86     page_id = none_or(page_id, int)
87     check_archive = bool(check_archive)
88     before = none_or(before, Timestamp)
89
90     # If we are searching the archive, we'll need to use `all_revisions`.
91     if check_archive:
92         dbrevs = db.all_revisions
93     else:
94         dbrevs = db.revisions
95
96     # If we don't have the sha1 or page_id, we're going to need to look them up
97     if page_id is None:
98         row = dbrevs.get(rev_id=rev_id)
99         page_id = row['rev_page']
100
101     # Load history and current rev
102     current_and_past_revs = list(dbrevs.query(
103         page_id=page_id,
104         limit=radius + 1,
105         before_id=rev_id + 1,  # Ensures that we capture the current revision
106         direction="older"
107     ))
108
109     try:
110         # Extract current rev and reorder history
111         current_rev, past_revs = (
112             current_and_past_revs[0],  # Current rev is the first one returned
113             reversed(current_and_past_revs[1:])  # The rest are past revs, but they are in the wrong order
114         )
115     except IndexError:
116         # Only way to get here is if there isn't enough history.  Couldn't be
117         # reverted.  Just return None.
118         return None
119
120     if window is not None and before is None:
121         before = Timestamp(current_rev['rev_timestamp']) + window
122
123     # Load future revisions
124     future_revs = dbrevs.query(
125         page_id=page_id,
126         limit=radius,
127         after_id=rev_id,
128         before=before,
129         direction="newer"
130     )
131
132     # Convert to an iterable of (checksum, rev) pairs for detect() to consume
133     checksum_revisions = chain(
134         ((rev['rev_sha1'] if rev['rev_sha1'] is not None \
135           else DummyChecksum(), rev)
136          for rev in past_revs),
137         [(current_rev['rev_sha1'] or DummyChecksum(), current_rev)],
138         ((rev['rev_sha1'] if rev['rev_sha1'] is not None \
139           else DummyChecksum(), rev)
140          for rev in future_revs)
141     )
142
143     for revert in detect(checksum_revisions, radius=radius):
144         # Check that this is a relevant revert
145         if rev_id in [rev['rev_id'] for rev in revert.reverteds]:
146             return revert
147
148     return None

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