ScoDoc-PE/app/entreprises/app_relations_entreprises.py

288 lines
8.6 KiB
Python
Raw Normal View History

# -*- 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,
EntrepriseSite,
)
from app import email, db
from app.scodoc import sco_preferences
from app.scodoc import sco_excel
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 get_dept_acronym_by_id(id):
dept = Departement.query.filter_by(id=id).first()
if dept is not None:
return dept.acronym
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]: civilite
correspondant_data[1]: nom
correspondant_data[2]: prenom
correspondant_data[3]: telephone
correspondant_data[4]: mail
correspondant_data[5]: poste
correspondant_data[6]: service
correspondant_data[7]: origine
correspondant_data[8]: notes
correspondant_data[9]: entreprise_siret
"""
# champs obligatoires
if (
correspondant_data[0].strip() == ""
or correspondant_data[1].strip() == ""
or correspondant_data[2].strip() == ""
or correspondant_data[9].strip() == ""
):
return False
# civilite entre H ou F
if correspondant_data[0].strip() not in ["H", "F"]:
return False
# entreprise_id existant
entreprise = Entreprise.query.filter_by(siret=correspondant_data[9].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[1].strip(),
prenom=correspondant_data[2].strip(),
entreprise_id=entreprise.id,
).first()
if correspondant is not None:
return False
if (
correspondant_data[3].strip() == "" and correspondant_data[4].strip() == ""
): # 1 moyen de contact
return False
return True
def verif_entreprise_data(entreprise_data):
"""
Verifie les données d'une ligne Excel (entreprise)
"""
2022-02-25 09:47:42 +01:00
if EntreprisePreferences.get_check_siret():
for data in entreprise_data: # champs obligatoires
if data.strip() == "":
2022-02-25 09:47:42 +01:00
return False
else:
for data in entreprise_data[1:]: # champs obligatoires
if data.strip() == "":
2022-02-25 09:47:42 +01:00
return False
2022-02-25 09:45:14 +01:00
if EntreprisePreferences.get_check_siret():
siret = entreprise_data[0].strip().replace(" ", "") # vérification sur le siret
2022-02-25 09:45:14 +01:00
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
2022-02-25 09:45:14 +01:00
except requests.ConnectionError:
return False
2022-02-25 09:45:14 +01:00
entreprise = Entreprise.query.filter_by(siret=siret).first()
if entreprise is not None:
return False
2022-02-25 10:40:51 +01:00
return True
def get_excel_book_are(export=False):
entreprises_keys = [
"siret",
"nom_entreprise",
"adresse",
"ville",
"code_postal",
"pays",
"association",
"visible",
"active",
"notes_active",
]
sites_keys = [
"siret_entreprise",
"nom_site",
"adresse",
"ville",
"code_postal",
"pays",
]
entreprises_titles = entreprises_keys[:]
sites_titles = sites_keys[:]
wb = sco_excel.ScoExcelBook()
ws1 = wb.create_sheet("Entreprises")
ws1.append_row(
[
ws1.make_cell(it, style)
for (it, style) in zip(
entreprises_titles,
[sco_excel.excel_make_style(bold=True)] * len(entreprises_titles),
)
]
)
ws2 = wb.create_sheet("Sites")
ws2.append_row(
[
ws2.make_cell(it, style)
for (it, style) in zip(
sites_titles,
[sco_excel.excel_make_style(bold=True)] * len(sites_titles),
)
]
)
if export:
entreprises = Entreprise.query.filter_by(visible=True).all()
sites = (
db.session.query(EntrepriseSite)
.join(Entreprise, EntrepriseSite.entreprise_id == Entreprise.id)
.filter_by(visible=True)
.all()
)
entreprises_lines = [
[entreprise.to_dict().get(k, "") for k in entreprises_keys]
for entreprise in entreprises
]
sites_lines = [
[site.to_dict().get(k, "") for k in sites_keys] for site in sites
]
for line in entreprises_lines:
cells = []
for it in line:
cells.append(ws1.make_cell(it))
ws1.append_row(cells)
for line in sites_lines:
cells = []
for it in line:
cells.append(ws2.make_cell(it))
ws2.append_row(cells)
return wb