# -*- coding: utf-8 -*- """ Ingestion du contenu du site via l'API REST WordPress (public + non public), puis construction d'un index vectoriel FAISS. Ingest website content via the WordPress REST API (public + non-public), then build a FAISS index. Le contenu NON public (pages/articles prives, brouillons, medias proteges) est recupere uniquement si des identifiants WordPress valides sont fournis (mot de passe d'application). / Non-public content is fetched only when valid WordPress credentials (Application Password) are provided. """ import re import json import pickle import html import time import requests from requests.auth import HTTPBasicAuth import config # -------------------------------------------------------------------------- # Session authentifiee / authenticated session # -------------------------------------------------------------------------- def _make_session(): session = requests.Session() session.headers.update({"User-Agent": "VizyonAyiti360-Assistant/1.0"}) authenticated = False if (config.INCLUDE_NON_PUBLIC and config.WP_USERNAME and config.WP_APP_PASSWORD): # Le mot de passe d'application peut contenir des espaces : on les garde. session.auth = HTTPBasicAuth(config.WP_USERNAME, config.WP_APP_PASSWORD) authenticated = True return session, authenticated # -------------------------------------------------------------------------- # Nettoyage HTML / HTML cleaning # -------------------------------------------------------------------------- def _strip_html(raw: str) -> str: if not raw: return "" raw = re.sub(r"(?s)<(script|style).*?", " ", raw) raw = re.sub(r"(?s)", " ", raw) text = re.sub(r"(?s)<[^>]+>", " ", raw) text = html.unescape(text) text = re.sub(r"\s+", " ", text).strip() return text # -------------------------------------------------------------------------- # Recuperation d'un type de contenu / fetch a content type # -------------------------------------------------------------------------- def fetch_content_type(rest_base, label, session, authenticated): items = [] page = 1 params_base = {"per_page": 100, "_fields": "id,link,title,content,excerpt,date,type,status"} # Si authentifie : contexte edition + statuts non publics. if authenticated: params_base["context"] = "edit" params_base["status"] = config.WP_STATUSES while True: params = dict(params_base, page=page) url = f"{config.WP_API_BASE}/{rest_base}" try: r = session.get(url, params=params, timeout=30) except requests.RequestException as e: print(f" [!] {rest_base} page {page}: {e}") break if r.status_code in (400, 401, 403): # 400 = au-dela de la derniere page ; 401/403 = pas les droits # pour ce statut -> on retente en public seulement. if authenticated and page == 1 and r.status_code in (401, 403): print(f" [!] {rest_base}: acces non-public refuse, repli public.") params_base.pop("context", None) params_base.pop("status", None) authenticated = False continue break if r.status_code != 200: print(f" [!] {rest_base}: HTTP {r.status_code}") break batch = r.json() if not isinstance(batch, list) or not batch: break for it in batch: title = _strip_html((it.get("title") or {}).get("rendered", "") or (it.get("title") or {}).get("raw", "")) content = it.get("content") or {} body = _strip_html(content.get("rendered", "") or content.get("raw", "")) if not body: excerpt = it.get("excerpt") or {} body = _strip_html(excerpt.get("rendered", "") or excerpt.get("raw", "")) if not body and not title: continue status = it.get("status", "publish") items.append({ "id": f"{rest_base}-{it.get('id')}", "type": label, "status": status, "title": title or "(sans titre)", "url": it.get("link", config.SITE_URL), "date": it.get("date", ""), "text": f"{title}. {body}".strip(), }) total_pages = int(r.headers.get("X-WP-TotalPages", page)) if page >= total_pages: break page += 1 time.sleep(0.2) npub = sum(1 for i in items if i["status"] != "publish") print(f" [+] {label}: {len(items)} element(s) ({npub} non-public(s))") return items def fetch_site(): session, authenticated = _make_session() if authenticated: print(" [i] Authentification WordPress active (contenu non-public inclus).") else: print(" [i] Mode public uniquement (pas d'identifiants WordPress).") documents = [] for rest_base, label in config.WP_CONTENT_TYPES.items(): documents.extend(fetch_content_type(rest_base, label, session, authenticated)) return documents # -------------------------------------------------------------------------- # Decoupage / chunking # -------------------------------------------------------------------------- def chunk_text(text, size, overlap): text = text.strip() if len(text) <= size: return [text] if text else [] chunks, start = [], 0 while start < len(text): end = start + size chunk = text[start:end] if end < len(text): last_dot = chunk.rfind(". ") if last_dot > size * 0.5: chunk = chunk[:last_dot + 1] end = start + last_dot + 1 chunks.append(chunk.strip()) start = end - overlap return [c for c in chunks if c] def build_chunks(documents): chunks = [] for doc in documents: for i, part in enumerate(chunk_text(doc["text"], config.CHUNK_SIZE, config.CHUNK_OVERLAP)): chunks.append({ "chunk_id": f"{doc['id']}-{i}", "type": doc["type"], "status": doc.get("status", "publish"), "title": doc["title"], "url": doc["url"], "date": doc["date"], "text": part, }) return chunks # -------------------------------------------------------------------------- # Index FAISS # -------------------------------------------------------------------------- def build_index(progress_cb=None): from sentence_transformers import SentenceTransformer import faiss def _say(msg): print(msg) if progress_cb: progress_cb(msg) _say("Recuperation du contenu du site (API REST WordPress)...") documents = fetch_site() if not documents: raise RuntimeError("Aucun contenu recupere depuis le site.") _say(f"Decoupage de {len(documents)} document(s)...") chunks = build_chunks(documents) _say(f"{len(chunks)} passage(s) a indexer.") _say(f"Chargement des embeddings : {config.EMBEDDING_MODEL}") model = SentenceTransformer(config.EMBEDDING_MODEL) _say("Calcul des vecteurs...") texts = [c["text"] for c in chunks] vectors = model.encode(texts, batch_size=32, show_progress_bar=False, normalize_embeddings=True, convert_to_numpy=True).astype("float32") dim = vectors.shape[1] index = faiss.IndexFlatIP(dim) index.add(vectors) faiss.write_index(index, str(config.INDEX_FILE)) with open(config.CHUNKS_FILE, "wb") as f: pickle.dump(chunks, f) meta = { "built_at": time.strftime("%Y-%m-%d %H:%M:%S"), "site": config.SITE_URL, "documents": len(documents), "chunks": len(chunks), "non_public": sum(1 for c in chunks if c.get("status") != "publish"), "embedding_model": config.EMBEDDING_MODEL, "dim": dim, } with open(config.META_FILE, "w", encoding="utf-8") as f: json.dump(meta, f, ensure_ascii=False, indent=2) _say(f"Index construit : {len(chunks)} passages.") return meta def index_exists(): return config.INDEX_FILE.exists() and config.CHUNKS_FILE.exists() def load_meta(): if config.META_FILE.exists(): with open(config.META_FILE, encoding="utf-8") as f: return json.load(f) return None if __name__ == "__main__": print(json.dumps(build_index(), ensure_ascii=False, indent=2))