A Democratic Social Network. Currently available at https://democraticnet.de The Test Server is available at https://test.democraticnet.de
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
DemNet/clock.py

136 lines
4.8 KiB

"""Running a scheduler to run jobs
on regular intervals.
"""
import json
import time
import subprocess
import schedule
from Server.Database import *
from Server.election import count
GIT_REPOSITORY = os.environ["GIT_REPOSITORY"]
GIT_ACCESS_TOKEN = os.environ["GIT_ACCESS_TOKEN"]
DEMNET_LOCATION_WITHOUT_PROTOCOL = os.environ["DEMNET_LOCATION_WITHOUT_PROTOCOL"]
def init():
"""Setup the Git Repository at $GIT_REPOSITORY
to the current version of the Source Code.
"""
subprocess.run( [ "bash", "init_git.sh" ]
, env = { "DEMNET_LOCATION_WITHOUT_PROTOCOL" : DEMNET_LOCATION_WITHOUT_PROTOCOL
, "GIT_ACCESS_TOKEN" : GIT_ACCESS_TOKEN
, "GIT_REPOSITORY" : GIT_REPOSITORY
}
)
def check_election(election : Election):
"""Check any election past their
voting phase and enforce their results.
"""
try:
init()
assert election.proposal_phase_start + length_voting_phase + length_proposal_phase < datetime.date.today()
users_count = User.select().count()
options = list(map(lambda p: p.id, election.proposals))
votes = list(map(lambda v: [ int(v_) for v_ in json.loads(v.choice)][::-1], election.votes))
results = count(votes, options, all_participants = users_count)
winner = results["winner"]
election.winner = winner
election.total_users = users_count
election.save()
# Automatically merge and the CI will take care of the rest.
if winner != "NoneOfTheOtherOptions":
proposal = Proposal.get_by_id(winner)
print(f"Election result: {proposal.title} won the election from {election.proposal_phase_start}")
r = subprocess.run( [ "git am --whitespace=fix -s && git push" ]
, input = proposal.patch.encode("utf-8")
, cwd = GIT_REPOSITORY
, shell = True
).returncode
if r != 0:
print("Election result couldn't be applied")
apply_to_books(proposal)
else:
print(f"Election result: NoneOfTheOtherOptions won the election from {election.proposal_phase_start}")
except AssertionError:
print("check_election called with premature election")
except Exception as e:
raise e
finally:
if not database.is_closed():
database.close()
return
def apply_to_books(proposal):
"""Apply the
patches of the books to those books.
"""
for book_proposal in proposal.book_proposals:
book = book_proposal.book
if Book.select().where(Book.name == book).exists():
book = Book.get_by_id(book)
else:
book = Book.create(name = book, text = "", since = datetime.date.today())
for user in User.select():
AcceptedLastChanges.create( user = user
, book = book
, accepted = False
)
book.text = book_proposal.new_text
if book.text == "":
book.delete_instance()
else:
book.save()
(AcceptedLastChanges
.update(accepted = False)
.where(AcceptedLastChanges.book == book)
.execute())
def check_count_and_create_election():
"""
Run this daily and all elections will be properly counted.
"""
try:
if database.is_closed():
database.connect()
elections = list(Election.select().where(Election.winner == None).where(Election.proposal_phase_start + length_voting_phase + length_proposal_phase < datetime.date.today()).order_by(-Election.proposal_phase_start))
for election in elections:
check_election(election)
current_election = Election.select().order_by(-Election.proposal_phase_start).get()
if current_election.proposal_phase_start + length_voting_phase + length_proposal_phase < datetime.date.today():
current_election = Election.create(proposal_phase_start = last_sunday(), id = Election.select().order_by(-Election.id).get().id + 1)
except Exception as e:
raise e
finally:
if not database.is_closed():
database.close()
def last_sunday():
"""Returns the date of the last sunday before the current day.
If datetime.date.today() == Mon 12. Oct 2020
then last_sunday() == Sun 11. Oct 2020
"""
return datetime.date.today() - datetime.timedelta(days = datetime.date.today().isoweekday() % 7)
def run():
"""Running every day
"""
check_count_and_create_election()
if __name__ == "__main__":
run()