colin/colibre
1
0
Fork 0
Code Issues 30 Pull Requests Actions Packages Projects Releases 33 Activity

Capitalisation des constantes

Browse Source
This commit is contained in:
Colin Maudry
2026-04-18 18:09:42 +02:00
parent accdfe1384
commit e352624c02
14 changed files with 128 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app.py
+6 -6
View File
@@ -14,10 +14,10 @@ load_dotenv()
# if os.getenv("PYTEST_CURRENT_TEST"):
# os.environ["DATA_FILE_PARQUET_PATH"]
# TODO: Importer LOGGER et DEVELOPMENT depuis utils.init
DEVELOPMENT = os.getenv("DEVELOPMENT", "").lower() == "true"
development = os.getenv("DEVELOPMENT", "").lower() == "true"
meta_tags = [
META_TAGS = [
{"name": "viewport", "content": "width=device-width, initial-scale=1"},
{
"name": "keywords",
@@ -25,14 +25,14 @@ meta_tags = [
},
]
if development:
meta_tags.append({"name": "robots", "content": "noindex"})
if DEVELOPMENT:
META_TAGS.append({"name": "robots", "content": "noindex"})
app: Dash = Dash(
title="decp.info",
use_pages=True,
compress=True,
meta_tags=meta_tags,
meta_tags=META_TAGS,
)
cache.init_app(
src/db.py
+5 -4
View File
@@ -9,7 +9,8 @@ import polars as pl
import polars.selectors as cs
from polars.exceptions import ComputeError
logger = logging.getLogger("decp.info")
# TODO: Importer le logger depuis utils
LOGGER = logging.getLogger("decp.info")
def should_rebuild(db_path: Path, parquet_path: Path) -> bool:
@@ -33,7 +34,7 @@ def _load_source_frame(parquet_path: Path) -> pl.DataFrame:
try:
lff: pl.LazyFrame = pl.scan_parquet(str(parquet_path))
except ComputeError:
logger.info("Lecture du parquet échouée, nouvelle tentative dans 10s...")
LOGGER.info("Lecture du parquet échouée, nouvelle tentative dans 10s...")
sleep(10)
lff = pl.scan_parquet(str(parquet_path))
@@ -71,7 +72,7 @@ def build_database(db_path: Path, parquet_path: Path) -> None:
if tmp_path.exists():
tmp_path.unlink()
logger.info(f"Construction de la base DuckDB à partir de {parquet_path}...")
LOGGER.info(f"Construction de la base DuckDB à partir de {parquet_path}...")
frame = _load_source_frame(parquet_path)
# Write transformed frame as parquet so DuckDB can read it natively
@@ -107,7 +108,7 @@ def build_database(db_path: Path, parquet_path: Path) -> None:
staging_parquet.unlink()
os.replace(tmp_path, db_path)
logger.info(f"Base DuckDB construite : {db_path}")
LOGGER.info(f"Base DuckDB construite : {db_path}")
def _resolve_db_path() -> Path:
src/figures.py
+10 -10
View File
@@ -13,10 +13,10 @@ from dash import dash_table, dcc, html
from dash_extensions.javascript import Namespace
from src.db import schema
from src.utils import (
from src.old_utils import (
DATA_SCHEMA,
DEPARTEMENTS_GEOJSON,
add_links,
data_schema,
departements_geojson,
format_number,
setup_table_columns,
)
@@ -260,8 +260,8 @@ class DataTable(dash_table.DataTable):
style_cell_common = {"fontFamily": "Inter", "fontSize": "16px"}
for key in data_schema.keys():
field = data_schema[key]
for key in DATA_SCHEMA.keys():
field = DATA_SCHEMA[key]
if field["type"] in ["number", "integer"]:
rule = {
"if": {"column_id": field["name"]},
@@ -514,7 +514,7 @@ def make_chloropleth_map(region: dict) -> dcc.Graph:
fig = px.choropleth(
df_map,
geojson=departements_geojson,
geojson=DEPARTEMENTS_GEOJSON,
locations="Département",
color="uid",
color_continuous_scale="Reds",
@@ -722,7 +722,7 @@ def make_donut(
nulls="?",
potentially_many_names: bool = False,
):
title = data_schema[names_col]["title"]
title = DATA_SCHEMA[names_col]["title"]
lff = lff.rename({names_col: title})
lff = lff.select("uid", title)
@@ -771,8 +771,8 @@ def make_column_picker(page: str):
table_columns = [
{
"id": col,
"name": data_schema[col]["title"],
"description": data_schema[col]["description"],
"name": DATA_SCHEMA[col]["title"],
"description": DATA_SCHEMA[col]["description"],
}
for col in schema.names()
]
@@ -780,7 +780,7 @@ def make_column_picker(page: str):
new_column = {
"id": column["id"],
"name": column["name"],
"description": data_schema[column["id"]]["description"],
"description": DATA_SCHEMA[column["id"]]["description"],
}
table_data.append(new_column)
src/utils.py → src/old_utils.py
+16 -16
View File
@@ -21,8 +21,8 @@ logging.basicConfig(
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("decp.info")
development = os.getenv("DEVELOPMENT", "False").lower() == "true"
if development:
DEVELOPMENT = os.getenv("DEVELOPMENT", "False").lower() == "true"
if DEVELOPMENT:
logger.setLevel(logging.DEBUG)
logging.getLogger("httpx").setLevel("WARNING")
@@ -266,8 +266,8 @@ def get_departement_region(code_postal):
code_departement = code_postal[:3]
else:
code_departement = code_postal[:2]
nom_departement = departements[code_departement]["departement"]
nom_region = departements[code_departement]["region"]
nom_departement = DEPARTEMENTS[code_departement]["departement"]
nom_region = DEPARTEMENTS[code_departement]["region"]
return code_departement, nom_departement, nom_region
@@ -358,7 +358,7 @@ def setup_table_columns(
for column_id in dff.columns:
if exclude and column_id in exclude:
continue
column_object = data_schema.get(column_id)
column_object = DATA_SCHEMA.get(column_id)
if column_object:
column_name = column_object.get("title")
else:
@@ -455,7 +455,7 @@ def get_data_schema() -> dict:
def track_search(query, category):
if len(query) >= 4 and not development and os.getenv("MATOMO_DOMAIN"):
if len(query) >= 4 and not DEVELOPMENT and os.getenv("MATOMO_DOMAIN"):
url = "https://decp.info"
params = {
"idsite": os.getenv("MATOMO_ID_SITE"),
@@ -776,7 +776,7 @@ def get_button_properties(height):
def get_enum_values_as_dict(column_name):
try:
options = {}
for value in data_schema[column_name]["enum"]:
for value in DATA_SCHEMA[column_name]["enum"]:
options[value] = value
return options
except KeyError:
@@ -854,22 +854,22 @@ def _build_org_frame(org_type: str) -> pl.DataFrame:
return get_cursor().execute(sql).pl()
df_acheteurs = _build_org_frame("acheteur")
df_titulaires = _build_org_frame("titulaire")
DF_ACHETEURS = _build_org_frame("acheteur")
DF_TITULAIRES = _build_org_frame("titulaire")
columns = schema.names()
COLUMNS = schema.names()
departements = get_departements()
departements_geojson = get_departements_geojson()
domain_name = (
DEPARTEMENTS = get_departements()
DEPARTEMENTS_GEOJSON = get_departements_geojson()
DOMAIN_NAME = (
"test.decp.info" if os.getenv("DEVELOPMENT").lower() == "true" else "decp.info"
)
meta_content = {
"image_url": f"https://{domain_name}/assets/decp.info.png",
META_CONTENT = {
"image_url": f"https://{DOMAIN_NAME}/assets/decp.info.png",
"title": "decp.info - exploration des marchés publics français",
"description": (
"Explorez et analysez les données des marchés publics français avec cet outil libre et gratuit. "
"Pour une commande publique accessible à toutes et tous."
),
}
data_schema = get_data_schema()
DATA_SCHEMA = get_data_schema()
src/pages/a-propos.py
+4 -4
View File
@@ -3,9 +3,9 @@ import os
from dash import dcc, html, register_page
from src.figures import get_sources_tables
from src.utils import meta_content
from src.old_utils import META_CONTENT
name = "À propos"
NAME = "À propos"
register_page(
__name__,
@@ -13,14 +13,14 @@ register_page(
title="À propos | decp.info",
name="À propos",
description="En savoir plus sur decp.info, l'outil d'exploration des données essentielles de la commande publique.",
image_url=meta_content["image_url"],
image_url=META_CONTENT["image_url"],
order=5,
)
layout = html.Div(
className="container",
children=[
html.H2(name),
html.H2(NAME),
html.Div(
className="a-propos-container",
children=[
src/pages/acheteur.py
+11 -11
View File
@@ -24,23 +24,23 @@ from src.figures import (
make_column_picker,
point_on_map,
)
from src.utils import (
columns,
df_acheteurs,
from src.old_utils import (
COLUMNS,
DF_ACHETEURS,
META_CONTENT,
filter_table_data,
format_number,
get_annuaire_data,
get_button_properties,
get_default_hidden_columns,
get_departement_region,
meta_content,
prepare_table_data,
sort_table_data,
)
def get_title(acheteur_id: str | None = None) -> str:
acheteur_nom = df_acheteurs.filter(pl.col("acheteur_id") == acheteur_id).select(
acheteur_nom = DF_ACHETEURS.filter(pl.col("acheteur_id") == acheteur_id).select(
"acheteur_nom"
)
if acheteur_nom.height > 0:
@@ -54,11 +54,11 @@ register_page(
title=get_title,
name="Acheteur",
description="Consultez les marchés publics attribués par cet acheteur.",
image_url=meta_content["image_url"],
image_url=META_CONTENT["image_url"],
order=5,
)
datatable = html.Div(
DATATABLE = html.Div(
className="marches_table",
children=DataTable(
dtid="acheteur_datatable",
@@ -229,7 +229,7 @@ layout = [
scrollable=True,
size="xl",
),
datatable,
DATATABLE,
],
),
],
@@ -460,8 +460,8 @@ clientside_callback(
)
def update_hidden_columns_from_checkboxes(selected_columns):
if selected_columns:
selected_columns = [columns[i] for i in selected_columns]
hidden_columns = [col for col in columns if col not in selected_columns]
selected_columns = [COLUMNS[i] for i in selected_columns]
hidden_columns = [col for col in COLUMNS if col not in selected_columns]
return hidden_columns
else:
return []
@@ -489,7 +489,7 @@ def update_checkboxes_from_hidden_columns(hidden_cols, current_checkboxes):
hidden_cols = hidden_cols or get_default_hidden_columns("acheteur")
# Show all columns that are NOT hidden
visible_cols = [columns.index(col) for col in columns if col not in hidden_cols]
visible_cols = [COLUMNS.index(col) for col in COLUMNS if col not in hidden_cols]
return visible_cols
src/pages/arbre/departement.py
+5 -5
View File
@@ -1,17 +1,17 @@
from dash import Input, Output, callback, dcc, html, register_page
from src.db import get_cursor
from src.utils import departements
from src.old_utils import DEPARTEMENTS
name = "Département"
NAME = "Département"
def get_title(code):
return f"Marchés publics de {departements[code]['departement']} | decp.info"
return f"Marchés publics de {DEPARTEMENTS[code]['departement']} | decp.info"
def get_description(code):
return f"Marchés publics passés dans le département {departements[code]['departement']} | decp.info"
return f"Marchés publics passés dans le département {DEPARTEMENTS[code]['departement']} | decp.info"
register_page(
@@ -20,7 +20,7 @@ register_page(
title=get_title,
description=get_description,
order=50,
name=name,
name=NAME,
)
layout = html.Div(
src/pages/arbre/departements.py
+3 -3
View File
@@ -1,8 +1,8 @@
from dash import dcc, html, register_page
from src.utils import departements
from src.old_utils import DEPARTEMENTS
name = "Départements"
NAME = "Départements"
register_page(
__name__,
@@ -18,7 +18,7 @@ layout = html.Div(
html.Ul(
[
html.Li(dcc.Link(d["departement"], href=f"/departements/{k}"))
for k, d in departements.items()
for k, d in DEPARTEMENTS.items()
]
),
]
src/pages/arbre/liste_marches_org.py
+5 -5
View File
@@ -2,17 +2,17 @@ import polars as pl
from dash import Input, Output, callback, dcc, html, register_page
from src.db import get_cursor
from src.utils import df_acheteurs, df_titulaires
from src.old_utils import DF_ACHETEURS, DF_TITULAIRES
name = "Liste des marchés publics"
NAME = "Liste des marchés publics"
def make_org_nom_verbe(org_type, org_id) -> tuple:
if org_type == "titulaire":
df = df_titulaires
df = DF_TITULAIRES
verbe = "remportés"
elif org_type == "acheteur":
df = df_acheteurs
df = DF_ACHETEURS
verbe = "attribués"
else:
raise ValueError
@@ -44,7 +44,7 @@ register_page(
title=get_title,
description=get_description,
order=40,
name=name,
name=NAME,
)
layout = html.Div(
src/pages/marche.py
+5 -5
View File
@@ -6,11 +6,11 @@ from dash import Input, Output, callback, dcc, html, register_page
from polars import selectors as cs
from src.db import query_marches
from src.utils import (
data_schema,
from src.old_utils import (
DATA_SCHEMA,
META_CONTENT,
format_values,
make_org_jsonld,
meta_content,
unformat_montant,
)
@@ -25,7 +25,7 @@ register_page(
title=get_title,
name="Marché",
description="Consultez les détails de ce marché public : montant, acheteur, titulaires, modifications, etc.",
image_url=meta_content["image_url"],
image_url=META_CONTENT["image_url"],
order=7,
)
@@ -110,7 +110,7 @@ def get_marche_data(url) -> tuple[dict, list]:
)
def update_marche_info(marche, titulaires):
def make_parameter(col, bold=True):
column_object = data_schema.get(col)
column_object = DATA_SCHEMA.get(col)
column_name = column_object.get("title") if column_object else col
if marche[col]:
src/pages/observatoire.py
+27 -27
View File
@@ -30,45 +30,45 @@ from src.figures import (
make_column_picker,
make_donut,
)
from src.utils import (
columns,
departements,
df_acheteurs,
df_titulaires,
from src.old_utils import (
COLUMNS,
DEPARTEMENTS,
DF_ACHETEURS,
DF_TITULAIRES,
META_CONTENT,
get_default_hidden_columns,
get_enum_values_as_dict,
logger,
meta_content,
prepare_dashboard_data,
prepare_table_data,
)
name = "Observatoire"
NAME = "Observatoire"
register_page(
__name__,
path="/observatoire",
title="Observatoire | decp.info",
name=name,
name=NAME,
description="Visualisez l'état de la publication des données essentielles des marchés publics en France.",
image_url=meta_content["image_url"],
image_url=META_CONTENT["image_url"],
order=3,
)
options_years = []
OPTIONS_YEARS = []
for year in reversed(range(2017, datetime.now().year + 1)):
option_year = {
"label": str(year),
"value": year,
}
options_years.append(option_year)
OPTIONS_YEARS.append(option_year)
options_departements = []
for code in departements.keys():
OPTIONS_DEPARTEMENTS = []
for code in DEPARTEMENTS.keys():
departement = {
"label": f"{departements[code]['departement']} ({code})",
"label": f"{DEPARTEMENTS[code]['departement']} ({code})",
"value": code,
}
options_departements.append(departement)
OPTIONS_DEPARTEMENTS.append(departement)
OBSERVATOIRE_COLUMNS = [
col
@@ -124,7 +124,7 @@ Alors, on fait comment ?
html.Div(
className="container-fluid",
children=[
html.H2(children=[name], id="page_title"),
html.H2(children=[NAME], id="page_title"),
dcc.Loading(
overlay_style={"visibility": "visible", "filter": "blur(2px)"},
id="loading-statistques",
@@ -142,7 +142,7 @@ Alors, on fait comment ?
dbc.Col(
dcc.Dropdown(
id="dashboard_year",
options=options_years,
options=OPTIONS_YEARS,
placeholder="12 derniers mois",
persistence=True,
persistence_type="local",
@@ -182,7 +182,7 @@ Alors, on fait comment ?
searchable=True,
multi=True,
placeholder="Département",
options=options_departements,
options=OPTIONS_DEPARTEMENTS,
persistence=True,
persistence_type="local",
),
@@ -221,7 +221,7 @@ Alors, on fait comment ?
searchable=True,
multi=True,
placeholder="Département",
options=options_departements,
options=OPTIONS_DEPARTEMENTS,
persistence=True,
persistence_type="local",
),
@@ -822,20 +822,20 @@ def add_organization_name_in_title(acheteur_id, titulaire_id):
return match[nom_col].item(0) if match.height >= 1 else None
if acheteur_id and len(acheteur_id) == 14:
if nom := lookup_nom(df_acheteurs, "acheteur_id", "acheteur_nom", acheteur_id):
if nom := lookup_nom(DF_ACHETEURS, "acheteur_id", "acheteur_nom", acheteur_id):
return [
name,
NAME,
html.Small(nom, className="text-muted d-block fw-normal fs-5"),
]
elif titulaire_id and len(titulaire_id) == 14:
if nom := lookup_nom(
df_titulaires, "titulaire_id", "titulaire_nom", titulaire_id
DF_TITULAIRES, "titulaire_id", "titulaire_nom", titulaire_id
):
return [
name,
NAME,
html.Small(nom, className="text-muted d-block fw-normal fs-5"),
]
return name
return NAME
@callback(
@@ -899,8 +899,8 @@ def populate_preview_table(
)
def update_hidden_columns_from_checkboxes(selected_columns):
if selected_columns:
selected_columns = [columns[i] for i in selected_columns]
hidden_columns = [col for col in columns if col not in selected_columns]
selected_columns = [COLUMNS[i] for i in selected_columns]
hidden_columns = [col for col in COLUMNS if col not in selected_columns]
return hidden_columns
else:
return []
@@ -928,7 +928,7 @@ def update_checkboxes_from_hidden_columns(hidden_cols, current_checkboxes):
hidden_cols = hidden_cols or get_default_hidden_columns("tableau")
# Show all columns that are NOT hidden
visible_cols = [columns.index(col) for col in columns if col not in hidden_cols]
visible_cols = [COLUMNS.index(col) for col in COLUMNS if col not in hidden_cols]
return visible_cols
src/pages/recherche.py
+9 -9
View File
@@ -2,23 +2,23 @@ import dash_bootstrap_components as dbc
from dash import Input, Output, State, callback, dcc, html, register_page
from src.figures import DataTable
from src.utils import (
df_acheteurs,
df_titulaires,
meta_content,
from src.old_utils import (
DF_ACHETEURS,
DF_TITULAIRES,
META_CONTENT,
search_org,
setup_table_columns,
)
name = "Recherche"
NAME = "Recherche"
register_page(
__name__,
path="/",
title="Recherche de marchés publics | decp.info",
name=name,
name=NAME,
description="Explorez et analysez les données des marchés publics français avec cet outil libre et gratuit. Pour une commande publique accessible à toutes et tous.",
image_url=meta_content["image_url"],
image_url=META_CONTENT["image_url"],
order=0,
)
@@ -97,9 +97,9 @@ def update_search_results(n_submit, n_clicks, query):
for org_type in ["acheteur", "titulaire"]:
if org_type == "acheteur":
dff = df_acheteurs
dff = DF_ACHETEURS
elif org_type == "titulaire":
dff = df_titulaires
dff = DF_TITULAIRES
else:
raise ValueError(f"{org_type} is not supported")
src/pages/tableau.py
+11 -11
View File
@@ -21,13 +21,13 @@ from dash import (
from src.db import query_marches, schema
from src.figures import DataTable, make_column_picker
from src.utils import (
columns,
from src.old_utils import (
COLUMNS,
META_CONTENT,
filter_table_data,
get_default_hidden_columns,
invert_columns,
logger,
meta_content,
prepare_table_data,
sort_table_data,
)
@@ -37,18 +37,18 @@ update_date = datetime.fromtimestamp(update_date_timestamp).strftime("%d/%m/%Y")
update_date_iso = datetime.fromtimestamp(update_date_timestamp).isoformat()
name = "Tableau"
NAME = "Tableau"
register_page(
__name__,
path="/tableau",
title="Tableau des marchés publics | decp.info",
name=name,
name=NAME,
description="Consultez, filtrez et exportez les données essentielles de la commande publique sous forme de tableau.",
image_url=meta_content["image_url"],
image_url=META_CONTENT["image_url"],
order=1,
)
datatable = html.Div(
DATATABLE = html.Div(
className="marches_table",
children=DataTable(
dtid="tableau_datatable",
@@ -272,7 +272,7 @@ layout = [
scrollable=True,
size="xl",
),
datatable,
DATATABLE,
],
),
]
@@ -480,8 +480,8 @@ def toggle_tableau_help(click_open, click_close, is_open):
)
def update_hidden_columns_from_checkboxes(selected_columns):
if selected_columns:
selected_columns = [columns[i] for i in selected_columns]
hidden_columns = [col for col in columns if col not in selected_columns]
selected_columns = [COLUMNS[i] for i in selected_columns]
hidden_columns = [col for col in COLUMNS if col not in selected_columns]
return hidden_columns
else:
return []
@@ -509,7 +509,7 @@ def update_checkboxes_from_hidden_columns(hidden_cols, current_checkboxes):
hidden_cols = hidden_cols or get_default_hidden_columns("tableau")
# Show all columns that are NOT hidden
visible_cols = [columns.index(col) for col in columns if col not in hidden_cols]
visible_cols = [COLUMNS.index(col) for col in COLUMNS if col not in hidden_cols]
return visible_cols
src/pages/titulaire.py
+11 -11
View File
@@ -23,23 +23,23 @@ from src.figures import (
make_column_picker,
point_on_map,
)
from src.utils import (
columns,
df_titulaires,
from src.old_utils import (
COLUMNS,
DF_TITULAIRES,
META_CONTENT,
filter_table_data,
format_number,
get_annuaire_data,
get_button_properties,
get_default_hidden_columns,
get_departement_region,
meta_content,
prepare_table_data,
sort_table_data,
)
def get_title(titulaire_id: str = None) -> str:
titulaire_nom = df_titulaires.filter(pl.col("titulaire_id") == titulaire_id).select(
titulaire_nom = DF_TITULAIRES.filter(pl.col("titulaire_id") == titulaire_id).select(
"titulaire_nom"
)
if titulaire_nom.height > 0:
@@ -53,11 +53,11 @@ register_page(
title=get_title,
name="Titulaire",
description="Consultez les marchés publics remportés par ce titulaire.",
image_url=meta_content["image_url"],
image_url=META_CONTENT["image_url"],
order=5,
)
datatable = html.Div(
DATATABLE = html.Div(
className="marches_table",
children=DataTable(
dtid="titulaire_datatable",
@@ -239,7 +239,7 @@ layout = [
scrollable=True,
size="xl",
),
datatable,
DATATABLE,
],
),
],
@@ -476,8 +476,8 @@ clientside_callback(
)
def update_hidden_columns_from_checkboxes(selected_columns):
if selected_columns:
selected_columns = [columns[i] for i in selected_columns]
hidden_columns = [col for col in columns if col not in selected_columns]
selected_columns = [COLUMNS[i] for i in selected_columns]
hidden_columns = [col for col in COLUMNS if col not in selected_columns]
return hidden_columns
else:
return []
@@ -505,7 +505,7 @@ def update_checkboxes_from_hidden_columns(hidden_cols, current_checkboxes):
hidden_cols = hidden_cols or get_default_hidden_columns("titulaire")
# Show all columns that are NOT hidden
visible_cols = [columns.index(col) for col in columns if col not in hidden_cols]
visible_cols = [COLUMNS.index(col) for col in COLUMNS if col not in hidden_cols]
return visible_cols