forked from ScoDoc/ScoDoc
210 lines
6.4 KiB
Python
210 lines
6.4 KiB
Python
# -*- 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, correspondant: EntrepriseCorrespondant
|
|
):
|
|
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}",
|
|
"",
|
|
"Correspondant:",
|
|
f"nom: {correspondant.nom}",
|
|
f"prenom: {correspondant.prenom}",
|
|
f"telephone: {correspondant.telephone}",
|
|
f"mail: {correspondant.mail}",
|
|
f"poste: {correspondant.poste}",
|
|
f"service: {correspondant.service}",
|
|
]
|
|
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_id
|
|
"""
|
|
# champs obligatoires
|
|
if (
|
|
correspondant_data[0] == ""
|
|
or correspondant_data[1] == ""
|
|
or correspondant_data[6] == ""
|
|
):
|
|
return False
|
|
|
|
# entreprise_id existant
|
|
entreprise = Entreprise.query.filter_by(siret=correspondant_data[6]).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],
|
|
prenom=correspondant_data[1],
|
|
entreprise_id=entreprise.id,
|
|
).first()
|
|
if correspondant is not None:
|
|
return False
|
|
|
|
if (
|
|
correspondant_data[2] == "" and correspondant_data[3] == ""
|
|
): # 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 == "":
|
|
return False
|
|
else:
|
|
for data in entreprise_data[1:]: # champs obligatoires
|
|
if data == "":
|
|
return False
|
|
if EntreprisePreferences.get_check_siret():
|
|
siret = entreprise_data[0].strip() # 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:
|
|
print("no internet")
|
|
entreprise = Entreprise.query.filter_by(siret=siret).first()
|
|
if entreprise is not None:
|
|
return False
|
|
return True
|