import re import gradio as gr import pandas as pd import numpy as np import markdown as md_lib from datasets import load_dataset from sentence_transformers import SentenceTransformer # --- Chargement du modèle (une seule fois au démarrage) --- print("Chargement du modèle BAAI/bge-m3...") model = SentenceTransformer("BAAI/bge-m3") print("Modèle prêt.") # --- Chargement des données --- ds = load_dataset("opt-nc/avps", split="train") df = ds.to_pandas() embeddings_matrix = np.array(df["embedding"].tolist(), dtype=np.float32) norms = np.linalg.norm(embeddings_matrix, axis=1, keepdims=True) embeddings_norm = embeddings_matrix / (norms + 1e-10) job_index = {str(row["id"]): dict(row) for _, row in df.iterrows()} # --- TF-IDF léger pour extraire les mots-clés saillants de chaque annonce --- import math from collections import Counter STOPWORDS = set(""" le la les un une des de du en et à au aux ce qui que qu est par sur dans pour avec ou si bien plus mais aussi car dont où ses son sa leur leurs nos vos il elle ils elles je tu nous vous me te se lui y être avoir faire pouvoir devoir vouloir aller savoir tout tous toute toutes cette cet ces même très peu """.split()) def tokenize(text): words = re.sub(r'[^a-zàâçéèêëîïôûùüÿæœ]', ' ', text.lower()).split() return [w for w in words if len(w) > 3 and w not in STOPWORDS] # Calcul IDF sur le corpus corpus_tokens = [tokenize(str(row.get("text", ""))) for _, row in df.iterrows()] N = len(corpus_tokens) df_counts = Counter() for tokens in corpus_tokens: df_counts.update(set(tokens)) idf = {w: math.log(N / (1 + c)) for w, c in df_counts.items()} # Calcul TF-IDF par annonce → top 3 mots def top_keywords(tokens, n=3): tf = Counter(tokens) total = len(tokens) or 1 scores = {w: (tf[w]/total) * idf.get(w, 0) for w in tf} return [w for w, _ in sorted(scores.items(), key=lambda x: -x[1])[:n]] # Pré-calcul pour toutes les annonces keywords_by_id = {} for (_, row), tokens in zip(df.iterrows(), corpus_tokens): keywords_by_id[str(row["id"])] = top_keywords(tokens) _model_ready = False def encode_query(text: str): global _model_ready try: vec = model.encode(text, normalize_embeddings=True) _model_ready = True return np.array(vec, dtype=np.float32) except Exception as e: print(f"encode_query error: {e}") return None def fmt_date(d): if not d or str(d) in ("NaT", "None", "nan"): return "" try: dt = pd.to_datetime(d) delta = (dt - pd.Timestamp.now()).days urgence = " 🔴" if delta <= 7 else (" 🟡" if delta <= 14 else "") return dt.strftime("%d/%m/%Y") + urgence except Exception: return str(d)[:10] def score_widget(score): pct = int(round(score * 100)) if pct >= 70: bg, color, label = "#d1fae5", "#065f46", f"✦ {pct}%" tip = f"{pct}% — très bonne correspondance" elif pct >= 50: bg, color, label = "#fef3c7", "#92400e", f"◈ {pct}%" tip = f"{pct}% — correspondance correcte" else: bg, color, label = "#f3f4f6", "#6b7280", f"○ {pct}%" tip = f"{pct}% — correspondance partielle" tooltip = ( f"{tip}. Score de similarité sémantique entre votre profil " "et le texte de l'annonce (missions, compétences), " "calculé par le modèle BAAI/bge-m3. 100% = correspondance parfaite." ) return ( f'{label} match' ) GRAY = "color:#6b7280 !important" def md_to_html_highlighted(text_md, qwords): """Convertit le markdown en HTML, force les couleurs inline, surligne les mots-clés.""" if not text_md: return "" html = md_lib.markdown(text_md, extensions=["nl2br", "sane_lists"]) for tag in ("p", "li", "ul", "ol", "h1", "h2", "h3", "h4", "strong", "em"): html = html.replace(f"<{tag}>", f'<{tag} style="{GRAY}">') html = re.sub(f'<{tag} ((?!style=)[^>]*)>', f'<{tag} style="{GRAY}" \\1>', html) if qwords: for w in qwords: html = re.sub( r"(?\w])", r'\1', html, flags=re.IGNORECASE ) return html def render_cards(results_df, scores=None, qwords=None): if results_df.empty: return "

Aucun résultat.

" cards = [] for i, (_, row) in enumerate(results_df.iterrows()): job_id = str(row.get("id") or "") titre = row.get("titre") or "Poste sans titre" numero = row.get("numero") or job_id direction = row.get("direction_interne_acronyme") or "" service = row.get("service_acronyme") or "" grade = row.get("corps_grade") or "" lieu = row.get("lieu_travail") or "" immed = row.get("disponible_immediatement", False) cloture = fmt_date(row.get("date_cloture")) url = row.get("url") or "" texte_md = row.get("text") or "" score = scores[i] if scores is not None else None dir_str = direction + (f" › {service}" if service else "") meta = " · ".join(p for p in [dir_str, grade, lieu] if p) badges = "" if score is not None: badges += score_widget(score) + " " if immed: badges += ('⚡ Immédiat') cloture_html = f'🗓 {cloture}' if cloture else "" ref_html = f'#{numero}' # Tags TF-IDF kw = keywords_by_id.get(job_id, []) tags_html = "" if kw: tags = " ".join( f'' f'#{w}' for w in kw ) tags_html = f'
{tags}
' # Aperçu HTML du markdown avec surlignage preview_html = md_to_html_highlighted(texte_md, qwords) preview_block = f"""
{preview_html}
""" annonce_btn = ( f'Voir →' if url else "" ) cards.append(f"""
{titre}
{badges}

{meta}

{cloture_html}{ref_html}
{tags_html} {preview_block}
{annonce_btn}
""") n = len(results_df) header = f'

{n} offre{"s" if n > 1 else ""} trouvée{"s" if n > 1 else ""}

' return header + "\n".join(cards) def render_detail(job_id: str): row = job_index.get(str(job_id)) if not row: return "" titre = row.get("titre") or "Poste sans titre" direction= row.get("direction_interne") or "" service = row.get("service") or "" grade = row.get("corps_grade") or "" lieu = row.get("lieu_travail") or "" immed = row.get("disponible_immediatement", False) cloture = fmt_date(row.get("date_cloture")) url = row.get("url") or "" url_pdf = row.get("url_pdf") or "" texte_md = row.get("text") or "" numero = row.get("numero") or job_id body_html = md_lib.markdown(texte_md, extensions=["nl2br", "sane_lists"]) dir_str = direction + (f" › {service}" if service else "") meta = " · ".join(p for p in [dir_str, grade, lieu] if p) immed_html = ( '⚡ Disponible immédiatement ' if immed else "" ) cloture_html = f'🗓 Clôture : {cloture}' if cloture else "" btns = "" if url: btns += (f'🌐 Voir l\'annonce') if url_pdf: btns += (f'📎 PDF') share_url = f"https://opt-nc.github.io/avps/{numero}/" share_text = f"AVP OPT-NC — {titre} ({numero})" from urllib.parse import quote se = quote(share_text) ue = quote(share_url) share_html = f"""

📤 Partager

✉️ Email 💬 WhatsApp
""" return f"""

#{numero}

{titre}

{meta}

{immed_html}{cloture_html}
{btns}
{share_html}
{body_html}
""" BANNER_SEMANTIC = """
🧠 Recherche sémantique — résultats classés par similarité avec votre profil (modèle local BAAI/bge-m3)
""" BANNER_KEYWORDS = """
⚠️ Mode mots-clés — le modèle sémantique est en cours de chargement ou indisponible. Résultats filtrés par mots-clés, relancez la recherche dans quelques secondes pour activer la recherche sémantique.
""" BANNER_EMPTY = "" def _search(query: str, threshold: float = 50) -> str: """ Recherche des Avis de Vacances de Poste (AVP) de l'OPT-NC par similarité sémantique. Décrivez librement votre profil, vos compétences ou un intitulé de poste. Les résultats sont classés par similarité avec le modèle BAAI/bge-m3. Args: query: Profil ou mots-clés à rechercher (ex: "ingénieur réseau télécoms management") threshold: Score minimum de similarité en % (0-100, défaut 50) Returns: Liste des annonces correspondantes au format JSON (titre, direction, lieu, score, url) """ import json min_score = threshold / 100.0 qwords = [w for w in query.lower().split() if len(w) > 2] if query.strip() else [] if not query.strip(): return render_cards(df), "", BANNER_EMPTY q_vec = encode_query(query) if q_vec is not None: sims = embeddings_norm @ q_vec order = np.argsort(sims)[::-1] results = df.iloc[order].copy() scores = sims[order].tolist() mask = [s >= min_score for s in scores] filtered = results[mask] filtered_scores = [s for s, m in zip(scores, mask) if m] return render_cards(filtered, filtered_scores, qwords), "", BANNER_SEMANTIC else: def score_row(row): text = f"{row.get('titre','')} {row.get('text','')} {row.get('corps_grade','')}".lower() return sum(1 for w in qwords if w in text) df2 = df.copy() df2["_s"] = df2.apply(score_row, axis=1) df2 = df2[df2["_s"] > 0].sort_values("_s", ascending=False) return render_cards(df2, scores=df2["_s"].tolist(), qwords=qwords), "", BANNER_KEYWORDS def on_card_click(job_id): if not job_id or not job_id.strip(): return "" return render_detail(job_id.strip()) with gr.Blocks(title="AVPs OPT-NC") as demo: gr.HTML("""""") gr.HTML(f"""

📋 AVPs OPT-NC

Décrivez votre profil ou saisissez des mots-clés

""") query_input = gr.Textbox( elem_id="query", label="", placeholder='Ex: "Cadre expérimenté en gestion de projets SI et management" — Entrée pour rechercher', lines=1, max_lines=6, show_label=False, ) with gr.Row(): search_btn = gr.Button("🔍 Rechercher", variant="primary", elem_id="search-btn", scale=1) threshold_slider = gr.Slider(minimum=0, maximum=100, value=50, step=5, label="Score minimum (%)", scale=2) mode_banner = gr.HTML() with gr.Row(): with gr.Column(scale=1, elem_id="results-col"): results_html = gr.HTML() with gr.Column(scale=1, elem_id="detail-panel"): detail_html = gr.HTML( '
' 'Cliquez sur une vignette pour afficher le détail
' ) hidden_id_input = gr.Textbox(visible=False, elem_id="hidden-job-id") gr.HTML("""""") hidden_id_input.change(fn=on_card_click, inputs=hidden_id_input, outputs=detail_html) inputs = [query_input, threshold_slider] outputs = [results_html, detail_html, mode_banner] search_btn.click(fn=_search, inputs=inputs, outputs=outputs, api_name=False) query_input.submit(fn=_search, inputs=inputs, outputs=outputs, api_name=False) threshold_slider.release(fn=_search, inputs=inputs, outputs=outputs, api_name=False) demo.load(fn=_search, inputs=inputs, outputs=outputs, api_name=False) def search_avps(query: str, threshold: float = 50) -> str: """ Recherche les Avis de Vacances de Poste (AVP) de l'OPT-NC (Office des Postes et Télécommunications de Nouvelle-Calédonie). Utilisation optimale : - Décrire un profil complet plutôt que des mots-clés isolés améliore la précision - Mentionner le domaine métier, le niveau hiérarchique et les compétences clés - Baisser threshold (ex: 30) pour élargir les résultats, monter (ex: 70) pour affiner Exemples de requêtes efficaces : - "ingénieur réseau senior expérience cybersécurité SOC et management d'équipe" - "cadre administratif pilotage budgétaire ressources humaines et conduite du changement" - "technicien de maintenance réseaux télécoms fibre optique intervention terrain" - "chef de projet SI transformation digitale MOA" Chaque résultat contient : - titre, numero, direction, service, grade, lieu - disponible_immediatement (bool), date_cloture (YYYY-MM-DD) - score : similarité cosinus [0.0-1.0] — au-dessus de 0.7 = très pertinent - url : page web de l'annonce - url_markdown : texte brut Markdown lisible directement par un LLM - keywords : 3 mots-clés TF-IDF caractérisant l'annonce parmi le corpus Args: query: Description libre du profil ou intitulé de poste recherché threshold: Score minimum de similarité en % entre 0 et 100 (défaut: 50) Returns: JSON array, 10 résultats max, triés par score décroissant """ import json min_score = threshold / 100.0 if not query.strip(): return json.dumps([], ensure_ascii=False) q_vec = encode_query(query) if q_vec is not None: sims = embeddings_norm @ q_vec order = np.argsort(sims)[::-1] results = df.iloc[order].copy() scores = sims[order].tolist() mask = [s >= min_score for s in scores] filtered = results[mask].copy() filtered_scores = [s for s, m in zip(scores, mask) if m] else: qwords = [w for w in query.lower().split() if len(w) > 2] def score_row(row): text = f"{row.get('titre','')} {row.get('text','')}".lower() return sum(1 for w in qwords if w in text) / (len(qwords) + 1) filtered = df.copy() filtered["_s"] = filtered.apply(score_row, axis=1) filtered = filtered[filtered["_s"] > 0].sort_values("_s", ascending=False) filtered_scores = filtered["_s"].tolist() def safe_str(v): """Convertit toute valeur en string, gère NaN, None, bool, numpy types.""" if v is None: return "" try: if pd.isna(v): return "" except (TypeError, ValueError): pass return str(v) out = [] for i, (_, row) in enumerate(filtered.iterrows()): job_id = safe_str(row.get("id")) numero = safe_str(row.get("numero")) or job_id out.append({ "titre": safe_str(row.get("titre")), "numero": numero, "direction": safe_str(row.get("direction_interne_acronyme")), "service": safe_str(row.get("service_acronyme")), "grade": safe_str(row.get("corps_grade")), "lieu": safe_str(row.get("lieu_travail")), "disponible_immediatement": bool(row.get("disponible_immediatement")), "date_cloture": safe_str(row.get("date_cloture")), "score": round(float(filtered_scores[i]), 3), "url": safe_str(row.get("url")), "url_markdown": "https://raw.githubusercontent.com/opt-nc/avps/refs/heads/main/data/" + numero + ".md" if numero else "", "keywords": keywords_by_id.get(job_id, []), }) return json.dumps(out[:10], ensure_ascii=False, indent=2) # --- Endpoint MCP search_avps via gr.api (pas de composants UI nécessaires) --- with demo: gr.api(search_avps, api_name="search_avps") if __name__ == "__main__": demo.launch(mcp_server=True)