""" scraper.py - Module de scraping intelligent et d'analyse LLM Version backend headless (sans Streamlit). """ from __future__ import annotations import asyncio import json import os import re from urllib.parse import urljoin, urlparse import httpx from backend.env import load_backend_env from backend.mistral_compat import build_mistral_client, chat_complete from backend.config import ( ABOUT_KEYWORDS, EXCLUDED_NAF_CODES, EXCLUDED_URL_DOMAINS, EXCLUDED_URL_KEYWORDS, EXCLUDED_URL_MESSAGE, EXCLUDED_URL_SUBSTRINGS, HOSTING_SIRENS, LEGAL_KEYWORDS, MISTRAL_LLM_MODEL, NAF_2_DIGIT_MAPPING, NAF_EXCLUDED_SECTORS, PRODUCTS_KEYWORDS, SIREN_PATTERNS, TOKEN_LIMIT, ) # ============================================================================ # INITIALISATION / ENVIRONNEMENT # --------------------------------------------------------------------------- # Chargement des variables d'environnement puis helpers de lecture de secrets. # ============================================================================ load_backend_env() def _sanitize_env_value(raw: str) -> str: # Harmonise la lecture env (quotes + commentaires inline). value = (raw or "").strip() value = re.sub(r"\s+#.*$", "", value).strip() value = value.strip("'\"") return value def _get_secret(key: str, *, required: bool = True, default: str = "") -> str: # Lecture centralisee des secrets backend. value = _sanitize_env_value(os.getenv(key) or default or "") if required and not value: raise ValueError(f"{key} doit etre configure en variable d'environnement") return value # ============================================================================ # PARAMETRES SCRAPING + COUTS # --------------------------------------------------------------------------- # Reglages Jina Reader et estimation de cout/token pour monitoring. # ============================================================================ JINA_API_KEY = _get_secret("JINA_API_KEY", required=False, default="") # Garde-fou volumetrique: on tronque le markdown avant LLM. JINA_MAX_CHARS = 10000 JINA_REMOVE_SELECTORS = "header, footer, nav" JINA_MAX_RETRIES = 3 JINA_TIMEOUT = 15.0 COST_JINA_1M = 0.05 COST_MISTRAL_3B_1M = 0.04 COST_EMBED_1M = 0.01 # ============================================================================ # MONITORING COUT / TOKENS # --------------------------------------------------------------------------- # Collecte locale des tokens/couts pour la requete en cours. # ============================================================================ class _CostTracker: # Monitoring cout/tokens pour suivre la conso IA. def __init__(self): self._entries = [] def reset(self): self._entries = [] def log(self, step_name: str, char_count: int, cost_per_1m: float): tokens = char_count / 4 cost = (tokens / 1_000_000) * cost_per_1m self._entries.append({"step": step_name, "tokens": int(tokens), "cost": cost}) print(f"[MONITOR] {step_name:<25} | Tokens: {int(tokens):>8,} | Cout: {cost:.6f} EUR".replace(",", " ")) def print_total(self): total_tokens = sum(e["tokens"] for e in self._entries) total_cost = sum(e["cost"] for e in self._entries) print(f"[MONITOR] TOTAL REQUEST COST | Tokens: {total_tokens:,} | Cout: {total_cost:.6f} EUR".replace(",", " ")) _cost_tracker = _CostTracker() def log_usage(step_name: str, char_count: int, cost_per_1m: float): # Point d'entree unique pour tracer les couts par etape. _cost_tracker.log(step_name, char_count, cost_per_1m) # ============================================================================ # UTILITAIRES HTTP JINA # --------------------------------------------------------------------------- # Fallback automatique vers Jina public si quota/auth avec cle API. # ============================================================================ def _extract_total_tokens(usage_obj) -> int: if usage_obj is None: return 0 if hasattr(usage_obj, "total_tokens"): return int(getattr(usage_obj, "total_tokens") or 0) if isinstance(usage_obj, dict): return int(usage_obj.get("total_tokens") or 0) return 0 def _extract_chat_content(response) -> str: # Defensive parsing pour differences de format entre versions SDK. try: content = response.choices[0].message.content except Exception: content = "" if isinstance(content, list): parts: list[str] = [] for chunk in content: if isinstance(chunk, dict): text = chunk.get("text") if isinstance(text, str): parts.append(text) elif isinstance(chunk, str): parts.append(chunk) return "".join(parts).strip() return str(content or "").strip() def _should_fallback_to_public_jina(response: httpx.Response) -> bool: # Detecte quota/auth KO pour retenter sans cle API. if response.status_code in {401, 402, 403, 429}: return True body = (response.text or "").lower() return any(token in body for token in ("unauthorized", "forbidden", "quota", "token", "payment")) async def _jina_get_with_fallback(client: httpx.AsyncClient, jina_url: str, public_headers: dict) -> httpx.Response: # Tente avec API key puis bascule en mode public si quota/auth KO. if JINA_API_KEY: auth_headers = dict(public_headers) auth_headers["Authorization"] = f"Bearer {JINA_API_KEY}" resp = await client.get(jina_url, headers=auth_headers) if _should_fallback_to_public_jina(resp): print(f"[JINA] Fallback public active (HTTP {resp.status_code})") return await client.get(jina_url, headers=public_headers) return resp return await client.get(jina_url, headers=public_headers) # ============================================================================ # SCRAPING PRINCIPAL # --------------------------------------------------------------------------- # Recuperation markdown, retries reseau, et normalisation de sortie. # ============================================================================ async def crawl_with_jina(url: str) -> str | None: # Scraping markdown principal avec retry progressif. jina_url = f"https://r.jina.ai/{url}" headers = { "Accept": "text/markdown", "User-Agent": "Mozilla/5.0 (compatible; ValuationBot/1.0)", "X-Remove-Selector": JINA_REMOVE_SELECTORS, "X-Retain-Images": "none", "X-With-Links-Summary": "false", } for attempt in range(1, JINA_MAX_RETRIES + 1): try: async with httpx.AsyncClient(timeout=JINA_TIMEOUT, follow_redirects=True) as client: response = await _jina_get_with_fallback(client, jina_url, headers) if response.status_code == 200 and len(response.text.strip()) > 100: # Troncature defensive pour limiter latence/cout aval. truncated = response.text[:JINA_MAX_CHARS] log_usage("Jina Reader", len(truncated), COST_JINA_1M) return truncated except (httpx.TimeoutException, httpx.ConnectError, httpx.ReadError): pass except Exception: pass if attempt < JINA_MAX_RETRIES: await asyncio.sleep(attempt * 2) return None async def scrape_url(url: str) -> tuple[str | None, None]: # Signature conservee pour compatibilite avec l'ancienne version. markdown = await crawl_with_jina(url) if markdown: return markdown, None return None, None # ============================================================================ # PREPROCESSING TEXTE / LIENS # --------------------------------------------------------------------------- # Nettoyage texte et detection heuristique de pages importantes. # ============================================================================ def normalize_text(text: str) -> str: # Uniformise les espaces/tirets Unicode pour fiabiliser regex SIREN. if not text: return "" text = text.replace("\u00a0", " ").replace("\u202f", " ").replace("\u2007", " ").replace("\u2009", " ") text = text.replace("\u200b", "").replace("\ufeff", "") text = text.replace("\u2013", "-").replace("\u2014", "-").replace("\u2012", "-") text = re.sub(r"[ \t]+", " ", text) return text def find_links_in_markdown(markdown: str, base_url: str) -> dict: # Detection heuristique des pages about/products/legal depuis markdown. if not markdown: return {} found = {"about": None, "products": None, "legal": None} base_domain = urlparse(base_url).netloc md_links = re.findall(r"\[([^\]]*)\]\(([^)]+)\)", markdown) raw_urls = re.findall(r"(https?://[^\s\)\"\'<>]+)", markdown) all_links = list(md_links) + [("", u) for u in raw_urls] for anchor_text, href in all_links: href = href.strip().rstrip("/") full_url = urljoin(base_url, href) parsed = urlparse(full_url) if parsed.netloc and parsed.netloc != base_domain: continue if href.startswith("#") or href.startswith("mailto:") or href.startswith("tel:"): continue combined = f"{anchor_text.lower()} {href.lower()}" if not found["about"] and any(kw in combined for kw in ABOUT_KEYWORDS): found["about"] = full_url if not found["products"] and any(kw in combined for kw in PRODUCTS_KEYWORDS): found["products"] = full_url if not found["legal"] and any(kw in combined for kw in LEGAL_KEYWORDS): found["legal"] = full_url return found # ============================================================================ # EXTRACTION SIREN DEPUIS PAGES LEGALES # --------------------------------------------------------------------------- # Heuristiques de discovery URL + parsing HTML -> texte -> regex SIREN. # ============================================================================ _LEGAL_URL_SUFFIXES = [ "/mentions-legales", "/mentions-legales/", "/legal", "/legal/", "/mentions-legales.html", "/legal-notice", "/informations-legales", "/informations-legales/", "/cgu", "/cgu/", "/cgv", "/cgv/", "/conditions-generales", "/conditions-generales/", "/politique-de-confidentialite", "/politique-de-confidentialite/", ] _LEGAL_SIGNALS = [ "siren", "siret", "rcs", "immatricul", "mentions légales", "mentions legales", "éditeur", "editeur", "hébergeur", "hebergeur", "capital social", "registre du commerce", ] _HEADERS_RAW = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "fr-FR,fr;q=0.9,en;q=0.5", } def _html_to_text(html: str) -> str: # Nettoyage HTML minimaliste sans dependance externe (bs4, etc.). text = re.sub(r"]*>[\s\S]*?", " ", html, flags=re.IGNORECASE) text = re.sub(r"]*>[\s\S]*?", " ", text, flags=re.IGNORECASE) text = re.sub(r"<[^>]+>", " ", text) text = re.sub(r" ", " ", text) text = re.sub(r"&", "&", text) text = re.sub(r"&[a-zA-Z]+;", " ", text) text = re.sub(r"&#\d+;", " ", text) text = re.sub(r"\s+", " ", text) return text.strip() async def _fetch_legal_page_raw(base_url: str) -> str | None: # Brute-force des pages legales pour extraire un SIREN fiable. parsed = urlparse(base_url) root = f"{parsed.scheme}://{parsed.netloc}" async with httpx.AsyncClient(timeout=12.0, follow_redirects=True, headers=_HEADERS_RAW) as client: for suffix in _LEGAL_URL_SUFFIXES: test_url = root + suffix try: resp = await client.get(test_url) if resp.status_code == 200 and len(resp.text.strip()) > 300: text = _html_to_text(resp.text) # On valide seulement les pages avec signaux "legaux" forts. if any(sig in text.lower() for sig in _LEGAL_SIGNALS): return text except Exception: continue return None def extract_siren_from_text(text: str) -> str | None: # Extraction SIREN robuste a partir des patterns metiers. if not text: return None cleaned = normalize_text(text) for pattern in SIREN_PATTERNS: matches = re.findall(pattern, cleaned) for match in matches: siren = "".join(match) if isinstance(match, tuple) else match siren = re.sub(r"\D", "", siren)[:9] if len(siren) == 9 and siren.isdigit() and siren not in HOSTING_SIRENS: return siren return None # ============================================================================ # PREPARATION CONTENU LLM # --------------------------------------------------------------------------- # Reduction du markdown pour respecter budget token et limiter le bruit. # ============================================================================ def reduce_markdown(text: str, max_tokens: int | None = None) -> str: # Nettoyage + troncature pour rester dans le budget token LLM. if max_tokens is None: max_tokens = TOKEN_LIMIT if not text: return "" lines = text.split("\n") cleaned = [] empty_count = 0 skip_patterns = [ r"^\s*!\[", r"^\s*\|.*\|.*\|", r"^\s*[-=]{3,}", r"^\s* max_chars * 0.8: truncated = truncated[: last_period + 1] return truncated + "\n\n[... contenu tronque pour respecter la limite de tokens ...]" # ============================================================================ # ETAPE LLM + ENRICHISSEMENT API GOUV # --------------------------------------------------------------------------- # Extraction semantique (name/summary/keywords) puis enrichissement officiel. # ============================================================================ def analyze_with_llm(markdown_content: str) -> dict: # Extraction structuree name/summary/keywords via Mistral. client = build_mistral_client(_get_secret("MISTRAL_API_KEY")) system_prompt = """Tu es un expert en analyse d'entreprise specialise en M&A. Analyse le contenu web d'une entreprise et extrais les informations suivantes. Reponds UNIQUEMENT avec un JSON valide, sans texte avant ni apres. Format attendu : { "name": "Nom officiel de l'entreprise", "summary": "Description concise de l'activite en 2 phrases maximum.", "keywords": "Mot1, Mot2, Mot3, Mot4, Mot5" } NE cherche PAS de numero SIREN. Concentre-toi uniquement sur name, summary et keywords.""" try: response = chat_complete( client, model=MISTRAL_LLM_MODEL, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"Contenu web de l'entreprise a analyser :\n\n{markdown_content}"}, ], response_format={"type": "json_object"}, temperature=0.1, max_tokens=500, ) # Le contrat JSON strict simplifie le parsing backend. content = _extract_chat_content(response) log_usage("Mistral LLM (input)", len(system_prompt) + len(markdown_content), COST_MISTRAL_3B_1M) log_usage("Mistral LLM (output)", len(content), COST_MISTRAL_3B_1M) tokens_llm = _extract_total_tokens(getattr(response, "usage", None)) result = json.loads(content) return { "name": result.get("name") or None, "summary": result.get("summary") or None, "keywords": result.get("keywords") or None, "tokens_llm": tokens_llm, } except Exception as exc: print(f"Erreur LLM Mistral : {exc}") return {"name": None, "summary": None, "keywords": None, "tokens_llm": 0} def get_entreprise_data(siren: str) -> dict | None: # Enrichissement officiel via API Recherche Entreprises (gouv). if not siren: return None siren_clean = re.sub(r"\D", "", siren.strip())[:9] if len(siren_clean) != 9: return None url = f"https://recherche-entreprises.api.gouv.fr/search?q={siren_clean}" try: response = httpx.get(url, timeout=15.0) if response.status_code != 200: return None data = response.json() results = data.get("results", []) if not results: return None entreprise = results[0] # On prend le premier resultat (normalement exact sur un SIREN unique). denomination = entreprise.get("nom_complet") or entreprise.get("nom_raison_sociale") naf_code = entreprise.get("activite_principale") date_creation = entreprise.get("date_creation") tranche_effectif = entreprise.get("tranche_effectif_salarie") nature_juridique = entreprise.get("nature_juridique") ca = None resultat_net = None finances = entreprise.get("finances") if finances and isinstance(finances, dict): derniere_annee = max(finances.keys()) donnees = finances[derniere_annee] ca = donnees.get("ca") resultat_net = donnees.get("resultat_net") return { "naf_code": naf_code.strip() if naf_code else None, "date_creation": date_creation.strip() if date_creation else None, "denomination": denomination.strip() if denomination else None, "tranche_effectif": tranche_effectif, "nature_juridique": nature_juridique, "ca": ca, "resultat_net": resultat_net, } except Exception: return None def process_naf_code(naf_code: str) -> tuple[str | None, str | None, bool]: # Classifie le NAF et determine si l'analyse doit etre stoppee. if not naf_code: return None, None, False naf_clean = naf_code.replace(".", "").replace(" ", "").strip().upper() if naf_clean in EXCLUDED_NAF_CODES: return naf_code, "Holding / Societe de participation (NAF ignore)", False naf_2 = naf_clean[:2] if len(naf_clean) >= 2 else None if not naf_2: return naf_code, None, False if naf_2 in NAF_EXCLUDED_SECTORS: sector_label = NAF_EXCLUDED_SECTORS[naf_2] return naf_code, sector_label, True sector_label = NAF_2_DIGIT_MAPPING.get(naf_2, f"Secteur NAF {naf_2}") return naf_code, sector_label, False # ============================================================================ # GARDE-FOUS METIER # --------------------------------------------------------------------------- # Verification du perimetre URL autorise avant de consommer IA/API. # ============================================================================ def _check_url_excluded(url: str) -> str | None: # Blocage de domaines/keywords hors perimetre produit. parsed = urlparse(url) domain = parsed.netloc.lower().replace("www.", "") url_lower = url.lower() for excluded in EXCLUDED_URL_DOMAINS: if domain == excluded or domain.endswith("." + excluded): return f"domaine exclu ({excluded})" for substring in EXCLUDED_URL_SUBSTRINGS: if substring in domain: return f"domaine contient '{substring}'" for keyword in EXCLUDED_URL_KEYWORDS: if keyword in url_lower: return f"mot-cle interdit dans l'URL ({keyword})" return None # ============================================================================ # ORCHESTRATION DE BOUT EN BOUT # --------------------------------------------------------------------------- # Pipeline complet de scraping/analyse puis wrapper sync pour FastAPI. # ============================================================================ async def scrape_and_analyze(url: str, progress_callback=None) -> dict: # Pipeline complet: validation URL -> SIREN -> Jina -> LLM -> enrichissement. result = { "name": None, "summary": None, "keywords": None, "tokens_llm": 0, "siren": None, "naf_code": None, "naf_2": None, "naf_label": None, "date_creation": None, "must_stop": False, "stop_reason": None, } rejection = _check_url_excluded(url) if rejection: # Politique metier: refus explicite des URL hors scope. result["must_stop"] = True result["stop_reason"] = EXCLUDED_URL_MESSAGE return result if progress_callback: progress_callback("Recherche du SIREN via mentions legales...") legal_text = await _fetch_legal_page_raw(url) siren_found = None if legal_text: siren_found = extract_siren_from_text(normalize_text(legal_text)) if not siren_found: # Fallback: tentative de detection SIREN sur la homepage brute. try: async with httpx.AsyncClient(timeout=12.0, follow_redirects=True, headers=_HEADERS_RAW) as client: resp = await client.get(url) if resp.status_code == 200: siren_found = extract_siren_from_text(normalize_text(_html_to_text(resp.text))) except Exception: pass gov_data_cached = None if siren_found: # Prefetch des donnees gouv pour eviter un second appel plus tard. gov_data_cached = get_entreprise_data(siren_found) if gov_data_cached: result["siren"] = siren_found if progress_callback: progress_callback("Exploration de la page web...") main_markdown, _ = await scrape_url(url) if not main_markdown: # Echec de recuperation: stop "fonctionnel" (pas exception serveur). result["must_stop"] = True result["stop_reason"] = ( "Analyse indisponible pour ce site. " "Le contenu de la page n'a pas pu etre recupere apres plusieurs tentatives." ) return result reduced_content = reduce_markdown(main_markdown) if progress_callback: progress_callback("Analyse du contenu par l'intelligence artificielle...") llm_result = analyze_with_llm(reduced_content) result["name"] = llm_result.get("name") result["summary"] = llm_result.get("summary") result["keywords"] = llm_result.get("keywords") result["tokens_llm"] = llm_result.get("tokens_llm", 0) if result["siren"]: if progress_callback: progress_callback("Enrichissement des donnees via l'API Entreprises...") gov_data = gov_data_cached if gov_data: result["date_creation"] = gov_data.get("date_creation") if not result["name"] and gov_data.get("denomination"): result["name"] = gov_data["denomination"] naf_code = gov_data.get("naf_code") if naf_code: # Le NAF peut imposer un arret si secteur exclu de la methode. naf_raw, naf_label, must_stop = process_naf_code(naf_code) result["naf_code"] = naf_raw result["naf_label"] = naf_label result["naf_2"] = naf_raw.replace(".", "").replace(" ", "")[:2] if naf_raw else None if must_stop: result["must_stop"] = True result["stop_reason"] = ( f"Le secteur identifie est {naf_label} (code NAF : {naf_raw}). " "Notre outil ne peut pas produire de resultats pertinents pour ce type d'activite." ) if gov_data.get("ca"): result["ca_gov"] = gov_data["ca"] return result def run_scraper(url: str, progress_callback=None) -> dict: # Wrapper sync pour simplifier l'appel depuis les endpoints FastAPI. _cost_tracker.reset() try: result = asyncio.run(scrape_and_analyze(url, progress_callback)) except RuntimeError: # Cas notebook/event loop deja active: on cree une loop dediee. loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete(scrape_and_analyze(url, progress_callback)) finally: loop.close() _cost_tracker.print_total() return result