ScoDoc/app/formations/formation_io.py

# -*- mode: python -*-
# -*- coding: utf-8 -*-

##############################################################################
#
# Gestion scolarite IUT
#
# Copyright (c) 1999 - 2024 Emmanuel Viennet.  All rights reserved.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
#
#   Emmanuel Viennet      emmanuel.viennet@viennet.net
#
##############################################################################

"""Import / Export de formations
"""
import xml.dom.minidom

import flask
from flask import flash, g, request, url_for
from flask_login import current_user

import app.scodoc.sco_utils as scu
from app import db
from app import log
from app.formations import edit_ue
from app.models import Formation, FormSemestre, Matiere, Module, UniteEns
from app.models import ScolarNews
from app.models.but_refcomp import (
    ApcAppCritique,
    ApcCompetence,
    ApcNiveau,
    ApcParcours,
    ApcReferentielCompetences,
)
from app.scodoc import sco_cache
from app.scodoc import codes_cursus
from app.scodoc import sco_preferences
from app.scodoc import sco_xml
from app.scodoc.gen_tables import GenTable
from app.scodoc.sco_exceptions import ScoValueError, ScoFormatError
from app.scodoc.sco_permissions import Permission
import sco_version


def formation_export_dict(
    formation: Formation,
    export_ids=False,
    export_tags=True,
    export_external_ues=False,
    export_codes_apo=True,
    ac_as_list=False,
    ue_reference_style="id",
) -> dict:
    """Get a formation, with UE, matieres, modules...
    as a deep dict.
    ac_as_list spécifie le format des Appentissages Critiques.
    """
    f_dict = formation.to_dict(with_refcomp_attrs=True)
    if not export_ids:
        del f_dict["id"]
        del f_dict["formation_id"]
        del f_dict["dept_id"]
    ues = formation.ues
    if not export_external_ues:
        ues = ues.filter_by(is_external=False)
    ues = ues.all()
    ues.sort(key=lambda u: (u.semestre_idx or 0, u.numero or 0, u.acronyme))
    f_dict["ue"] = []
    ue: UniteEns
    for ue in ues:
        ue_dict = ue.to_dict()
        f_dict["ue"].append(ue_dict)
        ue_dict.pop("module_ue_coefs", None)
        if formation.is_apc():
            # BUT: indique niveau de compétence associé à l'UE
            if ue.niveau_competence:
                ue_dict["apc_niveau_competence_titre"] = (
                    ue.niveau_competence.competence.titre
                )
                ue_dict["apc_niveau_libelle"] = ue.niveau_competence.libelle
                ue_dict["apc_niveau_annee"] = ue.niveau_competence.annee
                ue_dict["apc_niveau_ordre"] = ue.niveau_competence.ordre

        # pour les coefficients:
        ue_dict["reference"] = ue.id if ue_reference_style == "id" else ue.acronyme
        if not export_ids:
            for id_id in (
                "id",
                "ue_id",
                "formation_id",
                "parcour_id",
                "niveau_competence_id",
            ):
                ue_dict.pop(id_id, None)

        if not export_codes_apo:
            ue_dict.pop("code_apogee", None)
            ue_dict.pop("code_apogee_rcue", None)
        if ue_dict.get("ects") is None:
            ue_dict.pop("ects", None)
        mats = ue.matieres.all()
        mats.sort(key=lambda m: m.numero)
        mats_dict = [mat.to_dict() for mat in mats]
        ue_dict["matiere"] = mats_dict
        for mat_d in mats_dict:
            matiere_id = mat_d["matiere_id"]
            if not export_ids:
                del mat_d["id"]
                del mat_d["matiere_id"]
                del mat_d["ue_id"]
            mat = db.session.get(Matiere, matiere_id)
            mods = mat.modules.all()
            mods.sort(key=lambda m: (m.numero, m.code))
            mat_d["module"] = [mod.to_dict(convert_objects=True) for mod in mods]
            for module, mod_d in zip(mods, mat_d["module"]):
                if export_tags:
                    tags = [t.title for t in module.tags]
                    if tags:
                        mod_d["tags"] = [{"name": x} for x in tags]
                #
                if module.is_apc():
                    # Exporte les coefficients
                    if ue_reference_style == "id":
                        mod_d["coefficients"] = [
                            {"ue_reference": str(ue_id), "coef": str(coef)}
                            for (ue_id, coef) in module.get_ue_coef_dict().items()
                        ]
                    else:
                        mod_d["coefficients"] = [
                            {"ue_reference": ue_acronyme, "coef": str(coef)}
                            for (
                                ue_acronyme,
                                coef,
                            ) in module.get_ue_coef_dict_acronyme().items()
                        ]
                    # Et les parcours
                    mod_d["parcours"] = [
                        p.to_dict(with_annees=False) for p in module.parcours
                    ]
                    # Et les AC
                    if ac_as_list:
                        # XML préfère une liste
                        mod_d["app_critiques"] = [
                            x.to_dict(with_code=True) for x in module.app_critiques
                        ]
                    else:
                        mod_d["app_critiques"] = {
                            x.code: x.to_dict() for x in module.app_critiques
                        }
                if not export_ids:
                    del mod_d["id"]
                    del mod_d["ue_id"]
                    del mod_d["matiere_id"]
                    del mod_d["module_id"]
                    del mod_d["formation_id"]
                if not export_codes_apo:
                    del mod_d["code_apogee"]
                if mod_d["ects"] is None:
                    del mod_d["ects"]
    return f_dict


def formation_export(
    formation_id,
    export_ids=False,
    export_tags=True,
    export_external_ues=False,
    export_codes_apo=True,
    fmt=None,
) -> flask.Response | dict:
    """Get a formation, with UE, matieres, modules
    in desired format
    """
    if fmt not in (None, "xml", "json"):
        raise ScoValueError("Format invalide")
    formation = Formation.get_formation(formation_id)
    f_dict = formation_export_dict(
        formation,
        export_ids=export_ids,
        export_tags=export_tags,
        export_external_ues=export_external_ues,
        export_codes_apo=export_codes_apo,
        ac_as_list=fmt == "xml",
    )
    if fmt is None:
        return f_dict

    filename = f"""scodoc_formation_{formation.departement.acronym
        }_{formation.acronyme or ''}_v{formation.version}"""
    return scu.sendResult(
        f_dict,
        name="formation",
        fmt=fmt,
        force_outer_xml_tag=False,
        attached=True,
        filename=filename,
    )


def _formation_retreive_refcomp(f_dict: dict) -> int:
    """Recherche si on un référentiel de compétence chargé pour
    cette formation: utilise comme clé (version_orebut, specialite, type_titre)
    Retourne: referentiel_competence_id ou None
    """
    refcomp_version_orebut = f_dict.get("refcomp_version_orebut")
    refcomp_specialite = f_dict.get("refcomp_specialite")
    refcomp_type_titre = f_dict.get("refcomp_type_titre")
    if all((refcomp_version_orebut, refcomp_specialite, refcomp_type_titre)):
        refcomp = ApcReferentielCompetences.query.filter_by(
            dept_id=g.scodoc_dept_id,
            type_titre=refcomp_type_titre,
            specialite=refcomp_specialite,
            version_orebut=refcomp_version_orebut,
        ).first()
        if refcomp:
            return refcomp.id
        flash(
            f"""Impossible de trouver le référentiel de compétence pour {
                refcomp_specialite} : est-il chargé ?"""
        )
    return None


def _formation_retreive_apc_niveau(
    referentiel_competence_id: int, ue_dict: dict
) -> int | None:
    """Recherche dans le ref. de comp. un niveau pour cette UE.
    Utilise (libelle, annee, ordre) comme clé, ou
    (competence_titre, libelle, annee, ordre) si présent.
    """
    libelle = ue_dict.get("apc_niveau_libelle")
    annee = ue_dict.get("apc_niveau_annee")
    ordre = ue_dict.get("apc_niveau_ordre")
    competence_titre = ue_dict.get("apc_niveau_competence_titre")
    niveau = None
    if all((competence_titre, libelle, annee, ordre)):
        niveau = (
            ApcNiveau.query.filter_by(libelle=libelle, annee=annee, ordre=ordre)
            .join(ApcCompetence)
            .filter_by(referentiel_id=referentiel_competence_id, titre=competence_titre)
        ).first()

    elif all((libelle, annee, ordre)):
        niveau = (
            ApcNiveau.query.filter_by(libelle=libelle, annee=annee, ordre=ordre)
            .join(ApcCompetence)
            .filter_by(referentiel_id=referentiel_competence_id)
        ).first()

    return niveau.id if niveau is not None else None


def formation_import_xml(doc: str | bytes, import_tags=True, use_local_refcomp=False):
    """Create a formation from XML representation
    (format dumped by formation_export( fmt='xml' ))
    XML may contain object (UE, modules) ids: this function returns two
    dicts mapping these ids to the created ids.

    Args:
        doc:    str, xml data
        import_tags: if false, does not import tags on modules.
        use_local_refcomp: if True, utilise les id vers les ref. de compétences.
    Returns:
        formation_id, modules_old2new, ues_old2new
    """
    from app.formations import edit_formation

    if isinstance(doc, bytes):
        doc = doc.decode(scu.SCO_ENCODING)
    try:
        dom = xml.dom.minidom.parseString(sco_xml.remove_control_characters(doc))
    except Exception as exc:
        log(f"formation_import_xml: invalid XML data:\n{exc}")
        raise ScoValueError(f"Fichier XML invalide {exc}") from exc

    try:
        f = dom.getElementsByTagName("formation")[0]  # or dom.documentElement
        xml_dicts = sco_xml.xml_to_dicts(f)
    except Exception as exc:
        raise ScoFormatError(
            """Ce document xml ne correspond pas à un programme exporté par ScoDoc.
            (élément 'formation' inexistant par exemple)."""
        ) from exc
    assert xml_dicts[0] == "formation"
    f_dict = xml_dicts[1]
    f_dict["dept_id"] = g.scodoc_dept_id
    # Pour les clonages, on prend le refcomp_id donné:
    referentiel_competence_id = (
        f_dict.get("referentiel_competence_id") if use_local_refcomp else None
    )
    # Sinon, on cherche a retrouver le ref. comp.
    if referentiel_competence_id is None:
        referentiel_competence_id = _formation_retreive_refcomp(f_dict)
    f_dict["referentiel_competence_id"] = referentiel_competence_id
    # find new version number
    acronyme_lower = f_dict["acronyme"].lower() if f_dict["acronyme"] else ""
    titre_lower = f_dict["titre"].lower() if f_dict["titre"] else ""
    formations: list[Formation] = Formation.query.filter_by(
        dept_id=f_dict["dept_id"]
    ).filter(
        db.func.lower(Formation.acronyme) == acronyme_lower,
        db.func.lower(Formation.titre) == titre_lower,
    )
    if formations.count():
        version = max(f.version or 0 for f in formations)
    else:
        version = 0
    f_dict["version"] = version + 1

    # create formation
    formation = edit_formation.do_formation_create(f_dict)
    log(f"formation {formation.id} created")

    ues_old2new = {}  # xml ue_id : new ue_id
    modules_old2new = {}  # xml module_id : new module_id
    # (nb: mecanisme utilise pour cloner semestres seulement, pas pour I/O XML)

    ue_reference_to_id = {}  # pour les coefs APC (map reference -> ue_id)
    modules_a_coefficienter = []  # Liste des modules avec coefs APC
    with sco_cache.DeferredSemCacheManager():
        # -- create UEs
        for ue_info in xml_dicts[2]:
            assert ue_info[0] == "ue"
            ue_info[1]["formation_id"] = formation.id
            if "ue_id" in ue_info[1]:
                xml_ue_id = int(ue_info[1]["ue_id"])
                del ue_info[1]["ue_id"]
            else:
                xml_ue_id = None
            if referentiel_competence_id is None:
                if "niveau_competence_id" in ue_info[1]:
                    del ue_info[1]["niveau_competence_id"]
            else:
                ue_info[1]["niveau_competence_id"] = _formation_retreive_apc_niveau(
                    referentiel_competence_id, ue_info[1]
                )
            # Note: si le code est indiqué "" dans le xml, il faut le conserver vide
            # pour la comparaison ultérieure des formations XXX
            ue = edit_ue.do_ue_create(ue_info[1], allow_empty_ue_code=True)
            assert ue
            if xml_ue_id:
                ues_old2new[xml_ue_id] = ue.id

            # élément optionnel présent dans les exports BUT:
            ue_reference = ue_info[1].get("reference")
            if ue_reference:
                ue_reference_to_id[int(ue_reference)] = ue.id

            # -- Create matieres
            for mat_info in ue_info[2]:
                # Backward compat: un seul parcours par UE (ScoDoc < 9.4.71)
                if mat_info[0] == "parcour":
                    # Parcours (BUT)
                    code_parcours = mat_info[1]["code"]
                    parcour = ApcParcours.query.filter_by(
                        code=code_parcours,
                        referentiel_id=referentiel_competence_id,
                    ).first()
                    if parcour:
                        ue.parcours = [parcour]
                        db.session.add(ue)
                    else:
                        flash(f"Attention: parcours {code_parcours} inexistant !")
                        log(f"Warning: parcours {code_parcours} inexistant !")
                    continue
                elif mat_info[0] == "parcours":
                    # Parcours (BUT), liste (ScoDoc > 9.4.70), avec ECTS en option
                    code_parcours = mat_info[1]["code"]
                    ue_parcour_ects = mat_info[1].get("ects")
                    parcour = ApcParcours.query.filter_by(
                        code=code_parcours,
                        referentiel_id=referentiel_competence_id,
                    ).first()
                    if parcour:
                        ue.parcours.append(parcour)
                    else:
                        flash(f"Attention: parcours {code_parcours} inexistant !")
                        log(f"Warning: parcours {code_parcours} inexistant !")
                    if ue_parcour_ects is not None:
                        ue.set_ects(ue_parcour_ects, parcour)
                    db.session.add(ue)
                    continue

                assert mat_info[0] == "matiere"
                mat_info[1]["ue_id"] = ue.id
                mat = Matiere.create_from_dict(mat_info[1])
                mat_id = mat.id
                # -- create modules
                for mod_info in mat_info[2]:
                    assert mod_info[0] == "module"
                    if "module_id" in mod_info[1]:
                        xml_module_id = int(mod_info[1]["module_id"])
                        del mod_info[1]["module_id"]
                    else:
                        xml_module_id = None
                    mod_info[1]["formation_id"] = formation.id
                    mod_info[1]["matiere_id"] = mat_id
                    mod_info[1]["ue_id"] = ue.id
                    if not "module_type" in mod_info[1]:
                        mod_info[1]["module_type"] = scu.ModuleType.STANDARD
                    module = Module.create_from_dict(
                        mod_info[1], news=True, inval_cache=True
                    )
                    if xml_module_id:
                        modules_old2new[int(xml_module_id)] = module.id
                    if len(mod_info) > 2:
                        tag_names = []
                        ue_coef_dict = {}
                        for child in mod_info[2]:
                            if child[0] == "tags" and import_tags:
                                tag_names.append(child[1]["name"])
                            elif child[0] == "coefficients":
                                ue_reference = int(child[1]["ue_reference"])
                                coef = float(child[1]["coef"])
                                ue_coef_dict[ue_reference] = coef
                            elif child[0] == "app_critiques" and (
                                referentiel_competence_id is not None
                            ):
                                ac_code = child[1]["code"]
                                ac = (
                                    ApcAppCritique.query.filter_by(code=ac_code)
                                    .join(ApcNiveau)
                                    .join(ApcCompetence)
                                    .filter_by(referentiel_id=referentiel_competence_id)
                                ).first()
                                if ac is not None:
                                    module.app_critiques.append(ac)
                                    db.session.add(module)
                                else:
                                    log(f"Warning: AC {ac_code} inexistant !")

                            elif child[0] == "parcours":
                                # Si on a un référentiel de compétences,
                                # associe les parcours de ce module (BUT)
                                if referentiel_competence_id is not None:
                                    code_parcours = child[1]["code"]
                                    parcour = ApcParcours.query.filter_by(
                                        code=code_parcours,
                                        referentiel_id=referentiel_competence_id,
                                    ).first()
                                    if parcour:
                                        module.parcours.append(parcour)
                                        db.session.add(module)
                                    else:
                                        log(
                                            f"Warning: parcours {code_parcours} inexistant !"
                                        )
                        if import_tags and tag_names:
                            module.set_tags(tag_names)
                        if module.is_apc() and ue_coef_dict:
                            modules_a_coefficienter.append((module, ue_coef_dict))
        # Fixe les coefs APC (à la fin pour que les UE soient créées)
        for module, ue_coef_dict_ref in modules_a_coefficienter:
            # remap ue ids:
            ue_coef_dict = {
                ue_reference_to_id[k]: v for (k, v) in ue_coef_dict_ref.items()
            }
            module.set_ue_coef_dict(ue_coef_dict)
        db.session.commit()
    return formation.id, modules_old2new, ues_old2new


def formation_list_table(detail: bool) -> GenTable:
    """List formation, grouped by titre and sorted by versions
    and listing associated semestres.
    If detail, add column with more details.
    returns a table
    """
    formations: list[Formation] = Formation.query.filter_by(dept_id=g.scodoc_dept_id)
    title = "Formations (programmes pédagogiques)"
    lockicon = scu.icontag(
        "lock32_img", title="Comporte des semestres verrouillés", border="0"
    )
    suppricon = scu.icontag(
        "delete_small_img", border="0", alt="supprimer", title="Supprimer"
    )
    editicon = scu.icontag(
        "edit_img", border="0", alt="modifier", title="Modifier titres et code"
    )

    editable = current_user.has_permission(Permission.EditFormation)
    can_implement = current_user.has_permission(Permission.EditFormSemestre)

    # Traduit/ajoute des champs à afficher:
    rows = []
    for formation in formations:
        acronyme_no_spaces = formation.acronyme.lower().replace(" ", "-")
        row = {
            "acronyme": formation.acronyme,
            "parcours_name": codes_cursus.get_cursus_from_code(
                formation.type_parcours
            ).NAME,
            "titre": formation.titre,
            "_titre_target": url_for(
                "notes.ue_table",
                scodoc_dept=g.scodoc_dept,
                formation_id=formation.id,
            ),
            "_titre_link_class": "stdlink",
            "_titre_id": f"""titre-{acronyme_no_spaces}""",
            "version": formation.version or 0,
            "commentaire": formation.commentaire or "",
            "referentiel": (
                f"""{formation.referentiel_competence.specialite} {
                    formation.referentiel_competence.get_version()}"""
                if formation.referentiel_competence
                else ""
            ),
            "_referentiel_target": (
                url_for(
                    "notes.refcomp_show",
                    scodoc_dept=g.scodoc_dept,
                    refcomp_id=formation.referentiel_competence.id,
                )
                if formation.referentiel_competence
                else ""
            ),
        }
        # Ajoute les semestres associés à chaque formation:
        row["formsemestres"] = formation.formsemestres.order_by(
            FormSemestre.date_debut
        ).all()
        row["sems_list_txt"] = ", ".join(s.session_id() for s in row["formsemestres"])
        row["_sems_list_txt_html"] = ", ".join(
            [
                f"""<a class="discretelink" href="{
                url_for("notes.formsemestre_status",
                    scodoc_dept=g.scodoc_dept, formsemestre_id=s.id
                )}">{s.session_id()}</a>"""
                for s in row["formsemestres"]
            ]
            + (
                [
                    f"""<a class="stdlink"
                href="{ url_for("notes.formsemestre_createwithmodules",
                scodoc_dept=g.scodoc_dept, formation_id=formation.id, semestre_id=1
                )
                }">ajouter</a>
            """
                ]
                if can_implement
                else []
            )
        )
        # Répartition des UEs dans les semestres
        # utilise pour voir si la formation couvre tous les semestres
        row["semestres_ues"] = ", ".join(
            "S" + str(x if (x is not None and x > 0) else "-")
            for x in sorted({(ue.semestre_idx or 0) for ue in formation.ues})
        )
        # Date surtout utilisées pour le tri:
        if row["formsemestres"]:
            row["date_fin_dernier_sem"] = row["formsemestres"][-1].date_fin.isoformat()
            row["annee_dernier_sem"] = row["formsemestres"][-1].date_fin.year
        else:
            row["date_fin_dernier_sem"] = ""
            row["annee_dernier_sem"] = 0
        #
        if formation.has_locked_sems():
            but_locked = lockicon
            but_suppr = '<span class="but_placeholder"></span>'
        else:
            but_locked = '<span class="but_placeholder"></span>'
            if editable:
                but_suppr = f"""<a class="stdlink" href="{
                        url_for("notes.formation_delete",
                        scodoc_dept=g.scodoc_dept, formation_id=formation.id
                    )}" id="delete-formation-{acronyme_no_spaces}">{suppricon}</a>"""
            else:
                but_suppr = '<span class="but_placeholder"></span>'
        if editable:
            but_edit = f"""<a class="stdlink" href="{
                url_for("notes.formation_edit", scodoc_dept=g.scodoc_dept, formation_id=formation.id)
                }" id="edit-formation-{acronyme_no_spaces}">{editicon}</a>"""
        else:
            but_edit = '<span class="but_placeholder"></span>'
        row["buttons"] = ""
        row["_buttons_html"] = but_locked + but_suppr + but_edit
        rows.append(row)
    # Tri par annee_dernier_sem, type, acronyme, titre, version décroissante
    # donc plus récemment utilisée en tête
    rows.sort(
        key=lambda row: (
            -row["annee_dernier_sem"],
            row["parcours_name"],
            row["acronyme"],
            row["titre"],
            -row["version"],
        )
    )
    for i, row in enumerate(rows):
        row["_buttons_order"] = f"{i:05d}"

    #
    columns_ids = (
        "buttons",
        "acronyme",
        "parcours_name",
        "formation_code",
        "version",
        "titre",
        "referentiel",
        "commentaire",
        "sems_list_txt",
    )
    if detail:
        columns_ids += ("annee_dernier_sem", "semestres_ues")
    titles = {
        "buttons": "",
        "commentaire": "Commentaire",
        "acronyme": "Acro.",
        "parcours_name": "Type",
        "titre": "Titre",
        "version": "Version",
        "formation_code": "Code",
        "sems_list_txt": "Semestres",
        "referentiel": "Réf. Comp.",
        "date_fin_dernier_sem": "Fin dernier sem.",
        "annee_dernier_sem": "Année dernier sem.",
        "semestres_ues": "Semestres avec UEs",
    }
    return GenTable(
        base_url=f"{request.base_url}" + ("?detail=on" if detail else ""),
        caption=title,
        columns_ids=columns_ids,
        html_caption=title,
        html_class="formation_list_table table_leftalign",
        html_sortable=True,
        html_with_td_classes=True,
        origin=f"Généré par {sco_version.SCONAME} le {scu.timedate_human_repr()}",
        page_title=title,
        pdf_title=title,
        preferences=sco_preferences.SemPreferences(),
        rows=rows,
        table_id="formation_list_table",
        titles=titles,
    )


def formation_create_new_version(formation_id, redirect=True):
    "duplicate formation, with new version number"
    formation = Formation.get_or_404(formation_id)
    resp = formation_export(
        formation_id, export_ids=True, export_external_ues=True, fmt="xml"
    )
    xml_data = resp.get_data(as_text=True)
    new_id, modules_old2new, ues_old2new = formation_import_xml(
        xml_data, use_local_refcomp=True
    )
    # news
    ScolarNews.add(
        typ=ScolarNews.NEWS_FORM,
        obj=new_id,
        text=f"Nouvelle version de la formation {formation.acronyme}",
        max_frequency=0,
    )
    if redirect:
        flash("Nouvelle version !")
        return flask.redirect(
            url_for(
                "notes.ue_table",
                scodoc_dept=g.scodoc_dept,
                formation_id=new_id,
                msg="Nouvelle version !",
            )
        )
    return new_id, modules_old2new, ues_old2new