Colin Maudry
297 lines
10 KiB
Python
297 lines
10 KiB
Python
import datetime
|
|
|
|
import polars as pl
|
|
from dash import Input, Output, State, callback, dash_table, dcc, html, register_page
|
|
|
|
from src.callbacks import get_top_org_table
|
|
from src.figures import point_on_map
|
|
from src.utils import (
|
|
add_links_in_dict,
|
|
df,
|
|
format_number,
|
|
format_values,
|
|
get_annuaire_data,
|
|
get_departement_region,
|
|
meta_content,
|
|
setup_table_columns,
|
|
)
|
|
|
|
register_page(
|
|
__name__,
|
|
path_template="/acheteurs/<acheteur_id>",
|
|
title=meta_content["title"],
|
|
name="Acheteur",
|
|
description=meta_content["description"],
|
|
image_url=meta_content["image_url"],
|
|
order=5,
|
|
)
|
|
|
|
layout = [
|
|
dcc.Store(id="acheteur_data", storage_type="memory"),
|
|
dcc.Location(id="url", refresh="callback-nav"),
|
|
html.Div(
|
|
className="container",
|
|
children=[
|
|
html.Div(
|
|
className="wrapper",
|
|
children=[
|
|
html.H2(
|
|
className="org_title",
|
|
children=[
|
|
html.Span(id="acheteur_siret"),
|
|
" - ",
|
|
html.Span(id="acheteur_nom"),
|
|
],
|
|
),
|
|
html.Div(
|
|
className="org_year",
|
|
children=dcc.Dropdown(
|
|
id="acheteur_year",
|
|
options=["Toutes"]
|
|
+ [
|
|
str(year)
|
|
for year in range(
|
|
2018, int(datetime.date.today().year) + 1
|
|
)
|
|
],
|
|
placeholder="Année",
|
|
),
|
|
),
|
|
html.Div(
|
|
className="org_infos",
|
|
children=[
|
|
# TODO: ajouter le type d'acheteur : commune, CD, CR, etc.
|
|
html.P(["Commune : ", html.Strong(id="acheteur_commune")]),
|
|
html.P(
|
|
[
|
|
"Département : ",
|
|
html.Strong(id="acheteur_departement"),
|
|
]
|
|
),
|
|
html.P(["Région : ", html.Strong(id="acheteur_region")]),
|
|
html.A(
|
|
id="acheteur_lien_annuaire",
|
|
children="Plus de détails sur l'Annuaire des entreprises",
|
|
target="_blank",
|
|
),
|
|
],
|
|
),
|
|
html.Div(
|
|
className="org_stats",
|
|
children=[
|
|
html.P(id="acheteur_titre_stats"),
|
|
html.P(id="acheteur_marches_attribues"),
|
|
html.P(id="acheteur_titulaires_differents"),
|
|
html.Button(
|
|
"Téléchargement au format Excel",
|
|
id="btn-download-acheteur-data",
|
|
),
|
|
dcc.Download(id="download-acheteur-data"),
|
|
],
|
|
),
|
|
html.Div(className="org_map", id="acheteur_map"),
|
|
html.Div(
|
|
className="org_top",
|
|
children=[
|
|
html.H3("Top titulaires"),
|
|
html.Div(className="marches_table", id="top10_titulaires"),
|
|
],
|
|
),
|
|
],
|
|
),
|
|
# récupérer les données de l'acheteur sur l'api annuaire
|
|
html.H3("Derniers marchés publics attribués"),
|
|
html.Div(id="acheteur_last_marches", children=""),
|
|
],
|
|
),
|
|
]
|
|
|
|
|
|
@callback(
|
|
Output(component_id="acheteur_siret", component_property="children"),
|
|
Output(component_id="acheteur_nom", component_property="children"),
|
|
Output(component_id="acheteur_commune", component_property="children"),
|
|
Output(component_id="acheteur_map", component_property="children"),
|
|
Output(component_id="acheteur_departement", component_property="children"),
|
|
Output(component_id="acheteur_region", component_property="children"),
|
|
Output(component_id="acheteur_lien_annuaire", component_property="href"),
|
|
Input(component_id="url", component_property="pathname"),
|
|
)
|
|
def update_acheteur_infos(url):
|
|
acheteur_siret = url.split("/")[-1]
|
|
if len(acheteur_siret) != 14:
|
|
acheteur_siret = (
|
|
f"Le SIRET renseigné doit faire 14 caractères ({acheteur_siret})"
|
|
)
|
|
data = get_annuaire_data(acheteur_siret)
|
|
data_etablissement = data["matching_etablissements"][0]
|
|
acheteur_map = point_on_map(
|
|
data_etablissement["latitude"], data_etablissement["longitude"]
|
|
)
|
|
code_departement, nom_departement, nom_region = get_departement_region(
|
|
data_etablissement["code_postal"]
|
|
)
|
|
departement = f"{nom_departement} ({code_departement})"
|
|
lien_annuaire = (
|
|
f"https://annuaire-entreprises.data.gouv.fr/etablissement/{acheteur_siret}"
|
|
)
|
|
return (
|
|
acheteur_siret,
|
|
data["nom_raison_sociale"],
|
|
data_etablissement["libelle_commune"],
|
|
acheteur_map,
|
|
departement,
|
|
nom_region,
|
|
lien_annuaire,
|
|
)
|
|
|
|
|
|
@callback(
|
|
Output(component_id="acheteur_marches_attribues", component_property="children"),
|
|
Output(
|
|
component_id="acheteur_titulaires_differents", component_property="children"
|
|
),
|
|
Input(component_id="acheteur_data", component_property="data"),
|
|
)
|
|
def update_acheteur_stats(data):
|
|
dff = pl.DataFrame(data)
|
|
if dff.height == 0:
|
|
dff = pl.DataFrame(schema=df.collect_schema())
|
|
df_marches = dff.unique("id")
|
|
nb_marches = format_number(df_marches.height)
|
|
# somme_marches = format_number(int(df_marches.select(pl.sum("montant")).item()))
|
|
marches_attribues = [html.Strong(nb_marches), " marchés et accord-cadres attribués"]
|
|
# + ", pour un total de ", html.Strong(somme_marches + " €")]
|
|
del df_marches
|
|
|
|
nb_titulaires = dff.unique("titulaire_id").height
|
|
nb_titulaires = [
|
|
html.Strong(format_number(nb_titulaires)),
|
|
" titulaires (SIRET) différents",
|
|
]
|
|
del dff
|
|
|
|
return marches_attribues, nb_titulaires
|
|
|
|
|
|
@callback(
|
|
Output(component_id="acheteur_data", component_property="data"),
|
|
Input(component_id="url", component_property="pathname"),
|
|
Input(component_id="acheteur_year", component_property="value"),
|
|
)
|
|
def get_acheteur_marches_data(url, acheteur_year: str) -> list[dict]:
|
|
acheteur_siret = url.split("/")[-1]
|
|
lff = df.lazy()
|
|
lff = lff.filter(pl.col("acheteur_id") == acheteur_siret)
|
|
lff = lff.select(
|
|
"id",
|
|
"uid",
|
|
"objet",
|
|
"dateNotification",
|
|
"titulaire_id",
|
|
"titulaire_typeIdentifiant",
|
|
"titulaire_nom",
|
|
"distance",
|
|
"montant",
|
|
"codeCPV",
|
|
"dureeMois",
|
|
)
|
|
if acheteur_year and acheteur_year != "Toutes":
|
|
lff = lff.filter(
|
|
pl.col("dateNotification").cast(pl.String).str.starts_with(acheteur_year)
|
|
)
|
|
lff = lff.sort(["dateNotification", "id"], descending=True, nulls_last=True)
|
|
|
|
data = lff.collect(engine="streaming").to_dicts()
|
|
return data
|
|
|
|
|
|
@callback(
|
|
Output(component_id="acheteur_last_marches", component_property="children"),
|
|
Input(component_id="acheteur_data", component_property="data"),
|
|
)
|
|
def get_last_marches_table(data) -> html.Div:
|
|
dff = pl.DataFrame(data)
|
|
if dff.height == 0:
|
|
return html.Div(html.P("Aucun marché trouvé."))
|
|
dff = dff.cast(pl.String)
|
|
dff = dff.fill_null("")
|
|
dff = format_values(dff)
|
|
columns, tooltip = setup_table_columns(
|
|
dff,
|
|
hideable=False,
|
|
exclude=["titulaire_id", "titulaire_typeIdentifiant", "uid"],
|
|
)
|
|
data = dff.to_dicts()
|
|
data = add_links_in_dict(data, "titulaire")
|
|
|
|
table = html.Div(
|
|
className="marches_table",
|
|
id="acheteur_datatable",
|
|
children=dash_table.DataTable(
|
|
data=data,
|
|
markdown_options={"html": True},
|
|
page_action="native",
|
|
filter_action="native",
|
|
filter_options={"case": "insensitive", "placeholder_text": "Filtrer..."},
|
|
columns=columns,
|
|
tooltip_header=tooltip,
|
|
tooltip_duration=8000,
|
|
tooltip_delay=350,
|
|
cell_selectable=False,
|
|
page_size=10,
|
|
style_cell_conditional=[
|
|
{
|
|
"if": {"column_id": "objet"},
|
|
"minWidth": "300px",
|
|
"textAlign": "left",
|
|
"overflow": "hidden",
|
|
"lineHeight": "18px",
|
|
"whiteSpace": "normal",
|
|
},
|
|
{
|
|
"if": {"column_id": "titulaire_nom"},
|
|
"minWidth": "200px",
|
|
"textAlign": "left",
|
|
"overflow": "hidden",
|
|
"lineHeight": "18px",
|
|
"whiteSpace": "normal",
|
|
},
|
|
],
|
|
),
|
|
)
|
|
return table
|
|
|
|
|
|
@callback(
|
|
Output(component_id="top10_titulaires", component_property="children"),
|
|
Input(component_id="acheteur_data", component_property="data"),
|
|
)
|
|
def get_top_titulaires(data):
|
|
return get_top_org_table(data, "titulaire")
|
|
|
|
|
|
@callback(
|
|
Output("download-acheteur-data", "data"),
|
|
Input("btn-download-acheteur-data", "n_clicks"),
|
|
State(component_id="acheteur_data", component_property="data"),
|
|
State(component_id="acheteur_nom", component_property="children"),
|
|
State(component_id="acheteur_year", component_property="value"),
|
|
prevent_initial_call=True,
|
|
)
|
|
def download_acheteur_data(
|
|
n_clicks,
|
|
data: [dict],
|
|
acheteur_nom: str,
|
|
annee: str,
|
|
):
|
|
df_to_download = pl.DataFrame(data)
|
|
|
|
def to_bytes(buffer):
|
|
df_to_download.write_excel(
|
|
buffer, worksheet="DECP" if annee in ["Toutes", None] else annee
|
|
)
|
|
|
|
date = datetime.datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
|
|
return dcc.send_bytes(to_bytes, filename=f"decp_{acheteur_nom}_{date}.xlsx")
|