diff options
Diffstat (limited to 'app.py')
| -rw-r--r-- | app.py | 688 |
1 files changed, 391 insertions, 297 deletions
@@ -1,29 +1,36 @@ -from dotenv import load_dotenv -load_dotenv() - import datetime import os +import queue +import subprocess import sys -import traceback -import uuid import threading -import queue import time -from typing import Callable -from user_agents import parse as parse_user_agent +import traceback +import uuid from dataclasses import dataclass -from flask import Flask, render_template, redirect, request, make_response +from typing import Callable + +from dotenv import load_dotenv +from flask import Flask, make_response, redirect, render_template, request from flask_compress import Compress +from jinja2.exceptions import TemplateNotFound +from user_agents import parse as parse_user_agent + +_ = load_dotenv() import db import email_client BASE_URL = os.environ["BASE_URL"] +POLICY_INSTANCE_DOMAIN = os.environ["POLICY_INSTANCE_DOMAIN"] +POLICY_CONTACT_EMAIL = os.environ["POLICY_CONTACT_EMAIL"] + @dataclass class ChoicesByVoter: - name: str - votes: list[bool] + name: str + votes: list[bool] + TITLE_MAX_LENGTH = 100 DESCRIPTION_MAX_LENGTH = 1000 @@ -36,392 +43,479 @@ background_tasks_queue: queue.Queue[Task] = queue.Queue() ### Init + def background_thread(): - print("Background thread started") - while True: - task = background_tasks_queue.get() - task() - time.sleep(0.1) + print("Background thread started") + while True: + task = background_tasks_queue.get() + task() + time.sleep(0.1) + + +def get_version() -> str | None: + try: + return ( + subprocess.check_output( + ["git", "describe", "--tags", "--always"], + stderr=subprocess.DEVNULL, + text=True, + ).strip() + or None + ) + except Exception: + return None + + +APP_VERSION = get_version() + def create_app() -> Flask: threading.Thread(target=background_thread, daemon=True).start() app = Flask(__name__) Compress(app) + app.jinja_env.globals["app_version"] = APP_VERSION return app + app = create_app() + +def policy_page_exists(): + try: + _ = app.jinja_env.get_template("policy.html.j2") + return True + except TemplateNotFound: + return False + + +POLICY_PAGE_EXISTS = policy_page_exists() + ### Routes + def voter_selection_on_choice(voter_name: str, choice: db.Choice) -> int | None: - for vote in choice.votes: - if vote.voter_name == voter_name: - return vote.value - return None + for vote in choice.votes: + if vote.voter_name == voter_name: + return vote.value + return None + def error_page(message: str, code: int = 400): - return render_template("error.html.j2", error=message), code + return render_template("error.html.j2", error=message), code + @app.errorhandler(404) -def not_found(e): - return error_page("Not found", 404) +def not_found(_e: BaseException | None = None): + return error_page("Not found", 404) + @app.errorhandler(Exception) -def error_handler(e): - traceback.print_exception(e, file=sys.stderr) - return error_page("Internal server error", 500) +def error_handler(e: BaseException): + traceback.print_exception(e, file=sys.stderr) + return error_page("Internal server error", 500) + def validate_uuid(s: str) -> bool: - try: - uuid.UUID(s) - return True - except ValueError: - return False + try: + _ = uuid.UUID(s) + return True + except ValueError: + return False + + +@app.route("/policy") +def privacy(): + if not POLICY_PAGE_EXISTS: + return not_found() + + return render_template( + "policy.html.j2", + instance_domain=POLICY_INSTANCE_DOMAIN, + contact_email=POLICY_CONTACT_EMAIL, + ) + @app.route("/") def index(): - created_poll_codes = [] - for k, _ in request.cookies.items(): - if k.startswith("diddle_manage_code_"): - created_poll_codes.append(k.replace("diddle_manage_code_", "")) + created_poll_codes: list[str] = [] + for k, _ in request.cookies.items(): + if k.startswith("diddle_manage_code_"): + created_poll_codes.append(k.replace("diddle_manage_code_", "")) + + created_polls = ( + db.get_polls_by_codes(created_poll_codes) if len(created_poll_codes) > 0 else [] + ) - created_polls = db.get_polls_by_codes(created_poll_codes) if len(created_poll_codes) > 0 else [] + return render_template( + "index.html.j2", + created_polls=created_polls, + now=datetime.datetime.now(), + has_policy=POLICY_PAGE_EXISTS, + ) - return render_template('index.html.j2', - created_polls=created_polls, - now=datetime.datetime.now()) @app.post("/poll/create") def create(): - form = request.form + form = request.form - title = form.get("title") - title = title.strip() if title is not None else None - if title is None or len(title) == 0: - return error_page("Title is required") - if len(title) > TITLE_MAX_LENGTH: - return error_page(f"Title must be {TITLE_MAX_LENGTH} characters or fewer") - description = form.get("description") - description = description.strip() if description is not None else None - if description is not None and len(description) > DESCRIPTION_MAX_LENGTH: - return error_page(f"Description must be {DESCRIPTION_MAX_LENGTH} characters or fewer") - author_name = form.get("author_name") - author_name = author_name.strip() if author_name is not None else None - if author_name is None or len(author_name) == 0: - return error_page("Author name is required") - if len(author_name) > AUTHOR_NAME_MAX_LENGTH: - return error_page(f"Author name must be {AUTHOR_NAME_MAX_LENGTH} characters or fewer") - author_email = form.get("author_email") - author_email = author_email.strip() if author_email is not None else None - if author_email is not None and len(author_email) > AUTHOR_EMAIL_MAX_LENGTH: - return error_page(f"Author email must be {AUTHOR_EMAIL_MAX_LENGTH} characters or fewer") + title = form.get("title") + title = title.strip() if title is not None else None + if title is None or len(title) == 0: + return error_page("Title is required") + if len(title) > TITLE_MAX_LENGTH: + return error_page(f"Title must be {TITLE_MAX_LENGTH} characters or fewer") + description = form.get("description") + description = description.strip() if description is not None else None + if description is not None and len(description) > DESCRIPTION_MAX_LENGTH: + return error_page( + f"Description must be {DESCRIPTION_MAX_LENGTH} characters or fewer" + ) + author_name = form.get("author_name") + author_name = author_name.strip() if author_name is not None else None + if author_name is None or len(author_name) == 0: + return error_page("Author name is required") + if len(author_name) > AUTHOR_NAME_MAX_LENGTH: + return error_page( + f"Author name must be {AUTHOR_NAME_MAX_LENGTH} characters or fewer" + ) + author_email = form.get("author_email") + author_email = author_email.strip() if author_email is not None else None + if author_email is not None and len(author_email) > AUTHOR_EMAIL_MAX_LENGTH: + return error_page( + f"Author email must be {AUTHOR_EMAIL_MAX_LENGTH} characters or fewer" + ) - poll = db.create_poll( - title, - description, - author_name, - author_email, - "is_whole_day" in form, - ) + poll = db.create_poll( + title, + description, + author_name, + author_email, + "is_whole_day" in form, + ) - if email_client.email_enabled: - def task(): - email_client.send_poll_created_email(poll_id=poll.id) + if email_client.email_enabled: - background_tasks_queue.put(task) + def task(): + email_client.send_poll_created_email(poll_id=poll.id) - resp = make_response( - redirect(f"/manage/{poll.manage_code}") - ) - resp.set_cookie(f"diddle_manage_code_{poll.manage_code}", "1", - samesite="Strict", secure=False) - return resp + background_tasks_queue.put(task) + + resp = make_response(redirect(f"/manage/{poll.manage_code}")) + resp.set_cookie( + f"diddle_manage_code_{poll.manage_code}", "1", samesite="Strict", secure=False + ) + return resp VoterNameChoiceIdPair = tuple[str, str] + + @app.get("/poll/<id>") -def poll(id): - if not validate_uuid(id): - return error_page("Invalid poll ID", 400) +def poll(id: str): + if not validate_uuid(id): + return error_page("Invalid poll ID", 400) - prefill_voter_name = request.args.get("prefill_voter_name") + prefill_voter_name = request.args.get("prefill_voter_name") - voter_codes: list[str] = [] - for k, _ in request.cookies.items(): - if k.startswith("diddle_voter_code_"): - voter_codes.append(k.replace("diddle_voter_code_", "")) + voter_codes: list[str] = [] + for k, _ in request.cookies.items(): + if k.startswith("diddle_voter_code_"): + voter_codes.append(k.replace("diddle_voter_code_", "")) - poll = db.get_poll(id) - if poll is None: - return error_page("Poll not found", 404) + poll = db.get_poll(id) + if poll is None: + return error_page("Poll not found", 404) - display_mode_cookie = request.cookies.get("diddle_display_mode") - if display_mode_cookie is None: - user_agent = parse_user_agent(request.user_agent.string) - if user_agent.is_mobile or user_agent.is_tablet: - display_mode = "list" + display_mode_cookie = request.cookies.get("diddle_display_mode") + if display_mode_cookie is None: + user_agent = parse_user_agent(request.user_agent.string) + if user_agent.is_mobile or user_agent.is_tablet: + display_mode = "list" + else: + display_mode = "table" else: - display_mode = "table" - else: - display_mode = display_mode_cookie + display_mode = display_mode_cookie - voter_names_set: set[str] = set() - selections: dict[VoterNameChoiceIdPair, int] = {} - managed_voter_names: dict[str, str] = {} - for choice in poll.choices: - for vote in choice.votes: - selections[(vote.voter_name, choice.id)] = vote.value - voter_names_set.add(vote.voter_name) + voter_names_set: set[str] = set() + selections: dict[VoterNameChoiceIdPair, int] = {} + managed_voter_names: dict[str, str] = {} + for choice in poll.choices: + for vote in choice.votes: + selections[(vote.voter_name, choice.id)] = vote.value + voter_names_set.add(vote.voter_name) - if vote.manage_code in voter_codes: - managed_voter_names[vote.voter_name] = vote.manage_code + if vote.manage_code in voter_codes: + managed_voter_names[vote.voter_name] = vote.manage_code - voter_names = list(voter_names_set) - voter_names.sort() + voter_names = list(voter_names_set) + voter_names.sort() - most_voted_choice_ids: set[str] = set() - most_voted_value = 1 # start from 1 to avoid marking 0 votes as most voted - for choice in poll.choices: - n_votes = 0 - for vote in choice.votes: - n_votes += vote.value + most_voted_choice_ids: set[str] = set() + most_voted_value = 1 # start from 1 to avoid marking 0 votes as most voted + for choice in poll.choices: + n_votes = 0 + for vote in choice.votes: + n_votes += vote.value - if n_votes > most_voted_value: - most_voted_choice_ids = { choice.id } - most_voted_value = n_votes - elif n_votes == most_voted_value: - most_voted_choice_ids.add(choice.id) + if n_votes > most_voted_value: + most_voted_choice_ids = {choice.id} + most_voted_value = n_votes + elif n_votes == most_voted_value: + most_voted_choice_ids.add(choice.id) - resp = make_response( - render_template("poll.html.j2", - poll=poll, - selections=selections, - choices=poll.choices, - most_voted_choice_ids=most_voted_choice_ids, - prefill_voter_name=prefill_voter_name, - voter_names=voter_names, - managed_voter_names=managed_voter_names, - now=datetime.datetime.now(), - display_mode=display_mode)) + resp = make_response( + render_template( + "poll.html.j2", + poll=poll, + selections=selections, + choices=poll.choices, + most_voted_choice_ids=most_voted_choice_ids, + prefill_voter_name=prefill_voter_name, + voter_names=voter_names, + managed_voter_names=managed_voter_names, + now=datetime.datetime.now(), + display_mode=display_mode, + ) + ) - resp.set_cookie("diddle_display_mode", display_mode, - samesite="Lax", secure=False) - return resp + resp.set_cookie("diddle_display_mode", display_mode, samesite="Lax", secure=False) + return resp @app.post("/poll/<id>/vote") -def vote_poll(id): - if not validate_uuid(id): - return error_page("Invalid poll ID", 400) +def vote_poll(id: str): + if not validate_uuid(id): + return error_page("Invalid poll ID", 400) - form = request.form - voter_name = form.get("voter_name") - voter_name = voter_name.strip() if voter_name is not None else None - if voter_name is None or len(voter_name) == 0: - return error_page("Voter name is required") - if len(voter_name) > VOTER_NAME_MAX_LENGTH: - return error_page(f"Voter name must be {VOTER_NAME_MAX_LENGTH} characters or fewer") + form = request.form + voter_name = form.get("voter_name") + voter_name = voter_name.strip() if voter_name is not None else None + if voter_name is None or len(voter_name) == 0: + return error_page("Voter name is required") + if len(voter_name) > VOTER_NAME_MAX_LENGTH: + return error_page( + f"Voter name must be {VOTER_NAME_MAX_LENGTH} characters or fewer" + ) - poll = db.get_poll(id) - if poll is None: - return error_page("Poll not found") + poll = db.get_poll(id) + if poll is None: + return error_page("Poll not found") - selections: dict[str, int] = {} - for choice in poll.choices: - selections[choice.id] = 0 + selections: dict[str, int] = {} + for choice in poll.choices: + selections[choice.id] = 0 - for k in form.keys(): - if k.startswith("choice_"): - choice_id = k.replace("choice_", "") - selections[choice_id] = 1 + for k in form.keys(): + if k.startswith("choice_"): + choice_id = k.replace("choice_", "") + selections[choice_id] = 1 - manage_code = db.vote_poll(id, voter_name, selections) - if manage_code is None: - return error_page("That name is already in use") + manage_code = db.vote_poll(id, voter_name, selections) + if manage_code is None: + return error_page("That name is already in use") - if email_client.email_enabled: - def task(): - email_client.send_participation_email(poll_id=id, voter_name=voter_name) - background_tasks_queue.put(task) + if email_client.email_enabled: + + def task(): + email_client.send_participation_email(poll_id=id, voter_name=voter_name) + + background_tasks_queue.put(task) + + response = make_response(redirect(f"/poll/{id}")) + response.set_cookie( + f"diddle_voter_code_{manage_code}", "1", samesite="Strict", secure=False + ) + return response - response = make_response( - redirect(f"/poll/{id}") - ) - response.set_cookie(f"diddle_voter_code_{manage_code}", "1", - samesite="Strict", secure=False) - return response @app.post("/poll/<id>/delete_voter") -def delete_voter(id): - if not validate_uuid(id): - return error_page("Invalid poll ID", 400) +def delete_voter(id: str): + if not validate_uuid(id): + return error_page("Invalid poll ID", 400) + + voter_code = request.form["voter_code"] + voter_name = db.get_voter_name_by_manage_code(voter_code) - voter_code = request.form["voter_code"] - voter_name = db.get_voter_name_by_manage_code(voter_code) + return render_template( + "voter_confirm_delete.html.j2", + voter_name=voter_name, + voter_code=voter_code, + poll_id=id, + ) - return render_template("voter_confirm_delete.html.j2", - voter_name=voter_name, - voter_code=voter_code, - poll_id=id) @app.post("/poll/<id>/confirm_delete_voter") -def confirm_delete_voter(id): - if not validate_uuid(id): - return error_page("Invalid poll ID", 400) +def confirm_delete_voter(id: str): + if not validate_uuid(id): + return error_page("Invalid poll ID", 400) - voter_manage_code = request.form["voter_code"] + voter_manage_code = request.form["voter_code"] - voter_name = db.get_voter_name_by_manage_code(voter_manage_code) - if voter_name is None: - return error_page("Voter not found", 404) + voter_name = db.get_voter_name_by_manage_code(voter_manage_code) + if voter_name is None: + return error_page("Voter not found", 404) - db.delete_voter(voter_manage_code) + db.delete_voter(voter_manage_code) + + resp = make_response(redirect(f"/poll/{id}?prefill_voter_name={voter_name}")) + resp.set_cookie( + f"diddle_voter_code_{voter_manage_code}", + "", + expires=0, + samesite="Strict", + secure=False, + ) + return resp - resp = make_response( - redirect(f"/poll/{id}?prefill_voter_name={voter_name}") - ) - resp.set_cookie(f"diddle_voter_code_{voter_manage_code}", "", expires=0, - samesite="Strict", secure=False) - return resp @app.post("/manage/<code>/update_info") -def update_poll_info(code): - if not validate_uuid(code): - return error_page("Invalid manage code", 400) +def update_poll_info(code: str): + if not validate_uuid(code): + return error_page("Invalid manage code", 400) + + form = request.form - form = request.form + title = form.get("title") + title = title.strip() if title is not None else None + if title is None or len(title) == 0: + return error_page("Title is required") + if len(title) > TITLE_MAX_LENGTH: + return error_page(f"Title must be {TITLE_MAX_LENGTH} characters or fewer") + description = form.get("description") + description = description.strip() if description is not None else None + if description is not None and len(description) > DESCRIPTION_MAX_LENGTH: + return error_page( + f"Description must be {DESCRIPTION_MAX_LENGTH} characters or fewer" + ) + author_name = form.get("author_name") + author_name = author_name.strip() if author_name is not None else None + if author_name is None or len(author_name) == 0: + return error_page("Author name is required") + if len(author_name) > AUTHOR_NAME_MAX_LENGTH: + return error_page( + f"Author name must be {AUTHOR_NAME_MAX_LENGTH} characters or fewer" + ) + author_email = form.get("author_email") + author_email = author_email.strip() if author_email is not None else None + if author_email is not None and len(author_email) > AUTHOR_EMAIL_MAX_LENGTH: + return error_page( + f"Author email must be {AUTHOR_EMAIL_MAX_LENGTH} characters or fewer" + ) - title = form.get("title") - title = title.strip() if title is not None else None - if title is None or len(title) == 0: - return error_page("Title is required") - if len(title) > TITLE_MAX_LENGTH: - return error_page(f"Title must be {TITLE_MAX_LENGTH} characters or fewer") - description = form.get("description") - description = description.strip() if description is not None else None - if description is not None and len(description) > DESCRIPTION_MAX_LENGTH: - return error_page(f"Description must be {DESCRIPTION_MAX_LENGTH} characters or fewer") - author_name = form.get("author_name") - author_name = author_name.strip() if author_name is not None else None - if author_name is None or len(author_name) == 0: - return error_page("Author name is required") - if len(author_name) > AUTHOR_NAME_MAX_LENGTH: - return error_page(f"Author name must be {AUTHOR_NAME_MAX_LENGTH} characters or fewer") - author_email = form.get("author_email") - author_email = author_email.strip() if author_email is not None else None - if author_email is not None and len(author_email) > AUTHOR_EMAIL_MAX_LENGTH: - return error_page(f"Author email must be {AUTHOR_EMAIL_MAX_LENGTH} characters or fewer") + changed = db.update_poll_info( + code, + title, + description, + author_name, + author_email, + "is_whole_day" in form, + ) - changed = db.update_poll_info( - code, - title, - description, - author_name, - author_email, - "is_whole_day" in form, - ) + if changed is None: + return error_page("Poll not found", 404) - if changed is None: - return error_page("Poll not found", 404) + return redirect(f"/manage/{code}") - return redirect(f"/manage/{code}") @app.post("/manage/<code>/add_choice") -def add_choice(code): - if not validate_uuid(code): - return error_page("Invalid manage code", 400) +def add_choice(code: str): + if not validate_uuid(code): + return error_page("Invalid manage code", 400) + + form = request.form + if "start_datetime" not in form or len(form["start_datetime"]) == 0: + return error_page("Start datetime is required") + if "end_datetime" not in form or len(form["end_datetime"]) == 0: + return error_page("End datetime is required") + if form["start_datetime"] > form["end_datetime"]: + return error_page("Start datetime must be before end datetime") - form = request.form - if "start_datetime" not in form or len(form["start_datetime"]) == 0: - return error_page("Start datetime is required") - if "end_datetime" not in form or len(form["end_datetime"]) == 0: - return error_page("End datetime is required") - if form["start_datetime"] > form["end_datetime"]: - return error_page("Start datetime must be before end datetime") + start_datetime = form["start_datetime"] + if len(start_datetime) == 10: + start_datetime += " 00:00:00" + else: + start_datetime = start_datetime.replace("T", " ") + start_datetime += ":00" - start_datetime = form["start_datetime"] - if len(start_datetime) == 10: - start_datetime += " 00:00:00" - else: - start_datetime = start_datetime.replace("T", " ") - start_datetime += ":00" + end_datetime = form["end_datetime"] + if len(end_datetime) == 10: + end_datetime += " 23:59:00" + else: + end_datetime = end_datetime.replace("T", " ") + end_datetime += ":00" - end_datetime = form["end_datetime"] - if len(end_datetime) == 10: - end_datetime += " 23:59:00" - else: - end_datetime = end_datetime.replace("T", " ") - end_datetime += ":00" + db.add_choice_to_poll( + code, + start_datetime, + end_datetime, + ) - db.add_choice_to_poll( - code, - start_datetime, - end_datetime, - ) + return redirect(f"/manage/{code}?focus_next=1") - return redirect(f"/manage/{code}?focus_next=1") @app.post("/manage/<code>/delete_choice/<choice_id>") -def delete_choice(code, choice_id): - if not validate_uuid(code): - return error_page("Invalid manage code", 400) +def delete_choice(code: str, choice_id: str): + if not validate_uuid(code): + return error_page("Invalid manage code", 400) - poll = db.get_poll_by_code(code) - if poll is None: - return error_page("Poll not found", code=404) + poll = db.get_poll_by_code(code) + if poll is None: + return error_page("Poll not found", code=404) - db.delete_choice(choice_id) + db.delete_choice(choice_id) + + return redirect(f"/manage/{code}?focus_next=1") - return redirect(f"/manage/{code}?focus_next=1") @app.get("/manage/<code>") -def manage(code): - if not validate_uuid(code): - return error_page("Invalid manage code", 400) +def manage(code: str): + if not validate_uuid(code): + return error_page("Invalid manage code", 400) + + poll = db.get_poll_by_code(code) + if poll is None: + return error_page("Poll not found") - poll = db.get_poll_by_code(code) - if poll is None: - return error_page("Poll not found") + last_choice_id = poll.choices[-1].id if len(poll.choices) > 0 else None + resp = make_response( + render_template("manage.html.j2", poll=poll, last_choice_id=last_choice_id) + ) + resp.set_cookie(f"diddle_manage_code_{code}", "1", samesite="Strict", secure=False) + return resp - last_choice_id = poll.choices[-1].id if len(poll.choices) > 0 else None - resp = make_response( - render_template("manage.html.j2", - poll=poll, - last_choice_id=last_choice_id)) - resp.set_cookie(f"diddle_manage_code_{code}", "1", - samesite="Strict", secure=False) - return resp @app.post("/manage/<code>/delete") -def delete_poll(code): - if not validate_uuid(code): - return error_page("Invalid manage code", 400) +def delete_poll(code: str): + if not validate_uuid(code): + return error_page("Invalid manage code", 400) + + poll = db.get_poll_by_code(code) + return render_template("poll_confirm_delete.html.j2", poll=poll) - poll = db.get_poll_by_code(code) - return render_template("poll_confirm_delete.html.j2", poll=poll) @app.post("/manage/<code>/confirm_delete") -def confirm_delete_poll(code): - if not validate_uuid(code): - return error_page("Invalid manage code", 400) +def confirm_delete_poll(code: str): + if not validate_uuid(code): + return error_page("Invalid manage code", 400) + + db.delete_poll(code) + resp = make_response(redirect("/")) + resp.set_cookie( + f"diddle_manage_code_{code}", "", expires=0, samesite="Strict", secure=False + ) + return resp - db.delete_poll(code) - resp = make_response(redirect("/")) - resp.set_cookie(f"diddle_manage_code_{code}", "", expires=0, - samesite="Strict", secure=False) - return resp @app.post("/options/toggle_display_mode") def toggle_display_mode(): - poll_id = request.form["poll_id"] - redirect_url = f"/poll/{poll_id}" - display_mode = request.cookies.get("diddle_display_mode", "table") - if display_mode == "table": - display_mode = "list" - else: - display_mode = "table" + poll_id = request.form["poll_id"] + redirect_url = f"/poll/{poll_id}" + display_mode = request.cookies.get("diddle_display_mode", "table") + if display_mode == "table": + display_mode = "list" + else: + display_mode = "table" - resp = make_response(redirect(redirect_url)) - resp.set_cookie("diddle_display_mode", display_mode, - samesite="Lax", secure=False) - return resp + resp = make_response(redirect(redirect_url)) + resp.set_cookie("diddle_display_mode", display_mode, samesite="Lax", secure=False) + return resp |