Using Crystal Reports with Limesurvey

While Limesurvey is a very nice tool to create and manage web surveys, it’s a bit lacking in the reporting area. The functions are a limited and even if you want a quick and dirty, but presentable report, you must export the data and use other tools to format and present the answers to the survey.

The problem is the way how Limesurvey stores its data, more specifically the recorded answers. Every column in the table contains theanswer for the question identified by the column and each line (or record) contains a respondent. It seems like a very sensible way to store the respondent’s answers, but the trouble is that it’s very difficult to construct a generic report template in reporting tools with this kind of table schema. To be of any use, the table data have to be converted in a usable form. Furthermore, the schema used by Limesurvey is less than optimal and doesn’t use foreign keys to link tables. This complicates everything.

To convert the answer table data, I’ve written a little python script. It should be pretty plug-and-play. To use it, you have to change the database access variables. The script only takes one argument, the survey id. You can get the survey id in Limesurvey’s administration panel. It should follow the survey title when you select a survey in the administration panel.

When the script runs, it creates a table with 12 fields (or columns). Without going too much into details, it creates a group id column, a question id column, a question type column and an answer column. These columns contain the fundamental data and structure of the survey. From there the data should be in a usable format by reporting engines.

Crystal Reports need to import the tables and define the links between the tables. It needs to import these tables lime_surveys_languagesettings, lime_groups, limesurvey_crystal_report and lime_questions. You might need to install the MySQL OCDB connector to access your database from Crystal Reports. Don’t forget configure your firewall so it only access connection from your IP or use a SSH tunnel. As you probably know, it’s a very bad idea to expose an open MySQL port to the internet.

Here’s the code:
[sourcecode language=”python”]
”’
Created on 2009-10-21
olivier@olihb.com
”’
import sys
import MySQLdb as mysql
#database settings
dbport=3306
dbaddr=’server.com’
dbuser=’user’
dbdb=’database’
dbpass=’password’
dboutputtable=’limesurvey_crystal_report’
dbtableprefix=’lime_survey_’
#exception
class InputException(Exception):
pass
class Ddict(dict):
def __init__(self, default=None):
self.default = default
def __getitem__(self, key):
if not self.has_key(key):
self[key] = self.default()
return dict.__getitem__(self, key)
def xint(s):
if s is None: return None
try:
ret = int(s)
except ValueError:
ret=None
return ret
def xtupletodict(l):
d={}
for x,y in l:
d.setdefault(x,[]).append(y)
return d
def xtupleto2dict(l):
d = Ddict(dict)
for x,y,z in l:
d[x][y]=z
return d
class Question(object):
def __init__(self,data):
self.data=data
self.qid=data[0]
self.gid=data[1]
self.type=data[2]
self.sid=data[3]
def getColumns(db,dbinputsurveyid):
cursor = db.cursor()
cursor.execute("Show columns from "+dbtableprefix+dbinputsurveyid)
data = cursor.fetchall ()
return list(zip(*data)[0])
def getAllRespondants(db,dbinputsurveyid):
cursor = db.cursor()
cursor.execute("select * from "+dbtableprefix+dbinputsurveyid)
data = cursor.fetchall();
return data
def getQuestions(db,dbinputsurveyid):
cursor = db.cursor()
sql = """SELECT qid,gid,type,sid FROM lime_questions where sid=%s order by gid,question_order asc;""" % dbinputsurveyid
cursor.execute(sql)
data = cursor.fetchall();
return data
def getAllAnswers(db,dbinputsurveyid):
cursor = db.cursor()
sql="""SELECT lime_answers.qid,answer FROM lime_answers left join
lime_questions on lime_answers.qid=lime_questions.qid
where lime_questions.sid=%s;""" % dbinputsurveyid
cursor.execute(sql)
data = cursor.fetchall();
return data
def getAllAnswersCode(db,dbinputsurveyid):
cursor = db.cursor()
sql="""SELECT lime_answers.qid,code FROM lime_answers left join
lime_questions on lime_answers.qid=lime_questions.qid
where lime_questions.sid=%s;""" % dbinputsurveyid
cursor.execute(sql)
data = cursor.fetchall();
return data
def getAllAnswersOrder(db,dbinputsurveyid):
cursor = db.cursor()
sql="""SELECT lime_answers.qid,code,sortorder FROM lime_answers left join
lime_questions on lime_answers.qid=lime_questions.qid
where lime_questions.sid=%s;""" % dbinputsurveyid
cursor.execute(sql)
data = cursor.fetchall();
return data
def addAnswers(db,respondantid,surveyid,groupid,qstid,qsttype,answer,answtxt,answorder,subanswer,subanswtxt,subansworder):
cursor = db.cursor()
sql = "INSERT INTO "+dboutputtable+""" (
respondantid,surveyid,groupid,qstid,
qsttype,answer,answtext,answorder,subanwser,subanswtext,subansworder)
VALUE(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);"""
cursor.execute(sql,(respondantid,surveyid,groupid,qstid,qsttype,answer,answtxt,answorder,subanswer,subanswtxt,subansworder))
return
def createOrReplaceOutputTable(db):
cursor = db.cursor()
cursor.execute("""show tables like %s""",dboutputtable)
data = cursor.fetchone();
if data is None:
#Create table
sql = """CREATE TABLE %s (
id INT NOT NULL AUTO_INCREMENT,
respondantid INT NOT NULL,
surveyid INT NOT NULL,
groupid INT NOT NULL,
qstid INT NOT NULL,
qsttype TEXT NOT NULL,
answer VARCHAR(1000),
answtext TEXT,
answorder INT,
subanwser VARCHAR(1000),
subanswtext TEXT,
subansworder INT,
PRIMARY KEY ( id ) )
ENGINE = MYISAM CHARACTER SET utf8 COLLATE utf8_unicode_ci;""" % dboutputtable
cursor.execute(sql)
else:
#Empty table
cursor.execute("TRUNCATE " + dboutputtable)
return
def main(argv=None):
# extract input table
if argv==None:
argv=sys.argv
if len(argv)>2:
raise InputException("Needs a input survey id")
dbinputsurveyid = argv[1]
else:
dbinputsurveyid = argv
#connect to database
db=mysql.connect(host=dbaddr,user=dbuser,passwd=dbpass,db=dbdb,port=dbport)
#create or delete output table
createOrReplaceOutputTable(db)
#get data from database
questions=getQuestions(db, dbinputsurveyid)
columns=getColumns(db, dbinputsurveyid)
data=zip(*getAllRespondants(db, dbinputsurveyid))
answers = xtupletodict(getAllAnswers(db,dbinputsurveyid))
answersCode = xtupletodict(getAllAnswersCode(db,dbinputsurveyid))
answersOrder = xtupleto2dict(getAllAnswersOrder(db,dbinputsurveyid))
#looping over questions
for q in questions:
question = Question(q)
#List choice
if question.type == ‘L’ :
qstid = dbinputsurveyid+’X’+str(question.gid)+’X’+str(question.qid)
index=columns.index(qstid)
for i,a in enumerate(data[index]):
if a!=None and a!=”:
answer=int(data[index][i])
answerCodes=list(answersCode[question.qid])
answerindex = answerCodes.index(str(answer))
answerorder = answersOrder[question.qid][a]
addAnswers(db, data[0][i], question.sid, question.gid, question.qid, question.type, answers[question.qid][answerindex], None,answerorder, None, None,None)
# Multiple choices(M)
elif question.type ==’M’ :
for i,a in enumerate(answersCode[question.qid]):
qstid = dbinputsurveyid+’X’+str(question.gid)+’X’+str(question.qid)+a
qsttxt = answers[question.qid][i]
index=columns.index(qstid)
answerorder = answersOrder[question.qid][a]
for j,r in enumerate(data[index]):
if r == ‘Y’:
addAnswers(db, data[0][j] , question.sid, question.gid, question.qid, question.type, qsttxt, None, answerorder,1,None,None)
# Array-10(B)
elif question.type ==’B’ :
for i,a in enumerate(answersCode[question.qid]):
qstid = dbinputsurveyid+’X’+str(question.gid)+’X’+str(question.qid)+a
qsttxt = answers[question.qid][i]
index = columns.index(qstid)
for j,r in enumerate(data[index]):
if r!=None and r!=”:
addAnswers(db, data[0][j] , question.sid, question.gid, question.qid, question.type, qsttxt, None, answerorder, r, None,None)
#Text or Numeric
elif question.type ==’T’ or question.type ==’N’:
qstid=dbinputsurveyid+’X’+str(question.gid)+’X’+str(question.qid)
index=columns.index(qstid)
for i,a in enumerate(data[index]):
if a!=None and a!=”:
addAnswers(db, data[0][i], question.sid, question.gid, question.qid, question.type, None, a, None,None, None,None)
return
if __name__ == ‘__main__’:
main()
[/sourcecode]