Ajout du JSON-LD #66

src/pages/marche.py

@@ -1,3 +1,4 @@
+import json
 from datetime import datetime
 
 import dash_bootstrap_components as dbc
@@ -5,7 +6,14 @@ import polars as pl
 from dash import Input, Output, callback, dcc, html, register_page
 from polars import selectors as cs
 
-from src.utils import data_schema, df, format_values, meta_content
+from src.utils import (
+    data_schema,
+    df,
+    format_values,
+    make_org_jsonld,
+    meta_content,
+    unformat_montant,
+)
 
 
 def get_title(uid: str = None) -> str:
@@ -26,6 +34,9 @@ layout = [
     dcc.Store(id="marche_data"),
     dcc.Store(id="titulaires_data"),
     dcc.Location(id="marche_url", refresh="callback-nav"),
+    html.Script(
+        type="application/ld+json", id="marche_jsonld", children=['{"test": "1"}']
+    ),
     dbc.Container(
         className="marche_infos",
         children=[
@@ -208,3 +219,51 @@ def update_marche_info(marche, titulaires):
         titulaires_lines.append(content)
 
     return marche_objet, marche_infos[:half], marche_infos[half:], titulaires_lines
+
+
+@callback(
+    Output(component_id="marche_jsonld", component_property="children"),
+    Input("marche_data", "data"),
+    Input("titulaires_data", "data"),
+)
+def get_marche_jsonld(marche, titulaires) -> str:
+    acheteur_id = marche.get("acheteur_id")
+    type_order = (
+        "Service" if marche.get("categorie") in ["Services", "Travaux"] else "Product"
+    )
+    result = []
+
+    for titulaire in titulaires:
+        jsonld = {
+            "@context": "https://schema.org",
+            "@type": "Order",
+            "@id": f"https://decp.info/marches/{marche.get('uid')}",
+            "name": f"{marche.get('nature')} conclu par {marche.get('acheteur_nom')} le {marche.get('dateNotification')}",
+            "description": marche.get("objet"),
+            "orderNumber": marche.get("uid"),
+            "orderDate": marche.get("dateNotification"),
+            "price": unformat_montant(marche.get("montant")),
+            "priceCurrency": "EUR",
+            "customer": make_org_jsonld(
+                acheteur_id, org_name=marche.get("acheteur_nom"), org_type="acheteur"
+            ),
+            "seller": make_org_jsonld(
+                titulaire.get("titulaire_id"),
+                org_name=titulaire.get("titulaire_nom"),
+                org_type="titulaire",
+                type_org_id=titulaire.get("titulaire_typeIdentifiant"),
+            ),
+            "orderedItem": {
+                "@type": type_order,
+                "name": marche.get("objet"),
+                "category": {
+                    "@type": "CategoryCode",
+                    "propertyID": "cpv",
+                    "codeValue": marche.get("codeCPV"),
+                    # "description": "Description du code CPV"
+                },
+                # "serviceType": "Description du code CPV"
+            },
+        }
+        result.append(jsonld)
+    return json.dumps(result, indent=2)

src/utils.py

@@ -145,6 +145,13 @@ def format_number(number) -> str:
     return number
 
 
+def unformat_montant(number: str) -> int:
+    number = number.replace(" €", "")
+    number = number.replace(" €", "").replace(" ", "")
+    number = number.strip()
+    return int(number)
+
+
 def format_values(dff: pl.DataFrame) -> pl.DataFrame:
     def format_montant(expr, scale=None):
         # https://stackoverflow.com/a/78636786
@@ -713,3 +720,42 @@ meta_content = {
     ),
 }
 data_schema = get_data_schema()
+
+
+def make_org_jsonld(org_id, org_type, org_name=None, type_org_id="SIRET") -> dict:
+    org_types = {"acheteur": "GovernmentOrganization", "titulaire": "Organization"}
+    address = None
+    if type_org_id.lower() == "siret" and len(org_id) == 14:
+        annuaire_data = get_annuaire_data(org_id)
+        annuaire_address = annuaire_data["matching_etablissements"][0]
+        code_postal = annuaire_address["code_postal"]
+        commune = annuaire_address["libelle_commune"]
+
+        address = (
+            {
+                "@type": "PostalAddress",
+                "streetAddress": annuaire_address.get("adresse", "")
+                .replace(code_postal, "")
+                .replace(commune, "")
+                .strip(),
+                "addressLocality": commune,
+                "postalCode": code_postal,
+                "addressCountry": "FR",
+            },
+        )
+
+    jsonld = {
+        "@type": org_types[org_type],
+        "name": org_name,
+        "url": f"https://decp.info/titulaires/{org_id}",
+        "identifier": {
+            "@type": "PropertyValue",
+            "propertyID": type_org_id.lower(),
+            "value": org_id,
+        },
+    }
+
+    if address:
+        jsonld["address"] = address
+
+    return jsonld