# Trouve-KA — construction du document d'index # Author: Simon-Pierre Boucher # Contact: contact@spboucher.ai """Transforme une page parsée + signaux Québec en document OpenSearch (étape 1).""" from datetime import UTC, datetime from typing import Any from trouveka.types import ParsedPage, QuebecSignals # Catégorisation grossière par domaine — raffinée en asynchrone (étape 2) _GOV_SUFFIXES = (".gouv.qc.ca", ".gc.ca", ".quebec.ca") _GOV_DOMAINS = {"quebec.ca", "canada.ca", "montreal.ca", "laval.ca", "gatineau.ca", "sherbrooke.ca"} _NEWS_DOMAINS = { "lapresse.ca", "ledevoir.com", "journaldemontreal.com", "journaldequebec.com", "radio-canada.ca", "tvanouvelles.ca", "lesoleil.com", "ledroit.com", "latribune.ca", "lenouvelliste.ca", "lequotidien.com", "lavoixdelest.ca", "noovo.info", "24heures.ca", "montrealgazette.com", } _EDU_SUFFIXES = (".ulaval.ca", ".umontreal.ca", ".mcgill.ca", ".uqam.ca", ".usherbrooke.ca", ".concordia.ca", ".polymtl.ca", ".etsmtl.ca", ".hec.ca") def categorize_domain(domain: str) -> list[str]: d = domain.lower() cats: list[str] = [] if d in _GOV_DOMAINS or any(d.endswith(s) for s in _GOV_SUFFIXES) or ".gouv." in d: cats.append("government") if d in _NEWS_DOMAINS: cats.append("news") if any(d.endswith(s) or d == s.lstrip(".") for s in _EDU_SUFFIXES) or d.endswith(".edu"): cats.append("education") return cats def build_search_document( page: ParsedPage, signals: QuebecSignals, *, domain: str, domain_quebec_score: float, authority_score: float = 0.0, ) -> dict[str, Any]: return { "url": page.url, "canonical_url": page.canonical_url or page.url, "domain": domain, "title": page.title, "description": page.description, "body": page.body[:100_000], "headings": page.headings, "language": page.language, "page_quebec_score": signals.score, "domain_quebec_score": round(domain_quebec_score, 4), "locations": signals.locations, "organizations": [], # enrichissement étape 2 "people": [], # enrichissement étape 2 "categories": categorize_domain(domain), "image_url": page.image_url, "published_at": page.published_at.isoformat() if page.published_at else None, "crawled_at": datetime.now(UTC).isoformat(), "authority_score": round(authority_score, 4), "freshness_score": 0.0, # enrichissement étape 3 "quality_score": 0.0, # enrichissement étape 3 "spam_score": 0.0, # enrichissement étape 3 }