| Index: trunk/tools/editor_trends/etl/transformer.py |
| — | — | @@ -34,17 +34,20 @@ |
| 35 | 35 | A simple class takes care of fetching an editor from the queue and start |
| 36 | 36 | processing its edits. |
| 37 | 37 | ''' |
| 38 | | - def __init__(self, rts, tasks): |
| | 38 | + def __init__(self, rts, tasks, db_raw, db_dataset): |
| 39 | 39 | super(EditorConsumer, self).__init__(rts, tasks) |
| | 40 | + self.db_raw = db_raw |
| | 41 | + self.db_dataset = db_dataset |
| 40 | 42 | |
| 41 | 43 | def run(self): |
| 42 | 44 | while True: |
| 43 | | - new_editor = self.tasks.get() |
| | 45 | + editor = self.tasks.get() |
| 44 | 46 | self.tasks.task_done() |
| 45 | 47 | print '%s editors to go...' % messages.show(self.tasks.qsize) |
| 46 | | - if new_editor == None: |
| | 48 | + if editor == None: |
| 47 | 49 | break |
| 48 | | - new_editor() |
| | 50 | + editor = Editor(self.db_raw, self.db_dataset, editor) |
| | 51 | + editor() |
| 49 | 52 | |
| 50 | 53 | |
| 51 | 54 | class Editor: |
| — | — | @@ -52,12 +55,12 @@ |
| 53 | 56 | self.editor_id = editor_id |
| 54 | 57 | self.db_raw = db_raw #storage.init_database(self.rts.storage, self.rts.dbname, self.rts.editors_raw) |
| 55 | 58 | self.db_dataset = db_dataset #storage.init_database(self.rts.storage, self.rts.dbname, self.rts.editors_dataset) |
| | 59 | + self.cutoff = 9 |
| 56 | 60 | |
| 57 | 61 | def __str__(self): |
| 58 | 62 | return '%s' % (self.editor_id) |
| 59 | 63 | |
| 60 | 64 | def __call__(self): |
| 61 | | - cutoff = 9 |
| 62 | 65 | editor = self.db_raw.find_one('editor', self.editor_id) |
| 63 | 66 | if editor == None: |
| 64 | 67 | return |
| — | — | @@ -74,7 +77,7 @@ |
| 75 | 78 | character_count = determine_edit_volume(edits, first_year, final_year) |
| 76 | 79 | revert_count = determine_number_reverts(edits, first_year, final_year) |
| 77 | 80 | |
| 78 | | - edits = sort_edits(edits) |
| | 81 | + |
| 79 | 82 | edit_count = determine_number_edits(edits, first_year, final_year) |
| 80 | 83 | |
| 81 | 84 | totals = {} |
| — | — | @@ -84,18 +87,21 @@ |
| 85 | 88 | totals = calculate_totals(totals, counts, article_count, 'article_count') |
| 86 | 89 | totals = calculate_totals(totals, counts, edit_count, 'edit_count') |
| 87 | 90 | |
| 88 | | - if len(edits) > cutoff: |
| 89 | | - new_wikipedian = edits[cutoff]['date'] |
| | 91 | + if len(edits) > self.cutoff: |
| | 92 | + new_wikipedian = edits[self.cutoff]['date'] |
| 90 | 93 | else: |
| 91 | 94 | new_wikipedian = False |
| 92 | | - cum_edit_count = len(edits) |
| | 95 | + cum_edit_count_main_ns, cum_edit_count_other_ns = calculate_cum_edits(edits) |
| | 96 | + |
| | 97 | + edits = sort_edits(edits) |
| 93 | 98 | first_edit = edits[0]['date'] |
| 94 | 99 | final_edit = edits[-1]['date'] |
| 95 | 100 | |
| 96 | 101 | data = {'editor': self.editor_id, |
| 97 | 102 | 'username': username, |
| 98 | 103 | 'new_wikipedian': new_wikipedian, |
| 99 | | - 'cum_edit_count': cum_edit_count, |
| | 104 | + 'cum_edit_count_main_ns': cum_edit_count_main_ns, |
| | 105 | + 'cum_edit_count_other_ns': cum_edit_count_other_ns, |
| 100 | 106 | 'final_edit': final_edit, |
| 101 | 107 | 'first_edit': first_edit, |
| 102 | 108 | 'last_edit_by_year': last_edit_by_year, |
| — | — | @@ -109,6 +115,7 @@ |
| 110 | 116 | } |
| 111 | 117 | self.db_dataset.insert(data) |
| 112 | 118 | |
| | 119 | + |
| 113 | 120 | def cleanup_datacontainer(dc, variable_type): |
| 114 | 121 | ''' |
| 115 | 122 | valid variable_type are either a {}, a [] or 0. |
| — | — | @@ -154,15 +161,27 @@ |
| 155 | 162 | ''' |
| 156 | 163 | dc = data_converter.create_datacontainer(first_year, final_year) |
| 157 | 164 | dc = data_converter.add_months_to_datacontainer(dc, 'dict') |
| 158 | | - for edit in edits: |
| 159 | | - ns = edit['ns'] |
| 160 | | - year, month = str(edit['date'].year), edit['date'].month |
| 161 | | - dc[year][month].setdefault(ns, 0) |
| 162 | | - dc[year][month][ns] += 1 |
| | 165 | + for year in edits: |
| | 166 | + for edit in edits[year]: |
| | 167 | + ns = edit['ns'] |
| | 168 | + month = edit['date'].month |
| | 169 | + dc[year][month].setdefault(ns, 0) |
| | 170 | + dc[year][month][ns] += 1 |
| 163 | 171 | dc = cleanup_datacontainer(dc, {}) |
| 164 | 172 | return dc |
| 165 | 173 | |
| | 174 | +def calculate_cum_edits(edits): |
| | 175 | + cum_edit_count_main_ns = 0 |
| | 176 | + cum_edit_count_other_ns = 0 |
| | 177 | + for year in edits: |
| | 178 | + for edit in edits[year]: |
| | 179 | + if edit['ns'] == 0: |
| | 180 | + cum_edit_count_main_ns += 1 |
| | 181 | + else: |
| | 182 | + cum_edit_count_other_ns += 1 |
| 166 | 183 | |
| | 184 | + return cum_edit_count_main_ns, cum_edit_count_other_ns |
| | 185 | + |
| 167 | 186 | def determine_articles_workedon(edits, first_year, final_year): |
| 168 | 187 | ''' |
| 169 | 188 | This function creates a list of article_ids that an editor has worked on in |
| — | — | @@ -264,9 +283,9 @@ |
| 265 | 284 | for edit in edits[year]: |
| 266 | 285 | date = str(edit['date'].year) |
| 267 | 286 | if dc[date] == 0: |
| 268 | | - dc[date] = edit |
| 269 | | - elif dc[date] < edit: |
| 270 | | - dc[date] = edit |
| | 287 | + dc[date] = edit['date'] |
| | 288 | + elif dc[date] < edit['date']: |
| | 289 | + dc[date] = edit['date'] |
| 271 | 290 | return dc |
| 272 | 291 | |
| 273 | 292 | |
| — | — | @@ -290,51 +309,52 @@ |
| 291 | 310 | return sorted(edits, key=itemgetter('date')) |
| 292 | 311 | |
| 293 | 312 | |
| | 313 | +def setup_database(rts): |
| | 314 | + ''' |
| | 315 | + Initialize the database, including setting indexes and dropping the older |
| | 316 | + version of the collection. |
| | 317 | + ''' |
| | 318 | + db_raw = storage.init_database(rts.storage, rts.dbname, rts.editors_raw) |
| | 319 | + db_dataset = storage.init_database(rts.storage, rts.dbname, rts.editors_dataset) |
| | 320 | + db_dataset.drop_collection() |
| | 321 | + editors = db_raw.retrieve_editors() |
| | 322 | + return db_raw, db_dataset, editors |
| | 323 | + |
| | 324 | + |
| 294 | 325 | def transform_editors_multi_launcher(rts): |
| 295 | | - tasks = multiprocessing.JoinableQueue() |
| 296 | 326 | db_raw, db_dataset, editors = setup_database(rts) |
| 297 | | - transformers = [EditorConsumer(rts, tasks) for i in xrange(rts.number_of_processes)] |
| | 327 | + transformers = [EditorConsumer(rts, editors, db_raw, db_dataset) for i in xrange(rts.number_of_processes)] |
| 298 | 328 | |
| 299 | | - for editor in editors: |
| 300 | | - tasks.put(Editor(rts, editor)) |
| 301 | 329 | |
| 302 | 330 | for x in xrange(rts.number_of_processes): |
| 303 | | - tasks.put(None) |
| | 331 | + editors.put(None) |
| 304 | 332 | |
| 305 | | - print messages.show(tasks.qsize) |
| 306 | 333 | for transformer in transformers: |
| 307 | 334 | transformer.start() |
| 308 | 335 | |
| 309 | | - tasks.join() |
| | 336 | + editors.join() |
| 310 | 337 | |
| 311 | 338 | db_dataset.add_index('editor') |
| 312 | 339 | db_dataset.add_index('new_wikipedian') |
| 313 | 340 | |
| 314 | 341 | |
| 315 | | -def setup_database(rts): |
| 316 | | - ''' |
| 317 | | - Initialize the database, including setting indexes and dropping the older |
| 318 | | - version of the collection. |
| 319 | | - ''' |
| 320 | | - db_raw = storage.init_database(rts.storage, rts.dbname, rts.editors_raw) |
| 321 | | - db_dataset = storage.init_database(rts.storage, rts.dbname, rts.editors_dataset) |
| 322 | | - db_dataset.drop_collection() |
| 323 | | - editors = [] |
| 324 | | - #editors = db_raw.retrieve_distinct_keys('editor') |
| 325 | | - #db_dataset.add_index('editor') |
| 326 | | - #db_dataset.add_index('new_wikipedian') |
| 327 | | - |
| 328 | | - return db_raw, db_dataset, editors |
| 329 | | - |
| 330 | | - |
| 331 | 342 | def transform_editors_single_launcher(rts): |
| 332 | 343 | print rts.dbname, rts.editors_raw |
| 333 | 344 | db_raw, db_dataset, editors = setup_database(rts) |
| 334 | 345 | n = db_raw.count() |
| 335 | 346 | pbar = progressbar.ProgressBar(maxval=n).start() |
| 336 | | - for editor in db_raw.find(): |
| | 347 | + |
| | 348 | + for x in xrange(rts.number_of_processes): |
| | 349 | + editors.put(None) |
| | 350 | + |
| | 351 | + while True: |
| | 352 | + editor = editors.get() |
| | 353 | + editors.task_done() |
| | 354 | + if editor == None: |
| | 355 | + break |
| 337 | 356 | editor = Editor(db_raw, db_dataset, editor) |
| 338 | 357 | editor() |
| | 358 | + |
| 339 | 359 | pbar.update(pbar.currval + 1) |
| 340 | 360 | |
| 341 | 361 | db_dataset.add_index('editor') |
| Index: trunk/tools/editor_trends/etl/sort.py |
| — | — | @@ -56,9 +56,9 @@ |
| 57 | 57 | for x, d in enumerate(data): |
| 58 | 58 | d = d.strip().split('\t') |
| 59 | 59 | #TEMP FIX: |
| 60 | | - editor = d[2] |
| 61 | | - d[2] = d[0] |
| 62 | | - d[0] = editor |
| | 60 | + #editor = d[2] |
| | 61 | + #d[2] = d[0] |
| | 62 | + #d[0] = editor |
| 63 | 63 | #END TEMP FIX |
| 64 | 64 | data[x] = d |
| 65 | 65 | #data = [d.strip() for d in data] |
| Index: trunk/tools/editor_trends/classes/queue.py |
| — | — | @@ -0,0 +1,58 @@ |
| | 2 | +#!/usr/bin/python |
| | 3 | +# coding=utf-8 |
| | 4 | +''' |
| | 5 | +Copyright (C) 2010 by Diederik van Liere (dvanliere@gmail.com) |
| | 6 | +This program is free software; you can redistribute it and/or |
| | 7 | +modify it under the terms of the GNU General Public License version 2 |
| | 8 | +as published by the Free Software Foundation. |
| | 9 | +This program is distributed in the hope that it will be useful, |
| | 10 | +but WITHOUT ANY WARRANTY; without even the implied warranty of |
| | 11 | +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. |
| | 12 | +See the GNU General Public License for more details, at |
| | 13 | +http,//www.fsf.org/licenses/gpl.html |
| | 14 | +''' |
| | 15 | + |
| | 16 | + |
| | 17 | +__author__ = '''\n'''.join(['Diederik van Liere (dvanliere@gmail.com)']) |
| | 18 | +__email__ = 'dvanliere at gmail dot com' |
| | 19 | +__date__ = '2011-04-21' |
| | 20 | +__version__ = '0.1' |
| | 21 | + |
| | 22 | + |
| | 23 | +from multiprocessing.queues import JoinableQueue, Queue |
| | 24 | +import errno |
| | 25 | + |
| | 26 | +def retry_on_eintr(function, *args, **kw): |
| | 27 | + while True: |
| | 28 | + try: |
| | 29 | + return function(*args, **kw) |
| | 30 | + except IOError, e: |
| | 31 | + if e.errno == errno.EINTR: |
| | 32 | + continue |
| | 33 | + else: |
| | 34 | + raise |
| | 35 | + |
| | 36 | +class RetryQueue(Queue): |
| | 37 | + """Queue which will retry if interrupted with EINTR.""" |
| | 38 | + def get(self, block=True, timeout=None): |
| | 39 | + return retry_on_eintr(Queue.get, self, block, timeout) |
| | 40 | + |
| | 41 | + def qsize(self): |
| | 42 | + try: |
| | 43 | + return self.qsize() |
| | 44 | + except: |
| | 45 | + #OSX does not support the qsize function so we return unknown |
| | 46 | + return 'unknown' |
| | 47 | + |
| | 48 | + |
| | 49 | +class JoinableRetryQueue(JoinableQueue): |
| | 50 | + """Queue which will retry if interrupted with EINTR.""" |
| | 51 | + def get(self, block=True, timeout=None): |
| | 52 | + return retry_on_eintr(Queue.get, self, block, timeout) |
| | 53 | + |
| | 54 | + def qsize(self): |
| | 55 | + try: |
| | 56 | + return self.qsize() |
| | 57 | + except: |
| | 58 | + #OSX does not support the qsize function so we return unknown |
| | 59 | + return 'unknown' |
| Property changes on: trunk/tools/editor_trends/classes/queue.py |
| ___________________________________________________________________ |
| Added: svn:eol-style |
| 1 | 60 | + native |
| Index: trunk/tools/editor_trends/classes/storage.py |
| — | — | @@ -23,10 +23,11 @@ |
| 24 | 24 | if '..' not in sys.path: |
| 25 | 25 | sys.path.append('..') |
| 26 | 26 | |
| 27 | | -from classes import settings |
| | 27 | +import settings |
| 28 | 28 | settings = settings.Settings() |
| 29 | 29 | |
| 30 | | -from classes import exceptions |
| | 30 | +import exceptions |
| | 31 | +import queue |
| 31 | 32 | from utils import file_utils |
| 32 | 33 | |
| 33 | 34 | import_error = 0 |
| — | — | @@ -81,7 +82,7 @@ |
| 82 | 83 | '''Update an observation in a collection''' |
| 83 | 84 | |
| 84 | 85 | @abstractmethod |
| 85 | | - def find(self, key, value, qualifier): |
| | 86 | + def find(self, key, qualifier): |
| 86 | 87 | '''Find multiple observations in a collection''' |
| 87 | 88 | |
| 88 | 89 | @abstractmethod |
| — | — | @@ -146,7 +147,7 @@ |
| 147 | 148 | assert isinstance(data, dict), 'You need to feed me dictionaries.' |
| 148 | 149 | self.db[self.collection].update({key: value}, data, upsert=True) |
| 149 | 150 | |
| 150 | | - def find(self, key=None, value=1, qualifier=None): |
| | 151 | + def find(self, key=None, qualifier=None): |
| 151 | 152 | if qualifier == 'min': |
| 152 | 153 | return self.db[self.collection].find({ |
| 153 | 154 | key : {'$ne' : False}}).sort(key, pymongo.ASCENDING).limit(1)[0] |
| — | — | @@ -154,7 +155,7 @@ |
| 155 | 156 | return self.db[self.collection].find({ |
| 156 | 157 | key : {'$ne' : False}}).sort(key, pymongo.DESCENDING).limit(1)[0] |
| 157 | 158 | elif key != None: |
| 158 | | - return self.db[self.collection].find({key: value}) |
| | 159 | + return self.db[self.collection].find({}, fields=[key]) |
| 159 | 160 | else: |
| 160 | 161 | return self.db[self.collection].find() |
| 161 | 162 | |
| — | — | @@ -171,6 +172,13 @@ |
| 172 | 173 | def count(self): |
| 173 | 174 | return self.db[self.collection].count() |
| 174 | 175 | |
| | 176 | + def retrieve_editors(self): |
| | 177 | + q = queue.JoinableRetryQueue() |
| | 178 | + cursor = self.find('editor') |
| | 179 | + for editor in cursor: |
| | 180 | + q.put(editor['editor']) |
| | 181 | + return q |
| | 182 | + |
| 175 | 183 | def retrieve_distinct_keys(self, key, force_new=False): |
| 176 | 184 | ''' |
| 177 | 185 | TODO: figure out how big the index is and then take appropriate action, |
| — | — | @@ -290,7 +298,7 @@ |
| 291 | 299 | def update(self, key, value, data): |
| 292 | 300 | return |
| 293 | 301 | |
| 294 | | - def find(self, key, value, qualifier=None): |
| | 302 | + def find(self, key, qualifier=None): |
| 295 | 303 | return |
| 296 | 304 | |
| 297 | 305 | def save(self, data): |