aboutsummaryrefslogtreecommitdiff
path: root/uploader/entry.py
blob: 82034ed0a4c532cc20b3f34e795bd24aa225dc97 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
"""Entry-point module"""
import os
import mimetypes
from typing import Tuple
from zipfile import ZipFile, is_zipfile

from werkzeug.utils import secure_filename
from flask import (
    flash,
    request,
    url_for,
    redirect,
    Blueprint,
    render_template,
    current_app as app,
    send_from_directory)

from uploader.db import species
from uploader.authorisation import require_login
from uploader.db_utils import with_db_connection
from uploader.oauth2.client import user_logged_in

entrybp = Blueprint("entry", __name__)

@entrybp.route("/favicon.ico", methods=["GET"])
def favicon():
    """Return the favicon."""
    return send_from_directory(os.path.join(app.root_path, "static"),
                               "images/CITGLogo.png",
                               mimetype="image/png")


def errors(rqst) -> Tuple[str, ...]:
    """Return a tuple of the errors found in the request `rqst`. If no error is
    found, then an empty tuple is returned."""
    def __filetype_error__():
        return (
            ("Invalid file type provided.",)
            if rqst.form.get("filetype") not in ("average", "standard-error")
            else tuple())

    def __file_missing_error__():
        return (
            ("No file was uploaded.",)
            if ("qc_text_file" not in rqst.files or
                rqst.files["qc_text_file"].filename == "")
            else tuple())

    def __file_mimetype_error__():
        text_file = rqst.files["qc_text_file"]
        return (
            (
                ("Invalid file! Expected a tab-separated-values file, or a zip "
                 "file of the a tab-separated-values file."),)
            if text_file.mimetype not in (
                    "text/plain", "text/tab-separated-values",
                    "application/zip")
            else tuple())

    return (
        __filetype_error__() +
        (__file_missing_error__() or __file_mimetype_error__()))

def zip_file_errors(filepath, upload_dir) -> Tuple[str, ...]:
    """Check the uploaded zip file for errors."""
    zfile_errors: Tuple[str, ...] = tuple()
    if is_zipfile(filepath):
        with ZipFile(filepath, "r") as zfile:
            infolist = zfile.infolist()
            if len(infolist) != 1:
                zfile_errors = zfile_errors + (
                    ("Expected exactly one (1) member file within the uploaded zip "
                     f"file. Got {len(infolist)} member files."),)
            if len(infolist) == 1 and infolist[0].is_dir():
                zfile_errors = zfile_errors + (
                    ("Expected a member text file in the uploaded zip file. Got a "
                     "directory/folder."),)

            if len(infolist) == 1 and not infolist[0].is_dir():
                zfile.extract(infolist[0], path=upload_dir)
                mime = mimetypes.guess_type(f"{upload_dir}/{infolist[0].filename}")
                if mime[0] != "text/tab-separated-values":
                    zfile_errors = zfile_errors + (
                        ("Expected the member text file in the uploaded zip file to"
                         " be a tab-separated file."),)

    return zfile_errors

@entrybp.route("/", methods=["GET"])
def index():
    """Load the landing page"""
    return render_template("index.html" if user_logged_in() else "login.html")

@entrybp.route("/upload", methods=["GET", "POST"])
@require_login
def upload_file():
    """Enables uploading the files"""
    if request.method == "GET":
        return render_template(
            "select_species.html", species=with_db_connection(species))

    upload_dir = app.config["UPLOAD_FOLDER"]
    request_errors = errors(request)
    if request_errors:
        for error in request_errors:
            flash(error, "alert-danger error-expr-data")
        return redirect(url_for("entry.upload_file"))

    filename = secure_filename(request.files["qc_text_file"].filename)
    if not os.path.exists(upload_dir):
        os.mkdir(upload_dir)

    filepath = os.path.join(upload_dir, filename)
    request.files["qc_text_file"].save(os.path.join(upload_dir, filename))

    zip_errors = zip_file_errors(filepath, upload_dir)
    if zip_errors:
        for error in zip_errors:
            flash(error, "alert-danger error-expr-data")
        return redirect(url_for("entry.upload_file"))

    return redirect(url_for("parse.parse",
                            speciesid=request.form["speciesid"],
                            filename=filename,
                            filetype=request.form["filetype"]))

@entrybp.route("/data-review", methods=["GET"])
@require_login
def data_review():
    """Provide some help on data expectations to the user."""
    return render_template("data_review.html")