"""Adaptateur Mangools KWFinder : l'export CSV **de l'interface** (spec Imports §5, T7). Le format documenté en ligne (`kw, sv, lid, cpc…`) est celui de l'**API** ; le CSV que l'utilisateur télécharge depuis KWFinder n'a rien à voir. C'est ce dernier qui est implémenté ici, sur l'export réel `tests/fixtures/real/kwfinder_usa_top.csv` (relevé 2026-08-31, 147 colonnes) — jamais sur une doc. Ce que le fichier apporte, et que rien d'autre n'apportait : - **`Keyword interest growth (%)`** n'est *pas* la tendance canonique : c'est **une** croissance en pourcentage, là où `trend` attend les douze valeurs d'une sparkline (§3.2). C'est pour cela que `suggest.py` refuse de la proposer en mapping manuel — et c'est exactement ce que cet adaptateur débloque : les **133 colonnes mensuelles** `Search Volume M/YYYY` (7/2015 → 7/2026 dans la fixture) portent la vraie série, dont on garde les **douze derniers mois connus de la ligne**, ramenés en 0–100 comme les douze flottants de Semrush (`TREND_SCALE`). L'heuristique générique ne pouvait pas les construire : une colonne canonique ne se remplit pas avec cent trente-trois colonnes. - **Quatre moyennes de volume** (`Last Known Values`, `Last 12 months`, `Last 6 months`, `Last 3 months`) : seule celle de douze mois vaut un volume mensuel comparable aux autres sources, d'où sa place en tête des en-têtes acceptés. Les trois autres restent dans `raw`. Choix de lecture, tous adossés à la fixture : - **`Search Intent`** écrit déjà le vocabulaire canonique (`commercial`, `informational, commercial, transactional`, vide) : `to_intent` suffit, aucune table de correspondance à inventer — en écrire une donnerait l'illusion d'un mapping là où il n'y a qu'une normalisation de casse et de séparateurs. - **`Serp Features`** est une énumération séparée par des virgules (`Map Pack, AI Overview, People Also Ask`) — même forme que Semrush. - **`CPC/USD`** porte la devise du compte dans l'en-tête, comme le `CPC (USD)` de Semrush mais avec une autre ponctuation : motif ancré plutôt qu'énumération close. - **Ni URL ni position** : KWFinder ne mesure pas un classement, tous les mots-clés importés seront orphelins (§0), et les deux champs canoniques restent `None`. - **`Location`** porte le **pays de la ligne**, en toutes lettres (`United States`, `United Kingdom`) et parfois une ville (`New York` — KWFinder cible aussi des localités). Un export réel en mélange onze. C'est la colonne pays de l'adaptateur (`country_column = "Location"`, spec « Multi-sources et cloisonnement pays » §4.1) : elle est recensée à l'aperçu et sert à cloisonner l'import par cible. La résolution du libellé en code ISO appartient à `app.keywords.countries`, jamais à cet adaptateur — un libellé inconnu ou une ville rendent `None` et la ligne est **ignorée avec son motif**, jamais devinée. La valeur reste par ailleurs dans `raw`, comme toutes les colonnes. `Content Type`, `PPC`, `Keyword Opportunities`, `Keyword interest growth (%)` et les 133 colonnes mensuelles n'ont pas d'équivalent canonique : elles vivent dans `raw`, jamais perdues. """ import re from collections.abc import Iterator, Mapping, Sequence from dataclasses import replace from pathlib import Path from typing import Any from app.keywords.normalize import normalize_keyword from app.keywords.sources.base import ( TREND_LENGTH, TREND_SCALE, Adapter, CanonicalKeywordRow, ColumnSignature, Detection, ParsedRow, RowError, normalize_header, to_decimal, to_int, to_intent, to_serp_features, ) from app.keywords.sources.csvfile import CsvScan # En-têtes constatés sur l'export réel (interface en anglais — c'est la seule langue # d'export de KWFinder observée ; une variante localisée demandera sa propre signature, # comme pour Semrush). _EN: Mapping[str, tuple[str, ...]] = { "keyword": ("keyword",), # Ordre = priorité : la moyenne douze mois d'abord, les trois autres en repli si # l'utilisateur a masqué des colonnes à l'export. "volume": ( "avg. search volume (last 12 months)", "avg. search volume (last known values)", "avg. search volume (last 6 months)", "avg. search volume (last 3 months)", ), "kd": ("keyword difficulty",), "intent": ("search intent",), "serp_features": ("serp features",), # Résolus hors signature : `cpc` par motif (devise du compte), `trend` par les colonnes # mensuelles (voir `_monthly_columns`). Déclarés vides pour compter dans la confiance. "cpc": (), "trend": (), } # `CPC/USD`, `CPC/EUR`, `CPC/GBP` : la devise du compte est dans l'en-tête. _CPC_PATTERN = {"cpc": r"cpc/[a-z]{3}"} # `Search Volume 7/2015` … `Search Volume 7/2026` : la série mensuelle, en clair dans # l'en-tête. Ancré (`fullmatch` sur l'en-tête normalisé) pour ne jamais attraper # `Avg. Search Volume (Last 12 months)`. _MONTHLY = re.compile(r"search volume (\d{1,2})/(\d{4})") _SIGNATURE = ColumnSignature( language="en", columns=_EN, # Le discriminant est l'en-tête verbeux de la moyenne : aucune autre source ne l'écrit, # et aucune signature Semrush ne le reconnaît (leur `volume` est un mot seul). required=frozenset({"keyword", "volume"}), patterns=_CPC_PATTERN, ) def _monthly_columns(headers: Sequence[str]) -> list[tuple[int, str]]: """Les colonnes de volume mensuel, **triées chronologiquement**, ordre du fichier ignoré. Rendues avec leur rang en mois (`année × 12 + mois`), la seule forme qui permette de mesurer un écart entre deux colonnes. L'export les écrit déjà dans l'ordre, mais s'y fier ferait dépendre la tendance d'une convention de tableur : `3/2026` avant `12/2025` en tri lexicographique suffirait à fabriquer une sparkline fausse, sans que rien ne le signale. """ dated: list[tuple[int, str]] = [] for header in headers: matched = _MONTHLY.fullmatch(normalize_header(header)) if matched is not None: dated.append((int(matched[2]) * 12 + int(matched[1]), header)) dated.sort(key=lambda item: item[0]) return dated # Écart maximal, en mois, entre la fin de la fenêtre d'une ligne et le mois le plus récent du # **fichier**. Au-delà, pas de tendance : voir `_trend`. Six mois — la moitié de la fenêtre — # laisse passer les queues vides réelles (un à deux mois non encore publiés, parfois plus sur # un mot-clé de faible volume) et arrête les mots-clés éteints depuis des années. _TREND_MAX_LAG_MONTHS = 6 def _trend(cells: Mapping[str, str | None], monthly: Sequence[tuple[int, str]]) -> list[int] | None: """Les douze derniers mois **connus** de la ligne, en 0–100 (part du maximum). Deux façons dont l'export troue sa série, toutes deux constatées sur la fixture : - **queue vide** — le mois courant et le précédent ne sont pas encore publiés (19 lignes sur 21 en ont au moins un) ; exiger les douze dernières *colonnes* priverait presque tout le fichier de tendance, pour une donnée qui n'existe chez personne ; - **trous internes** — un mot-clé éteint n'a pas de volume pour des dizaines de mois. Les mois manquants sont sautés, pas comblés par des zéros : un zéro inventé se lirait comme un effondrement du volume. La fenêtre est donc celle de la ligne, pas celle du fichier — chaque sparkline montre les douze derniers mois dont **ce** mot-clé a une valeur. Moins de douze mois connus : `None`, comme toute tendance incomplète (ornement d'affichage, jamais une ligne en erreur). **Mais une fenêtre trop vieille est un mensonge**, pas un ornement : `jewlery casting` s'arrête en 1/2020 dans un fichier qui va jusqu'à 7/2026, et rendrait une sparkline pimpante de 2019, indiscernable à l'écran d'une tendance actuelle — sur un mot-clé dont le volume est vide, donc mort. La fenêtre doit se terminer à moins de `_TREND_MAX_LAG_MONTHS` du mois le plus récent **du fichier** (et non de la date du jour : c'est l'export qui dit jusqu'où il sait). Au-delà, `None` — les volumes mensuels restent dans `raw`. """ known = [ (rank, value) for rank, header in monthly if (value := to_int(cells.get(header))) is not None ] window = known[-TREND_LENGTH:] if len(window) < TREND_LENGTH: return None if monthly[-1][0] - window[-1][0] > _TREND_MAX_LAG_MONTHS: return None values = [value for _, value in window] peak = max(values) if peak <= 0: # Douze mois à zéro : la part du maximum n'est pas définie, et une sparkline plate à # zéro n'apprendrait rien. Les volumes bruts restent dans `raw`. return None return [round(value * TREND_SCALE / peak) for value in values] class MangoolsKwfinder(Adapter): """L'export CSV de l'interface KWFinder (§5). Ni URL ni position : jamais un classement.""" name = "mangools" signatures = (_SIGNATURE,) # Hors signature, délibérément (voir `Adapter.country_column`) : `Location` n'entre pas # dans le calcul de confiance, et un export qui l'aurait masquée à l'export reste reconnu # — il n'a simplement aucun pays à recenser. country_column = "Location" def detect(self, headers: Sequence[str]) -> Detection | None: """La détection de base, plus la colonne mensuelle qui **termine** la fenêtre. `trend` ne se résout pas comme les autres champs : il ne vient pas d'une colonne mais de cent trente-trois. La plus récente est rendue dans `columns` pour que l'aperçu (T3) ne compte pas la série mensuelle parmi les colonnes ignorées « conservées dans les données brutes » — elle est bel et bien lue. """ detection = super().detect(headers) if detection is None: return None monthly = _monthly_columns(headers) if not monthly: return detection columns = {**detection.columns, "trend": monthly[-1][1]} return replace(detection, columns=columns, confidence=len(columns) / len(_EN)) def parse_csv( self, path: Path, *, options: Mapping[str, Any] | None = None ) -> Iterator[ParsedRow]: with CsvScan(path) as scan: detection = self.detect(scan.headers) if detection is None: yield RowError(1, f"en-têtes non reconnus par l'adaptateur {self.name}") return headers = scan.headers monthly = _monthly_columns(headers) columns = detection.columns # Même repère que les autres adaptateurs : ligne 1 = en-tête, donc la première # ligne de données est la 2 — en **rang de données** (`to_physical_lines`). line = 1 for batch in scan.batches(): for values in batch: line += 1 raw = {header: value for header, value in zip(headers, values, strict=False)} yield _row(line, raw, columns, monthly) for rejected in scan.rejects(): yield RowError(rejected.line, rejected.reason, physical=True) def _row( line: int, cells: Mapping[str, str | None], columns: Mapping[str, str], monthly: Sequence[tuple[int, str]], ) -> ParsedRow: def cell(field: str) -> str | None: header = columns.get(field) return None if header is None else cells.get(header) keyword = normalize_keyword(cell("keyword") or "") if not keyword: return RowError(line, "mot-clé vide") return CanonicalKeywordRow( line=line, keyword=keyword, volume=to_int(cell("volume")), kd=to_decimal(cell("kd")), cpc=to_decimal(cell("cpc")), intent=to_intent(cell("intent")), serp_features=to_serp_features(cell("serp_features")), trend=_trend(cells, monthly), # KWFinder n'observe aucun classement : les deux champs restent vides par nature, # pas par défaut de mapping. url=None, position=None, raw=dict(cells), )