Milimetric has submitted this change and it was merged.
Change subject: Setup celery task workflow to handle running reports for the
ProgramMetrics API
......................................................................
Setup celery task workflow to handle running reports for the ProgramMetrics API
The form to trigger the Program Metrics reports can be found at
/reports/program-global-metrics
Bug:T118308
Change-Id: I1d4a6bba9f81962d01990754299a22a077ba61df
---
M tests/test_controllers/test_reports.py
A tests/test_models/test_run_program_metrics_report.py
M wikimetrics/api/__init__.py
M wikimetrics/api/cohorts.py
A wikimetrics/api/reports.py
M wikimetrics/controllers/cohorts.py
M wikimetrics/controllers/reports.py
M wikimetrics/forms/__init__.py
D wikimetrics/forms/global_metrics_form.py
A wikimetrics/forms/program_metrics_form.py
M wikimetrics/models/report_nodes/__init__.py
M wikimetrics/models/report_nodes/report.py
A wikimetrics/models/report_nodes/run_program_metrics_report.py
M wikimetrics/models/report_nodes/run_report.py
M wikimetrics/models/report_nodes/sum_aggregate_by_user_report.py
A wikimetrics/models/report_nodes/validate_program_metrics_report.py
R wikimetrics/static/js/programMetricsReportUpload.js
R wikimetrics/templates/forms/program_metrics_report_upload.html
R wikimetrics/templates/program_metrics_reports.html
19 files changed, 715 insertions(+), 145 deletions(-)
Approvals:
Milimetric: Verified; Looks good to me, approved
jenkins-bot: Verified
diff --git a/tests/test_controllers/test_reports.py
b/tests/test_controllers/test_reports.py
index 8703186..509e7f7 100644
--- a/tests/test_controllers/test_reports.py
+++ b/tests/test_controllers/test_reports.py
@@ -2,7 +2,6 @@
import celery
import time
import unittest
-import os.path
from mock import Mock, MagicMock, patch
from contextlib import contextmanager
from flask import appcontext_pushed, g
@@ -19,7 +18,7 @@
from wikimetrics.controllers.reports import (
get_celery_task,
)
-from wikimetrics.configurables import app, get_absolute_path, db, queue
+from wikimetrics.configurables import app, queue
@contextmanager
@@ -788,3 +787,56 @@
mock_report = ReportStore()
failure = mock_report.get_result_safely('')
assert_equal(failure['failure'], 'result not available')
+
+
+class ProgramMetricsReportControllerTest(ControllerAsyncTest):
+
+ def setUp(self):
+ ControllerAsyncTest.setUp(self)
+
+ def test_report_fails_with_invalid_cohort(self):
+ response = self.client.post('/reports/program-global-metrics',
data=dict(
+ name='TestCohort1',
+ project='wiki',
+ centralauth=True,
+ validate_as_user_ids=False,
+ paste_ids_or_names='test-specific-0\ntest-specific-1',
+ start_date='2015-11-01 00:00:00',
+ end_date='2015-11-30 00:00:00',
+ ))
+
+ assert_equal(response.status_code, 302)
+ assert_true(response.data.find('/reports/') >= 0)
+
+ # Wait for the task to get processed
+ time.sleep(3)
+
+ response = self.client.get('/reports/list/')
+ parsed = json.loads(response.data)
+ result_key = parsed['reports'][-1]['result_key']
+ task, report = get_celery_task(result_key)
+ assert_true(task is not None)
+
+ def test_report_create_and_result(self):
+
+ response = self.client.post('/reports/program-global-metrics',
data=dict(
+ name='TestCohort2',
+ project='wiki',
+ centralauth=True,
+ validate_as_user_ids=False,
+ paste_ids_or_names='Editor test-specific-0\nEditor
test-specific-1',
+ start_date='2015-11-01 00:00:00',
+ end_date='2015-11-30 00:00:00',
+ ))
+
+ assert_equal(response.status_code, 302)
+ assert_true(response.data.find('/reports/') >= 0)
+
+ # Wait for the task to get processed
+ time.sleep(3)
+
+ response = self.client.get('/reports/list/')
+ parsed = json.loads(response.data)
+ result_key = parsed['reports'][-1]['result_key']
+ task, report = get_celery_task(result_key)
+ assert_true(task is not None)
diff --git a/tests/test_models/test_run_program_metrics_report.py
b/tests/test_models/test_run_program_metrics_report.py
new file mode 100644
index 0000000..3b5206f
--- /dev/null
+++ b/tests/test_models/test_run_program_metrics_report.py
@@ -0,0 +1,120 @@
+from nose.tools import assert_equals, raises, assert_true, assert_is_not_none
+
+from tests.fixtures import QueueDatabaseTest, d
+from wikimetrics.models import RunProgramMetricsReport, ReportStore,
WikiUserStore
+from wikimetrics.utils import parse_pretty_date, format_pretty_date
+from wikimetrics.enums import Aggregation
+from wikimetrics.api import CohortService
+
+cohort_service = CohortService()
+
+
+def make_pending(report):
+ report.status = 'PENDING'
+
+
+class RunProgramMetricsReportWithInvalidCohort(QueueDatabaseTest):
+ def setUp(self):
+ QueueDatabaseTest.setUp(self)
+ self.common_cohort_1()
+
+ @raises(Exception)
+ def test_raises_unvalidated_cohort(self):
+ self.cohort.validated = False
+ jr = RunProgramMetricsReport(self.cohort.id,
+ parse_pretty_date('2015-01-01 00:00:00'),
+ parse_pretty_date('2015-01-31 00:00:00'),
+ self.owner_user_id)
+ jr.task.delay(jr).get()
+
+ def test_invalid_cohort_returns_failure(self):
+ self.cohort.validated = True
+ wikiusers = self.session.query(WikiUserStore) \
+ .filter(WikiUserStore.validating_cohort == self.cohort.id) \
+ .all()
+ # Valid cohorts have >=50% of the users valid, so we make half
+ # of them invalid to test
+ for wu in wikiusers[:(len(wikiusers) / 2 + 1)]:
+ wu.valid = False
+ self.session.commit()
+ jr = RunProgramMetricsReport(self.cohort.id,
+ parse_pretty_date('2015-01-01 00:00:00'),
+ parse_pretty_date('2015-01-31 00:00:00'),
+ self.owner_user_id)
+ results = jr.task.delay(jr).get()
+ self.session.commit()
+ result_key = self.session.query(ReportStore) \
+ .get(jr.persistent_id) \
+ .result_key
+ results = results[result_key]
+ assert_is_not_none(results['FAILURE'])
+
+
+class RunProgramMetricsReportTest(QueueDatabaseTest):
+ def setUp(self):
+ QueueDatabaseTest.setUp(self)
+ # registration for all the editors below
+ self.r = r = 20150101000000
+ # exactly 30 days after registration
+ self.m = m = 20150131000000
+ self.r_plus_30 = format_pretty_date(d(self.m))
+
+ self.create_test_cohort(
+ editor_count=5,
+ revisions_per_editor=8,
+ revision_timestamps=[
+ # this one will make 5 edits within 30 days of self.r_plus_30
+ [r + 1, r + 2, r + 3, r + 4, r + 5, m + 6, m + 7, m + 8],
+ # this one will make 3 edits within 30 days of self.r_plus_30
+ [r + 1, r + 2, r + 3, m + 4, m + 5, m + 6, m + 7, m + 8],
+ # this one will make 8 edits within 30 days of self.r_plus_30
+ [r + 1, r + 2, r + 3, r + 4, r + 5, r + 6, r + 7, r + 8],
+ # this one will make 0 edits within 30 days of self.r_plus_30
+ [m + 1, m + 2, m + 3, m + 4, m + 5, m + 6, m + 7, m + 8],
+ # this one will make the 5th edit right on self.r_plus_30
+ [r + 1, r + 2, r + 3, r + 4, m + 0, m + 6, m + 7, m + 8],
+ ],
+ user_registrations=r,
+ revision_lengths=10
+ )
+
+ @raises(Exception)
+ def test_empty_response(self):
+ """
+ Case where user tries to submit form with no cohorts / metrics
+ should be handled client side, on server side an exception will be
+ thrown if RunProgramMetricsReport object cannot be created
+ """
+ jr = RunProgramMetricsReport(None, None, None,
user_id=self.owner_user_id)
+ jr.task.delay(jr).get()
+
+ def test_basic_response(self):
+ jr = RunProgramMetricsReport(self.cohort.id,
+ parse_pretty_date('2015-01-01 00:00:00'),
+ parse_pretty_date('2015-01-31 00:00:00'),
+ self.owner_user_id)
+ results = jr.task.delay(jr).get()
+ self.session.commit()
+
+ # Make sure the cohort is validated
+ assert_true(self.cohort.validated)
+
+ result_key = self.session.query(ReportStore) \
+ .get(jr.persistent_id) \
+ .result_key
+ results = results[result_key]
+ assert_equals(
+ len(results[Aggregation.SUM]),
+ 6)
+ assert_equals(
+ results[Aggregation.SUM]['pages_created'], 1)
+ assert_equals(
+ results[Aggregation.SUM]['pages_edited'], 4)
+ assert_equals(
+ results[Aggregation.SUM]['newly_registered'], 0)
+ assert_equals(
+ results[Aggregation.SUM]['existing_editors'], 5)
+ assert_equals(
+ results[Aggregation.SUM]['rolling_active_editor'], 0)
+ assert_equals(
+ results[Aggregation.SUM]['bytes_added'], 10)
diff --git a/wikimetrics/api/__init__.py b/wikimetrics/api/__init__.py
index 15e69bb..7d7367a 100644
--- a/wikimetrics/api/__init__.py
+++ b/wikimetrics/api/__init__.py
@@ -3,7 +3,7 @@
from cohorts import *
from tags import *
from replication_lag import *
-
+from reports import *
from batch import *
# ignore flake8 because of F403 violation
diff --git a/wikimetrics/api/cohorts.py b/wikimetrics/api/cohorts.py
index e3d0c88..3b797cb 100644
--- a/wikimetrics/api/cohorts.py
+++ b/wikimetrics/api/cohorts.py
@@ -1,5 +1,5 @@
from sqlalchemy.orm.exc import NoResultFound
-from sqlalchemy import func
+from sqlalchemy import func, distinct
from wikimetrics.configurables import db
from wikimetrics.exceptions import Unauthorized, InvalidCohort, DatabaseError
from wikimetrics.models import cohort_classes, ValidatedCohort, WikiCohort
@@ -100,6 +100,16 @@
"""
db_session = db.get_session()
return db_session.query(CohortStore).get(cohort.id)
+
+ def fetch_by_id(self, cohort_id):
+ """
+ Fetches a CohortStore object from the database, without checking
permissions
+
+ Parameters
+ cohort : a logical Cohort object
+ """
+ db_session = db.get_session()
+ return db_session.query(CohortStore).get(cohort_id)
def get_cohort_by_name(self, db_session, name):
"""
@@ -276,7 +286,7 @@
session.add(search)
session.commit()
- def get_validation_info(self, cohort, session):
+ def get_validation_info(self, cohort, session, unique_users=False):
"""
Returns
If the cohort has no validation information, an empty dictionary
@@ -286,6 +296,7 @@
valid_count : number of valid users
total_count : number of users in the cohort
not_validated_count : users in the cohort not yet validated at all
+ percentage_valid : percentage of valid users in the cohort
"""
if not cohort.has_validation_info:
# make UI work as little as possible, do not return nulls
@@ -327,8 +338,21 @@
stats['total_count'] = sum(stats.values())
stats['validated_count'] = stats['valid_count'] +
stats['invalid_count']
-
+ stats['percentage_valid'] = (stats['valid_count'] /
+ float(stats['total_count']) * 100) \
+ if stats['total_count'] > 0 else 0
+ if unique_users:
+ stats['unique_users'] = self.get_unique_users(cohort, session)
return stats
+
+ def get_unique_users(self, cohort, session):
+ """
+ Returns:
+ the number of users in this cohort
+ """
+ return
session.query(func.count(distinct(WikiUserStore.mediawiki_username))) \
+ .filter(WikiUserStore.validating_cohort == cohort.id) \
+ .one()[0]
def get_membership(self, cohort, session):
wikiusers = (
@@ -432,3 +456,66 @@
)
db_session.add(cohort_tag)
db_session.commit()
+
+ def delete_owner_cohort(self, db_session, cohort_id):
+ """
+ Deletes the cohort and all associate records with that cohort if user
is the
+ owner.
+
+ Raises an error if it cannot delete the cohort.
+ """
+ db_session = db.get_session() if not db_session else db_session
+ # Check that there's only one owner and delete it
+ cu = db_session.query(CohortUserStore) \
+ .filter(CohortUserStore.cohort_id == cohort_id) \
+ .filter(CohortUserStore.role == CohortUserRole.OWNER) \
+ .delete()
+
+ if cu != 1:
+ db_session.rollback()
+ raise DatabaseError('No owner or multiple owners in cohort.')
+ else:
+ try:
+ # Delete all other non-owners from cohort_user
+ db_session.query(CohortUserStore) \
+ .filter(CohortUserStore.cohort_id == cohort_id) \
+ .delete()
+ db_session.query(CohortWikiUserStore) \
+ .filter(CohortWikiUserStore.cohort_id == cohort_id) \
+ .delete()
+
+ db_session.query(WikiUserStore) \
+ .filter(WikiUserStore.validating_cohort == cohort_id) \
+ .delete()
+
+ # Delete tags related to the cohort
+ db_session.query(CohortTagStore) \
+ .filter(CohortTagStore.cohort_id == cohort_id) \
+ .delete()
+
+ c = db_session.query(CohortStore) \
+ .filter(CohortStore.id == cohort_id) \
+ .delete()
+ if c < 1:
+ raise DatabaseError
+ except DatabaseError:
+ db_session.rollback()
+ raise DatabaseError('Owner attempt to delete a cohort failed.')
+
+ def delete_viewer_cohort(self, db_session, user_id, cohort_id):
+ """
+ Used when deleting a user's connection to a cohort. Currently used
when user
+ is a VIEWER of a cohort and want to remove that cohort from their list.
+
+ Raises exception when viewer is duplicated, nonexistent, or can not be
deleted.
+ """
+ db_session = db.get_session() if not db_session else db_session
+ cu = db_session.query(CohortUserStore) \
+ .filter(CohortUserStore.cohort_id == cohort_id) \
+ .filter(CohortUserStore.user_id == user_id) \
+ .filter(CohortUserStore.role == CohortUserRole.VIEWER) \
+ .delete()
+
+ if cu != 1:
+ db_session.rollback()
+ raise DatabaseError('Viewer attempt delete cohort failed.')
diff --git a/wikimetrics/api/reports.py b/wikimetrics/api/reports.py
new file mode 100644
index 0000000..3c8c96f
--- /dev/null
+++ b/wikimetrics/api/reports.py
@@ -0,0 +1,22 @@
+from batch import write_report_task
+from wikimetrics.models.storage.report import ReportStore
+
+
+class ReportService(object):
+
+ def write_report_to_file(self, report, results, db_session):
+ """
+ Helper to write results of a report to file.
+
+ The post_process method in RunReport and RunProgramMetricsReport have
+ the uses this functionality to write public reports, so we abstract it
here.
+ """
+ db_report = db_session.query(ReportStore).get(report.persistent_id)
+
+ data = db_report.get_json_result(results)
+
+ # code below schedules an async task on celery to write the file
+ report_id_to_write = report.persistent_id
+ if report.recurrent_parent_id is not None:
+ report_id_to_write = report.recurrent_parent_id
+ write_report_task.delay(report_id_to_write, report.created, data)
diff --git a/wikimetrics/controllers/cohorts.py
b/wikimetrics/controllers/cohorts.py
index af707e4..0274629 100644
--- a/wikimetrics/controllers/cohorts.py
+++ b/wikimetrics/controllers/cohorts.py
@@ -159,7 +159,7 @@
if not task_key:
validation['validation_status'] = 'UNKNOWN'
- #TODO do these defaults really make sense?
+ # TODO do these defaults really make sense?
validation['validated_count'] = number_of_wikiusers
validation['total_count'] = number_of_wikiusers
validation['valid_count'] = validation['total_count']
@@ -295,69 +295,6 @@
raise DatabaseError('No role found in cohort user.')
-def delete_viewer_cohort(session, cohort_id):
- """
- Used when deleting a user's connection to a cohort. Currently used when
user
- is a VIEWER of a cohort and want to remove that cohort from their list.
-
- Raises exception when viewer is duplicated, nonexistent, or can not be
deleted.
- """
- cu = session.query(CohortUserStore) \
- .filter(CohortUserStore.cohort_id == cohort_id) \
- .filter(CohortUserStore.user_id == current_user.id) \
- .filter(CohortUserStore.role == CohortUserRole.VIEWER) \
- .delete()
-
- if cu != 1:
- session.rollback()
- raise DatabaseError('Viewer attempt delete cohort failed.')
-
-
-def delete_owner_cohort(session, cohort_id):
- """
- Deletes the cohort and all associate records with that cohort if user is
the
- owner.
-
- Raises an error if it cannot delete the cohort.
- """
- # Check that there's only one owner and delete it
- cu = session.query(CohortUserStore) \
- .filter(CohortUserStore.cohort_id == cohort_id) \
- .filter(CohortUserStore.role == CohortUserRole.OWNER) \
- .delete()
-
- if cu != 1:
- session.rollback()
- raise DatabaseError('No owner or multiple owners in cohort.')
- else:
- try:
- # Delete all other non-owners from cohort_user
- session.query(CohortUserStore) \
- .filter(CohortUserStore.cohort_id == cohort_id) \
- .delete()
- session.query(CohortWikiUserStore) \
- .filter(CohortWikiUserStore.cohort_id == cohort_id) \
- .delete()
-
- session.query(WikiUserStore) \
- .filter(WikiUserStore.validating_cohort == cohort_id) \
- .delete()
-
- # Delete tags related to the cohort
- session.query(CohortTagStore) \
- .filter(CohortTagStore.cohort_id == cohort_id) \
- .delete()
-
- c = session.query(CohortStore) \
- .filter(CohortStore.id == cohort_id) \
- .delete()
- if c < 1:
- raise DatabaseError
- except DatabaseError:
- session.rollback()
- raise DatabaseError('Owner attempt to delete a cohort failed.')
-
-
@app.route('/cohorts/delete/<int:cohort_id>', methods=['POST'])
def delete_cohort(cohort_id):
"""
@@ -373,13 +310,13 @@
# Owner wants to delete, no other viewers or
# Owner wants to delete, have other viewers, delete from other
viewer's lists too
if owner_and_viewers >= 1 and role == CohortUserRole.OWNER:
- delete_owner_cohort(session, cohort_id)
+ g.cohort_service.delete_owner_cohort(session, cohort_id)
session.commit()
return json_redirect(url_for('cohorts_index'))
# Viewer wants to delete cohort from their list, doesn't delete cohort
from db;l,
elif owner_and_viewers > 1 and role == CohortUserRole.VIEWER:
- delete_viewer_cohort(session, cohort_id)
+ g.cohort_service.delete_viewer_cohort(session, current_user.id,
cohort_id)
session.commit()
return json_redirect(url_for('cohorts_index'))
diff --git a/wikimetrics/controllers/reports.py
b/wikimetrics/controllers/reports.py
index b62021e..a762faf 100644
--- a/wikimetrics/controllers/reports.py
+++ b/wikimetrics/controllers/reports.py
@@ -4,19 +4,19 @@
from StringIO import StringIO
from sqlalchemy import or_
from sqlalchemy.orm.exc import NoResultFound
-from flask import render_template, request, redirect, url_for, Response,
abort, g
+from flask import render_template, request, redirect, url_for, Response, g,
flash
from flask.ext.login import current_user
-from sqlalchemy.exc import SQLAlchemyError
from wikimetrics.configurables import app, db
+from wikimetrics.forms import ProgramMetricsForm
from wikimetrics.models import (
- Report, RunReport, ReportStore, WikiUserStore, WikiUserKey, TaskErrorStore
+ Report, RunReport, RunProgramMetricsReport, ReportStore,
+ WikiUserKey, TaskErrorStore, ValidateCohort
)
from wikimetrics.utils import (
- json_response, json_error, json_redirect, thirty_days_ago, stringify
+ json_response, json_error, json_redirect, thirty_days_ago
)
from wikimetrics.enums import Aggregation, TimeseriesChoices
-from wikimetrics.exceptions import UnauthorizedReportAccessError
-from wikimetrics.api import PublicReportFileManager, CohortService
+from wikimetrics.api import PublicReportFileManager, CohortService,
CentralAuthService
@app.before_request
@@ -29,8 +29,11 @@
app.logger,
app.absolute_path_to_app_root)
cohort_service = getattr(g, 'cohort_service', None)
+ centralauth_service = getattr(g, 'centralauth_service', None)
if cohort_service is None:
g.cohort_service = CohortService()
+ if centralauth_service is None:
+ g.centralauth_service = CentralAuthService()
@app.route('/reports/unset-public/<int:report_id>', methods=['POST'])
@@ -127,6 +130,42 @@
# TODO fix json_response to deal with ReportStore objects
return json_response(reports=reports)
+
+
+@app.route('/reports/program-global-metrics', methods=['GET', 'POST'])
+def program_metrics_reports_request():
+ """
+ Renders a page that facilitates kicking off a new ProgramMetrics report
+ """
+ form = ProgramMetricsForm()
+
+ if request.method == 'POST':
+ form = ProgramMetricsForm.from_request(request)
+ try:
+ if not form.validate():
+ flash('Please fix validation problems.', 'warning')
+
+ else:
+ form.parse_records()
+ vc = ValidateCohort.from_upload(form, current_user.id)
+ gm = RunProgramMetricsReport(vc.cohort_id,
+ form.start_date.data,
+ form.end_date.data,
+ current_user.id)
+ # Validate the cohort, and on success, call the
+ # RunProgramMetricsReport task. No parameters are passed
+ # from ValidateCohort to the report, so we are using
+ # an immutable task signature in the link param.
+ vc.task.apply_async([vc], link=gm.task.si(gm))
+ return redirect(url_for('reports_index'))
+ except Exception, e:
+ app.logger.exception(str(e))
+ flash('Server error while processing your request', 'error')
+
+ return render_template(
+ 'program_metrics_reports.html',
+ form=form,
+ )
def get_celery_task(result_key):
@@ -434,22 +473,22 @@
return json_response(message='Report scheduled for rerun')
-#@app.route('/reports/kill/<result_key>')
-#def report_kill(result_key):
- #return 'not implemented'
- #db_session = db.get_session()
- #db_report = db_session.query(ReportStore).get(result_key)
- #if not db_report:
- # return json_error('no task exists with id: {0}'.format(result_key))
- #celery_task = Report.task.AsyncResult(db_report.result_key)
- #app.logger.debug('revoking task: %s', celery_task.id)
- #from celery.task.control import revoke
- #celery_task.revoke()
- # TODO figure out how to terminate tasks. this throws an error
- # which I believe is related to
https://github.com/celery/celery/issues/1153
- # and which is fixed by a patch. however, I can't get things running
- # with development version
- #revoke(celery_task.id, terminate=True)
- #return json_response(status=celery_task.status)
+# @app.route('/reports/kill/<result_key>')
+# def report_kill(result_key):
+# return 'not implemented'
+# db_session = db.get_session()
+# db_report = db_session.query(ReportStore).get(result_key)
+# if not db_report:
+# return json_error('no task exists with id: {0}'.format(result_key))
+# celery_task = Report.task.AsyncResult(db_report.result_key)
+# app.logger.debug('revoking task: %s', celery_task.id)
+# from celery.task.control import revoke
+# celery_task.revoke()
+# TODO figure out how to terminate tasks. this throws an error
+# which I believe is related to
https://github.com/celery/celery/issues/1153
+# and which is fixed by a patch. however, I can't get things running
+# with development version
+# revoke(celery_task.id, terminate=True)
+# return json_response(status=celery_task.status)
-######## Internal functions not available via HTTP
################################
+# ####### Internal functions not available via HTTP
################################
diff --git a/wikimetrics/forms/__init__.py b/wikimetrics/forms/__init__.py
index bafa288..be63cfb 100644
--- a/wikimetrics/forms/__init__.py
+++ b/wikimetrics/forms/__init__.py
@@ -1,5 +1,6 @@
from secure_form import *
from cohort_upload import *
+from program_metrics_form import *
from fields import *
# ignore flake8 because of F403 violation
diff --git a/wikimetrics/forms/global_metrics_form.py
b/wikimetrics/forms/global_metrics_form.py
deleted file mode 100644
index a6b1d38..0000000
--- a/wikimetrics/forms/global_metrics_form.py
+++ /dev/null
@@ -1,28 +0,0 @@
-import uuid
-
-from flask import g
-from wikimetrics.configurables import db
-from wtforms import HiddenField
-from wikimetrics.forms.fields import BetterBooleanField, BetterDateTimeField
-from wtforms.validators import Required
-from wikimetrics.forms.validators import NotGreater
-from wikimetrics.utils import parse_username, thirty_days_ago, today,
format_pretty_date
-from cohort_upload import CohortUpload
-from validators import (
- CohortNameUnused, CohortNameLegalCharacters, ProjectExists, RequiredIfNot
-)
-
-
-class GlobalMetricsForm(CohortUpload):
- """
- Defines the fields necessary to upload inputs to calculate
- the global metrics
- """
- # Override cohort name and default project
- # The user for the Global API doesn't have to define these
- name = HiddenField(default='GlobalCohort_' +
str(uuid.uuid1()))
- project = HiddenField(default='enwiki')
- validate_as_user_ids = HiddenField(default='False')
- start_date = BetterDateTimeField(
- default=thirty_days_ago, validators=[NotGreater('end_date')])
- end_date = BetterDateTimeField(default=today)
diff --git a/wikimetrics/forms/program_metrics_form.py
b/wikimetrics/forms/program_metrics_form.py
new file mode 100644
index 0000000..3724e2d
--- /dev/null
+++ b/wikimetrics/forms/program_metrics_form.py
@@ -0,0 +1,23 @@
+import uuid
+
+from wtforms import HiddenField
+from wikimetrics.forms.fields import BetterDateTimeField
+from wikimetrics.forms.validators import NotGreater
+from wikimetrics.utils import thirty_days_ago, today
+from cohort_upload import CohortUpload
+
+
+class ProgramMetricsForm(CohortUpload):
+ """
+ Defines the fields necessary to upload inputs to calculate
+ the ProgramMetrics
+ """
+ # Override cohort name and default project
+ # The user for the ProgramMetrics API doesn't have to define these
+ name = HiddenField(
+ default='ProgramGlobalMetricsCohort_' + str(uuid.uuid1()))
+ project = HiddenField(default='enwiki')
+ validate_as_user_ids = HiddenField(default='False')
+ start_date = BetterDateTimeField(
+ default=thirty_days_ago, validators=[NotGreater('end_date')])
+ end_date = BetterDateTimeField(default=today)
diff --git a/wikimetrics/models/report_nodes/__init__.py
b/wikimetrics/models/report_nodes/__init__.py
index c26abca..9a52fba 100644
--- a/wikimetrics/models/report_nodes/__init__.py
+++ b/wikimetrics/models/report_nodes/__init__.py
@@ -4,6 +4,7 @@
from sum_aggregate_by_user_report import *
from report import *
from run_report import *
-
+from run_program_metrics_report import *
+from validate_program_metrics_report import *
# ignore flake8 because of F403 violation
# flake8: noqa
diff --git a/wikimetrics/models/report_nodes/report.py
b/wikimetrics/models/report_nodes/report.py
index 41c7b82..10c709e 100644
--- a/wikimetrics/models/report_nodes/report.py
+++ b/wikimetrics/models/report_nodes/report.py
@@ -168,7 +168,6 @@
"""
self.set_status(celery.states.STARTED, task_id=current_task.request.id)
results = []
-
if self.children:
try:
child_results = [child.run() for child in self.children]
diff --git a/wikimetrics/models/report_nodes/run_program_metrics_report.py
b/wikimetrics/models/report_nodes/run_program_metrics_report.py
new file mode 100644
index 0000000..177d555
--- /dev/null
+++ b/wikimetrics/models/report_nodes/run_program_metrics_report.py
@@ -0,0 +1,269 @@
+from celery.utils.log import get_task_logger
+from datetime import timedelta
+
+from wikimetrics.enums import Aggregation
+from wikimetrics.metrics import metric_classes
+from report import ReportNode
+from aggregate_report import AggregateReport
+from validate_program_metrics_report import ValidateProgramMetricsReport
+from sum_aggregate_by_user_report import SumAggregateByUserReport
+from wikimetrics.api import ReportService, CohortService
+from wikimetrics.configurables import db
+
+__all__ = ['RunProgramMetricsReport']
+task_logger = get_task_logger(__name__)
+cohort_service = CohortService()
+
+
+class RunProgramMetricsReport(ReportNode):
+ """
+ This class launches the reports for the 4 metrics needed to
+ calculate the global metrics, collects the results and
+ writes them to storage.
+ """
+
+ show_in_ui = True
+
+ def __init__(self,
+ cohort_id,
+ start_date,
+ end_date,
+ user_id=0):
+ """
+ Parameters:
+ cohort_id : ID of the cohort
+ start_date : Start date for the GlobalMetrics report
+ end_date : End date for the GlobalMetrics report
+ user_id : The user running this report
+ """
+ self.cohort_id = cohort_id
+ self.start_date = start_date
+ self.end_date = end_date
+ self.user_id = user_id
+ self.recurrent_parent_id = None
+ self.persistent_id = None
+
+ def run(self):
+ """
+ This initializes the cohort and any parameters not known at init time
+ for this ReportNode, and initializes and calls it's super class' run
method.
+
+ Raises:
+ KeyError if required parameters are missing
+ """
+ cohort_store_object = cohort_service.fetch_by_id(self.cohort_id)
+ # First make sure this is a valid cohort
+ if cohort_store_object is not None and cohort_store_object.validated:
+ self.cohort = cohort_service.convert(cohort_store_object)
+ validate_report = ValidateProgramMetricsReport(self.cohort,
+ db.get_session(),
+
user_id=self.user_id)
+ self.cohort.size = validate_report.unique_users
+ self.parameters = {
+ 'name': 'Program Global Metrics Report',
+ 'cohort': {
+ 'id': self.cohort.id,
+ 'name': self.cohort.name,
+ 'size': self.cohort.size,
+ },
+ 'user_id': self.user_id,
+ 'metric': {
+ 'name': 'ProgramGlobalMetrics',
+ 'end_date': self.end_date
+ },
+ }
+
+ super(RunProgramMetricsReport, self).__init__(
+ name=self.parameters['name'],
+ user_id=self.user_id,
+ parameters=self.parameters,
+ public=False,
+ recurrent=False,
+ recurrent_parent_id=self.recurrent_parent_id,
+ created=None,
+ store=True,
+ persistent_id=self.persistent_id,
+ )
+
+ if validate_report.valid():
+ self.children = [self.get_active_editors_report(),
+ self.get_new_editors_report(),
+ self.get_pages_created_report(),
+ self.get_pages_edited_report(),
+ self.get_bytes_added_report()]
+ else:
+ self.children = [validate_report]
+ return super(RunProgramMetricsReport, self).run()
+
+ else:
+ # This should never happen, unless it's a test where
RunProgramMetricsReport
+ # is being directly instantiated.
+ task_logger.error("Cohort not validated")
+ # Clean up cohort anyway
+ cohort_service.delete_owner_cohort(None, self.cohort_id)
+ raise Exception("Cohort not validated")
+
+ def finish(self, aggregated_results):
+ # Delete the cohort - we don't want to store these cohorts permanently
+ cohort_service.delete_owner_cohort(None, self.cohort_id)
+ if len(aggregated_results) > 1:
+ # Get all the results into the desired shape and return them
+ new_editors_count =
aggregated_results[1][Aggregation.SUM]['newly_registered']
+
+ # Existing editors can be calculated by subtracting the new
editors count
+ # from the size of the cohort, we calculate this here and add it to
+ # the list of aggregated_results.
+ existing_editors_count = self.cohort.size - new_editors_count
+ aggregated_results.append({Aggregation.SUM:
+ {'existing_editors':
existing_editors_count}})
+
+ # Manually rename absolute_sum to bytes_added - this looks silly,
there
+ # is probably a better way to do it
+ bytes_added_count =
aggregated_results[4][Aggregation.SUM]['absolute_sum']
+ aggregated_results[4] = ({Aggregation.SUM:
+ {'bytes_added': bytes_added_count}})
+
+ # At this point aggregated_results is a list of dicts that looks
like:
+ # [{Aggregation.SUM: {'newly_registered':3}},
+ # {Aggregation.SUM: {'existing_editors':3}}]
+ # We convert this into a single dict with key Sum,
+ # and all the submetrics as values
+ # Like: {Aggregation.SUM: {'newly_registered': 3,
'existing_editors': 3}}
+ submetrics = [s[Aggregation.SUM] for s in aggregated_results]
+ result = {}
+ for s in submetrics:
+ result.update(s)
+ return self.report_result({Aggregation.SUM: result})
+ else:
+ result = self.report_result(aggregated_results[0])
+ return result
+
+ def post_process(self, results):
+ """
+ If the report is public and this task went well,
+ it will create a file on disk asynchronously.
+
+ Results are of this form:
+
+ Parameters:
+ results : data to write to disk, in this form:
+ {'5cab8d55-da19-436f-b675-1d2a3fca3481':
+ {'Sum': {'pages_created': Decimal('0.0000')}}
+ }
+ """
+ if self.public is False:
+ return
+ rs = ReportService()
+ rs.write_report_to_file(self, results, db.get_session())
+
+ def get_aggregate_by_user_report(self, parameters):
+ metric_dict = parameters['metric']
+ metric = metric_classes[metric_dict['name']](**metric_dict)
+ return SumAggregateByUserReport(self.cohort,
+ metric,
+ parameters=parameters,
+ user_id=self.user_id)
+
+ def get_aggregate_report(self, parameters):
+ metric_dict = parameters['metric']
+ metric = metric_classes[metric_dict['name']](**metric_dict)
+ return AggregateReport(metric,
+ self.cohort,
+ options=parameters['options'],
+ user_id=self.user_id)
+
+ def get_active_editors_report(self):
+ return self.get_aggregate_by_user_report({
+ 'name': 'Active Editors report',
+ 'cohort': {
+ 'id': self.cohort.id,
+ 'name': self.cohort.name,
+ },
+ 'metric': {
+ 'name': 'RollingActiveEditor',
+ 'end_date': self.start_date,
+ 'individualResults': True,
+ },
+ })
+
+ def get_new_editors_report(self):
+ return self.get_aggregate_by_user_report({
+ 'name': 'New Editors report',
+ 'cohort': {
+ 'id': self.cohort.id,
+ 'name': self.cohort.name,
+ },
+ 'metric': {
+ 'name': 'NewlyRegistered',
+ 'start_date': self.end_date - timedelta(weeks=2),
+ 'end_date': self.end_date,
+ 'individualResults': True,
+ },
+ })
+
+ def get_pages_created_report(self):
+ return self.get_aggregate_report({
+ 'name': 'Pages created report',
+ 'cohort': {
+ 'id': self.cohort.id,
+ 'name': self.cohort.name,
+ },
+ 'metric': {
+ 'name': 'PagesCreated',
+ 'namespaces': [0],
+ 'start_date': self.start_date,
+ 'end_date': self.end_date,
+ 'aggregateResults': True,
+ 'aggregateSum': True,
+ },
+ 'options': {
+ 'aggregateResults': True,
+ 'aggregateSum': True,
+ }
+ })
+
+ def get_pages_edited_report(self):
+ return self.get_aggregate_by_user_report({
+ 'name': 'Pages Edited report',
+ 'cohort': {
+ 'id': self.cohort.id,
+ 'name': self.cohort.name,
+ },
+ 'metric': {
+ 'name': 'PagesEdited',
+ 'namespaces': [0],
+ 'start_date': self.start_date,
+ 'end_date': self.end_date,
+ 'aggregateResults': True,
+ 'aggregateSum': True,
+ },
+ 'options': {
+ 'aggregateResults': True,
+ 'aggregateSum': True,
+ }
+ })
+
+ def get_bytes_added_report(self):
+ return self.get_aggregate_report({
+ 'name': 'Bytes Added report',
+ 'cohort': {
+ 'id': self.cohort.id,
+ 'name': self.cohort.name,
+ },
+ 'metric': {
+ 'name': 'BytesAdded',
+ 'namespaces': [0],
+ 'start_date': self.start_date,
+ 'end_date': self.end_date,
+ 'aggregateResults': True,
+ 'aggregateSum': True,
+ 'absolute_sum': True,
+ 'negative_only_sum': False,
+ 'positive_only_sum': False,
+ 'net_sum': False,
+ },
+ 'options': {
+ 'aggregateResults': True,
+ 'aggregateSum': True,
+ }
+ })
diff --git a/wikimetrics/models/report_nodes/run_report.py
b/wikimetrics/models/report_nodes/run_report.py
index bcc3968..720bd36 100644
--- a/wikimetrics/models/report_nodes/run_report.py
+++ b/wikimetrics/models/report_nodes/run_report.py
@@ -18,7 +18,7 @@
from null_report import NullReport
from validate_report import ValidateReport
from metric_report import MetricReport
-from wikimetrics.api import write_report_task, CohortService
+from wikimetrics.api import ReportService, CohortService
from wikimetrics.utils import stringify
from wikimetrics.schedules import recurring_reports
@@ -145,16 +145,8 @@
if self.public is False:
return
- session = db.get_session()
- db_report = session.query(ReportStore).get(self.persistent_id)
-
- data = db_report.get_json_result(results)
-
- # code below schedules an async task on celery to write the file
- report_id_to_write = self.persistent_id
- if self.recurrent_parent_id is not None:
- report_id_to_write = self.recurrent_parent_id
- write_report_task.delay(report_id_to_write, self.created, data)
+ rs = ReportService()
+ rs.write_report_to_file(self, results, db.get_session())
# TODO, this method belongs on a different class and it should not be a
class method
@classmethod
diff --git a/wikimetrics/models/report_nodes/sum_aggregate_by_user_report.py
b/wikimetrics/models/report_nodes/sum_aggregate_by_user_report.py
index 142782c..b41813c 100644
--- a/wikimetrics/models/report_nodes/sum_aggregate_by_user_report.py
+++ b/wikimetrics/models/report_nodes/sum_aggregate_by_user_report.py
@@ -13,7 +13,7 @@
class SumAggregateByUserReport(ReportNode):
"""
A node responsible for aggregating the results of MultiProjectMetricReport
- by user, and formatting them as expected by RunGlobalReport. It
specifically
+ by user, and formatting them as expected by RunProgramMetricsReport. It
specifically
knows how to aggregate rolling active editor and newly registered metrics.
"""
show_in_ui = False
@@ -43,6 +43,7 @@
# The way of aggregating results accross different projects
# is applying the OR operator. Read more in the Wikitech docs:
#
https://wikitech.wikimedia.org/wiki/Analytics/Wikimetrics/Global_metrics
+ # (TODO: Rename page to Program_metrics)
aggregated_results = defaultdict(lambda: defaultdict(lambda: 0))
for key_str, result in results.iteritems():
key = WikiUserKey.fromstr(key_str)
@@ -57,4 +58,4 @@
summed_results[metric_name] += value
# Encapsulate the results to be consistent with other metrics.
- return {Aggregation.SUM: summed_results}
+ return {Aggregation.SUM: dict(summed_results)}
diff --git a/wikimetrics/models/report_nodes/validate_program_metrics_report.py
b/wikimetrics/models/report_nodes/validate_program_metrics_report.py
new file mode 100644
index 0000000..a205690
--- /dev/null
+++ b/wikimetrics/models/report_nodes/validate_program_metrics_report.py
@@ -0,0 +1,55 @@
+import celery
+from report import ReportLeaf
+from celery import current_task
+from wikimetrics.api import CohortService
+
+
+class ValidateProgramMetricsReport(ReportLeaf):
+ """
+ Checks if the cohort is valid (it should be validated, and >=50% of the
+ cohort members must be valid. If not, this ReportLeaf
+ can be added to a Report hierarchy and will report the appropriate errors
+ """
+
+ show_in_ui = False
+
+ def __init__(self, cohort, session, *args, **kwargs):
+ """
+ Parameters
+ cohort : an instance of CohortStore
+ session : DB session instance
+ """
+ super(ValidateProgramMetricsReport, self).__init__(*args, **kwargs)
+ self.cohort = cohort
+ self.session = session
+ self.validation_info = self.get_validation_info()
+ self.unique_users = self.validation_info['unique_users']
+ self.cohort_valid = self.is_cohort_valid()
+
+ def get_validation_info(self):
+ cohort_service = CohortService()
+ return cohort_service.get_validation_info(self.cohort,
+ self.session,
+ True)
+
+ def is_cohort_valid(self):
+ if self.validation_info['percentage_valid'] >= 50.0:
+ return True
+ else:
+ return False
+
+ def valid(self):
+ return self.cohort_valid
+
+ def run(self):
+ """
+ This will get executed if the instance is added into a Report node
hierarchy
+ It outputs failure messages due to any invalid configuration. None of
these
+ failures should happen unless the user tries to hack the system.
+ """
+ self.set_status(celery.states.STARTED, task_id=current_task.request.id)
+
+ message = ''
+ if not self.cohort_valid:
+ message += 'Cohort invalid: >=50% of the cohort members are not
valid'
+ return {'FAILURE': message or 'False'}
diff --git a/wikimetrics/static/js/globalReportUpload.js
b/wikimetrics/static/js/programMetricsReportUpload.js
similarity index 100%
rename from wikimetrics/static/js/globalReportUpload.js
rename to wikimetrics/static/js/programMetricsReportUpload.js
diff --git a/wikimetrics/templates/forms/global_report_upload.html
b/wikimetrics/templates/forms/program_metrics_report_upload.html
similarity index 94%
rename from wikimetrics/templates/forms/global_report_upload.html
rename to wikimetrics/templates/forms/program_metrics_report_upload.html
index c171b25..08ff09f 100644
--- a/wikimetrics/templates/forms/global_report_upload.html
+++ b/wikimetrics/templates/forms/program_metrics_report_upload.html
@@ -1,14 +1,14 @@
{% import 'forms/field_validation.html' as validation %}
<form enctype="multipart/form-data"
- action="{{url_for('global_reports_request')}}"
+ action="{{url_for('program_metrics_reports_request')}}"
method="POST"
class="upload-cohort form-horizontal">
{{ form.csrf_token }}
<div class="navbar-inner">
- <h3>Launch a global report</h3>
+ <h3>Launch a Program Global Metrics report</h3>
</div>
<!-- These fields are hidden since the user doesn't have to provide values
for them -->
<input id="name" name="name" type="hidden" value="{{form.name.data}}"/>
@@ -61,6 +61,6 @@
</div>
</div>
<div class="form-actions">
- <input type="submit" class="btn btn-primary" value="Launch Global
Report"/>
+ <input type="submit" class="btn btn-primary" value="Launch Program
Global Metrics Report"/>
</div>
-</form>
\ No newline at end of file
+</form>
diff --git a/wikimetrics/templates/global_reports.html
b/wikimetrics/templates/program_metrics_reports.html
similarity index 70%
rename from wikimetrics/templates/global_reports.html
rename to wikimetrics/templates/program_metrics_reports.html
index 9a60a27..7655c00 100644
--- a/wikimetrics/templates/global_reports.html
+++ b/wikimetrics/templates/program_metrics_reports.html
@@ -1,11 +1,11 @@
{% extends "layout.html" %}
{% block body %}
-{% include "forms/global_report_upload.html" %}
+{% include "forms/program_metrics_report_upload.html" %}
{% endblock %}
{% block scripts %}
<script
src="//ajax.aspnetcdn.com/ajax/jquery.validate/1.11.1/jquery.validate.min.js"></script>
<script src="{{ url_for('static',
filename='js/bootstrap-datetimepicker.min.js') }}"></script>
<script src="{{ url_for('static', filename='js/cohortUpload.js') }}"></script>
-<script src="{{ url_for('static', filename='js/globalReportUpload.js')
}}"></script>
+<script src="{{ url_for('static', filename='js/programMetricsReportUpload.js')
}}"></script>
{% endblock %}
\ No newline at end of file
--
To view, visit https://gerrit.wikimedia.org/r/253750
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: merged
Gerrit-Change-Id: I1d4a6bba9f81962d01990754299a22a077ba61df
Gerrit-PatchSet: 20
Gerrit-Project: analytics/wikimetrics
Gerrit-Branch: master
Gerrit-Owner: Madhuvishy <mviswanat...@wikimedia.org>
Gerrit-Reviewer: Madhuvishy <mviswanat...@wikimedia.org>
Gerrit-Reviewer: Mforns <mfo...@wikimedia.org>
Gerrit-Reviewer: Milimetric <dandree...@wikimedia.org>
Gerrit-Reviewer: Nuria <nu...@wikimedia.org>
Gerrit-Reviewer: jenkins-bot <>
_______________________________________________
MediaWiki-commits mailing list
MediaWiki-commits@lists.wikimedia.org
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
