EbookFoundation
/
fef-questionnaire
mirror of https://github.com/EbookFoundation/fef-questionnaire.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

normalize runid in db, fix bugs

dj111py38
eric 2016-11-03 11:43:08 -04:00
parent 72790b421d
commit 406aec970a
19 changed files with 299 additions and 148 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -14,6 +14,8 @@ It can be run either as a survey where subjects are solicited by email, or as a
In either mode, an instance can be linked to an arbitrary object via the django content-types module.
Try out the questionaire on the Unglue.it page for "Open Access Ebooks" https://unglue.it/work/82028/
History
-------

13
questionnaire/admin.py
View File

@ -61,7 +61,7 @@ class QuestionnaireAdmin(admin.ModelAdmin):
class RunInfoAdmin(admin.ModelAdmin):
list_display = ['random', 'runid', 'subject', 'created', 'emailsent', 'lastemailerror']
list_display = ['random', 'run', 'subject', 'created', 'emailsent', 'lastemailerror']
pass
@ -70,17 +70,18 @@ class RunInfoHistoryAdmin(admin.ModelAdmin):
class AnswerAdmin(admin.ModelAdmin):
search_fields = ['subject__email', 'runid', 'question__number', 'answer']
list_display = ['id', 'runid', 'subject', 'question']
list_filter = ['subject', 'runid']
ordering = [ 'id', 'subject', 'runid', 'question', ]
search_fields = ['subject__email', 'run__id', 'question__number', 'answer']
list_display = ['id', 'run', 'subject', 'question']
list_filter = ['subject', 'run__id']
ordering = [ 'id', 'subject', 'run__id', 'question', ]
from django.contrib import admin
# new in dj1.7
# @admin.register(Landing)
class LandingAdmin(admin.ModelAdmin):
pass
list_display = ('label', 'content_type', 'object_id', )
ordering = [ 'object_id', ]
adminsite.register(Questionnaire, QuestionnaireAdmin)
adminsite.register(Question, QuestionAdmin)

2
questionnaire/dependency_checker.py
View File

@ -121,7 +121,7 @@ def dep_check(expr, runinfo, answerdict):
else:
# retrieve from database
answer_object = Answer.objects.filter(question=check_question,
runid=runinfo.runid,
run=runinfo.run,
subject=runinfo.subject)
if answer_object:
actual_answer = answer_object[0].split_answer()

25
questionnaire/emails.py
View File

@ -9,7 +9,7 @@ from email.Header import Header
from email.Utils import formataddr, parseaddr
from django.core.mail import get_connection, EmailMessage
from django.contrib.auth.decorators import login_required
from django.template import Context, loader
from django.template import loader
from django.utils import translation
from django.conf import settings
from django.http import Http404, HttpResponse
@ -49,16 +49,11 @@ def _new_runinfo(subject, questionset):
"""
nextrun = subject.nextrun
runid = str(nextrun.year)
entries = list(RunInfo.objects.filter(runid=runid, subject=subject))
if len(entries)>0:
r = entries[0]
else:
r = RunInfo()
(run, created) = Run.objects.get_or_create(runid=runid)
(r, created) = RunInfo.objects.get_or_create(run=run, subject=subject)
if created:
r.random = _new_random(subject)
r.subject = subject
r.runid = runid
r.emailcount = 0
r.created = datetime.now()
r.questionset = questionset
r.save()
if nextrun.month == 2 and nextrun.day == 29: # the only exception?
@ -73,13 +68,13 @@ def _send_email(runinfo):
subject = runinfo.subject
translation.activate(subject.language)
tmpl = loader.get_template(settings.QUESTIONNAIRE_EMAIL_TEMPLATE)
c = Context()
c = {}
c['surname'] = subject.surname
c['givenname'] = subject.givenname
c['gender'] = subject.gender
c['email'] = subject.email
c['random'] = runinfo.random
c['runid'] = runinfo.runid
c['runid'] = runinfo.run.runid
c['created'] = runinfo.created
c['site'] = getattr(settings, 'QUESTIONNAIRE_URL', '(settings.QUESTIONNAIRE_URL not set)')
email = tmpl.render(c)
@ -143,18 +138,18 @@ def send_emails(request=None, qname=None):
WEEKAGO = time.time() - (60 * 60 * 24 * 7) # one week ago
outlog = []
for r in runinfos:
if r.runid.startswith('test:'):
if r.run.runid.startswith('test:'):
continue
if r.emailcount == -1:
continue
if r.emailcount == 0 or time.mktime(r.emailsent.timetuple()) < WEEKAGO:
try:
if _send_email(r):
outlog.append(u"[%s] %s, %s: OK" % (r.runid, r.subject.surname, r.subject.givenname))
outlog.append(u"[%s] %s, %s: OK" % (r.run.runid, r.subject.surname, r.subject.givenname))
else:
outlog.append(u"[%s] %s, %s: %s" % (r.runid, r.subject.surname, r.subject.givenname, r.lastemailerror))
outlog.append(u"[%s] %s, %s: %s" % (r.run.runid, r.subject.surname, r.subject.givenname, r.lastemailerror))
except Exception, e:
outlog.append("Exception: [%s] %s: %s" % (r.runid, r.subject.surname, str(e)))
outlog.append("Exception: [%s] %s: %s" % (r.run.runid, r.subject.surname, str(e)))
if request:
return HttpResponse("Sent Questionnaire Emails:\n "
+"\n ".join(outlog), content_type="text/plain")

23
questionnaire/fixtures/testQuestions.yaml
View File

@ -60,6 +60,21 @@
text_de: shown with testtag,
text_en: shown with testtag,
}
model: questionnaire.run
pk: 1
- fields: {
runid: 'test:test',
}
model: questionnaire.run
pk: 2
- fields: {
runid: 'test:withtags',
}
model: questionnaire.run
pk: 3
- fields: {
runid: 'test:withouttags',
}
model: questionnaire.questionset
pk: 4
- fields: {
@ -70,7 +85,7 @@
lastemailerror: null,
questionset: 1,
random: 'test:test',
runid: 'test:test',
run: 1,
state: '',
subject: 1
}
@ -84,7 +99,7 @@
lastemailerror: null,
questionset: 3,
random: 'test:withtags',
runid: 'test:withtags',
run: 2,
state: '',
tags: 'testtag',
subject: 1
@ -99,7 +114,7 @@
lastemailerror: null,
questionset: 3,
random: 'test:withouttags',
runid: 'test:withouttags',
run: 3,
state: '',
tags: '',
subject: 1
@ -108,7 +123,7 @@
pk: 3
- fields: {
completed: 2009-05-16,
runid: 'test:test',
run: 1,
subject: 1,
questionnaire: 1,
}

17
questionnaire/legacy_test_urls.py
View File

@ -1,18 +1,17 @@
# vim: set fileencoding=utf-8
import questionnaire
from django.conf.urls.defaults import *
from .views import *
from . import views
urlpatterns = patterns('',
urlpatterns = [
url(r'^q/(?P<runcode>[^/]+)/(?P<qs>\d+)/$',
'questionnaire.views.questionnaire', name='questionset'),
views.questionnaire, name='questionset'),
url(r'^q/([^/]+)/',
'questionnaire.views.questionnaire', name='questionset'),
views.questionnaire, name='questionset'),
url(r'^q/manage/csv/(\d+)/',
'questionnaire.views.export_csv'),
'views.export_csv),
url(r'^q/manage/sendemail/(\d+)/$',
'questionnaire.views.send_email'),
views.send_email),
url(r'^q/manage/manage/sendemails/$',
'questionnaire.views.send_emails'),
)
views.send_emails),
]

17
questionnaire/legacy_tests.py
View File

@ -40,8 +40,8 @@ class TypeTest(TestCase):
'question_12_multiple_1' : 'q12_choice1',# choice-multiple-freeform
'question_12_more_1' : 'blah', # choice-multiple-freeform
}
runinfo = self.runinfo = RunInfo.objects.get(runid='test:test')
self.runid = runinfo.runid
runinfo = self.runinfo = RunInfo.objects.get(run__runid='test:test')
self.runid = runinfo.run.runid
self.subject_id = runinfo.subject_id
@ -66,7 +66,7 @@ class TypeTest(TestCase):
response = self.client.get('/q/test:test/1/')
assert "Don't Know" in response.content
self.assertEqual(response.status_code, 200)
runinfo = RunInfo.objects.get(runid='test:test')
runinfo = RunInfo.objects.get(run__runid='test:test')
self.assertEqual(runinfo.subject.language, 'en')
response = self.client.get('/q/test:test/1/', {"lang" : "de"})
self.assertEqual(response.status_code, 302)
@ -74,7 +74,7 @@ class TypeTest(TestCase):
response = self.client.get('/q/test:test/1/')
assert "Weiss nicht" in response.content
self.assertEqual(response.status_code, 200)
runinfo = RunInfo.objects.get(runid='test:test')
runinfo = RunInfo.objects.get(run__runid='test:test')
self.assertEqual(runinfo.subject.language, 'de')
@ -105,9 +105,10 @@ class TypeTest(TestCase):
"POST complete answers for QuestionSet 1"
c = self.client
ansdict1 = self.ansdict1
runinfo = RunInfo.objects.get(runid='test:test')
runid = runinfo.random = runinfo.runid = '1real'
runinfo = RunInfo.objects.get(run__runid='test:test')
runid = runinfo.random = runinfo.run.runid = '1real'
runinfo.save()
runinfo.run.save()
response = c.get('/q/1real/1/')
self.assertEqual(response.status_code, 200)
@ -126,7 +127,7 @@ class TypeTest(TestCase):
self.assertEqual(response.status_code, 302)
self.assertEqual(response['Location'], 'http://testserver/')
self.assertEqual(RunInfo.objects.filter(runid='1real').count(), 0)
self.assertEqual(RunInfo.objects.filter(run__runid='1real').count(), 0)
# TODO: The format of these answers seems very strange to me. It was
# simpler before I changed it to get the test to work.
@ -148,7 +149,7 @@ class TypeTest(TestCase):
'12' : u'["q12_choice1", ["blah"]]',
}
for k, v in dbvalues.items():
ans = Answer.objects.get(runid=runid, subject__id=self.subject_id,
ans = Answer.objects.get(run__runid=runid, subject__id=self.subject_id,
question__number=k)
v = v.replace('\r', '\\r').replace('\n', '\\n')

36
questionnaire/migrations/0002_auto_20160929_1320.py Normal file
View File

@ -0,0 +1,36 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('questionnaire', '0001_initial'),
]
operations = [
migrations.CreateModel(
name='Run',
fields=[
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
('runid', models.CharField(max_length=32, null=True)),
],
),
migrations.AddField(
model_name='answer',
name='run',
field=models.ForeignKey(related_name='answers', to='questionnaire.Run', null=True),
),
migrations.AddField(
model_name='runinfo',
name='run',
field=models.ForeignKey(related_name='run_infos', to='questionnaire.Run', null=True),
),
migrations.AddField(
model_name='runinfohistory',
name='run',
field=models.ForeignKey(related_name='run_info_histories', to='questionnaire.Run', null=True),
),
]

36
questionnaire/migrations/0003_auto_20160929_1321.py Normal file
View File

@ -0,0 +1,36 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations, models
def models_to_migrate(apps):
return [
apps.get_model('questionnaire', 'RunInfo'),
apps.get_model('questionnaire', 'RunInfoHistory'),
apps.get_model('questionnaire', 'Answer'),
]
class Migration(migrations.Migration):
def move_runids(apps, schema_editor):
Run = apps.get_model('questionnaire', 'Run')
for model in models_to_migrate(apps):
for instance in model.objects.all():
(run, created) = Run.objects.get_or_create(runid=instance.runid)
instance.run = run
instance.save()
def unmove_runids(apps, schema_editor):
for model in models_to_migrate(apps):
for instance in model.objects.all():
instance.runid = instance.run.runid
instance.save()
dependencies = [
('questionnaire', '0002_auto_20160929_1320'),
]
operations = [
migrations.RunPython(move_runids, reverse_code=unmove_runids, hints={'questionnaire': 'Run'}),
]

47
questionnaire/migrations/0004_auto_20160929_1800.py Normal file
View File

@ -0,0 +1,47 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('questionnaire', '0003_auto_20160929_1321'),
]
operations = [
migrations.RemoveField(
model_name='runinfo',
name='runid',
),
migrations.RemoveField(
model_name='runinfohistory',
name='runid',
),
migrations.AlterField(
model_name='answer',
name='run',
field=models.ForeignKey(related_name='answers', default=1, to='questionnaire.Run'),
preserve_default=False,
),
migrations.AlterField(
model_name='runinfo',
name='run',
field=models.ForeignKey(related_name='run_infos', default=1, to='questionnaire.Run'),
preserve_default=False,
),
migrations.AlterField(
model_name='runinfohistory',
name='run',
field=models.ForeignKey(related_name='run_info_histories', to='questionnaire.Run'),
),
migrations.AlterIndexTogether(
name='answer',
index_together=set([('subject', 'run'), ('subject', 'run', 'id')]),
),
migrations.RemoveField(
model_name='answer',
name='runid',
),
]

26
questionnaire/models.py
View File

@ -5,7 +5,7 @@ import uuid
from datetime import datetime
from transmeta import TransMeta
from django.conf import settings
from django.contrib.contenttypes.generic import GenericForeignKey
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.db import models
from django.db.models.signals import post_save
@ -70,10 +70,10 @@ class Subject(models.Model):
return None
def history(self):
return RunInfoHistory.objects.filter(subject=self).order_by('runid')
return RunInfoHistory.objects.filter(subject=self).order_by('run__runid')
def pending(self):
return RunInfo.objects.filter(subject=self).order_by('runid')
return RunInfo.objects.filter(subject=self).order_by('run__runid')
class Meta:
index_together = [
@ -216,12 +216,14 @@ class QuestionSet(models.Model):
["questionnaire", "sortid"],
["sortid",]
]
class Run(models.Model):
runid = models.CharField(max_length=32, null=True)
class RunInfo(models.Model):
"Store the active/waiting questionnaire runs here"
subject = models.ForeignKey(Subject)
random = models.CharField(max_length=32) # probably a randomized md5sum
runid = models.CharField(max_length=32)
run = models.ForeignKey(Run, related_name='run_infos')
landing = models.ForeignKey(Landing, null=True, blank=True)
# questionset should be set to the first QuestionSet initially, and to null on completion
# ... although the RunInfo entry should be deleted then anyway.
@ -301,7 +303,7 @@ class RunInfo(models.Model):
return self.__cookiecache
def __unicode__(self):
return "%s: %s, %s" % (self.runid, self.subject.surname, self.subject.givenname)
return "%s: %s, %s" % (self.run.runid, self.subject.surname, self.subject.givenname)
class Meta:
verbose_name_plural = 'Run Info'
@ -311,7 +313,7 @@ class RunInfo(models.Model):
class RunInfoHistory(models.Model):
subject = models.ForeignKey(Subject)
runid = models.CharField(max_length=32)
run = models.ForeignKey(Run, related_name='run_info_histories')
completed = models.DateTimeField()
landing = models.ForeignKey(Landing, null=True, blank=True)
tags = models.TextField(
@ -325,11 +327,11 @@ class RunInfoHistory(models.Model):
questionnaire = models.ForeignKey(Questionnaire)
def __unicode__(self):
return "%s: %s on %s" % (self.runid, self.subject, self.completed)
return "%s: %s on %s" % (self.run.runid, self.subject, self.completed)
def answers(self):
"Returns the query for the answers."
return Answer.objects.filter(subject=self.subject, runid=self.runid)
return Answer.objects.filter(subject=self.subject, run=self.run)
class Meta:
verbose_name_plural = 'Run Info History'
@ -448,7 +450,7 @@ class Question(models.Model):
return self.type == 'comment'
def get_value_for_run_question(self, runid):
runanswer = Answer.objects.filter(runid=runid,question=self)
runanswer = Answer.objects.filter(run__runid=runid, question=self)
if len(runanswer) > 0:
return runanswer[0].answer
else:
@ -481,7 +483,7 @@ class Choice(models.Model):
class Answer(models.Model):
subject = models.ForeignKey(Subject, help_text = u'The user who supplied this answer')
question = models.ForeignKey(Question, help_text = u"The question that this is an answer to")
runid = models.CharField(u'RunID', help_text = u"The RunID (ie. year)", max_length=32)
run = models.ForeignKey(Run, related_name='answers')
answer = models.TextField()
def __unicode__(self):
@ -535,6 +537,6 @@ class Answer(models.Model):
class Meta:
index_together = [
['subject', 'runid'],
['subject', 'runid', 'id'],
['subject', 'run'],
['subject', 'run', 'id'],
]

8
questionnaire/page/views.py
View File

@ -1,5 +1,5 @@
# Create your views here.
from django.shortcuts import render_to_response
from django.shortcuts import render, render_to_response
from django.conf import settings
from django.template import RequestContext
from django import http
@ -10,14 +10,12 @@ def page(request, page_to_render):
try:
p = Page.objects.get(slug=page_to_render, public=True)
except Page.DoesNotExist:
return render_to_response("pages/{}.html".format(page_to_render),
return render(request, "pages/{}.html".format(page_to_render),
{ "request" : request,},
context_instance = RequestContext(request)
)
return render_to_response("page.html",
return render(request, "page.html",
{ "request" : request, "page" : p, },
context_instance = RequestContext(request)
)
def langpage(request, lang, page_to_trans):

2
questionnaire/qprocessors/choice.py
View File

@ -47,7 +47,7 @@ def process_choice(question, answer):
opt = answer['ANSWER'] or ''
if not opt and not question.type.endswith( '-optional'):
raise AnswerException(_(u'You must select an option'))
if opt == '_entry_' and question.type == 'choice-freeform':
if opt == '_entry_' and question.type.startswith('choice-freeform'):
opt = answer.get('comment','')
if not opt:
raise AnswerException(_(u'Field cannot be blank'))

4
questionnaire/templates/admin/questionnaire/subject/change_form.html
View File

@ -2,13 +2,13 @@
{% block after_field_sets %}
{% for x in original.pending %}
{% if forloop.first %}<h3>Pending:</h3><ul>{% endif %}
<li><b>{{ x.runid }}: created {{ x.created }}, last email sent {{ x.emailsent }}</b></li>
<li><b>{{ x.run.runid }}: created {{ x.created }}, last email sent {{ x.emailsent }}</b></li>
{% if forloop.last %}</ul>{% endif %}
{% endfor %}
{% for x in original.history %}
{% if forloop.first %}<h3>History:</h3><ul>{% endif %}
<li><b>{{ x.runid }}: completed {{ x.completed }}</b></li>
<li><b>{{ x.run.runid }}: completed {{ x.completed }}</b></li>
{% if forloop.last %}</ul>{% endif %}
{% endfor %}
{% endblock %}

2
questionnaire/templatetags/landings.py
View File

@ -1,5 +1,5 @@
import django.template
from django.template import Template, Context
from django.template import Template
register = django.template.Library()
@register.simple_tag(takes_context=True)

15
questionnaire/urls.py
View File

@ -4,8 +4,7 @@ from django.conf.urls import *
from .views import *
from .page.views import page, langpage
urlpatterns = patterns(
'',
urlpatterns = [
url(r'^$',
questionnaire, name='questionnaire_noargs'),
url(r'^csv/(?P<qid>\d+)/$',
@ -22,23 +21,21 @@ urlpatterns = patterns(
questionnaire, name='questionset'),
url(r'^q/manage/csv/(\d+)/',
export_csv, name="export_csv"),
)
]
if not use_session:
urlpatterns += patterns(
'',
urlpatterns += [
url(r'^(?P<runcode>[^/]+)/$',
questionnaire, name='questionnaire'),
url(r'^(?P<runcode>[^/]+)/(?P<qs>[-]{0,1}\d+)/prev/$',
redirect_to_prev_questionnaire,
name='redirect_to_prev_questionnaire'),
)
]
else:
urlpatterns += patterns(
'',
urlpatterns += [
url(r'^$',
questionnaire, name='questionnaire'),
url(r'^prev/$',
redirect_to_prev_questionnaire,
name='redirect_to_prev_questionnaire')
)
]

37
questionnaire/utils.py
View File

@ -1,4 +1,8 @@
#!/usr/bin/python
import codecs
import cStringIO
import csv
from django.conf import settings
try:
use_session = settings.QUESTIONNAIRE_USE_SESSION
@ -50,8 +54,39 @@ def get_runid_from_request(request):
if use_session:
return request.session.get('runcode', None)
else:
return request.runinfo.runid
return request.runinfo.run.runid
if __name__ == "__main__":
import doctest
doctest.testmod()
class UnicodeWriter:
"""
COPIED from http://docs.python.org/library/csv.html example:
A CSV writer which will write rows to CSV file "f",
which is encoded in the given encoding.
"""
def __init__(self, f, dialect=csv.excel, encoding="utf-8", **kwds):
# Redirect output to a queue
self.queue = cStringIO.StringIO()
self.writer = csv.writer(self.queue, dialect=dialect, **kwds)
self.stream = f
self.encoder = codecs.getincrementalencoder(encoding)()
def writerow(self, row):
self.writer.writerow([unicode(s).encode("utf-8") for s in row])
# Fetch UTF-8 output from the queue ...
data = self.queue.getvalue()
data = data.decode("utf-8")
# ... and reencode it into the target encoding
data = self.encoder.encode(data)
# write to the target stream
self.stream.write(data)
# empty queue
self.queue.truncate(0)
def writerows(self, rows):
for row in rows:
self.writerow(row)

2
questionnaire/utils_noncircular.py
View File

@ -6,7 +6,7 @@ def get_completed_answers_for_questions(questionnaire_id, question_list):
completed_questionnaire_runs = RunInfoHistory.objects.filter(questionnaire__id=questionnaire_id)
completed_answers = []
for run in completed_questionnaire_runs:
specific_answers = Answer.objects.filter(runid=run.runid, question_id__in=question_list)
specific_answers = Answer.objects.filter(run=run.run, question_id__in=question_list)
answer_set = []
for answer in specific_answers:
if answer.answer != '[]':

133
questionnaire/views.py
View File

@ -1,17 +1,27 @@
#!/usr/bin/python
# vim: set fileencoding=utf-8
import logging
import random
import re
import tempfile
from compat import commit_on_success, commit, rollback
from hashlib import md5
from uuid import uuid4
from django.http import HttpResponse, HttpResponseRedirect
from django.template import RequestContext
from django.core.urlresolvers import reverse
from django.core.cache import cache
from django.contrib.auth.decorators import permission_required
from django.shortcuts import render_to_response, get_object_or_404
from django.shortcuts import render, render_to_response, get_object_or_404
from django.views.generic.base import TemplateView
from django.db import transaction
from django.conf import settings
from datetime import datetime
from django.utils import translation
from django.utils.translation import ugettext_lazy as _
from . import QuestionProcessors
from . import questionnaire_start, questionset_start, questionset_done, questionnaire_done
from . import AnswerException
@ -21,15 +31,9 @@ from .models import *
from .parsers import *
from .parsers import BoolNot, BoolAnd, BoolOr, Checker
from .emails import _send_email, send_emails
from .utils import numal_sort, split_numal
from .utils import numal_sort, split_numal, UnicodeWriter
from .request_cache import request_cache
from .dependency_checker import dep_check
from compat import commit_on_success, commit, rollback
import logging
import random
from hashlib import md5
import re
from uuid import uuid4
try:
@ -46,7 +50,7 @@ except AttributeError:
def r2r(tpl, request, **contextdict):
"Shortcut to use RequestContext instead of Context in templates"
contextdict['request'] = request
return render_to_response(tpl, contextdict, context_instance=RequestContext(request))
return render(request, tpl, contextdict)
def get_runinfo(random):
@ -61,9 +65,9 @@ def get_question(number, questionset):
return res and res[0] or None
def delete_answer(question, subject, runid):
"Delete the specified question/subject/runid combination from the Answer table"
Answer.objects.filter(subject=subject, runid=runid, question=question).delete()
def delete_answer(question, subject, run):
"Delete the specified question/subject/run combination from the Answer table"
Answer.objects.filter(subject=subject, run=run, question=question).delete()
def add_answer(runinfo, question, answer_dict):
@ -77,7 +81,7 @@ def add_answer(runinfo, question, answer_dict):
answer = Answer()
answer.question = question
answer.subject = runinfo.subject
answer.runid = runinfo.runid
answer.run = runinfo.run
type = question.get_type()
@ -90,7 +94,7 @@ def add_answer(runinfo, question, answer_dict):
raise AnswerException("No Processor defined for question type %s" % type)
# first, delete all existing answers to this question for this particular user+run
delete_answer(question, runinfo.subject, runinfo.runid)
delete_answer(question, runinfo.subject, runinfo.run)
# then save the new answer to the database
answer.save(runinfo)
@ -479,7 +483,7 @@ def questionnaire(request, runcode=None, qs=None):
if not depparser.parse(depon):
# if check is not the same as answer, then we don't care
# about this question plus we should delete it from the DB
delete_answer(question, runinfo.subject, runinfo.runid)
delete_answer(question, runinfo.subject, runinfo.run)
if cd.get('store', False):
runinfo.set_cookie(question.number, None)
continue
@ -518,7 +522,7 @@ def questionnaire(request, runcode=None, qs=None):
def finish_questionnaire(request, runinfo, questionnaire):
hist = RunInfoHistory()
hist.subject = runinfo.subject
hist.runid = runinfo.runid
hist.run = runinfo.run
hist.completed = datetime.now()
hist.questionnaire = questionnaire
hist.tags = runinfo.tags
@ -532,11 +536,11 @@ def finish_questionnaire(request, runinfo, questionnaire):
redirect_url = questionnaire.redirect_url
for x, y in (('$LANG', lang),
('$SUBJECTID', runinfo.subject.id),
('$RUNID', runinfo.runid),):
('$RUNID', runinfo.run.runid),):
redirect_url = redirect_url.replace(x, str(y))
if runinfo.runid in ('12345', '54321') \
or runinfo.runid.startswith('test:'):
if runinfo.run.runid in ('12345', '54321') \
or runinfo.run.runid.startswith('test:'):
runinfo.questionset = QuestionSet.objects.filter(questionnaire=questionnaire).order_by('sortid')[0]
runinfo.save()
else:
@ -720,7 +724,7 @@ def show_questionnaire(request, runinfo, errors={}):
current_answers = []
if debug_questionnaire:
current_answers = Answer.objects.filter(subject=runinfo.subject, runid=runinfo.runid).order_by('id')
current_answers = Answer.objects.filter(subject=runinfo.subject, run=runinfo.run).order_by('id')
r = r2r("questionnaire/questionset.html", request,
@ -826,70 +830,51 @@ def _table_headers(questions):
columns.append(q.number)
return columns
default_extra_headings = [u'subject', u'run id']
def default_extra_entries(subject, run):
return ["%s/%s" % (subject.id, subject.ip_address), run.id]
@permission_required("questionnaire.export")
def export_csv(request, qid): # questionnaire_id
def export_csv(request, qid,
extra_headings=default_extra_headings,
extra_entries=default_extra_entries,
answer_filter=None,
filecode=0,
):
"""
For a given questionnaire id, generaete a CSV containing all the
For a given questionnaire id, generate a CSV containing all the
answers for all subjects.
qid -- questionnaire_id
extra_headings -- customize the headings for extra columns,
extra_entries -- function returning a list of extra column entries,
answer_filter -- custom filter for the answers
filecode -- code for filename
"""
import tempfile, csv, cStringIO, codecs
from django.core.servers.basehttp import FileWrapper
class UnicodeWriter:
"""
COPIED from http://docs.python.org/library/csv.html example:
A CSV writer which will write rows to CSV file "f",
which is encoded in the given encoding.
"""
def __init__(self, f, dialect=csv.excel, encoding="utf-8", **kwds):
# Redirect output to a queue
self.queue = cStringIO.StringIO()
self.writer = csv.writer(self.queue, dialect=dialect, **kwds)
self.stream = f
self.encoder = codecs.getincrementalencoder(encoding)()
def writerow(self, row):
self.writer.writerow([unicode(s).encode("utf-8") for s in row])
# Fetch UTF-8 output from the queue ...
data = self.queue.getvalue()
data = data.decode("utf-8")
# ... and reencode it into the target encoding
data = self.encoder.encode(data)
# write to the target stream
self.stream.write(data)
# empty queue
self.queue.truncate(0)
def writerows(self, rows):
for row in rows:
self.writerow(row)
fd = tempfile.TemporaryFile()
questionnaire = get_object_or_404(Questionnaire, pk=int(qid))
headings, answers = answer_export(questionnaire)
headings, answers = answer_export(questionnaire, answer_filter=answer_filter)
writer = UnicodeWriter(fd)
writer.writerow([u'subject', u'runid'] + headings)
for subject, runid, answer_row in answers:
row = ["%s/%s" % (subject.id, subject.state), runid] + [
writer.writerow(extra_headings + headings)
for subject, run, answer_row in answers:
row = extra_entries(subject, run) + [
a if a else '--' for a in answer_row]
writer.writerow(row)
response = HttpResponse(FileWrapper(fd), content_type="text/csv")
response['Content-Length'] = fd.tell()
response['Content-Disposition'] = 'attachment; filename="export-%s.csv"' % qid
fd.seek(0)
response = HttpResponse(fd, content_type="text/csv")
response['Content-Length'] = fd.tell()
response['Content-Disposition'] = 'attachment; filename="answers-%s-%s.csv"' % (qid, filecode)
return response
def answer_export(questionnaire, answers=None):
def answer_export(questionnaire, answers=None, answer_filter=None):
"""
questionnaire -- questionnaire model for export
answers -- query set of answers to include in export, defaults to all
answer_filter -- filter for the answers
Return a flat dump of column headings and all the answers for a
questionnaire (in query set answers) in the form (headings, answers)
@ -911,9 +896,11 @@ def answer_export(questionnaire, answers=None):
"""
if answers is None:
answers = Answer.objects.all()
if answer_filter:
answers = answer_filter(answers)
answers = answers.filter(
question__questionset__questionnaire=questionnaire).order_by(
'subject', 'runid', 'question__questionset__sortid', 'question__number')
'subject', 'run__runid', 'question__questionset__sortid', 'question__number')
answers = answers.select_related()
questions = Question.objects.filter(
questionset__questionnaire=questionnaire)
@ -930,11 +917,12 @@ def answer_export(questionnaire, answers=None):
runid = subject = None
out = []
row = []
run = None
for answer in answers:
if answer.runid != runid or answer.subject != subject:
if answer.run != run or answer.subject != subject:
if row:
out.append((subject, runid, row))
runid = answer.runid
out.append((subject, run, row))
run = answer.run
subject = answer.subject
row = [""] * len(headings)
ans = answer.split_answer()
@ -958,7 +946,7 @@ def answer_export(questionnaire, answers=None):
row[col] = choice
# and don't forget about the last one
if row:
out.append((subject, runid, row))
out.append((subject, run, row))
return headings, out
@ -1053,9 +1041,8 @@ def generate_run(request, questionnaire_id, subject_id=None, context={}):
str_to_hash += settings.SECRET_KEY
key = md5(str_to_hash).hexdigest()
landing = context.get('landing', None)
run = RunInfo(subject=su, random=key, runid=key, questionset=qs, landing=landing)
run.save()
r = Run.objects.create(runid=key)
run = RunInfo.objects.create(subject=su, random=key, run=r, questionset=qs, landing=landing)
if not use_session:
kwargs = {'runcode': key}
else: