"""Adaptateur `fetch()` de l'API Mangools KWFinder (BYOK, spec Imports §5 et T8). **Le nom de source est `mangools`, partagé avec l'adaptateur CSV** — décision actée 2026-09-01 (`sources/__init__.py`) : le CSV KWFinder et l'API KWFinder rendent les mêmes métriques sur les mêmes mots-clés, seul le tuyau change. Les distinguer découperait la consolidation en deux pour une différence invisible à l'écran. ## Le format, relevé sur les réponses réelles (2026-09-02) La page de référence (apidocs.mangools.com/kwfinder) documente l'authentification, les chemins et les quotas — **jamais le format JSON**. Tout ce qui suit vient donc de captures prises avec une vraie clé (`capture_mangools`). Ces captures brutes ne sont pas commitées (règle `tests/fixtures/real/mangools_api/` d'`api/.gitignore`) : elles portent l'état du compte et tous les mots-clés du domaine interrogé. Ce qui est versionné, et ce sur quoi les tests de ce module s'appuient, ce sont des **extraits réduits et anonymisés** dans `api/tests/fixtures/mangools_api/`. **`GET /kwfinder/limits`** décrit le plan, il ne compte pas un quota unique. À la racine : `competitor-keywords-rows` (le plafond de lignes d'un lookup de domaine, 2000 sur le plan relevé), `related-keywords-rows`, `reqs-per-short-period` (3 — le débit). `resources` est un dict `{nom: {limit, remaining, reset}}` ; `related-keywords` est le compteur qui bouge. **Mesuré, et c'est ce qui gouverne tout le reste : une unité de quota est consommée _par requête_, pas par mot-clé.** Sur la capture, cinq appels (limits, locations, competitor-keywords, keyword-imports, limits) n'ont coûté que **2 unités** — celles des deux appels qui rendent des mots-clés. `limits` et `locations` sont donc gratuits, ce qui permet de lire le plan avant chaque import sans rien payer. Le coût se modélise en **nombre d'appels** : un gros lot coûte le même prix qu'un petit, et la pagination est la seule chose qui coûte. **`GET /mangools/locations?query=`** — `query` est obligatoire et non vide, sinon 200 avec `[]` (le piège : ça se lit comme « aucune localisation »). Rend une liste plate où pays et subdivisions se mélangent ; seul `target_type == "Country"` est un pays, et `country_code` (ISO-2, majuscules) est le pont vers `app.keywords.countries`. `code`/`tld` ne conviennent pas : le Royaume-Uni y est `gb`/`uk` là où son ISO est `GB`. **`GET /kwfinder/competitor-keywords?url=…&location_id=…&limit=…&page=…`** — enveloppe `{url_type, location, keywords, totalCount, page, resultsSize, __hasNextPage, isAllowedPagination, …}`. **La localisation est globale à la réponse** ; le `lid` de chaque ligne ne fait que la répéter. **`POST /kwfinder/keyword-imports`** `{"keywords": [...], "location_id": …}` — enveloppe `{location, language, keywords, countKeywordsBeforeLimit}`, sans pagination. **`GET /kwfinder/lists`** — les listes déjà enregistrées dans le compte KWFinder, les plus récentes d'abord. Chaque entrée porte `_id`, `name`, `keyword_ids`, `user_id`, `creator`, `created_at`, `updated_at`, `__v`. **Le schéma publié ne mentionne pas `_id`** alors que la réponse le porte, et c'est lui qui adresse la liste : la réponse fait foi, pas le YAML. Les horodatages sont en **millisecondes** (mesuré : `1742442235000` = 2025-03-20), là où le schéma annonce un « UNIX timestamp » — encore la réponse qui fait foi. **`GET /kwfinder/lists/{list_id}?fields=…`** — la liste avec les métriques complètes de chacun de ses mots-clés. `fields` est **obligatoire** (voir `LIST_FIELDS`). Enveloppe `{list, keywords}` ; chaque mot-clé a la même forme que les autres endpoints **plus un objet `location` `{_id, code, label}`** qui dit le marché de *cette ligne-là*. **C'est tout le cloisonnement pays du chemin « liste », et il ne ressemble pas à celui des deux autres modes.** `competitor-keywords` et `keyword-imports` prennent un `location_id` en entrée : les lignes hors marché n'existent pas. Une liste, elle, se lit telle qu'elle a été enregistrée — le compte relevé en mélange manifestement (listes « USA » et « France ») —, donc son marché n'est pas un paramètre mais une **propriété de chaque ligne**. Le tri se fait donc en aval, exactement comme pour un CSV KWFinder qui mêle onze pays : la résolution du libellé (`location.label`) appartient à `app.keywords.countries`, et ce qui ne correspond pas à la cible est compté `rows_off_target`, jamais importé en silence. **Mesuré le 2026-09-02, et c'est ce qui gouverne le récapitulatif : ces deux endpoints ne consomment RIEN.** Cinq appels enchaînés (`limits`, `lists`, `limits`, `lists/{id}` sur une liste de 2 mots-clés, `limits`) ont laissé les vingt compteurs de `resources` **strictement identiques** — `related-keywords` est resté à 1188 du premier au dernier. Importer des listes déjà enregistrées ne coûte donc aucune unité de quota, quel que soit le nombre de listes ou de mots-clés. Seul le **débit** reste à respecter (`reqs-per-short-period`), d'où le pacing inchangé de `_pace`. ## Ce que chaque champ devient | champ | sort | |----------|-------------------------------------------------------------------------------| | `kw` | `keyword`, passé par `normalize_keyword` comme toutes les sources | | `sv` | `volume` | | `seo` | `kd` — **sparse sur competitor-keywords** (15 lignes sur 50 dans la capture) | | `cpc` | `cpc`, arrondi à 2 décimales (l'API sérialise du float32 : `2.940000057220459`)| | `msv` | `trend` — les 12 derniers mois connus, ramenés en 0–100 (`TREND_SCALE`) | | le reste | `raw`, jamais perdu | Les champs **volontairement non portés**, et pourquoi : - **`position`** : c'est un vrai classement du domaine interrogé, mais le réservoir ne sait ranger une position **qu'attachée à une URL** (`keyword_page_map.url_normalized` est NOT NULL) et cet endpoint n'en donne aucune. La porter dans `CanonicalKeywordRow.position` remplirait un champ que le pipeline n'ouvre que pour les lignes qui ont une URL : personne ne la lirait. Elle reste dans `raw`, comme tout le reste, et redeviendra utile le jour où l'adaptateur saura demander les URL. - **`msv` au-delà de douze mois** : la série va jusqu'à 131 points ; `keyword_metrics.trend` est un `SMALLINT[]` de douze (§3.2). L'historique complet reste dans `raw`. - **`et`** (trafic estimé), **`ppc`** (difficulté PPC), **`svs`**, **`svn`**, **`c`** : aucun équivalent canonique, et surtout **aucune sémantique confirmée**. `svn` n'est pas un doublon de `sv` (mesuré : `sv=3100` avec `svn=3900` sur la même ligne) et rien ne dit lequel est le volume mensuel comparable aux autres sources — inventer la réponse, c'est exactement le mode d'échec que la règle « fixture réelle d'abord » cherche à éviter. Ils vivent dans `raw`. - **`intent`** et **`serp_features`** : l'API ne les rend pas. `None`, pas une liste vide — « la source ne le dit pas » n'est pas « la source dit qu'il n'y en a pas ». Une valeur absente reste `None` : jamais 0, jamais une moyenne, jamais une reprise d'une autre ligne. """ import asyncio import hashlib import logging import time from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass from datetime import UTC, datetime from decimal import Decimal from pathlib import Path from typing import Any from urllib.parse import quote import httpx import pycountry from app.core.config import get_settings from app.keywords.normalize import normalize_keyword from app.keywords.sources.base import ( TREND_LENGTH, TREND_SCALE, CanonicalKeywordRow, ParsedRow, RowError, ) logger = logging.getLogger(__name__) SOURCE_NAME = "mangools" # En-tête d'authentification de l'API Mangools (relevé 2026-08-31). Sa valeur est le secret # du client : elle n'apparaît jamais dans un log, ni dans un message d'erreur. AUTH_HEADER = "x-access-token" LIMITS_PATH = "/kwfinder/limits" LOCATIONS_PATH = "/mangools/locations" COMPETITOR_KEYWORDS_PATH = "/kwfinder/competitor-keywords" RELATED_KEYWORDS_PATH = "/kwfinder/related-keywords" KEYWORD_IMPORTS_PATH = "/kwfinder/keyword-imports" LISTS_PATH = "/kwfinder/lists" # Les champs demandés au détail d'une liste. `fields` est **obligatoire** (422 sans lui), et # cette chaîne est **exactement celle qui a servi à la mesure du 2026-09-02** : la forme de # réponse décrite plus bas est celle qu'elle a rendue, pas celle qu'un autre jeu de champs # rendrait. La changer sans re-mesurer, c'est décrire une réponse qu'on n'a jamais vue. LIST_FIELDS = "kw,sv,cpc,ppc,seo,seo_ts,msv,svn,svs,ts,h,lid" # Où `capture_mangools` dépose les réponses brutes. **Jamais commitées** : la règle # `tests/fixtures/real/mangools_api/` d'`api/.gitignore` les exclut, parce qu'elles portent # l'état du compte qui a lancé la capture (plan, quotas restants) et la liste complète des # mots-clés du domaine interrogé. Elles ne peuvent donc pas servir de condition d'exécution — # voir `parsing_ready`. Ce que les tests lisent, ce sont les extraits réduits et versionnés de # `tests/fixtures/mangools_api/`. CAPTURE_DIR = Path(__file__).resolve().parents[3] / "tests" / "fixtures" / "real" / "mangools_api" REQUIRED_CAPTURES = ( "competitor_keywords.json", "keyword_imports.json", "locations.json", ) # **Le verrou, désormais ouvert.** Passé à `True` le 2026-09-02, dans le commit qui écrit le # parse sur les captures réelles — les deux gestes vont ensemble, et c'est le second qui # distingue « les fichiers sont là » de « quelqu'un les a lus ». PARSER_WRITTEN = True DEFAULT_TIMEOUT = 30.0 # Plafond demandé quand l'appelant n'en impose pas. Ramené au plafond du plan # (`competitor-keywords-rows`) au moment du fetch : c'est le plan qui a le dernier mot. DEFAULT_LIMIT = 1_000 # --- débit ---------------------------------------------------------------------------------- # # `reqs-per-short-period` vaut 3 sur le plan relevé, et l'API ne dit **pas** ce qu'est une # « courte période ». On prend dix secondes, généreusement : se tromper vers le haut coûte de # l'attente dans un job de fond, se tromper vers le bas coûte des 429 sur la clé du client. # # La règle appliquée n'est pas « trois d'un coup puis on attend » mais **un intervalle minimal # entre deux requêtes** (`période / débit`, soit ~3,3 s). C'est plus strict qu'une fenêtre # glissante, donc toujours conforme, et surtout ça se raisonne en une phrase : deux appels ne # sont jamais plus proches que ça. Aucune rafale n'est représentable. SHORT_PERIOD_SECONDS = 10.0 REQUESTS_PER_SHORT_PERIOD = 3 MIN_REQUEST_INTERVAL = SHORT_PERIOD_SECONDS / REQUESTS_PER_SHORT_PERIOD # **L'espacement est partagé par processus, pas par instance** (correctif 2026-09-02). Il vivait # sur l'instance, ce qui n'aurait été juste que si une instance vivait aussi longtemps que la # clé : or on en construit une par appel — une par déclenchement d'import, une par job. Trois # cibles lancées à la suite faisaient donc trois requêtes rapprochées sur la même clé, contre # trois autorisées par courte période, et le 429 tombait dans un job à `max_attempts=1`. # # L'horodatage de la **prochaine** requête autorisée est donc gardé ici, par empreinte de clé — # jamais la clé elle-même, ce module n'en garde aucune trace en mémoire partagée. Chaque # appelant *réserve* son créneau (lecture puis écriture, sans `await` entre les deux : en # asyncio c'est indivisible) puis dort jusqu'à lui. Deux coroutines parties ensemble se # retrouvent donc à la queue leu leu, sans verrou — un `asyncio.Lock` au niveau du module se # lierait à la boucle d'événements qui l'a acquis en premier, ce qui est un piège dans un # processus qui en change (tests, worker relancé). _NEXT_REQUEST_AT: dict[str, float] = {} def _key_fingerprint(api_key: str) -> str: """Empreinte non réversible d'une clé : de quoi la reconnaître, jamais de quoi la relire.""" return hashlib.sha256(api_key.encode("utf-8")).hexdigest()[:16] # Repli quand `/kwfinder/limits` est illisible : le plafond du plan relevé. Jamais utilisé # quand l'appel passe, ce qui est le cas courant (il est gratuit). FALLBACK_COMPETITOR_ROWS = 2_000 # Garde-fou de boucle : même si l'API répondait `__hasNextPage` indéfiniment, on s'arrête. À # 1000 lignes par page et 2000 de plafond, deux pages suffisent — cinq laissent de la marge à # un plan plus large sans jamais laisser la pagination s'emballer sur le quota du client. MAX_PAGES = 5 # Taille d'un lot d'enrichissement. Une unité de quota est consommée **par requête** : un gros # lot est donc strictement moins cher qu'une série de petits. Le plafond réel du corps n'a pas # été mesuré (la capture n'a envoyé que 3 mots-clés, consigne d'économie de quota) ; 100 reste # très en dessous des deux plafonds de plan visibles (`related-keywords-rows` 700, # `competitor-keywords-rows` 2000). Une troncature côté serveur resterait de toute façon # **visible** : `_fetch_keywords` compare ce qui revient à ce qui a été demandé et le loge. KEYWORD_IMPORTS_BATCH = 100 # Une localisation Mangools n'est un pays que sous ce `target_type` ; tout le reste (`State`, # `City`, `County`, `DMA Region`, `Province`) est une subdivision, et cibler une subdivision # ferait entrer dans une cible « pays » des observations d'un marché plus étroit. COUNTRY_TARGET_TYPE = "Country" # Les pays dont ni `name` ni `common_name` de pycountry ne donnent un terme cherchable dans un # index anglais (vérifié sur 26.2.16 : les trois n'ont pas de `common_name`). Même principe que # `countries.ALIASES`, dans l'autre sens — et volontairement **large** : le filtre par # `country_code` fait le tri derrière, donc un terme qui ramène trop vaut mieux qu'un terme qui # ne ramène rien. La table reste courte : on n'y ajoute qu'un pays dont on a constaté l'échec. _SEARCH_TERMS: dict[str, str] = { "RU": "Russia", # ISO : « Russian Federation » "TR": "Turkey", # ISO : « Türkiye » depuis 2022 "CD": "Congo", # ISO : « Congo, The Democratic Republic of the » — ramène aussi CG, sans risque } # Écart maximal, en mois, entre la fin de la fenêtre d'une ligne et le mois le plus récent de # **la réponse**. Même règle, même valeur et même raison que l'adaptateur CSV # (`sources/mangools.py`) : une sparkline de 2019 est indiscernable à l'écran d'une tendance # actuelle, donc une fenêtre trop vieille est un mensonge, pas un ornement. TREND_MAX_LAG_MONTHS = 6 class FormatNotCapturedError(RuntimeError): """Le parse n'est pas disponible : l'import refuse plutôt que de deviner. Levée **avant tout appel réseau** : rien ne justifie de consommer le quota du client pour obtenir un JSON qu'on ne saurait pas lire. Conservée après l'écriture du parse — c'est elle qui porterait un retour en arrière (`PARSER_WRITTEN = False`) si un changement de format devait être re-capturé avant d'être lu. """ class MangoolsApiError(RuntimeError): """L'API a refusé ou n'a pas répondu. Le message ne porte jamais la clé.""" @dataclass(frozen=True, slots=True) class FetchSpec: """Ce qu'un import API demande au fournisseur, pour une cible donnée. **Trois modes, un seul objet**, parce que le pipeline n'a qu'un `fetch()` et que les trois rendent exactement les mêmes lignes canoniques : - **liste enregistrée** (`list_id`) — `GET /kwfinder/lists/{id}` : les mots-clés d'une liste du compte KWFinder, avec leurs métriques. **Le seul mode exposé par l'interface** depuis la reprise du 2026-09-02 ; - **découverte d'un domaine** (`domain`) — `competitor-keywords` : les mots-clés sur lesquels le site est déjà vu. Rend volume, CPC, tendance… et un `seo` (le KD) **souvent absent**, parce que cet endpoint sert le cache de Mangools tel quel ; - **enrichissement d'une liste nommée** (`keywords`) — `keyword-imports` : les mêmes métriques sur des mots-clés qu'on nomme, KD compris et renseigné. Les deux derniers sont **dormants** : ils ne sont plus atteignables depuis le wizard (le parcours ne propose que les listes enregistrées) mais restent justes, testés et prêts à resservir — voir `docs/pre-prod-checklist.md`. Les trois sont exclusifs, et l'un des trois est obligatoire : un spec vide interrogerait le fournisseur pour rien, un spec double laisserait le choix à l'ordre des `if`. Le `ValueError` tombe à la construction, donc côté appelant, pas dans le worker. `locale` est le marché de la cible (`fr-FR`). Pour `domain` et `keywords`, c'est de lui qu'on déduit le `lid` Mangools quand `location_id` n'est pas forcé, **et le cloisonnement pays s'arrête là** : la requête ne demande qu'un marché, les lignes hors marché n'existent pas. Pour `list_id`, il ne sert à rien ici — une liste se lit telle qu'elle a été enregistrée, mélange de marchés compris, et c'est le pipeline qui trie ligne à ligne comme il le fait d'un CSV (voir la docstring du module). """ locale: str domain: str = "" keywords: tuple[str, ...] = () list_id: str = "" location_id: int | None = None limit: int = DEFAULT_LIMIT def __post_init__(self) -> None: if sum(1 for mode in (self.domain, self.keywords, self.list_id) if mode) != 1: raise ValueError( "FetchSpec attend exactement un mode : `list_id` (liste enregistrée), " "`domain` (découverte) ou `keywords` (enrichissement)." ) @property def is_discovery(self) -> bool: return bool(self.domain) def missing_captures() -> list[str]: """Les captures brutes absentes du poste. Diagnostic de développement uniquement. Sert à `capture_mangools` et au message d'erreur, **jamais** à décider si un import peut tourner : `api/.gitignore` exclut ce répertoire, ces fichiers n'existent donc que sur le poste qui a lancé la capture. Les faire gouverner `parsing_ready()` (ce qu'ils faisaient avant que le parse n'existe) reviendrait à interdire tout import API en production. """ return [name for name in REQUIRED_CAPTURES if not (CAPTURE_DIR / name).is_file()] def parsing_ready() -> bool: """`True` quand le parse est écrit. Rien d'autre n'entre en jeu — voir `missing_captures`.""" return PARSER_WRITTEN def not_captured_reason() -> str: """La phrase que verrait l'utilisateur si le verrou était refermé.""" return ( "Le parse de l'API Mangools est désactivé (PARSER_WRITTEN est à False dans " "app/keywords/sources/mangools_api.py). Aucun import API ne sera tenté tant que ce " "n'est pas rétabli : mieux vaut ne rien importer qu'importer des métriques devinées. " "Pour relever à nouveau le format : " "`uv run python -m app.keywords.sources.capture_mangools `." ) def redact(text: str, secret: str) -> str: """Filet de sécurité : la clé ne doit jamais fuir dans un message, un log ou une capture. Public parce que `capture_mangools` s'en sert pour caviarder les fichiers qu'il écrit : ces JSON sortent d'une clé réelle, et une clé qui s'y glisserait y resterait pour de bon. """ return text.replace(secret, "***") if secret else text # --- coercions ------------------------------------------------------------------------------ # # L'API rend du JSON typé, pas des cellules de tableur : il n'y a donc rien à « lire », juste à # refuser ce qui n'a pas le type attendu plutôt que de le forcer. Un champ d'un type inattendu # rend `None` — la valeur est absente, pas nulle, et surtout pas 0. def _to_int(value: Any) -> int | None: # `bool` est un `int` en Python : sans ce test, `True` deviendrait un volume de 1. if isinstance(value, bool) or not isinstance(value, int | float): return None return int(value) def _to_decimal(value: Any, *, places: int = 2) -> Decimal | None: """Un nombre JSON en `Decimal`, les artefacts de flottant en moins. L'API sérialise des **float32** élargis : `cpc` 2,94 arrive en `2.940000057220459` et `ppc` 2 en `1.9999999552965164`. Écrire ce chiffre tel quel dans une colonne `Numeric` serait *plus* faux que de l'arrondir — ce n'est pas une précision supplémentaire, c'est du bruit de représentation. Les entiers (`seo`, le KD) passent exacts, sans décimale inventée. """ if value is None or isinstance(value, bool): return None if isinstance(value, int): return Decimal(value) if isinstance(value, float): if value != value or value in (float("inf"), float("-inf")): return None return Decimal(f"{value:.{places}f}") return None def _month_rank(entry: Any) -> tuple[int, int] | None: """`[année, mois, volume]` → `(rang en mois, volume)`, ou `None` si la case est vide. **Le mois est indexé à partir de 0** dans la réponse (`[2024, 0, 3600]` = janvier 2024). Le rang `année × 12 + mois` s'en moque — c'est justement pourquoi on compare des rangs plutôt que des mois — mais quiconque relit ce module doit le savoir avant de convertir cette série en dates. """ if not isinstance(entry, list | tuple) or len(entry) != 3: return None year, month, volume = entry if isinstance(year, bool) or isinstance(month, bool) or not isinstance(year, int): return None if not isinstance(month, int): return None known = _to_int(volume) if known is None: return None return year * 12 + month, known def _newest_month(payloads: Sequence[Mapping[str, Any]]) -> int | None: """Le mois le plus récent de **toute** la réponse, en rang. C'est la réponse qui dit jusqu'où elle sait, pas la date du jour : un import rejoué demain doit juger les mêmes lignes de la même façon. """ ranks = [ dated[0] for payload in payloads for entry in (payload.get("msv") or []) if (dated := _month_rank(entry)) is not None ] return max(ranks) if ranks else None def _trend(raw: Any, newest: int | None) -> list[int] | None: """Les douze derniers mois **connus** de la ligne, en 0–100 (part du maximum). Même forme que la tendance des autres sources (`TREND_SCALE`, `TREND_LENGTH`) : c'est ce qui rend les sparklines comparables d'une provenance à l'autre. La série de l'API n'est ni bornée à douze points (12 à 131 selon les lignes relevées), ni garantie ordonnée, et **contient des mois sans volume**. On trie donc par rang, on saute les mois vides plutôt que de les combler par des zéros — un zéro inventé se lirait comme un effondrement du volume —, et on garde la fenêtre de la ligne, pas celle de la réponse. `None` dès que quelque chose manque : moins de douze mois connus, fenêtre trop vieille (voir `TREND_MAX_LAG_MONTHS`), ou douze mois à zéro (la part du maximum n'y est pas définie). Une tendance est un ornement d'affichage : son absence n'est jamais une raison de rejeter la ligne, et la série brute reste dans `raw`. """ if not isinstance(raw, list): return None known = sorted(dated for entry in raw if (dated := _month_rank(entry)) is not None) window = known[-TREND_LENGTH:] if len(window) < TREND_LENGTH: return None if newest is not None and newest - window[-1][0] > TREND_MAX_LAG_MONTHS: return None peak = max(volume for _, volume in window) if peak <= 0: return None return [round(volume * TREND_SCALE / peak) for _, volume in window] def parse_row(line: int, payload: Any, newest: int | None = None) -> ParsedRow: """Une ligne de l'API (les deux endpoints ont la même forme) en ligne canonique. `line` n'est pas un numéro de ligne de fichier — il n'y a pas de fichier — mais le rang du mot-clé dans la réponse. Il ne sert qu'au rapport et à `errors.csv`, où il répond à « la combientième ligne rendue par le fournisseur a posé problème ». """ if not isinstance(payload, Mapping): return RowError(line, "ligne Mangools illisible : objet JSON attendu") keyword = normalize_keyword(str(payload.get("kw") or "")) if not keyword: return RowError(line, "mot-clé vide dans la réponse Mangools") return CanonicalKeywordRow( line=line, keyword=keyword, volume=_to_int(payload.get("sv")), # Renseigné sur `keyword-imports`, souvent `null` sur `competitor-keywords` : le # `None` dit « cette source ne connaît pas le KD de ce mot-clé aujourd'hui », ce qui # est exact, et la consolidation ira le chercher chez une autre source si elle l'a. kd=_to_decimal(payload.get("seo")), cpc=_to_decimal(payload.get("cpc")), # L'API ne rend ni intention ni features SERP. `None` — pas une liste vide. intent=None, serp_features=None, trend=_trend(payload.get("msv"), newest), # Ni URL ni position portées : voir la docstring du module. `position` existe sur # `competitor-keywords` mais reste dans `raw`, faute d'URL à qui l'attacher. url=None, position=None, raw=dict(payload), ) def parse_rows(payloads: Sequence[Any]) -> list[ParsedRow]: """Les lignes d'une réponse, dans l'ordre, la fenêtre de tendance calibrée sur l'ensemble.""" objects = [payload for payload in payloads if isinstance(payload, Mapping)] newest = _newest_month(objects) return [parse_row(index, payload, newest) for index, payload in enumerate(payloads, start=1)] def country_locations(payload: Any) -> dict[str, int]: """`/mangools/locations` → `{code ISO-2: lid}`, **les pays seulement**. La réponse mélange pays et subdivisions dans une liste plate ; garder une subdivision ferait entrer dans une cible « France » les mots-clés d'Île-de-France, sans un mot. Le pont se fait par `country_code` (déjà en ISO-2 majuscules, comme `app.keywords.countries`) et non par `code`/`tld`, qui divergent de l'ISO — le Royaume-Uni y est `gb` et `uk`. """ if not isinstance(payload, list): return {} found: dict[str, int] = {} for entry in payload: if not isinstance(entry, Mapping) or entry.get("target_type") != COUNTRY_TARGET_TYPE: continue code = entry.get("country_code") lid = entry.get("_id") if isinstance(code, str) and len(code) == 2 and isinstance(lid, int): found.setdefault(code.upper(), lid) return found # --- listes enregistrées du compte ---------------------------------------------------------- @dataclass(frozen=True, slots=True) class KeywordListSummary: """Une liste du compte KWFinder, telle que l'écran de choix l'affiche. `keyword_count` vient de la longueur de `keyword_ids` : c'est le compte du **fournisseur**, pas celui de ce qui sera importé — le cloisonnement par pays en écartera une partie, et c'est le rapport d'import qui le dira. Annoncer ici un nombre déjà filtré demanderait d'ouvrir chaque liste pour l'afficher. """ id: str name: str keyword_count: int created_at: datetime | None updated_at: datetime | None # Frontière entre un horodatage en secondes et un en millisecondes. Les valeurs relevées sont # en **millisecondes** (`1742442235000` = 2025-03-20) là où le schéma publié annonce des # secondes ; plutôt que de choisir contre la mesure ou contre la doc, on lit les deux — 1e11 # secondes tombe en l'an 5138 et 1e11 millisecondes en 1973, donc aucune date plausible n'est # ambiguë. _MILLISECOND_THRESHOLD = 1e11 def _moment(value: Any) -> datetime | None: """Un horodatage Mangools en `datetime` UTC, ou `None` si la valeur n'en est pas un.""" if isinstance(value, bool) or not isinstance(value, int | float): return None seconds = value / 1000 if abs(value) >= _MILLISECOND_THRESHOLD else float(value) try: return datetime.fromtimestamp(seconds, tz=UTC) except (OverflowError, OSError, ValueError): return None def parse_lists(payload: Any) -> list[KeywordListSummary]: """`GET /kwfinder/lists` → les listes du compte, dans l'ordre rendu (les plus récentes d'abord). Une entrée sans `_id` ou sans `name` exploitable est **sautée** plutôt que devinée : sans identifiant elle n'est pas importable, et un nom fabriqué (« Liste 3 ») ne correspondrait à rien de ce que l'utilisateur voit dans KWFinder. """ if not isinstance(payload, list): return [] found: list[KeywordListSummary] = [] for entry in payload: if not isinstance(entry, Mapping): continue identifier = entry.get("_id") name = entry.get("name") if not isinstance(identifier, str) or not identifier.strip(): continue if not isinstance(name, str) or not name.strip(): continue ids = entry.get("keyword_ids") found.append( KeywordListSummary( id=identifier, name=name, keyword_count=len(ids) if isinstance(ids, list) else 0, created_at=_moment(entry.get("created_at")), updated_at=_moment(entry.get("updated_at")), ) ) return found # Clé de l'objet marché d'une ligne de liste, et clé de son libellé. Le pipeline s'en sert pour # cloisonner (`app.keywords.pipeline`), ce qui est la seule chose qui les rende publics. LOCATION_FIELD = "location" LOCATION_LABEL_FIELD = "label" def row_country_label(payload: Any) -> str: """Le libellé de marché d'une ligne de liste (« United States »), ou `""`. **Le libellé, pas le `code`.** C'est exactement l'entrée que `app.keywords.countries` sait résoudre — la même que la colonne `Location` du CSV KWFinder — donc les deux chemins partagent la table d'alias, les refus d'homonymes (« Georgia ») et le rapport. Le `code` de l'objet `location` diverge par ailleurs de l'ISO sur certains pays (`uk` pour GB dans l'index Mangools), ce qui en fait un pont plus fragile qu'il n'en a l'air. """ if not isinstance(payload, Mapping): return "" location = payload.get(LOCATION_FIELD) if not isinstance(location, Mapping): return "" label = location.get(LOCATION_LABEL_FIELD) return label.strip() if isinstance(label, str) else "" def plan_limits(payload: Any) -> tuple[int, float]: """`/kwfinder/limits` → `(plafond de lignes d'un lookup, intervalle minimal entre requêtes)`. Lu **à chaque fetch** parce que l'appel est gratuit (mesuré : il ne bouge aucun compteur de `resources`) et parce que ces deux valeurs appartiennent au plan du client, pas au nôtre. Un plan plus large paginera plus loin sans qu'on touche à une constante ; un plan plus étroit sera respecté sans qu'on ait à le deviner. Repli sur les constantes du module si la réponse n'a pas la forme attendue — jamais une exception : l'import doit pouvoir tourner. """ rows = FALLBACK_COMPETITOR_ROWS interval = MIN_REQUEST_INTERVAL if isinstance(payload, Mapping): declared = payload.get("competitor-keywords-rows") if isinstance(declared, int) and not isinstance(declared, bool) and declared > 0: rows = declared burst = payload.get("reqs-per-short-period") if isinstance(burst, int) and not isinstance(burst, bool) and burst > 0: interval = SHORT_PERIOD_SECONDS / burst return rows, interval # La ressource de `resources` qui compte les appels rendant des mots-clés — donc la seule qui # borne un import API. Relevé sur la capture : `{"limit": 1200, "remaining": 1192, "reset": …}`, # et c'est bien elle qui bouge après un `competitor-keywords` comme après un `keyword-imports`. QUOTA_RESOURCE = "related-keywords" def resource_quota(payload: Any, resource: str = QUOTA_RESOURCE) -> tuple[int | None, int | None]: """`/kwfinder/limits` → `(plafond, restant)` d'une ressource. `(None, None)` si illisible. **Aucune valeur inventée** : une ressource absente ou d'un type inattendu rend `None`, que l'écran affiche comme « quota inconnu ». Annoncer « 0 restant » découragerait un import que le fournisseur accepterait, et « 1200 » en promettrait un qu'il refuserait. Rappel de la mesure qui gouverne la lecture de ce nombre : **une unité est consommée par requête, pas par mot-clé**. `remaining` se compare donc à un nombre d'appels, jamais à la taille du lot demandé. """ if not isinstance(payload, Mapping): return None, None resources = payload.get("resources") if not isinstance(resources, Mapping): return None, None entry = resources.get(resource) if not isinstance(entry, Mapping): return None, None return _to_int(entry.get("limit")), _to_int(entry.get("remaining")) def _country_of(locale: str) -> str: """Le pays ISO-2 d'une locale `langue-PAYS`. C'est ce que `locale_of` écrit toujours.""" _, _, tail = locale.partition("-") code = tail.strip().upper() if len(code) != 2 or not code.isalpha(): raise MangoolsApiError( f"Cette cible n'a pas de pays exploitable ({locale!r}) : impossible de choisir le " "marché à interroger chez Mangools. Corrigez le pays de la cible dans Réglages → " "Sites." ) return code def _search_term(code: str) -> str: """Le terme à chercher dans `/mangools/locations` pour ce pays. **Le nom ISO ne convient pas** : `pycountry.name` rend « Russian Federation », « Viet Nam », « Korea, Republic of », « Iran, Islamic Republic of », « Lao People's Democratic Republic »… Aucun de ces libellés n'a de chance de tomber sur un index anglais qui, lui, écrit « Russia » et « Vietnam. » `common_name` existe précisément pour ça et couvre l'essentiel des cas (KR, KP, VN, IR, TW, BO, VE, MD, TZ, SY, LA — vérifié sur pycountry 26.2.16) ; les trois autres sont dans `_SEARCH_TERMS`. Ce qu'on cherche n'a pas à être exact, seulement **assez large pour ramener le pays** : le filtre, lui, porte sur `country_code` (`country_locations`). « Congo » qui rend à la fois CG et CD ne pose donc aucun problème — on prend celui dont le code correspond, et c'est aussi pourquoi un terme trop précis est plus dangereux qu'un terme trop vague. """ manual = _SEARCH_TERMS.get(code) if manual is not None: return manual entry = pycountry.countries.get(alpha_2=code) if entry is None: return code return getattr(entry, "common_name", None) or entry.name def _row_id(row: Any) -> str | None: """Le `_id` d'une ligne (un hash), ou `None`. Sert à repérer un recouvrement de pages.""" if not isinstance(row, Mapping): return None identifier = row.get("_id") return identifier if isinstance(identifier, str) and identifier else None def _chunks(values: Sequence[str], size: int) -> Iterable[Sequence[str]]: for start in range(0, len(values), size): yield values[start : start + size] class MangoolsApi: """Client HTTP de l'API KWFinder. Une instance par clé, jamais partagée entre organisations. **L'espacement des requêtes n'est pas porté par l'instance** mais par le processus, indexé sur l'empreinte de la clé (`_NEXT_REQUEST_AT`) : on construit une instance par appel, donc une mémoire d'instance ne voit jamais la requête que l'appel précédent vient de faire sur la même clé. Deux organisations qui importent en même temps ont deux clés et deux débits, donc toujours rien à se partager. """ name = SOURCE_NAME def __init__( self, api_key: str, *, base_url: str | None = None, client: httpx.AsyncClient | None = None, timeout: float = DEFAULT_TIMEOUT, min_request_interval: float | None = None, ) -> None: self._api_key = api_key self._base_url = (base_url or get_settings().mangools_api_base_url).rstrip("/") self._client = client self._timeout = timeout # `0.0` est une valeur légitime, et c'est celle des tests : `or` la mangerait. self._interval = ( MIN_REQUEST_INTERVAL if min_request_interval is None else min_request_interval ) self._fingerprint = _key_fingerprint(api_key) async def _pace(self) -> None: """Réserve le prochain créneau autorisé **pour cette clé**, puis attend jusqu'à lui. La mémoire est celle du processus (`_NEXT_REQUEST_AT`), pas celle de l'instance : voir le commentaire de cette table. Horloge **monotone** : un ajustement NTP pendant un import ne doit pas transformer l'attente en rafale (ni en attente de plusieurs heures). """ if self._interval <= 0: return now = time.monotonic() reserved = _NEXT_REQUEST_AT.get(self._fingerprint) # Réservation : le créneau est écrit **avant** le `sleep`, sinon deux coroutines # parties ensemble liraient le même passé et repartiraient ensemble. slot = now if reserved is None else max(now, reserved) _NEXT_REQUEST_AT[self._fingerprint] = slot + self._interval if (delay := slot - now) > 0: await asyncio.sleep(delay) async def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response: await self._pace() headers = {AUTH_HEADER: self._api_key, "accept": "application/json"} if self._client is not None: return await self._client.request( method, f"{self._base_url}{path}", headers=headers, **kwargs ) async with httpx.AsyncClient(timeout=self._timeout) as client: return await client.request( method, f"{self._base_url}{path}", headers=headers, **kwargs ) def _raise_for_status(self, response: httpx.Response) -> None: """Traduit un code HTTP en message pour l'utilisateur. **Jamais le corps brut.** Le corps d'une erreur peut renvoyer la requête en écho, en-têtes compris : le recopier remettrait la clé dans `imports.error`, que l'écran affiche. Les codes suffisent à dire quoi faire. """ if response.status_code in (401, 403): raise MangoolsApiError("Clé Mangools refusée par le fournisseur (401/403).") if response.status_code == 429: # **Un 429 dit le débit, pas le quota** : les endpoints de listes ne consomment # aucune unité et en prennent quand même quand les requêtes se rapprochent trop. # Annoncer un quota épuisé enverrait l'utilisateur racheter des crédits pour rien. raise MangoolsApiError( "Débit Mangools dépassé (429) : trop de requêtes rapprochées, réessayez dans " "quelques secondes." ) if response.status_code >= 400: raise MangoolsApiError(f"API Mangools en erreur (HTTP {response.status_code}).") async def _json(self, method: str, path: str, **kwargs: Any) -> Any: try: response = await self._request(method, path, **kwargs) except httpx.HTTPError as exc: raise MangoolsApiError( f"API Mangools injoignable : {redact(str(exc), self._api_key)}" ) from exc self._raise_for_status(response) try: return response.json() except ValueError as exc: raise MangoolsApiError( f"Réponse Mangools illisible (corps non JSON) sur {path}." ) from exc async def check(self) -> dict[str, Any]: """Confronte la clé au fournisseur : `GET /kwfinder/limits` (§5). L'appel le moins cher qui existe — il ne consomme aucune unité de quota (mesuré) — et le code HTTP suffit à trancher. Le corps est rendu tel quel quand c'est un objet JSON : il porte les plafonds du plan, que `plan_limits` sait lire. """ try: response = await self._request("GET", LIMITS_PATH) except httpx.HTTPError as exc: raise MangoolsApiError( f"API Mangools injoignable : {redact(str(exc), self._api_key)}" ) from exc self._raise_for_status(response) try: body = response.json() except ValueError: # Un 200 non-JSON reste un succès d'authentification : on ne prétend pas lire. return {} return body if isinstance(body, dict) else {} async def resolve_location(self, locale: str) -> int: """Le `lid` Mangools du pays de cette locale. Gratuit (aucune unité consommée). Échoue plutôt que de replier sur un pays voisin ou sur les États-Unis : importer le marché américain dans une cible française serait indétectable à l'écran — les mots-clés sont en anglais, les volumes plausibles, et rien ne dirait d'où ils viennent. """ code = _country_of(locale) payload = await self._json("GET", LOCATIONS_PATH, params={"query": _search_term(code)}) lid = country_locations(payload).get(code) if lid is None: # **Pas de conseil que l'utilisateur ne peut pas suivre.** `config.location_id` # est bien une échappatoire, mais l'écran Intégrations n'envoie jamais de # `config` : le lui recommander l'enverrait chercher un réglage qui n'existe pas # dans l'interface. On dit donc le fait et sa conséquence, rien de plus. raise MangoolsApiError( f"Mangools ne propose pas de marché « pays » pour {code} : cette cible ne peut " "pas être importée depuis ce fournisseur." ) return lid async def lists(self) -> list[KeywordListSummary]: """Les listes déjà enregistrées dans le compte. **Gratuit** (mesuré 2026-09-02). Aucune pagination : la réponse est la liste entière, les plus récentes d'abord. """ return parse_lists(await self._json("GET", LISTS_PATH)) async def list_keywords(self, list_id: str) -> list[Any]: """Les mots-clés bruts d'une liste, métriques et marché compris. **Gratuit** (mesuré). Rend les charges utiles telles quelles — c'est `parse_rows` qui les traduit et le pipeline qui trie par marché. `fields` est obligatoire : sans lui l'API rend une erreur, et avec un autre jeu que `LIST_FIELDS` la forme rendue n'est plus celle qui a été relevée. """ identifier = list_id.strip() if not identifier: raise MangoolsApiError("Identifiant de liste KWFinder vide : rien à importer.") body = await self._json( "GET", f"{LISTS_PATH}/{quote(identifier, safe='')}", params={"fields": LIST_FIELDS}, ) if not isinstance(body, Mapping): raise MangoolsApiError("Réponse `lists/{id}` inattendue : objet JSON attendu.") batch = body.get("keywords") if not isinstance(batch, list): raise MangoolsApiError( "Réponse `lists/{id}` sans liste `keywords` : la liste n'a pas pu être lue." ) return batch async def fetch(self, spec: FetchSpec) -> Sequence[CanonicalKeywordRow]: """Les mots-clés de la cible, en lignes canoniques. Ordre des appels, et pourquoi : `limits` d'abord (gratuit) pour connaître le plafond du plan et le débit **avant** de payer quoi que ce soit ; la localisation ensuite (gratuite aussi), sauf si l'intégration la force ; puis seulement les appels qui consomment. **Le mode « liste » court-circuite les deux.** Il ne paie rien (mesuré) et ne prend pas de `location_id` : lire le plan ou résoudre un marché ne servirait qu'à ajouter deux allers-retours — et à ralentir l'import du seul mode que l'interface expose, puisque `_pace` espace toutes les requêtes de l'instance. Les lignes en erreur ne sont pas levées mais **rendues** dans le flux, comme pour un CSV : c'est le pipeline qui les compte et les rapporte (§4.2). Une ligne illisible ne doit pas faire perdre l'import — ni le quota déjà dépensé pour l'obtenir. """ if not parsing_ready(): raise FormatNotCapturedError(not_captured_reason()) if spec.list_id: return parse_rows(await self.list_keywords(spec.list_id)) # type: ignore[return-value] rows_cap, interval = plan_limits(await self.check()) # Le débit annoncé par le plan prime sur la constante du module — sauf quand # l'appelant a explicitement demandé de ne pas espacer (tests, bouchons). if self._interval > 0: self._interval = interval location_id = spec.location_id if location_id is None: location_id = await self.resolve_location(spec.locale) payloads = ( await self._fetch_domain(spec, location_id, rows_cap) if spec.is_discovery else await self._fetch_keywords(spec, location_id) ) return parse_rows(payloads) # type: ignore[return-value] async def _fetch_domain(self, spec: FetchSpec, location_id: int, rows_cap: int) -> list[Any]: """`competitor-keywords`, paginé — **une unité de quota par page**. **La taille de page ne varie jamais d'un appel à l'autre**, et c'est le point délicat. Demander « ce qu'il me manque » (`budget - len(collected)`) paraît économe et fabrique un trou : si l'offset de la page *n* vaut `(n-1) × limit`, une page 1 à 1000 suivie d'une page 2 à 500 relit les lignes 500-999 — un doublon payé une unité — et saute les 1000-1499, définitivement. Le `page` échoé vaudrait pourtant bien 2 : rien ne le signalerait. On envoie donc toujours le même `limit` (le plafond du plan) et on tronque à la fin ; le surcoût est en octets, jamais en quota, et l'alignement est garanti quelle que soit la convention d'offset — qui, elle, n'a pas été capturée. Trois freins, et il en faut trois : le plafond voulu par l'appelant (`spec.limit`), le plafond du plan (`competitor-keywords-rows`) et `MAX_PAGES`, qui garantit la terminaison même si `__hasNextPage` restait vrai à tort. On s'arrête aussi dès qu'une page revient vide : une page sans ligne ne servira pas mieux la suivante. """ budget = min(spec.limit, rows_cap) # Constante pour tout le fetch. L'API rogne d'elle-même à sa vraie taille de page # (mesuré : `limit=50` demandé, `resultsSize: 1000` échoé), donc demander le plafond # du plan revient à demander « ta page entière », ce qui est exactement ce qu'il faut # pour que deux appels successifs s'enchaînent sans trou ni recouvrement. page_size = rows_cap collected: list[Any] = [] seen_ids: set[str] = set() for page in range(1, MAX_PAGES + 1): if len(collected) >= budget: break body = await self._json( "GET", COMPETITOR_KEYWORDS_PATH, params={ # `url` et non `domain` : l'API rend un 422 « Parameter url is mandatory ». # Elle accepte un domaine nu (`url_type: "RootDomain"`). "url": spec.domain, "location_id": location_id, "limit": page_size, "page": page, }, ) if not isinstance(body, Mapping): raise MangoolsApiError( "Réponse `competitor-keywords` inattendue : objet JSON attendu." ) batch = body.get("keywords") if not isinstance(batch, list) or not batch: break # **La pagination est le seul point déduit et non capturé** : la réponse porte # `page`, `__hasNextPage` et `isAllowedPagination`, donc l'API pagine — mais la # capture n'a pris qu'une page, et le *nom* du paramètre vient du champ rendu. # Deux gardes, parce qu'il y a deux façons de se tromper : # # 1. l'API ignore le paramètre et renvoie toujours la page 1 — le `page` échoé le # dit (comparé en `str` : un écho `"2"` ne doit pas contourner la garde) ; # 2. elle le comprend mais pas comme nous — l'offset ne tombe pas où l'on croit. # Aucun champ ne le dirait ; ce qui le dit, ce sont les **lignes elles-mêmes**, # déjà vues. `_id` est un hash présent sur chaque ligne : le recouvrement est # donc observable, et il vaut arrêt — mieux vaut un import court qu'un import # qui a payé pour relire, et qui a sauté ce qu'il croyait avoir lu. echoed = body.get("page") if page > 1 and echoed is not None and str(echoed) != str(page): logger.warning( "Mangools competitor-keywords : page %d demandée, page %s rendue — la " "pagination n'est pas pilotée par ce paramètre, arrêt à %d lignes", page, echoed, len(collected), ) break fresh = [row for row in batch if _row_id(row) not in seen_ids] if page > 1 and len(fresh) < len(batch): logger.warning( "Mangools competitor-keywords : la page %d recouvre la précédente " "(%d lignes déjà vues sur %d) — arrêt à %d lignes", page, len(batch) - len(fresh), len(batch), len(collected), ) break seen_ids.update(identifier for row in batch if (identifier := _row_id(row))) collected.extend(fresh) # `isAllowedPagination` est le mot du plan : faux, la page 2 serait payée pour # rien. `__hasNextPage` est le mot de la réponse. if not body.get("__hasNextPage") or not body.get("isAllowedPagination", True): break # La troncature vit **ici**, pas dans le `limit` envoyé : c'est ce qui permet à la # taille de page de rester constante sans jamais rendre plus que ce qu'on demandait. return collected[:budget] async def _fetch_keywords(self, spec: FetchSpec, location_id: int) -> list[Any]: """`keyword-imports` par lots — **une unité de quota par lot, pas par mot-clé**. C'est pourquoi les lots sont gros : découper 300 mots-clés en trois appels coûte trois unités, en un seul il en coûte une. Ce qui est surveillé, c'est la **troncature silencieuse** — l'API annonce un `countKeywordsBeforeLimit`, donc elle sait tronquer. Un lot qui revient plus court que demandé est logué (jamais avec la clé), pas masqué. """ collected: list[Any] = [] for chunk in _chunks(list(spec.keywords), KEYWORD_IMPORTS_BATCH): body = await self._json( "POST", KEYWORD_IMPORTS_PATH, json={"keywords": list(chunk), "location_id": location_id}, ) if not isinstance(body, Mapping): raise MangoolsApiError("Réponse `keyword-imports` inattendue : objet JSON attendu.") batch = body.get("keywords") if not isinstance(batch, list): # Aussi bruyant que la troncature ci-dessous : s'arrêter en silence ferait # passer un import amputé pour un import complet, ce qui est le seul mode # d'échec que ce module refuse par principe. logger.warning( "Mangools keyword-imports : réponse sans liste `keywords` — lot de %d " "mots-clés abandonné, import arrêté à %d lignes", len(chunk), len(collected), ) break if len(batch) < len(chunk): logger.warning( "Mangools keyword-imports : %d mots-clés demandés, %d rendus — plafond de " "plan probable, le lot sera importé incomplet", len(chunk), len(batch), ) collected.extend(batch) return collected