100 lines
2.7 KiB
Python
100 lines
2.7 KiB
Python
from bopytex.tasks import Task, activate_corr_on, compile_pdf, generate, join_pdfs
|
|
import bopytex.planner.naming as naming
|
|
from bopytex.planner.exceptions import PlannerMissingOption
|
|
import csv
|
|
|
|
|
|
def build_subject_list_from_infos(infos: list[dict]) -> list[dict]:
|
|
subjects = []
|
|
digit = len(str(len(infos)))
|
|
for i, infos in enumerate(infos):
|
|
subjects.append({"number": str(i + 1).zfill(digit), **infos})
|
|
return subjects
|
|
|
|
|
|
def build_subject_list_from_qty(qty: int) -> list[dict]:
|
|
subjects = []
|
|
digit = len(str(qty))
|
|
for i in range(qty):
|
|
subjects.append({"number": str(i + 1).zfill(digit)})
|
|
return subjects
|
|
|
|
|
|
def planner(options: dict) -> list[Task]:
|
|
try:
|
|
students_csv = options["students_csv"]
|
|
assert options["students_csv"] != ""
|
|
|
|
except (KeyError, AssertionError):
|
|
try:
|
|
quantity_subjects = options["quantity_subjects"]
|
|
assert options["quantity_subjects"] != 0
|
|
except (KeyError, AssertionError):
|
|
raise PlannerMissingOption("students_csv or quantity_subjects is required")
|
|
else:
|
|
options["subjects"] = build_subject_list_from_qty(qty=quantity_subjects)
|
|
|
|
else:
|
|
with open(students_csv, "r") as csv_file:
|
|
infos = csv.DictReader(csv_file)
|
|
options["subjects"] = build_subject_list_from_infos(infos)
|
|
|
|
return tasks_builder(options)
|
|
|
|
|
|
def tasks_builder(
|
|
options: dict,
|
|
) -> list[Task]:
|
|
|
|
opt = {
|
|
"corr": False,
|
|
"no_join": False,
|
|
"no_pdf": False,
|
|
}
|
|
opt.update(options)
|
|
|
|
try:
|
|
template = opt["template"]
|
|
subjects = opt["subjects"]
|
|
corr = opt["corr"]
|
|
no_join = opt["no_join"]
|
|
no_pdf = opt["no_pdf"]
|
|
except KeyError:
|
|
raise PlannerMissingOption("An option is missing")
|
|
|
|
tasks = []
|
|
|
|
pdfs = []
|
|
corr_pdfs = []
|
|
|
|
for subject in subjects:
|
|
source = naming.template2source(template, subject)
|
|
args = {**subject, **options}
|
|
|
|
tasks.append(generate(template, args, source))
|
|
|
|
if not no_pdf:
|
|
pdf = naming.source2pdf(source)
|
|
tasks.append(compile_pdf(source, pdf))
|
|
pdfs.append(pdf)
|
|
|
|
if corr:
|
|
corr_source = naming.corr(source)
|
|
tasks.append(activate_corr_on(source, corr_source))
|
|
|
|
if not no_pdf:
|
|
corr_pdf = naming.source2pdf(corr_source)
|
|
tasks.append(compile_pdf(corr_source, corr_pdf))
|
|
corr_pdfs.append(corr_pdf)
|
|
|
|
if not no_join:
|
|
joined = naming.join(template)
|
|
if pdfs:
|
|
tasks.append(join_pdfs(pdfs, joined))
|
|
|
|
if corr_pdfs:
|
|
corr_joined = naming.corr(joined)
|
|
tasks.append(join_pdfs(corr_pdfs, corr_joined))
|
|
|
|
return tasks
|