ScoDoc/app/but/import_refcomp.py

##############################################################################
# ScoDoc
# Copyright (c) 1999 - 2022 Emmanuel Viennet.  All rights reserved.
# See LICENSE
##############################################################################
from xml.etree import ElementTree

import sqlalchemy

from app import db

from app.models.but_refcomp import (
    ApcReferentielCompetences,
    ApcCompetence,
    ApcSituationPro,
    ApcAppCritique,
    ApcComposanteEssentielle,
    ApcNiveau,
    ApcParcours,
    ApcAnneeParcours,
    ApcParcoursNiveauCompetence,
)
from app.scodoc.sco_exceptions import ScoFormatError, ScoValueError


def orebut_import_refcomp(xml_data: str, dept_id: int, orig_filename=None):
    """Importation XML Orébut
    peut lever TypeError ou ScoFormatError
    Résultat: instance de ApcReferentielCompetences
    """
    # Vérifie que le même fichier n'a pas déjà été chargé:
    if ApcReferentielCompetences.query.filter_by(
        scodoc_orig_filename=orig_filename, dept_id=dept_id
    ).count():
        raise ScoValueError(
            f"""Un référentiel a déjà été chargé d'un fichier de même nom. 
            ({orig_filename})
            Supprimez-le ou changez le nom du fichier."""
        )

    try:
        root = ElementTree.XML(xml_data)
    except ElementTree.ParseError as exc:
        raise ScoFormatError(f"fichier XML Orébut invalide (2): {exc.args}")
    if root.tag != "referentiel_competence":
        raise ScoFormatError("élément racine 'referentiel_competence' manquant")
    args = ApcReferentielCompetences.attr_from_xml(root.attrib)
    args["dept_id"] = dept_id
    args["scodoc_orig_filename"] = orig_filename
    ref = ApcReferentielCompetences(**args)
    db.session.add(ref)
    competences = root.find("competences")
    if not competences:
        raise ScoFormatError("élément 'competences' manquant")
    for competence in competences.findall("competence"):
        try:
            c = ApcCompetence(**ApcCompetence.attr_from_xml(competence.attrib))
            db.session.flush()
        except sqlalchemy.exc.IntegrityError as exc:
            # ne devrait plus se produire car pas d'unicité de l'id: donc inutile
            db.session.rollback()
            raise ScoValueError(
                f"""Un référentiel a déjà été chargé avec les mêmes compétences ! ({competence.attrib["id"]})
                """
            ) from exc
        ref.competences.append(c)
        # --- SITUATIONS
        situations = competence.find("situations")
        for situation in situations:
            libelle = "".join(situation.itertext()).strip()
            s = ApcSituationPro(competence_id=c.id, libelle=libelle)
            c.situations.append(s)
        # --- COMPOSANTES ESSENTIELLES
        composantes = competence.find("composantes_essentielles")
        for composante in composantes:
            libelle = "".join(composante.itertext()).strip()
            compo_ess = ApcComposanteEssentielle(libelle=libelle)
            c.composantes_essentielles.append(compo_ess)
        # --- NIVEAUX (années)
        niveaux = competence.find("niveaux")
        for niveau in niveaux:
            niv = ApcNiveau(**ApcNiveau.attr_from_xml(niveau.attrib))
            c.niveaux.append(niv)
            acs = niveau.find("acs")
            for ac in acs:
                libelle = "".join(ac.itertext()).strip()
                code = ac.attrib["code"]
                niv.app_critiques.append(ApcAppCritique(code=code, libelle=libelle))
    # --- PARCOURS
    parcours = root.find("parcours")
    if not parcours:
        raise ScoFormatError("élément 'parcours' manquant")
    for parcour in parcours.findall("parcour"):
        parc = ApcParcours(**ApcParcours.attr_from_xml(parcour.attrib))
        ref.parcours.append(parc)
        for annee in parcour.findall("annee"):
            a = ApcAnneeParcours(**ApcAnneeParcours.attr_from_xml(annee.attrib))
            parc.annees.append(a)
            for competence in annee.findall("competence"):
                comp_id_orebut = competence.attrib["id"]
                niveau = int(competence.attrib["niveau"])
                # Retrouve la competence
                comp = ref.competences.filter_by(id_orebut=comp_id_orebut).first()
                if comp is None:
                    raise ScoFormatError(f"competence {comp_id_orebut} non définie")
                ass = ApcParcoursNiveauCompetence(
                    niveau=niveau, annee_parcours=a, competence=comp
                )
                db.session.add(ass)

    db.session.commit()
    return ref


"""
xmlfile = open("but-RT-refcomp-30112021.xml")
tree = ElementTree.parse(xmlfile)
# get root element
root = tree.getroot()
assert root.tag == "referentiel_competence"

ref = ApcReferentielCompetences(**ApcReferentielCompetences.attr_from_xml(root.attrib))

competences = root.find("competences")
if not competences:
    raise ScoFormatError("élément 'competences' manquant")

competence = competences.findall("competence")[0]  # XXX

from app.but.import_refcomp import *
dept_id = models.Departement.query.first().id
data = open("tests/data/but-RT-refcomp-exemple.xml").read()
ref = orebut_import_refcomp(data, dept_id)
#------
from app.but.import_refcomp import *
ref = ApcReferentielCompetences.query.first()
p = ApcParcours(code="PARC", libelle="Parcours Test")
ref.parcours.append(p)
annee = ApcAnneeParcours(numero=1)
p.annees.append(annee)
annee.competences
c = ref.competences.filter_by(titre="Administrer").first()
annee.competences.append(c)
"""