lab-signal-bot/create_task.py

#!/usr/bin/env python

import requests
from result import Err, Ok, Result
from models import Task
import argparse
from sqlmodel import Session, SQLModel, create_engine
from main import *

import sys
import dateparser

def create_hedgedoc_copy(old_url: str) -> Result[str, str]:
    # remove query params and #
    old_url = old_url.split("?")[0].split("#")[0]

    # remove trailing slash
    if old_url[-1] == "/":
        old_url = old_url[:-1]

    download_url = old_url + "/download"
    response = requests.get(download_url)

    if response.status_code != 200:
        return Err("Failed to download the document.")

    old_content = response.text
    content = old_content + "\n\n---\n\nThe template for this pad can be found [here](" + old_url + "). If there is something in the tasks that should be changed, you can alter the template carefully."

    pad_base_url = "/".join(old_url.split("/")[0:3])

    # do not follow redirects
    response = requests.post(
        pad_base_url + "/new",
        data=content,
        headers={"Content-Type": "text/markdown"},
        allow_redirects=False)

    if response.status_code != 302:
        return Err(response.text)

    return Ok(response.headers['Location'])


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Create a task')
    parser.add_argument('name', type=str, help='Name of the task')
    parser.add_argument('number_of_participants', type=int, help='Number of participants')
    parser.add_argument('due', type=dateparser.parse, help='Due date of the task')
    parser.add_argument('--timeout_seconds', type=int, help='Timeout in seconds (default = 1 day)', default=24*3600)
    parser.add_argument('--pad-template-url', type=str, help='URL to pad that contains the task description (will be copied)')

    args = parser.parse_args()

    if args.due is None:
        print("Invalid due date.", file=sys.stderr)
        exit(1)

    with Session(engine) as session:
        task = Task(
            name=args.name,
            required_number_of_participants=args.number_of_participants,
            due=args.due,
            timeout=args.timeout_seconds)

        if args.pad_template_url is not None:
            result = create_hedgedoc_copy(args.pad_template_url)

            if result.is_err():
                print("Error: " + result.unwrap_err(), file=sys.stderr)
                exit(1)

            task.pad_url = result.unwrap()

        session.add(task)
        session.commit()
        session.refresh(task)

        print("Created: " + repr(task))