ScoDoc/app/entreprises/app_relations_entreprises.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
#
#
##############################################################################

import os
from config import Config
import re
import requests
import glob

from flask_login import current_user

from app.entreprises.models import (
    Entreprise,
    EntrepriseCorrespondant,
    EntrepriseOffre,
    EntrepriseOffreDepartement,
    EntreprisePreferences,
)

from app import email
from app.scodoc import sco_preferences
from app.models import Departement
from app.scodoc.sco_permissions import Permission


def get_depts():
    """
    Retourne une liste contenant les l'id des départements des roles de l'utilisateur courant
    """
    depts = []
    for role in current_user.user_roles:
        dept_id = get_dept_id_by_acronym(role.dept)
        if dept_id is not None:
            depts.append(dept_id)
    return depts


def get_dept_id_by_acronym(acronym):
    """
    Retourne l'id d'un departement a l'aide de son acronym
    """
    dept = Departement.query.filter_by(acronym=acronym).first()
    if dept is not None:
        return dept.id
    return None


def check_offre_depts(depts, offre_depts):
    """
    Retourne vrai si l'utilisateur a le droit de visibilité sur l'offre
    """
    if current_user.has_permission(Permission.RelationsEntreprisesChange, None):
        return True
    for offre_dept in offre_depts:
        if offre_dept.dept_id in depts:
            return True
    return False


def get_offre_files_and_depts(offre: EntrepriseOffre, depts: list):
    """
    Retourne l'offre, les fichiers attachés a l'offre et les département liés
    """
    offre_depts = EntrepriseOffreDepartement.query.filter_by(offre_id=offre.id).all()
    correspondant = EntrepriseCorrespondant.query.filter_by(
        id=offre.correspondant_id
    ).first()
    if not offre_depts or check_offre_depts(depts, offre_depts):
        files = []
        path = os.path.join(
            Config.SCODOC_VAR_DIR,
            "entreprises",
            f"{offre.entreprise_id}",
            f"{offre.id}",
        )
        if os.path.exists(path):
            for dir in glob.glob(
                f"{path}/[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]-[0-9][0-9]-[0-9][0-9]-[0-9][0-9]"
            ):
                for _file in glob.glob(f"{dir}/*"):
                    file = [os.path.basename(dir), os.path.basename(_file)]
                    files.append(file)
        return [offre, files, offre_depts, correspondant]
    return None


def send_email_notifications_entreprise(subject, entreprise: Entreprise):
    txt = [
        "Une entreprise est en attente de validation",
        "Entreprise:",
        f"\tnom: {entreprise.nom}",
        f"\tsiret: {entreprise.siret}",
        f"\tadresse: {entreprise.adresse}",
        f"\tcode postal: {entreprise.codepostal}",
        f"\tville: {entreprise.ville}",
        f"\tpays: {entreprise.pays}",
    ]
    txt = "\n".join(txt)
    email.send_email(
        subject,
        sco_preferences.get_preference("email_from_addr"),
        [EntreprisePreferences.get_email_notifications],
        txt,
    )
    return txt


def verif_correspondant_data(correspondant_data):
    """
    Verifie les données d'une ligne Excel (correspondant)
    correspondant_data[0]: nom
    correspondant_data[1]: prenom
    correspondant_data[2]: telephone
    correspondant_data[3]: mail
    correspondant_data[4]: poste
    correspondant_data[5]: service
    correspondant_data[6]: entreprise_siret
    """
    # champs obligatoires
    if (
        correspondant_data[0].strip() == ""
        or correspondant_data[1].strip() == ""
        or correspondant_data[6].strip() == ""
    ):
        return False

    # entreprise_id existant
    entreprise = Entreprise.query.filter_by(siret=correspondant_data[6].strip()).first()
    if entreprise is None:
        return False

    # correspondant possède le meme nom et prénom dans la meme entreprise
    correspondant = EntrepriseCorrespondant.query.filter_by(
        nom=correspondant_data[0].strip(),
        prenom=correspondant_data[1].strip(),
        entreprise_id=entreprise.id,
    ).first()
    if correspondant is not None:
        return False

    if (
        correspondant_data[2].strip() == "" and correspondant_data[3].strip() == ""
    ):  # 1 moyen de contact
        return False

    return True


def verif_entreprise_data(entreprise_data):
    """
    Verifie les données d'une ligne Excel (entreprise)
    """
    if EntreprisePreferences.get_check_siret():
        for data in entreprise_data:  # champs obligatoires
            if data.strip() == "":
                return False
    else:
        for data in entreprise_data[1:]:  # champs obligatoires
            if data.strip() == "":
                return False
    if EntreprisePreferences.get_check_siret():
        siret = entreprise_data[0].strip().replace(" ", "")  # vérification sur le siret
        if re.match("^\d{14}$", siret) is None:
            return False
        try:
            req = requests.get(
                f"https://entreprise.data.gouv.fr/api/sirene/v1/siret/{siret}"
            )
            if req.status_code != 200:
                return False
        except requests.ConnectionError:
            return False
        entreprise = Entreprise.query.filter_by(siret=siret).first()
        if entreprise is not None:
            return False
    return True