initial working state
This commit is contained in:
Ville Rantanen
301
manager.py
Normal file
301
manager.py
Normal file
@@ -0,0 +1,301 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from datetime import datetime
|
||||
from utils import *
|
||||
import argparse
|
||||
import os
|
||||
import random
|
||||
import sqlite3
|
||||
import string
|
||||
import sys
|
||||
|
||||
|
||||
def insert_token(db, name, token):
|
||||
table_name = get_voter_table_name(name)
|
||||
cur = db.cursor()
|
||||
|
||||
cur.execute("""
|
||||
INSERT INTO `%s` VALUES (
|
||||
?,
|
||||
'false'
|
||||
);
|
||||
"""%( table_name, ),
|
||||
(
|
||||
token,
|
||||
)
|
||||
)
|
||||
db.commit()
|
||||
|
||||
|
||||
def add_token(options):
|
||||
if not is_key(options.name, options):
|
||||
raise Exception("%s does not exist, or is not a valid question set name"%( options.name, ))
|
||||
db = open_db(options.db)
|
||||
create_voter_table(db, options.name)
|
||||
for i in range(options.number):
|
||||
N = 32
|
||||
token = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(N))
|
||||
insert_token(db, options.name, token)
|
||||
print("%s/vote/%s/%s"%(
|
||||
options.prefix,
|
||||
options.name,
|
||||
token
|
||||
))
|
||||
|
||||
def open_db(db):
|
||||
return sqlite3.connect(db)
|
||||
|
||||
|
||||
def parse_options(database, questions):
|
||||
path_self = os.path.realpath(
|
||||
os.path.dirname(
|
||||
os.path.realpath(__file__)
|
||||
)
|
||||
)
|
||||
default_db = os.path.join(path_self, database)
|
||||
default_questions = os.path.join(path_self, questions)
|
||||
|
||||
parser = argparse.ArgumentParser(description='aBot vote manager')
|
||||
parser.add_argument('--db', action="store", dest="db", default = default_db,
|
||||
help = "Path to database [%(default)s]")
|
||||
parser.add_argument('--questions', action="store", dest="questions", default = default_questions,
|
||||
help = "Path to question folder [%(default)s]")
|
||||
|
||||
|
||||
subparsers = parser.add_subparsers(help='sub-command help', dest='subparser_name')
|
||||
## tokens
|
||||
parser_token = subparsers.add_parser('token', help = "Manage tokens")
|
||||
parser_token.add_argument(
|
||||
'-n',
|
||||
action="store",
|
||||
dest="number",
|
||||
default = 1, type = int,
|
||||
help = "Number of tokens to generate"
|
||||
)
|
||||
parser_token.add_argument(
|
||||
'--prefix',
|
||||
action="store",
|
||||
dest="prefix",
|
||||
default = "",
|
||||
help = "Prefix tokens with the server URL to automate emails etc.."
|
||||
)
|
||||
parser_token.add_argument(
|
||||
dest = "name",
|
||||
help = "Name of the question set"
|
||||
)
|
||||
|
||||
## summary of vote
|
||||
parser_summary = subparsers.add_parser('summary', help = "Vote results")
|
||||
parser_summary.add_argument(
|
||||
'--tsv',
|
||||
action="store_true",
|
||||
dest="tsv",
|
||||
default = False,
|
||||
help = "TSV output"
|
||||
)
|
||||
parser_summary.add_argument(
|
||||
dest = "name",
|
||||
help = "Name of the question set"
|
||||
)
|
||||
## clear
|
||||
parser_clear = subparsers.add_parser('clear', help = "Delete results")
|
||||
parser_clear.add_argument(
|
||||
'--really',
|
||||
action="store_true",
|
||||
dest="really",
|
||||
default = False,
|
||||
help = "Really delete results for the vote"
|
||||
)
|
||||
parser_clear.add_argument(
|
||||
'--tokens',
|
||||
action="store_true",
|
||||
dest="tokens",
|
||||
default = False,
|
||||
help = "Delete tokens too"
|
||||
)
|
||||
parser_clear.add_argument(
|
||||
dest = "name",
|
||||
help = "Name of the question set"
|
||||
)
|
||||
|
||||
## list
|
||||
parser_list = subparsers.add_parser('list', help = "List all question set names")
|
||||
|
||||
parsed = parser.parse_args()
|
||||
if parsed.subparser_name == None:
|
||||
parser.print_help()
|
||||
return parsed
|
||||
|
||||
|
||||
def list_question_sets(options):
|
||||
for f in os.listdir(options.questions):
|
||||
if not f.endswith(".txt"):
|
||||
continue
|
||||
print(f[0:-4])
|
||||
|
||||
|
||||
def summary(options):
|
||||
if not is_key(options.name, options):
|
||||
raise Exception("%s does not exist, or is not a valid question set name"%( options.name, ))
|
||||
db = open_db(options.db)
|
||||
cur = db.cursor()
|
||||
|
||||
cur.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name=?;",
|
||||
( options.name, )
|
||||
)
|
||||
matching_tables = cur.fetchall()
|
||||
if len(matching_tables) == 0:
|
||||
print("No votes yet")
|
||||
return
|
||||
token_table = get_voter_table_name(options.name)
|
||||
cur.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name=?;",
|
||||
( token_table, )
|
||||
)
|
||||
matching_tables = cur.fetchall()
|
||||
if len(matching_tables) == 0:
|
||||
used_tokens = 0
|
||||
unused_tokens = 0
|
||||
else:
|
||||
cur.execute(
|
||||
"SELECT count(*) FROM `%s` WHERE answered = 'true'"%(
|
||||
token_table,
|
||||
)
|
||||
)
|
||||
used_tokens = cur.fetchall()[0][0]
|
||||
cur.execute(
|
||||
"SELECT count(*) FROM `%s` WHERE answered = 'false'"%(
|
||||
token_table,
|
||||
)
|
||||
)
|
||||
unused_tokens = cur.fetchall()[0][0]
|
||||
tokens = {
|
||||
'unused': unused_tokens,
|
||||
'used': used_tokens,
|
||||
'total': used_tokens + unused_tokens
|
||||
}
|
||||
|
||||
questions = []
|
||||
answers = {}
|
||||
cur.execute(
|
||||
"SELECT question, answer, answer_type FROM `%s`"%(
|
||||
options.name,
|
||||
)
|
||||
)
|
||||
for row in cur:
|
||||
if row[0] not in answers.keys():
|
||||
questions.append(row[0])
|
||||
answers[row[0]] = {
|
||||
'answers': {},
|
||||
'answer_type': row[2]
|
||||
}
|
||||
|
||||
if row[1] not in answers[row[0]]['answers'].keys():
|
||||
answers[row[0]]['answers'][row[1]] = 0
|
||||
answers[row[0]]['answers'][row[1]] += 1
|
||||
|
||||
try:
|
||||
if options.tsv:
|
||||
summary_tsv(questions, answers, tokens)
|
||||
else:
|
||||
summary_list(questions, answers, tokens)
|
||||
except AttributeError:
|
||||
summary_list(questions, answers, tokens)
|
||||
|
||||
def summary_list(questions, answers, tokens):
|
||||
print(
|
||||
"""# Tokens for this question set:
|
||||
# used: {used}, unused: {unused}, total: {total}""".format_map(tokens)
|
||||
)
|
||||
|
||||
for q,a in zip(questions, answers):
|
||||
sum_answers = sum([answers[q]['answers'][x] for x in answers[q]['answers']])
|
||||
print("\n%s\n# Answers total: %d"%( q, sum_answers, ))
|
||||
sorted_answers = sorted(
|
||||
[(x, answers[q]['answers'][x]) for x in answers[q]['answers']],
|
||||
key = lambda i: -i[1]
|
||||
)
|
||||
for answer in sorted_answers:
|
||||
prefix = ""
|
||||
postfix = ""
|
||||
if answers[q]['answer_type'] == "single":
|
||||
prefix = "- "
|
||||
postfix = ": %d (%d%%) "%( answer[1], 100 * float(answer[1])/sum_answers)
|
||||
if answers[q]['answer_type'] == "multiple":
|
||||
prefix = "+ "
|
||||
postfix = ": %d (%d%%) "%( answer[1], 100 * float(answer[1])/sum_answers)
|
||||
if answers[q]['answer_type'] == "open":
|
||||
prefix = "----\n> "
|
||||
postfix = ""
|
||||
|
||||
print("%s%s%s"%(
|
||||
prefix,
|
||||
answer[0],
|
||||
postfix
|
||||
))
|
||||
|
||||
|
||||
def summary_tsv(questions, answers, tokens):
|
||||
print(
|
||||
'''Tokens\tUsed\tUnused\tTotal
|
||||
""\t"{used}"\t"{unused}"\t"{total}"'''.format_map(tokens)
|
||||
)
|
||||
print('"Question"\t"Question type"\t"Answer"\t"Count"')
|
||||
good_characters = dict.fromkeys(range(32))
|
||||
for q,a in zip(questions, answers):
|
||||
sum_answers = sum([answers[q]['answers'][x] for x in answers[q]['answers']])
|
||||
print('"%s"\t"%s"\t""\t"%d"'%( q, answers[q]['answer_type'], sum_answers, ))
|
||||
sorted_answers = sorted(
|
||||
[(x, answers[q]['answers'][x]) for x in answers[q]['answers']],
|
||||
key = lambda i: -i[1]
|
||||
)
|
||||
for answer in sorted_answers:
|
||||
|
||||
print('""\t""\t""%s\t"%d"'%(
|
||||
answer[0].translate(good_characters),
|
||||
answer[1],
|
||||
))
|
||||
|
||||
|
||||
def clear_votes(options):
|
||||
if not is_key(options.name, options):
|
||||
raise Exception("%s does not exist, or is not a valid question set name"%( options.name, ))
|
||||
summary(options)
|
||||
if not options.really:
|
||||
print("\nNot really deleting results")
|
||||
sys.exit(0)
|
||||
db = open_db(options.db)
|
||||
cur = db.cursor()
|
||||
|
||||
cur.execute(
|
||||
"DROP TABLE IF EXISTS `%s`"%( options.name, )
|
||||
)
|
||||
db.commit()
|
||||
print("\nDeleted votes for %s"%( options.name, ))
|
||||
|
||||
|
||||
def clear_tokens(options):
|
||||
if not is_key(options.name, options):
|
||||
raise Exception("%s does not exist, or is not a valid question set name"%( options.name, ))
|
||||
db = open_db(options.db)
|
||||
cur = db.cursor()
|
||||
cur.execute(
|
||||
"DROP TABLE IF EXISTS `%s`"%( get_voter_table_name(options.name,) )
|
||||
)
|
||||
db.commit()
|
||||
print("\nDeleted tokens for %s"%( options.name, ))
|
||||
|
||||
def main(database, questions):
|
||||
options = parse_options(database, questions)
|
||||
if options.subparser_name == "list":
|
||||
list_question_sets(options)
|
||||
if options.subparser_name == "token":
|
||||
add_token(options)
|
||||
if options.subparser_name == "summary":
|
||||
summary(options)
|
||||
if options.subparser_name == "clear":
|
||||
clear_votes(options)
|
||||
if options.tokens:
|
||||
clear_tokens(options)
|
||||
|
||||
Reference in New Issue
Block a user