initial commit

.gitignore

+.DS_Store
+__pycache__
+.*.swp
+.*.swo
+
+.idea/
+venv/
+.pytest_cache/
+
+.cache_fake_data.tar
+fake_data/
+
+*.egg-info/
+build/
+dist/

.gitlab-ci.yml

+image: "python:3.8"
+
+stages:
+  - linting
+  - publish
+
+black:
+  stage: linting
+  image: registry.gitlab.com/pipeline-components/black:latest
+  script:
+    - black --check --verbose -- .
+  tags:
+    - docker
+
+publish_package:
+  stage: publish
+  script:
+    - pip install twine
+    - rm -rf dist/
+    - python setup.py bdist_wheel
+    - TWINE_PASSWORD=${CI_JOB_TOKEN} TWINE_USERNAME=gitlab-ci-token python -m twine upload --repository-url ${CI_API_V4_URL}/projects/${CI_PROJECT_ID}/packages/pypi dist/*
+  tags:
+    - docker
+  only:
+    - tags

.pre-commit-config.yaml

+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.2.0
+    hooks:
+    -   id: trailing-whitespace
+    -   id: end-of-file-fixer
+  - repo: https://github.com/psf/black
+    rev: 22.3.0
+    hooks:
+      - id: black

README.md

+# Mercure
+
+Mercure is a webapp allowing teacher to publish students exam papers.
+
+## Installation
+
+## External dependencies
+
+`pdfinfo` and `pdftoppm` must be in the `PATH`. These are provided by
+`poppler-utils`.
+
+```bash
+apt install poppler-utils
+```
+
+### Production version
+
+```bash
+pip install --upgrade mercure --index-url https://gitlab.utc.fr/api/v4/groups/12819/-/packages/pypi/simple
+```
+
+### Development inplace version (for developing purpose only)
+
+```bash
+pip install -r requirements.txt --index-url https://gitlab.utc.fr/api/v4/groups/12819/-/packages/pypi/simple
+```
+
+Then:
+```
+pip install -e .
+```
+
+## Run server
+
+TODO: write conf with gunicorn
+
+## Run development server
+```bash
+mercure_dev_webserver
+```
+
+## Architecture
+
+Mercure is built on top of [Flask](https://flask.palletsprojects.com/en/2.1.x/) framework, using [Jinja2](https://jinja.palletsprojects.com/en/3.1.x/) to render HTML.
+It uses [Bulma](https://bulma.io/documentation/) CSS framework.
+
+
+## Contribute
+
+Get the source from git:
+
+```
+git clone https://gitlab.utc.fr/mercure/mercure
+```
+or
+```
+git clone git@gitlab.utc.fr:mercure/mercure
+```
+
+Install the developpement dependencies
+
+```
+pip install -r requirements-dev.txt
+```
+
+Install pre-commit-hooks
+```
+pre-commit install
+```
+
+Enjoy !

mercure_cli_sclo/__init__.py

+# Copyright 2022, Thibaud Le Graverend <thibaud@legraverend.fr>
+# Copyright 2022, Jean-Benoist Leger <jbleger@hds.utc.fr>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a
+# copy of this software and associated documentation files (the
+# "Software"), to deal in the Software without restriction, including
+# without limitation the rights to use, copy, modify, merge, publish,
+# distribute, sublicense, and/or sell copies of the Software, and to
+# permit persons to whom the Software is furnished to do so, subject to
+# the following conditions:
+#
+# The above copyright notice and this permission notice shall be included
+# in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+__title__ = "Mercure CLI SCLO"
+__author__ = "Mercure authors"
+__license__ = "MIT"
+
+version_info = (0, 1)
+__version__ = ".".join(map(str, version_info))
+
+from . import _send
+from . import _build_config

mercure_cli_sclo/_build_config.py

+import io
+import sys
+import csv
+import pathlib
+import argparse
+import textwrap
+import itertools
+
+import cchardet
+import yaml
+
+DOC_YAML = textwrap.dedent(
+    """
+    Éditer ce fichier pour donner l'échelle de notation.
+
+    Le nom de chaque exercice peut-être changé, mais le nom nom des questions NE
+    DOIT PAS CHANGER.
+
+    Pour chaque question, VOUS DEVEZ:
+     - vérifier la valeur `raw`. Cette valeur correspond au nombre de points sur
+       les cases de la copie. Cette valeur a été obtenue avec une heuristique
+       utilisant entre autres le max dans le csv. Dans le cas d'une echelle par
+       demi-point, il faut utiliser la même syntaxe que dans SCLO, c'est à dire
+       "4,2" désigne une echelle de 4 par demi-points.
+     - positionner dans `points` le nombre de points associé à la question.
+    """
+)
+
+
+def _can_be_a_grade(x):
+    try:
+        a = int(x)
+    except ValueError:
+        return False
+    if a == -10000:
+        print(
+            "The is a -10000 in the file. All incorrect value must be fixed before using this program.",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    if 0 <= a <= 1000:
+        return True
+
+
+def _infer_raw(raw_values):
+    max_value = max(int(r) for r in raw_values)
+    if max_value <= 1:
+        return 1
+    if max_value <= 2:
+        return 2
+    if max_value <= 4:
+        return 4
+    if max_value <= 5:
+        return 5
+    if max_value <= 8:
+        return 8
+    if max_value <= 10:
+        return 10
+    return max_value
+
+
+def build_config(infilename, outfilename):
+    csvcontent = open(infilename, "rb").read()
+    encoding = cchardet.detect(csvcontent)["encoding"]
+    if encoding is None:
+        print("CSV encoding error", file=sys.stderr)
+        sys.exit(1)
+    csvdecoded = csvcontent.decode(encoding)
+    dialect = csv.Sniffer().sniff(csvdecoded)
+    csvreader = csv.DictReader(io.StringIO(csvdecoded), dialect=dialect)
+    cols = csvreader.fieldnames
+    data = list(csvreader)
+
+    questions = tuple(
+        c
+        for i, c in enumerate(cols)
+        if i > 0 and "." in c and all(_can_be_a_grade(d[c]) for d in data)
+    )
+
+    exos = [
+        {
+            "name": exo,
+            "questions": {
+                q: {"raw": _infer_raw(d[q] for d in data), "points": 1}
+                for q in sorted(quests, key=lambda x: int(x.split(".")[-1]))
+            },
+        }
+        for exo, quests in itertools.groupby(
+            questions, key=lambda x: ".".join(x.split(".")[:-1])
+        )
+    ]
+
+    config = {"exam_name": "Hiver 1931 - ZZ02 - Examen médian", "config": exos}
+
+    config_yaml = yaml.safe_dump(
+        config, indent=4, sort_keys=False, allow_unicode=True
+    ).replace("\n-", "\n\n-")
+
+    with open(outfilename, "w") as f:
+        f.write(textwrap.indent(DOC_YAML, "# "))
+        f.write("\n\n\n")
+        f.write(config_yaml)
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description=textwrap.dedent(
+            """
+            Generate editable scale file from AMC csv export.
+            """
+        ),
+    )
+
+    parser.add_argument(
+        "--version", action="version", version="", help=argparse.SUPPRESS
+    )
+
+    parser.add_argument(
+        "-o",
+        "--output",
+        help="""
+                Output file. If the output file is not provided, the output
+                file will be the same as the input file with the '.yml' extension.
+                """,
+        default=None,
+    )
+
+    parser.add_argument(
+        "input_csv",
+        help="""
+                Input csv file from AMC export.
+            """,
+    )
+
+    args = parser.parse_args()
+
+    if args.output is not None:
+        fout = args.output
+    else:
+        fout = pathlib.Path(args.input_csv).with_suffix(".yml")
+
+    build_config(args.input_csv, fout)
+
+
+if __name__ == "__main__":
+    main()

mercure_cli_sclo/_send.py

+import os
+import re
+import io
+import sys
+import csv
+import json
+import argparse
+import textwrap
+import pathlib
+import itertools
+import multiprocessing
+from getpass import getpass
+from dataclasses import dataclass
+
+import yaml
+import cchardet
+import requests
+import mechanize
+import numpy as np
+from PIL import Image, ImageDraw, ImageFont
+
+import mercure_pdfimg
+
+
+@dataclass
+class StudentRes:
+    login: str
+    paper: int
+    questions: dict
+
+
+@dataclass
+class StudentPaperDesc:
+    login: str
+    paper_pdf: pathlib.Path
+    grade: float
+    abstracts: list
+
+
+@dataclass
+class StudentPaper:
+    login: str
+    grade: float
+    pages: list
+
+
+def _ech(x):
+    try:
+        y = int(x)
+        return y, 1
+    except ValueError:
+        pass
+    return tuple(int(w) for w in x.split(","))
+
+
+def _ech_max(x):
+    up, step = _ech(x)
+    return up * step
+
+
+def _all_scale(x):
+    up, step = _ech(x)
+    return tuple(
+        (i, (r // step if r % step == 0 else r / step))
+        for i, r in enumerate(range(0, step * up + 1))
+    )
+
+
+def compute_grades_stats(config, data):
+    hists = {
+        q: {
+            val: sum(1 for d in data if d.questions[q] == k)
+            for k, val in _all_scale(qconfig["raw"])
+        }
+        for q, qconfig in itertools.chain.from_iterable(
+            e["questions"].items() for e in config["config"]
+        )
+    }
+
+    grades = {
+        stud.login: [
+            sum(
+                stud.questions[q] / _ech_max(qconfig["raw"]) * qconfig["points"]
+                for q, qconfig in exercise["questions"].items()
+            )
+            for exercise in config["config"]
+        ]
+        for stud in data
+    }
+
+    for val in grades.values():
+        val.append(sum(val))
+
+    quantiles = tuple(
+        tuple(x)
+        for x in np.quantile(
+            np.array(list(grades.values())), (0, 0.25, 0.5, 0.75, 1), axis=0
+        ).T
+    )
+
+    return grades, hists, quantiles
+
+
+def sflo(x):
+    if x % 1 == 0:
+        return int(x)
+    return x
+
+
+def build_papersdesc(config, data, pdfs):
+    grades, hists, quantiles = compute_grades_stats(config, data)
+
+    papers = []
+    for stud in data:
+
+        abstracts = []
+
+        stud_grades = grades[stud.login]
+        for exercise, exercise_grade, exercise_quantiles in zip(
+            config["config"], stud_grades, quantiles
+        ):
+            txt = io.StringIO()
+            print(config["exam_name"], file=txt)
+            print("=" * len(config["exam_name"]), file=txt)
+            print("\n\n", file=txt)
+
+            subname = f"Exercice « {exercise['name']} »"
+            print(subname, file=txt)
+            print("-" * len(subname), file=txt)
+            print("\n", file=txt)
+
+            for i, (q, qconfig) in enumerate(exercise["questions"].items()):
+                up, step = _ech(qconfig["raw"])
+                print(f"  - Question {i+1}", file=txt)
+                print(
+                    f"        Score obtenu par l'étudiant : {sflo(stud.questions[q]/step)}/{sflo(up)}",
+                    file=txt,
+                )
+                print(f"        Score de la promotion : {hists[q]}", file=txt)
+                print(f"        Barème : {qconfig['points']} pts", file=txt)
+                print(
+                    f"        Points obtenus par l'étudiant : {stud.questions[q]/(up*step)*qconfig['points']} pts",
+                    file=txt,
+                )
+            print(file=txt)
+            print(
+                f"Total de l'exercice obtenu par l'étudiant : {exercise_grade} pts",
+                file=txt,
+            )
+            print(
+                f"Total de l'exercice obtenu par la promotion :\n"
+                f"    (min, q1, q2, q3, max) = {exercise_quantiles}",
+                file=txt,
+            )
+            abstracts.append(txt.getvalue())
+
+        txt = io.StringIO()
+        print(config["exam_name"], file=txt)
+        print("=" * len(config["exam_name"]), file=txt)
+        print("\n\n", file=txt)
+
+        subname = "Total"
+        print(subname, file=txt)
+        print("-" * len(subname), file=txt)
+        print("\n", file=txt)
+
+        print(f"Total obtenu par l'étudiant : {stud_grades[-1]} pts", file=txt)
+        print(
+            f"Total obtenu par la promotion :\n"
+            f"    (min, q1, q2, q3, max) = {quantiles[-1]}",
+            file=txt,
+        )
+        abstracts.append(txt.getvalue())
+
+        papers.append(
+            StudentPaperDesc(
+                login=stud.login,
+                abstracts=abstracts,
+                grade=stud_grades[-1],
+                paper_pdf=pdfs[stud.paper],
+            )
+        )
+
+    return papers
+
+
+def abstract_to_png(abstract):
+    out = Image.new("RGB", (2480 // 2, 3508 // 2), (255, 255, 255))
+    fnt = ImageFont.truetype("Pillow/Tests/fonts/FreeMono.ttf", 20)
+    d = ImageDraw.Draw(out)
+    d.multiline_text((40, 40), abstract, font=fnt, fill=(0, 0, 0))
+
+    out = out.convert("P", palette=Image.Palette.ADAPTIVE, colors=256)
+    stream = io.BytesIO()
+    out.save(stream, format="png", optimize=True)
+    return stream.getvalue()
+
+
+def build_paper(paperdesc: StudentPaperDesc) -> StudentPaper:
+    ret = StudentPaper(login=paperdesc.login, grade=paperdesc.grade, pages=[])
+    for p in mercure_pdfimg.RawPdf(open(paperdesc.paper_pdf, "rb").read()):
+        ret.pages.append(p.content)
+
+    for abstract in paperdesc.abstracts:
+        ret.pages.append(abstract_to_png(abstract))
+
+    return ret
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description=textwrap.dedent(
+            """
+            Send associated file with grade to mercure app
+            """
+        ),
+    )
+
+    parser.add_argument(
+        "-u",
+        "--url",
+        help="Url of the exam manage page. If not provided, the url is asked.",
+        default=None,
+    )
+
+    parser.add_argument(
+        "-l",
+        "--login",
+        help="""
+                Login used for authentification. If not provided, the login is
+                asked. If the environment var MERCURE_AUTH_PASSWD is set, the
+                value is used for auth, otherwise the passwed is asked.
+                """,
+        default=None,
+    )
+
+    parser.add_argument("-q", "--quiet", help="Quiet mode.", action="store_true")
+
+    parser.add_argument(
+        "config_yaml",
+        help="Exam configuration and grading scale.",
+    )
+
+    parser.add_argument(
+        "exam_csv",
+        help="Exam csv export from AMC.",
+    )
+
+    parser.add_argument(
+        "pdf_dir",
+        metavar="pdf_dir/",
+        help="Directory which contains exported pdf from AMC.",
+    )
+
+    args = parser.parse_args()
+
+    config = yaml.safe_load(open(args.config_yaml))
+    questions = tuple(
+        itertools.chain.from_iterable(x["questions"] for x in config["config"])
+    )
+
+    csvcontent = open(args.exam_csv, "rb").read()
+    encoding = cchardet.detect(csvcontent)["encoding"]
+    if encoding is None:
+        print("CSV encoding error", file=sys.stderr)
+        sys.exit(1)
+    csvdecoded = csvcontent.decode(encoding)
+    dialect = csv.Sniffer().sniff(csvdecoded)
+    csvreader = csv.DictReader(io.StringIO(csvdecoded), dialect=dialect)
+    cols = csvreader.fieldnames
+    logincols = [c for c in cols if "login" in c.lower()]
+    if not logincols:
+        print("csvfile: Login column not found", file=sys.stderr)
+        sys.exit(1)
+    if len(logincols) > 1:
+        print("csvfile: Login column is ambiguous", file=sys.stderr)
+        sys.exit(1)
+    logincol = logincols[0]
+    papercol = cols[0]
+    if not all(c in cols for c in questions):
+        print(
+            f"Some columns in config are not in csv file: {tuple(c for c in questions if c not in cols)!r}",
+            file=sys.stderr,
+        )
+        sys.exit()
+    data = [
+        StudentRes(
+            paper=int(row[papercol]),
+            login=row[logincol],
+            questions={q: int(row[q]) for q in questions},
+        )
+        for row in csvreader
+    ]
+
+    pdf_dir = pathlib.Path(args.pdf_dir)
+    pdfs = {
+        int(x.name.split("-")[0]): x
+        for x in pdf_dir.iterdir()
+        if x.is_file() and x.suffix == ".pdf" and "-" in x.name
+    }
+    if not pdfs:
+        print("No pdf found in directory", file=sys.stderr)
+        sys.exit(1)
+
+    not_found_pdf = tuple((s.paper, s.login) for s in data if s.paper not in pdfs)
+    if not_found_pdf:
+        print(f"Following pdf are not found: {not_found_pdf!r}.", file=sys.stderr)
+        sys.exit(1)
+
+    if args.url is None: