""" GuichetOI ML — Streamlit demo. One-page workflow: upload all files for a demande de localisation PAR (loose files OR a ZIP archive of the demande folder), and the recommendation engine produces a complétude verdict + a draft AR mail. Run: pip install -e .[ui] # one-time, installs guichetoi streamlit run apps/streamlit_demo.py """ from __future__ import annotations import io import sys import tempfile import zipfile from pathlib import Path import streamlit as st # Repo layout: apps/streamlit_demo.py → parents[1] = repo root → repo_root/src/ # When the package is installed via `pip install -e .` this is a no-op; we keep # the sys.path insert so the demo also runs straight from a fresh checkout. ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) from guichetoi import cms as cms_gen from guichetoi import inference, recommendation as reco @st.cache_resource(show_spinner="Préparation de l'analyse (≈30 s)…") def get_pipeline(): return inference.GuichetOIPipeline() @st.cache_resource(show_spinner=False) def get_engine(): return reco.RecommendationEngine(pipeline=get_pipeline()) # ──────────────────────────────────────────────────────────────────────────── # Demo samples — pre-cached verdicts so the demo recording stays snappy # ──────────────────────────────────────────────────────────────────────────── import json as _json @st.cache_data(show_spinner=False) def load_sample_verdicts() -> dict[str, dict]: """Read assets/sample_verdicts.json and index by ZIP basename.""" p = ROOT / "assets" / "sample_verdicts.json" if not p.exists(): return {} data = _json.loads(p.read_text(encoding="utf-8")) return {r["zip"]: r["verdict"] for r in data if r.get("verdict")} # Curated demo flow: one example per outcome, in narrative order DEMO_SAMPLES: list[tuple[str, str, str]] = [ ("✅ Demande complète — PIM résidentiel", "Cas standard : 1 logement, tous les champs extraits, CMS pré-rempli.", "PF0442402600168.zip"), ("✅ Demande complète — noms de fichiers atypiques", "Filenames ALL-CAPS sans préfixe PF : 'ARRETE PC', 'CERTIFICAT ADRESSAGE'. " "Les heuristiques de nom de fichier corrigent la classification.", "PF0331402600885.zip"), ("⚠️ Demande incomplète — collectif, champ manquant", "Projet collectif (14 logements). nb_log_totale non lisible sur la fiche → " "incomplète, mais le consultant peut toujours générer un CMS partiel.", "PF0335202600876.zip"), ("🔁 Hors-périmètre — dossier de récolement", "Fichiers post-installation (tranchées, points de raccordement). Détecté " "automatiquement et routé en vérification manuelle.", "PF0820002600007_Dossier-de-recolement_RAR-1-1_1.zip"), ] def verdict_from_dict(d: dict) -> "reco.Verdict": """Reconstruct a Verdict dataclass from its dict serialisation.""" docs = [] for doc_d in d.get("documents", []) or []: docs.append(reco.DocumentSummary( file=doc_d.get("file", ""), doc_class=doc_d.get("doc_class", ""), doc_confidence=float(doc_d.get("doc_confidence", 0.0) or 0.0), fields=doc_d.get("fields", {}) or {}, flags=list(doc_d.get("flags", []) or []), )) return reco.Verdict( status=d.get("status", ""), missing_documents=list(d.get("missing_documents", []) or []), incomplete_documents=list(d.get("incomplete_documents", []) or []), documents=docs, fiche_summary=d.get("fiche_summary", {}) or {}, manual_review_documents=list(d.get("manual_review_documents", []) or []), ar_mail_body=d.get("ar_mail_body", ""), ) # ──────────────────────────────────────────────────────────────────────────── # Constants — class icons, field names, expected doc set # ──────────────────────────────────────────────────────────────────────────── CLASS_ICON: dict[str, str] = { "fiche": "📋", "Autorisation": "📜", "Mandat": "✍️", "Certificat": "📌", "PlanMasse": "🗺️", "PlanSituation": "📍", } CLASS_LABEL: dict[str, str] = { "fiche": "Fiche de renseignement", "Autorisation": "Autorisation d'urbanisme", "Mandat": "Mandat", "Certificat": "Certificat d'adressage", "PlanMasse": "Plan de masse", "PlanSituation": "Plan de situation", } FIELD_LABEL_FR: dict[str, str] = { "Reference_Urbanisme": "N° d'urbanisme", "DLPI": "Date de livraison (DLPI)", "Disposition_Mandat": "Mandat de représentation", "Nombre_Logement_Lot_MacroLot": "Nb logements/lots/macrolots", "Nb_log_pro": "Bâtiments professionnels", "Nb_log_res": "Bâtiments résidentiels", "nb_log_totale": "Nb total de logements", "cabinet_conseil": "Cabinet conseil", "Representant_Nom_Complet": "Nom du représentant", "Representant_Telephone": "Téléphone", "Representant_Email": "Email", "Batiment_Adresse": "Adresse du bâtiment", } EXPECTED_CLASSES = ("fiche", "Autorisation", "PlanMasse", "PlanSituation", "Mandat") # ──────────────────────────────────────────────────────────────────────────── # Page setup + global CSS # ──────────────────────────────────────────────────────────────────────────── st.set_page_config( page_title="Orange · Guichet Accueil Infrastructures", page_icon="🟧", layout="wide", initial_sidebar_state="expanded", ) st.markdown( """ """, unsafe_allow_html=True, ) # ──────────────────────────────────────────────────────────────────────────── # UI helpers # ──────────────────────────────────────────────────────────────────────────── def conf_class(pct: float) -> str: if pct >= 0.85: return "conf-hi" if pct >= 0.60: return "conf-mid" return "conf-lo" def confidence_dot(pct: float) -> str: return f"{pct:.0%}" def class_pill(name: str, conf: float) -> str: icon = CLASS_ICON.get(name, "📄") label = CLASS_LABEL.get(name, name) return (f"{icon} {label}" f"{confidence_dot(conf)}") def verdict_banner(status: str, needs_review: bool = False): if status == "hors-périmètre": label = "🔁 HORS PÉRIMÈTRE — routage manuel requis" cls = "verdict-review" elif status.startswith("complèt"): if needs_review: label = "✅ COMPLÈTE — sous réserve de vérification manuelle" cls = "verdict-review" else: label = "✅ DEMANDE COMPLÈTE" cls = "verdict-ok" else: label = "⚠️ DEMANDE INCOMPLÈTE" cls = "verdict-bad" st.markdown(f"
{label}
", unsafe_allow_html=True) def render_field_row(field_name: str, value: str, confidence: float): pretty = FIELD_LABEL_FR.get(field_name, field_name) st.markdown( f"
" f"{pretty}" f"{value}" f"{confidence_dot(confidence)}" f"
", unsafe_allow_html=True, ) def render_page_preview(file_bytes: bytes, suffix: str, zoom: float = 1.2): try: import fitz from PIL import Image except ImportError: st.warning("PyMuPDF / Pillow non disponible — aperçu désactivé.") return if suffix.lower() == ".pdf": with fitz.open(stream=file_bytes, filetype="pdf") as doc: if len(doc) == 0: st.warning("PDF vide.") return pix = doc[0].get_pixmap(matrix=fitz.Matrix(zoom, zoom)) img = Image.frombytes("RGB", (pix.width, pix.height), pix.samples) else: img = Image.open(io.BytesIO(file_bytes)).convert("RGB") st.image(img, use_container_width=True) def write_uploaded_to_tempfile(uploaded) -> Path: suffix = Path(uploaded.name).suffix or ".bin" tmp = tempfile.NamedTemporaryFile(prefix="guichetoi_", suffix=suffix, delete=False) tmp.write(uploaded.getbuffer()) tmp.close() return Path(tmp.name) SUPPORTED_EXTS = {".pdf", ".png", ".jpg", ".jpeg", ".bmp", ".tif", ".tiff"} def collect_files(uploaded_files) -> list[Path]: """ Take Streamlit UploadedFile objects (regular docs and/or .zip archives) and return a flat list of paths on disk pointing at every supported document inside. ZIP contents are extracted to a temp directory. Hidden files and macOS resource forks (`__MACOSX/…`, `._foo`) are skipped. """ out: list[Path] = [] for f in uploaded_files: suffix = Path(f.name).suffix.lower() if suffix == ".zip": extract_dir = Path(tempfile.mkdtemp(prefix="guichetoi_zip_")) try: with zipfile.ZipFile(io.BytesIO(f.getbuffer())) as zf: zf.extractall(extract_dir) except zipfile.BadZipFile: st.error(f"« {f.name} » n'est pas un ZIP valide.") continue for p in extract_dir.rglob("*"): if not p.is_file(): continue if p.suffix.lower() not in SUPPORTED_EXTS: continue if p.name.startswith("._") or "__MACOSX" in p.parts: continue out.append(p) elif suffix in SUPPORTED_EXTS: out.append(write_uploaded_to_tempfile(f)) else: st.warning(f"Format non supporté ignoré : {f.name}") return out # ──────────────────────────────────────────────────────────────────────────── # Header # ──────────────────────────────────────────────────────────────────────────── col_logo, col_title = st.columns([1, 8]) with col_logo: logo_path = ROOT / "assets" / "fibergate_logo.svg" if logo_path.exists(): st.image(str(logo_path), width=140) else: # Inline CSS fallback (no asset required) — keeps the brand visible st.markdown( "", unsafe_allow_html=True, ) with col_title: st.markdown( "

Guichet Accueil Infrastructures

" "

Analyse automatique des demandes de " "localisation du Point d'Accès au Réseau (PAR). Téléversez les pièces — " "individuellement ou en archive ZIP — et récupérez le verdict de " "complétude et le brouillon d'accusé de réception.

", unsafe_allow_html=True, ) st.markdown("---") # ──────────────────────────────────────────────────────────────────────────── # Sidebar # ──────────────────────────────────────────────────────────────────────────── with st.sidebar: st.markdown("## 📘 Mode d'emploi") st.markdown( "1. **Téléversez** tous les fichiers de la demande " "(individuellement ou via un ZIP du dossier).\n" "2. Le moteur **identifie** chaque document.\n" "3. Il **extrait** les champs métier (n° d'urbanisme, " "DLPI, nb de logements, etc.).\n" "4. Il **détecte** les pièces manquantes ou incomplètes.\n" "5. Téléchargez le **brouillon de mail** d'accusé de réception." ) st.markdown("---") st.markdown("### Pièces attendues") for cls in EXPECTED_CLASSES: st.markdown(f"{CLASS_ICON[cls]} {CLASS_LABEL[cls]}") st.markdown("---") st.caption( "Modèle : LayoutLMv3 fine-tuné · 6 classes · 13 champs · " "post-traitement par règles." ) # ════════════════════════════════════════════════════════════════════════════ # Main view — upload + analyse + verdict # ════════════════════════════════════════════════════════════════════════════ st.markdown("### Vérification d'une demande de localisation PAR") st.caption( "Choisissez un échantillon de démonstration ci-dessous **ou** téléversez vos " "propres fichiers (un par un, en multi-sélection, ou en archive ZIP)." ) # ── Demo samples — one click, instant cached result ─────────────────────── samples_data = load_sample_verdicts() if samples_data: st.markdown("#### 🎬 Échantillons de démonstration") st.caption( "Cas de référence avec résultats précalculés — affichage instantané pour " "la présentation. Pour une analyse en direct, utilisez le téléversement plus bas." ) sample_cols = st.columns(2) for i, (label, blurb, zip_name) in enumerate(DEMO_SAMPLES): if zip_name not in samples_data: continue with sample_cols[i % 2]: if st.button(label, key=f"sample_btn_{i}", use_container_width=True, help=blurb): st.session_state["sample_verdict"] = samples_data[zip_name] st.session_state["sample_label"] = label st.session_state["sample_zip"] = zip_name st.caption(blurb) if st.session_state.get("sample_verdict"): if st.button("✖ Effacer l'échantillon", key="clear_sample"): for k in ("sample_verdict", "sample_label", "sample_zip"): st.session_state.pop(k, None) st.rerun() st.markdown("---") # ── File uploader (live analysis) ───────────────────────────────────────── st.markdown("#### 📤 Ou téléversez votre propre demande") uploaded_files = st.file_uploader( "Glissez-déposez vos fichiers ici (PDF, images ou archive ZIP)", type=["pdf", "png", "jpg", "jpeg", "bmp", "tif", "tiff", "zip"], accept_multiple_files=True, key="multi_upload", help=( "Vous pouvez téléverser :\n" "• un ou plusieurs documents (PDF / image)\n" "• une archive ZIP contenant tout le dossier de la demande\n" "Les sous-dossiers à l'intérieur du ZIP sont parcourus automatiquement." ), ) # Determine which source we're using: uploaded files take priority IF the # user has just uploaded; otherwise fall back to the selected sample. using_sample = bool(st.session_state.get("sample_verdict")) and not uploaded_files if not uploaded_files and not using_sample: st.info( "👆 Sélectionnez un échantillon ci-dessus pour la démonstration, " "ou téléversez les fichiers d'une demande réelle." ) st.stop() # ── Build the verdict, either from cache or by running the engine ───────── if using_sample: sample_label = st.session_state.get("sample_label", "") sample_zip = st.session_state.get("sample_zip", "") st.success( f"📦 Résultat précalculé — **{sample_label}** · source : `{sample_zip}`" ) verdict = verdict_from_dict(st.session_state["sample_verdict"]) # Inventory of the documents in the cached verdict with st.expander( f"Voir les {len(verdict.documents)} fichier(s) analysé(s)", expanded=False, ): for doc in verdict.documents: st.markdown(f"- `{Path(doc.file).name}`") else: # Live mode: extract files (ZIP → flat list), then run engine with st.spinner("📦 Préparation des fichiers…"): temp_paths = collect_files(uploaded_files) if not temp_paths: st.error("Aucun document exploitable trouvé dans les fichiers téléversés.") st.stop() n_zip = sum(1 for f in uploaded_files if Path(f.name).suffix.lower() == ".zip") header = f"📥 **{len(temp_paths)} document(s) à analyser**" if n_zip: header += f" · extraits depuis {n_zip} archive(s) ZIP" st.markdown(header) with st.expander("Voir la liste des fichiers", expanded=False): for p in temp_paths: st.markdown(f"- `{p.name}`") with st.spinner(f"🔍 Analyse de {len(temp_paths)} document(s) — peut prendre quelques minutes…"): engine = get_engine() verdict = engine.evaluate_files(temp_paths) # ── Verdict banner needs_review = bool(getattr(verdict, "manual_review_documents", None)) verdict_banner(verdict.status, needs_review=needs_review) # ── Doc checklist + counts by_class: dict[str, int] = {} for d in verdict.documents: by_class[d.doc_class] = by_class.get(d.doc_class, 0) + 1 st.markdown("#### 📋 Composition de la demande") cols = st.columns(len(EXPECTED_CLASSES)) for col, cls in zip(cols, EXPECTED_CLASSES): n = by_class.get(cls, 0) icon = CLASS_ICON[cls] label = CLASS_LABEL[cls] with col: if n > 0: st.metric(f"{icon}\n{label}", n, delta="Présent") else: st.metric(f"{icon}\n{label}", "—", delta="Manquant") st.markdown("---") # ── Missing / Incomplete details col_miss, col_inc = st.columns(2) with col_miss: st.markdown("#### 🚫 Documents manquants") if verdict.missing_documents: for m in verdict.missing_documents: st.error(m) else: st.success("Aucun document manquant") with col_inc: st.markdown("#### ⚠️ Documents incomplets") if verdict.incomplete_documents: for m in verdict.incomplete_documents: st.warning(m) else: st.success("Aucun document incomplet") # ── Manual review (separate — does NOT make the demande incomplète) if getattr(verdict, "manual_review_documents", None): st.markdown("---") st.markdown("#### 👤 Vérification manuelle requise") st.caption( "Ces documents sont fournis mais le modèle ne peut pas les analyser " "automatiquement avec certitude. La demande n'est **pas** marquée " "incomplète pour autant — un consultant doit confirmer manuellement." ) for m in verdict.manual_review_documents: st.info(m) # ── Fiche summary (always shown if any fiche was processed) if verdict.fiche_summary: st.markdown("---") st.markdown("#### 📋 Synthèse de la fiche de renseignement") for name, payload in sorted(verdict.fiche_summary.items()): render_field_row(name, str(payload["value"]), payload["confidence"]) # ── Per-document detail (collapsed by default) st.markdown("---") st.markdown("#### 🗂️ Détails par document") for d in verdict.documents: file_name = Path(d.file).name icon = CLASS_ICON.get(d.doc_class, "📄") header = f"{icon} **{file_name}** — classé {CLASS_LABEL.get(d.doc_class, d.doc_class)} ({d.doc_confidence:.0%})" with st.expander(header): st.markdown(class_pill(d.doc_class, d.doc_confidence), unsafe_allow_html=True) if d.flags: nice_flags = [] for flag in d.flags: if flag.startswith("class_overridden"): nice_flags.append("⚙️ classe ajustée par nom de fichier") elif flag == "plan_inexploitable": nice_flags.append("⚠️ plan possiblement inexploitable") elif flag == "low_classification_confidence": nice_flags.append("ℹ️ classification incertaine") else: nice_flags.append(flag) st.caption(" · ".join(nice_flags)) if d.fields: for fname, payload in sorted(d.fields.items()): render_field_row(fname, str(payload["value"]), payload["confidence"]) else: st.caption("(aucun champ extrait pour ce type de document)") # ── CMS file generation (only when the demande is complète) ────────────── verdict_dict = verdict.to_dict() # CMS generation is available for ALL statuses — the consultant chooses when # to pre-fill the spreadsheet. For non-complete demandes the file will simply # carry more gaps (listed below the download button) for manual completion. st.markdown("---") _is_complete = (verdict.status or "").startswith("complèt") _is_hors_perim = verdict.status == "hors-périmètre" st.markdown("#### 📊 Génération du fichier CMS IMMO 9 BANBOU") if _is_complete: st.caption( "La demande est **complète** — le moteur pré-remplit l'onglet " "*création IMB* (et *création syndic* pour les projets collectifs) " "avec les informations extraites. Les coordonnées XY (Géoréso), " "l'identifiant Mondofi et le SIRET restent à compléter manuellement." ) elif _is_hors_perim: st.warning( "Cette demande est **hors-périmètre** (dossier de récolement). " "Vous pouvez quand même générer un CMS si nécessaire, mais le " "fichier n'aura aucun sens métier — utilisez-le uniquement " "comme gabarit vide." ) else: st.info( "Cette demande n'est **pas marquée complète**. Vous pouvez quand " "même générer un CMS partiel pour le compléter manuellement — " "tous les champs manquants seront listés ci-dessous." ) # Preview of what will be filled in the CMS (regardless of status) cms_preview = cms_gen.summarise_cms_fields(verdict_dict) cms_cols = st.columns(3) keys = list(cms_preview.keys()) for i, k in enumerate(keys): v = cms_preview[k] cms_cols[i % 3].metric(k, str(v)) # Build the CMS xlsx into a temp file then surface as a download_button try: out_path = Path(tempfile.gettempdir()) / "GuichetOI_CMS_prerempli.xlsx" cms_result = cms_gen.fill_cms(verdict_dict, out_path) with open(out_path, "rb") as f: cms_bytes = f.read() btn_label = ( "⬇️ Télécharger le CMS pré-rempli (.xlsx)" if _is_complete else "⬇️ Télécharger le CMS partiel (.xlsx)" ) st.download_button( btn_label, data=cms_bytes, file_name="GuichetOI_CMS_prerempli.xlsx", mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", use_container_width=True, ) # ── Tell the consultant which cells still need attention ────────── missing_x = cms_result.get("missing_extractions") or [] manual_x = cms_result.get("manual_lookup") or [] if missing_x or manual_x: st.markdown("##### 🛠️ À compléter manuellement avant envoi") if missing_x: st.warning( f"**{len(missing_x)} champ(s) attendu(s) n'ont pas pu être " "extraits automatiquement** — vérifier dans les documents source " "et compléter dans le CMS :" ) for f in missing_x: st.markdown(f"- {f}") if manual_x: with st.expander( f"ℹ️ {len(manual_x)} champ(s) toujours saisis manuellement " "(Géoréso, Mondofi, Siret…)", expanded=False, ): for f in manual_x: st.markdown(f"- {f}") except FileNotFoundError as e: st.error(f"Modèle CMS introuvable : {e}") except Exception as e: st.error(f"Erreur lors de la génération du CMS : {e}") # ── Downloadable artefacts st.markdown("---") st.markdown("#### 📨 Brouillon de mail d'accusé de réception") st.text_area( "Corps du mail", value=verdict.ar_mail_body, height=320, help="Sélectionnez et copiez pour coller dans MSURVEY.", key="ar_mail_text", ) col_d1, col_d2 = st.columns(2) with col_d1: st.download_button( "⬇️ Télécharger le mail", data=verdict.ar_mail_body.encode("utf-8"), file_name="ar_mail.txt", mime="text/plain", use_container_width=True, ) with col_d2: import json as _json st.download_button( "⬇️ Télécharger le verdict JSON", data=_json.dumps(verdict.to_dict(), ensure_ascii=False, indent=2).encode("utf-8"), file_name="verdict.json", mime="application/json", use_container_width=True, ) with st.expander("📦 Verdict JSON brut"): st.json(verdict.to_dict())