ScoDoc/app/scodoc/sco_portal_apogee.py

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

##############################################################################
#
# Gestion scolarite IUT
#
# Copyright (c) 1999 - 2022 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
#
##############################################################################

"""Liaison avec le portail ENT (qui donne accès aux infos Apogée)
"""
import datetime
import os
import time
import urllib
import xml
import xml.sax.saxutils
import xml.dom.minidom

import app.scodoc.sco_utils as scu
from app import log
from app.scodoc.sco_exceptions import ScoValueError
from app.scodoc import sco_preferences

SCO_CACHE_ETAPE_FILENAME = os.path.join(scu.SCO_TMP_DIR, "last_etapes.xml")


def has_portal():
    "True if we are connected to a portal"
    return get_portal_url()


class PortalInterface(object):
    def __init__(self):
        self.first_time = True

    def get_portal_url(self):
        "URL of portal"
        portal_url = sco_preferences.get_preference("portal_url")
        if portal_url and not portal_url.endswith("/"):
            portal_url += "/"
        if self.first_time:
            if portal_url:
                log("Portal URL=%s" % portal_url)
            else:
                log("Portal not configured")
            self.first_time = False
        return portal_url

    def get_etapes_url(self):
        "Full URL of service giving list of etapes (in XML)"
        etapes_url = sco_preferences.get_preference("etapes_url")
        if not etapes_url:
            # Default:
            portal_url = self.get_portal_url()
            if not portal_url:
                return None
            api_ver = self.get_portal_api_version()
            if api_ver > 1:
                etapes_url = portal_url + "scodocEtapes.php"
            else:
                etapes_url = portal_url + "getEtapes.php"
        return etapes_url

    def get_etud_url(self):
        "Full URL of service giving list of students (in XML)"
        etud_url = sco_preferences.get_preference("etud_url")
        if not etud_url:
            # Default:
            portal_url = self.get_portal_url()
            if not portal_url:
                return None
            api_ver = self.get_portal_api_version()
            if api_ver > 1:
                etud_url = portal_url + "scodocEtudiant.php"
            else:
                etud_url = portal_url + "getEtud.php"
        return etud_url

    def get_photo_url(self):
        "Full URL of service giving photo of student"
        photo_url = sco_preferences.get_preference("photo_url")
        if not photo_url:
            # Default:
            portal_url = self.get_portal_url()
            if not portal_url:
                return None
            api_ver = self.get_portal_api_version()
            if api_ver > 1:
                photo_url = portal_url + "scodocPhoto.php"
            else:
                photo_url = portal_url + "getPhoto.php"
        return photo_url

    def get_maquette_url(self):
        """Full URL of service giving Apogee maquette pour une étape (fichier "CSV")"""
        maquette_url = sco_preferences.get_preference("maquette_url")
        if not maquette_url:
            # Default:
            portal_url = self.get_portal_url()
            if not portal_url:
                return None
            maquette_url = portal_url + "scodocMaquette.php"
        return maquette_url

    def get_portal_api_version(self):
        "API version of the portal software"
        api_ver = sco_preferences.get_preference("portal_api")
        if not api_ver:
            # Default:
            api_ver = 1
        return api_ver


_PI = PortalInterface()
get_portal_url = _PI.get_portal_url
get_etapes_url = _PI.get_etapes_url
get_etud_url = _PI.get_etud_url
get_photo_url = _PI.get_photo_url
get_maquette_url = _PI.get_maquette_url
get_portal_api_version = _PI.get_portal_api_version


def get_inscrits_etape(code_etape, anneeapogee=None, ntrials=2):
    """Liste des inscrits à une étape Apogée
    Result = list of dicts
    ntrials: try several time the same request, useful for some bad web services
    """
    log("get_inscrits_etape: code=%s anneeapogee=%s" % (code_etape, anneeapogee))
    if anneeapogee is None:
        anneeapogee = str(time.localtime()[0])

    etud_url = get_etud_url()
    api_ver = get_portal_api_version()
    if not etud_url:
        return []
    portal_timeout = sco_preferences.get_preference("portal_timeout")
    if api_ver > 1:
        req = (
            etud_url
            + "?"
            + urllib.parse.urlencode((("etape", code_etape), ("annee", anneeapogee)))
        )
    else:
        req = etud_url + "?" + urllib.parse.urlencode((("etape", code_etape),))
    actual_timeout = float(portal_timeout) / ntrials
    if portal_timeout > 0:
        actual_timeout = max(1, actual_timeout)
    for _ntrial in range(ntrials):
        doc = scu.query_portal(req, timeout=actual_timeout)
        if doc:
            break
    if not doc:
        raise ScoValueError(
            f"pas de réponse du portail ! <br>(timeout={portal_timeout}, requête: <tt>{req}</tt>)"
        )
    etuds = _normalize_apo_fields(xml_to_list_of_dicts(doc, req=req))

    # Filtre sur annee inscription Apogee:
    def check_inscription(e):
        if "inscription" in e:
            if e["inscription"] == anneeapogee:
                return True
            else:
                return False
        else:
            log(
                "get_inscrits_etape: pas inscription dans code_etape=%s e=%s"
                % (code_etape, e)
            )
            return False  # ??? pas d'annee d'inscription dans la réponse

    etuds = [e for e in etuds if check_inscription(e)]
    return etuds


def query_apogee_portal(**args):
    """Recupere les infos sur les etudiants nommés
    args: nom, prenom, code_nip
    (nom et prenom matchent des parties de noms)
    """
    etud_url = get_etud_url()
    api_ver = get_portal_api_version()
    if not etud_url:
        return []
    if api_ver > 1:
        if args["nom"] or args["prenom"]:
            # Ne fonctionne pas avec l'API 2 sur nom et prenom
            # XXX TODO : va poser problème pour la page modif données étudiants : A VOIR
            return []
    portal_timeout = sco_preferences.get_preference("portal_timeout")
    req = etud_url + "?" + urllib.parse.urlencode(list(args.items()))
    doc = scu.query_portal(req, timeout=portal_timeout)  # sco_utils
    return xml_to_list_of_dicts(doc, req=req)


def xml_to_list_of_dicts(doc, req=None):
    """Convert an XML 1.0 str to a list of dicts."""
    if not doc:
        return []
    # Fix for buggy XML returned by some APIs (eg USPN)
    invalid_entities = {
        "&CCEDIL;": "Ç",
        "& ": "&amp; ",  # only when followed by a space (avoid affecting entities)
        # to be completed...
    }
    for k in invalid_entities:
        doc = doc.replace(k, invalid_entities[k])
    #
    try:
        dom = xml.dom.minidom.parseString(doc)
    except xml.parsers.expat.ExpatError as e:
        # Find faulty part
        err_zone = doc.splitlines()[e.lineno - 1][e.offset : e.offset + 20]
        # catch bug: log and re-raise exception
        log(
            "xml_to_list_of_dicts: exception in XML parseString\ndoc:\n%s\n(end xml doc)\n"
            % doc
        )
        raise ScoValueError(
            'erreur dans la réponse reçue du portail ! (peut être : "%s")' % err_zone
        )
    infos = []
    try:
        if dom.childNodes[0].nodeName != "etudiants":
            raise ValueError
        etudiants = dom.getElementsByTagName("etudiant")
        for etudiant in etudiants:
            d = {}
            # recupere toutes les valeurs <valeur>XXX</valeur>
            for e in etudiant.childNodes:
                if e.nodeType == e.ELEMENT_NODE:
                    childs = e.childNodes
                    if len(childs):
                        d[str(e.nodeName)] = childs[0].nodeValue
            infos.append(d)
    except:
        log("*** invalid XML response from Etudiant Web Service")
        log("req=%s" % req)
        log("doc=%s" % doc)
        raise ValueError("invalid XML response from Etudiant Web Service\n%s" % doc)
    return infos


def get_infos_apogee_allaccents(nom, prenom):
    "essai recup infos avec differents codages des accents"
    if nom:
        nom_noaccents = scu.suppress_accents(nom)
    else:
        nom_noaccents = nom

    if prenom:
        prenom_noaccents = scu.suppress_accents(prenom)
    else:
        prenom_noaccents = prenom

    # avec accents
    infos = query_apogee_portal(nom=nom, prenom=prenom)
    # sans accents
    if nom != nom_noaccents or prenom != prenom_noaccents:
        infos += query_apogee_portal(nom=nom_noaccents, prenom=prenom_noaccents)
    return infos


def get_infos_apogee(nom, prenom):
    """recupere les codes Apogee en utilisant le web service CRIT"""
    if (not nom) and (not prenom):
        return []
    # essaie plusieurs codages: tirets, accents
    infos = get_infos_apogee_allaccents(nom, prenom)
    nom_st = nom.replace("-", " ")
    prenom_st = prenom.replace("-", " ")
    if nom_st != nom or prenom_st != prenom:
        infos += get_infos_apogee_allaccents(nom_st, prenom_st)
    # si pas de match et nom ou prenom composé, essaie en coupant
    if not infos:
        if nom:
            nom1 = nom.split()[0]
        else:
            nom1 = nom
        if prenom:
            prenom1 = prenom.split()[0]
        else:
            prenom1 = prenom
        if nom != nom1 or prenom != prenom1:
            infos += get_infos_apogee_allaccents(nom1, prenom1)
    return infos


def get_etud_apogee(code_nip):
    """Informations à partir du code NIP.
    None si pas d'infos sur cet etudiant.
    Exception si reponse invalide.
    """
    if not code_nip:
        return {}
    etud_url = get_etud_url()
    if not etud_url:
        return {}
    portal_timeout = sco_preferences.get_preference("portal_timeout")
    req = etud_url + "?" + urllib.parse.urlencode((("nip", code_nip),))
    doc = scu.query_portal(req, timeout=portal_timeout)
    d = _normalize_apo_fields(xml_to_list_of_dicts(doc, req=req))
    if not d:
        return None
    if len(d) > 1:
        raise ValueError("invalid XML response from Etudiant Web Service\n%s" % doc)
    return d[0]


def get_default_etapes():
    """Liste par défaut, lue du fichier de config"""
    filename = scu.SCO_TOOLS_DIR + "/default-etapes.txt"
    log("get_default_etapes: reading %s" % filename)
    f = open(filename)
    etapes = {}
    for line in f.readlines():
        line = line.strip()
        if line and line[0] != "#":
            dept, code, intitule = [x.strip() for x in line.split(":")]
            if dept and code:
                if dept in etapes:
                    etapes[dept][code] = intitule
                else:
                    etapes[dept] = {code: intitule}
    return etapes


def _parse_etapes_from_xml(doc):
    """
    may raise exception if invalid xml doc
    """
    xml_etapes_by_dept = sco_preferences.get_preference("xml_etapes_by_dept")
    # parser XML
    dom = xml.dom.minidom.parseString(doc)
    infos = {}
    if dom.childNodes[0].nodeName != "etapes":
        raise ValueError
    if xml_etapes_by_dept:
        # Ancien format XML avec des sections par departement:
        for d in dom.childNodes[0].childNodes:
            if d.nodeType == d.ELEMENT_NODE:
                dept = d.nodeName
                _xml_list_codes(infos, dept, d.childNodes)
    else:
        # Toutes les étapes:
        dept = ""
        _xml_list_codes(infos, "", dom.childNodes[0].childNodes)
    return infos


def get_etapes_apogee():
    """Liste des etapes apogee
    { departement : { code_etape : intitule } }
    Demande la liste au portail, ou si échec utilise liste
    par défaut
    """
    etapes_url = get_etapes_url()
    infos = {}
    if etapes_url:
        portal_timeout = sco_preferences.get_preference("portal_timeout")
        log(
            "get_etapes_apogee: requesting '%s' with timeout=%s"
            % (etapes_url, portal_timeout)
        )
        doc = scu.query_portal(etapes_url, timeout=portal_timeout)
        try:
            infos = _parse_etapes_from_xml(doc)
            # cache le resultat (utile si le portail repond de façon intermitente)
            if infos:
                log("get_etapes_apogee: caching result")
                with open(SCO_CACHE_ETAPE_FILENAME, "w") as f:
                    f.write(doc)
        except:
            log("invalid XML response from getEtapes Web Service\n%s" % etapes_url)
            # Avons nous la copie d'une réponse récente ?
            try:
                doc = open(SCO_CACHE_ETAPE_FILENAME).read()
                infos = _parse_etapes_from_xml(doc)
                log("using last saved version from " + SCO_CACHE_ETAPE_FILENAME)
            except:
                infos = {}
    else:
        # Pas de portail: utilise étapes par défaut livrées avec ScoDoc
        log("get_etapes_apogee: no configured URL (using default file)")
        infos = get_default_etapes()
    return infos


def _xml_list_codes(target_dict, dept, nodes):
    for e in nodes:
        if e.nodeType == e.ELEMENT_NODE:
            intitule = e.childNodes[0].nodeValue
            code = e.attributes["code"].value
            if dept in target_dict:
                target_dict[dept][code] = intitule
            else:
                target_dict[dept] = {code: intitule}


def get_etapes_apogee_dept():
    """Liste des etapes apogee pour ce departement.
    Utilise la propriete 'portal_dept_name' pour identifier le departement.

    Si xml_etapes_by_dept est faux (nouveau format XML depuis sept 2014),
    le departement n'est pas utilisé: toutes les étapes sont présentées.

    Returns [ ( code, intitule) ], ordonnée
    """
    xml_etapes_by_dept = sco_preferences.get_preference("xml_etapes_by_dept")
    if xml_etapes_by_dept:
        portal_dept_name = sco_preferences.get_preference("portal_dept_name")
        log('get_etapes_apogee_dept: portal_dept_name="%s"' % portal_dept_name)
    else:
        portal_dept_name = ""
        log("get_etapes_apogee_dept: pas de sections par departement")

    infos = get_etapes_apogee()
    if portal_dept_name and portal_dept_name not in infos:
        log(
            "get_etapes_apogee_dept: pas de section '%s' dans la reponse portail"
            % portal_dept_name
        )
        return []
    if portal_dept_name:
        etapes = list(infos[portal_dept_name].items())
    else:
        # prend toutes les etapes
        etapes = []
        for k in infos.keys():
            etapes += list(infos[k].items())

    etapes.sort()  # tri sur le code etape
    return etapes


def _portal_date_dmy2date(s):
    """date inscription renvoyée sous la forme dd/mm/yy
    renvoie un objet date, ou None
    """
    s = s.strip()
    if not s:
        return None
    else:
        d, m, y = [int(x) for x in s.split("/")]  # raises ValueError if bad format
        if y < 100:
            y += 2000  # 21ème siècle
        return datetime.date(y, m, d)


def _normalize_apo_fields(infolist):
    """
    infolist: liste de dict renvoyés par le portail Apogee

    recode les champs: paiementinscription (-> booleen), datefinalisationinscription (date)
    ajoute le champs 'paiementinscription_str' : 'ok', 'Non' ou '?'
    ajoute les champs 'etape' (= None) et 'prenom' ('') s'ils ne sont pas présents.
    """
    for infos in infolist:
        if "paiementinscription" in infos:
            infos["paiementinscription"] = (
                infos["paiementinscription"].lower() == "true"
            )
            if infos["paiementinscription"]:
                infos["paiementinscription_str"] = "ok"
            else:
                infos["paiementinscription_str"] = "Non"
        else:
            infos["paiementinscription"] = None
            infos["paiementinscription_str"] = "?"

        if "datefinalisationinscription" in infos:
            infos["datefinalisationinscription"] = _portal_date_dmy2date(
                infos["datefinalisationinscription"]
            )
            infos["datefinalisationinscription_str"] = infos[
                "datefinalisationinscription"
            ].strftime("%d/%m/%Y")
        else:
            infos["datefinalisationinscription"] = None
            infos["datefinalisationinscription_str"] = ""

        if "etape" not in infos:
            infos["etape"] = None

        if "prenom" not in infos:
            infos["prenom"] = ""

    return infolist


def check_paiement_etuds(etuds):
    """Interroge le portail pour vérifier l'état de "paiement" et l'étape d'inscription.

    Seuls les etudiants avec code NIP sont renseignés.

    Renseigne l'attribut booleen 'paiementinscription' dans chaque etud.

    En sortie: modif les champs de chaque etud
    'paiementinscription' : True, False ou None
    'paiementinscription_str' : 'ok', 'Non' ou '?' ou '(pas de code)'
    'etape' : etape Apogee ou None
    """
    # interrogation séquentielle longue...
    for etud in etuds:
        if "code_nip" not in etud:
            etud["paiementinscription"] = None
            etud["paiementinscription_str"] = "(pas de code)"
            etud["datefinalisationinscription"] = None
            etud["datefinalisationinscription_str"] = "NA"
            etud["etape"] = None
        else:
            # Modifie certains champs de l'étudiant:
            infos = get_etud_apogee(etud["code_nip"])
            if infos:
                for k in (
                    "paiementinscription",
                    "paiementinscription_str",
                    "datefinalisationinscription",
                    "datefinalisationinscription_str",
                    "etape",
                ):
                    etud[k] = infos[k]
            else:
                etud["datefinalisationinscription"] = None
                etud["datefinalisationinscription_str"] = "Erreur"
                etud["datefinalisationinscription"] = None
                etud["paiementinscription_str"] = "(pb cnx Apogée)"


def get_maquette_apogee(etape="", annee_scolaire="") -> str:
    """Maquette CSV Apogee pour une étape et une annee scolaire"""
    maquette_url = get_maquette_url()
    if not maquette_url:
        return None
    portal_timeout = sco_preferences.get_preference("portal_timeout")
    req = (
        maquette_url
        + "?"
        + urllib.parse.urlencode((("etape", etape), ("annee", annee_scolaire)))
    )
    doc = scu.query_portal(req, timeout=portal_timeout)
    return doc