# # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # import hashlib import json import os import time from lib import constants class FileBasedBackend: homedir = None def __init__(self, config): self.homedir = config.get("general", "homedir") def document_exists(self, election, *issue): "Returns True if an election/issue exists, False otherwise" elpath = os.path.join(self.homedir, "issues", election) if issue: elpath += "/" + issue[0] + ".json" return os.path.isfile(elpath) else: return os.path.isdir(elpath) def get_basedata(self, election): "Get base data from an election" elpath = os.path.join(self.homedir, "issues", election) if os.path.isdir(elpath): with open(elpath + "/basedata.json", "r") as f: data = f.read() f.close() basedata = json.loads(data) if hideHash and 'hash' in basedata: del basedata['hash'] basedata['id'] = election return basedata return None def close(self, election, reopen = False): "Mark an election as closed" basedata = self.get_basedata(election) elpath = os.path.join(self.homedir, "issues", election) if reopen: basedata['closed'] = False else: basedata['closed'] = True with open(elpath + "/basedata.json", "w") as f: f.write(json.dumps(basedata)) f.close() def issue_get(self, electionID, issueID): "Get JSON data from an issue" issuedata = None ihash = "" issuepath = os.path.join(self.homedir, "issues", electionID, issueID) + ".json" if os.path.isfile(issuepath): with open(issuepath, "r") as f: data = f.read() ihash = hashlib.sha224(data).hexdigest() f.close() issuedata = json.loads(data) return issuedata, ihash def votes_get(self, electionID, issueID): "Read votes from the vote file" rvotes = getVotesRaw(electionID, issueID) votes = {} for key in rvotes: votes[key] = rvotes[key]['vote'] return {} def votes_get_raw(self, electionID, issueID): issuepath = os.path.join(self.homedir, "issues", electionID, issueID) + ".json.votes" if os.path.isfile(issuepath): with open(issuepath, "r") as f: votes = json.loads(f.read()) f.close() return votes return {} def vote_history(self, electionID, issueID): issuepath = os.path.join(self.homedir, "issues", electionID, issueID) + ".json.history" if os.path.isfile(issuepath): with open(issuepath, "r") as f: votes = json.loads(f.read()) f.close() return votes return [] def election_create(self, eid, basedata): elpath = os.path.join(self.homedir, "issues", eid) os.mkdir(elpath) with open(elpath + "/basedata.json", "w") as f: f.write(json.dumps(basedata)) f.close() with open(elpath + "/voters.json", "w") as f: f.write("{}") f.close() def election_update(self, electionID, basedata): elpath = os.path.join(self.homedir, "issues", electionID) with open(elpath + "/basedata.json", "w") as f: f.write(json.dumps(basedata)) f.close() def issue_update(self, electionID, issueID, issueData): issuepath = os.path.join(self.homedir, "issues", electionID, issueID) + ".json" with open(issuepath, "w") as f: f.write(json.dumps(issueData)) f.close() def issue_list(self, election): "List all issues in an election" issues = [] elpath = os.path.join(self.homedir, "issues", election) if os.path.isdir(elpath): issues = [f.strip(".json") for f in os.listdir(elpath) if os.path.isfile(os.path.join(elpath, f)) and f != "basedata.json" and f != "voters.json" and f.endswith(".json")] return issues def election_list(self): "List all elections" elections = [] path = os.path.join(self.homedir, "issues") if os.path.isdir(path): elections = [f for f in os.listdir(path) if os.path.isdir(os.path.join(path, f))] return elections def vote(self, electionID, issueID, uid, vote): "Casts a vote on an issue" votes = {} now = time.time() issuepath = os.path.join(self.homedir, "issues", electionID, issueID) + ".json" if os.path.isfile(issuepath + ".votes"): with open(issuepath + ".votes", "r") as f: votes = json.loads(f.read()) f.close() votes[uid] = { 'vote': vote, 'timestamp': now } with open(issuepath + ".votes", "w") as f: f.write(json.dumps(votes)) f.close() # history backlog vote_history = [] issuepath = os.path.join(self.homedir, "issues", electionID, issueID) + ".json" if os.path.isfile(issuepath + ".history"): with open(issuepath + ".history", "r") as f: vote_history = json.loads(f.read()) f.close() vote_history.append({ 'key': uid, 'data': { 'vote': vote, 'timestamp': now } }) with open(issuepath + ".history", "w") as f: f.write(json.dumps(vote_history)) f.close() def issue_delete(self, electionID, issueID): "Deletes an issue if it exists" issuepath = os.path.join(self.homedir, "issues", electionID, issueID) + ".json" if os.path.isfile(issuepath): os.unlink(issuepath) if os.path.isfile(issuepath + ".votes"): os.unlink(issuepath + ".votes") def issue_create(self, electionID, issueID, data): issuepath = os.path.join(self.homedir, "issues", electionID, issueID) + ".json" with open(issuepath, "w") as f: f.write(json.dumps(data)) f.close() def voter_get_uid(self, electionID, votekey): "Get vote UID/email with a given vote key hash" elpath = os.path.join(self.homedir, "issues", electionID) with open(elpath + "/voters.json", "r") as f: voters = json.loads(f.read()) f.close() for voter in voters: if voters[voter] == xhash: return voter return None def voter_add(self, election, PID, xhash): elpath = os.path.join(self.homedir, "issues", election) with open(elpath + "/voters.json", "r") as f: voters = json.loads(f.read()) f.close() voters[PID] = xhash with open(elpath + "/voters.json", "w") as f: f.write(json.dumps(voters)) f.close() def voter_remove(self, election, UID): elpath = os.path.join(self.homedir, "issues", election) with open(elpath + "/voters.json", "r") as f: voters = json.loads(f.read()) f.close() if UID in voters: del voters[UID] with open(elpath + "/voters.json", "w") as f: f.write(json.dumps(voters)) f.close() def voter_has_voted(self, election, issue, uid): path = os.path.join(self.homedir, "issues", election, issue) votes = {} if os.path.isfile(path + ".json.votes"): with open(path + ".json.votes", "r") as f: votes = json.loads(f.read()) f.close() return True if uid in votes else False constants.appendBackend("file", FileBasedBackend)