"""Lecture d'un export CSV : encodage, séparateur, lignes rejetées (spec Imports §4.1). DuckDB en **librairie Python** du worker (canon data §3 : DuckDB uniquement dans l'import CSV), jamais dans une requête HTTP. Trois responsabilités et pas une de plus : 1. **l'encodage** — BOM, puis validation UTF-8 stricte, puis `charset-normalizer` en dernier recours (voir `detect_encoding`). Semrush exporte en UTF-16 selon le chemin d'export, et les tableurs recrachent du cp1252 : ce dernier n'est pas lisible par `read_csv` (qui **valide** ses encodages et rejette le fichier entier), il est donc transcrit en UTF-8 dans un temporaire (`_transcode`) ; 2. **le séparateur et le BOM** — laissés au renifleur de DuckDB, qui distingue `,`, `;` (« CSV Semicolon ») et la tabulation, et mange le BOM de l'en-tête ; 3. **les lignes que le CSV lui-même rejette** — `store_rejects` les met de côté avec leur numéro de ligne physique et la raison, au lieu de faire tomber la lecture entière. Ce troisième point n'est pas théorique : l'export Keyword Magic réel se termine par trois lignes de pied de page commerciales (« You can export only first 50000 rows… ») qui n'ont pas le bon nombre de colonnes. Sans `store_rejects`, `read_csv` lève et l'import de 50 000 lignes valides échoue pour un pied de page. Avec, elles ressortent en lignes en erreur — visibles dans le rapport et dans `errors.csv`, jamais avalées, jamais importées comme des mots-clés (ce que ferait `null_padding`). Tout est lu en `VARCHAR` : la conversion appartient aux coercions de `base.py`, où une cellule illisible produit `None` ou une ligne en erreur, jamais une exception de cast. """ import codecs import logging import tempfile from collections.abc import Iterator from dataclasses import dataclass from pathlib import Path from types import TracebackType import duckdb from charset_normalizer import from_path logger = logging.getLogger(__name__) # `read_csv` ne connaît nativement que ces trois encodages, et **il les vérifie** : ce n'est # pas une simple étiquette de décodage. Tout le reste passe par une transcription (voir # `_transcode`), jamais par une étiquette approximative. _DUCKDB_ENCODINGS = {"utf_8": "utf-8", "utf_16": "utf-16", "utf_16_le": "utf-16"} # Codec de repli **côté Python**, jamais passé à DuckDB. cp1252 est la sortie par défaut # d'Excel et des tableurs Windows en Europe de l'Ouest — c'est le cas réel, pas latin-1. _FALLBACK_CODEC = "cp1252" # cp1252 laisse cinq octets indéfinis (0x81, 0x8D, 0x8F, 0x90, 0x9D) ; latin-1 est total et ne # lève sur aucun octet. Il ferme la chaîne, pour qu'aucun fichier ne puisse faire échouer la # transcription. _TOTAL_CODEC = "latin-1" _READ_CSV = ( "read_csv(?, all_varchar = true, header = true, encoding = ?, " "store_rejects = true, rejects_scan = 'rs', rejects_table = 're')" ) @dataclass(frozen=True, slots=True) class RejectedLine: """Une ligne physique écartée par le lecteur CSV, avec son numéro et sa raison.""" line: int reason: str def detect_encoding(path: Path) -> str: """L'encodage du fichier, parmi ceux que `read_csv` accepte — sinon `""`. Trois étapes, dans cet ordre, parce que la reniflée statistique seule ne suffit pas : sur un export UTF-8 dont les premiers kilo-octets sont purement ASCII, `charset-normalizer` répond volontiers « cp1252 » ou « cp1250 » — verdict inoffensif tant que le fichier est ASCII, mais qui transforme le premier « é » en mojibake, et donc en **mot-clé distinct** de sa forme correcte. L'unicité `(org_id, locale, keyword)` ne protège de rien contre ça : elle voit deux mots-clés différents. 1. **BOM** — verdict certain, aucune heuristique ne le bat ; 2. **validation UTF-8 stricte du fichier entier** — l'UTF-8 s'auto-valide : du latin-1 accentué ne passe quasiment jamais pour de l'UTF-8 valide, et un fichier ASCII est de toute façon lu à l'identique par les deux. Une lecture séquentielle de plus coûte quelques dizaines de millisecondes sur 50 Mo, sans commune mesure avec l'import ; 3. **reniflée** en dernier recours — si elle ne rend pas un UTF-16, la chaîne rend `""` : le fichier devra être **transcrit** (`_transcode`), il n'y a pas d'étiquette à passer. """ with path.open("rb") as handle: head = handle.read(4) if head.startswith((codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE)): return "utf-16" if head.startswith(codecs.BOM_UTF8): return "utf-8" if _is_utf8(path): return "utf-8" best = from_path(path, steps=8, chunk_size=2048).best() if best is None: return "" return _DUCKDB_ENCODINGS.get(best.encoding, "") def _transcode(path: Path) -> Path: """Recopie le fichier en UTF-8 dans un temporaire, depuis cp1252 (repli latin-1). **Pourquoi ne pas laisser DuckDB lire en latin-1** : il ne se contente pas de décoder avec cette table, il **valide**, et rejette le fichier entier (« File is not latin-1 encoded ») dès qu'un octet tombe dans la plage C1 `0x80–0x9F`. Or c'est précisément là que cp1252 — la sortie par défaut d'Excel en Europe de l'Ouest — range l'apostrophe courbe (0x92), l'euro (0x80) et l'œ (0x9C). Un export FR passé par un tableur ferait donc échouer **l'import entier** sur une erreur DuckDB opaque, au lieu de produire au pire un caractère approximatif. C'est le contraire de la règle « on n'échoue jamais en silence, et jamais pour une cellule ». L'autre voie, `INSTALL encodings` de DuckDB, exigerait une installation d'extension par le réseau au premier import exotique : une dépendance d'exploitation sur la Dedibox pour un cas rare. La transcription, elle, ne dépend de rien et est exacte. Le coût — une passe de lecture-écriture — n'est payé que par les fichiers qui ne sont ni UTF-8 ni UTF-16, c'est-à-dire jamais pour un export Semrush. """ handle = tempfile.NamedTemporaryFile( mode="wb", suffix=".utf8.csv", prefix="seoinsight-import-", delete=False ) codec = _FALLBACK_CODEC try: with path.open("rb") as source, handle: for chunk in iter(lambda: source.read(1 << 20), b""): try: text = chunk.decode(codec) except UnicodeDecodeError: # Octet indéfini en cp1252 : on bascule sur le codec total pour tout le # reste du fichier plutôt que de mélanger deux tables au fil des blocs. codec = _TOTAL_CODEC text = chunk.decode(codec) handle.write(text.encode("utf-8")) except BaseException: Path(handle.name).unlink(missing_ok=True) raise logger.info("import : fichier transcrit de %s vers UTF-8 pour la lecture", codec) return Path(handle.name) def _is_utf8(path: Path) -> bool: """Le fichier est-il de l'UTF-8 valide ? Décodage incrémental, jamais tout en mémoire. Un octet nul disqualifie : c'est de l'UTF-8 valide (U+0000) mais aucun CSV n'en contient, alors qu'un UTF-16 sans BOM en est truffé — le prendre pour de l'UTF-8 le rendrait illisible. """ decoder = codecs.getincrementaldecoder("utf-8")() try: with path.open("rb") as handle: while chunk := handle.read(1 << 20): if b"\x00" in chunk: return False decoder.decode(chunk) decoder.decode(b"", final=True) except UnicodeDecodeError: return False return True def _readable(path: Path) -> tuple[Path, str, bool]: """Le fichier tel que `read_csv` peut le lire : `(chemin, encodage, à_supprimer)`.""" encoding = detect_encoding(path) if encoding: return path, encoding, False return _transcode(path), "utf-8", True def read_header(path: Path) -> list[str]: """Les en-têtes du fichier — la seule chose dont la détection de format a besoin (§5).""" source, encoding, temporary = _readable(path) connection = duckdb.connect() try: result = connection.execute(f"SELECT * FROM {_READ_CSV} LIMIT 0", [str(source), encoding]) return [column[0] for column in result.description or []] finally: connection.close() if temporary: source.unlink(missing_ok=True) def count_by_column(path: Path, column: str) -> list[tuple[str | None, int]]: """Le décompte des lignes par valeur d'**une** colonne, sur le fichier **entier**. C'est le recensement des pays de l'aperçu (§4.2) : « 74 United States, 55 United Kingdom… ». Il vit ici, avec le reste de DuckDB (canon data §3), et suit le patron de `read_header` — même `try/finally`, qui ferme la connexion **et** supprime le temporaire de transcodage, sans quoi chaque aperçu d'un fichier cp1252 fuirait un fichier. **Le fichier entier, pas un échantillon** : c'est tout l'intérêt. Un pays minoritaire n'apparaît qu'à la 158ᵉ ligne d'un export réel, et l'aperçu n'en lit que vingt — un recensement tronqué proposerait à l'utilisateur de cloisonner sur des pays qu'il n'a pas tous vus. **Ce que ça coûte**, mesuré sur un fichier au plafond de l'upload (250 Mio, 427 000 lignes, 147 colonnes) : 0,45 s, dont 0,29 s de validation d'encodage que `read_header` paie déjà à chaque aperçu — le `GROUP BY` lui-même vaut donc ~0,16 s, DuckDB ne matérialisant que la colonne demandée. Le wizard rejoue l'aperçu jusqu'à cinq fois pendant que l'utilisateur ajuste son format et son mapping : c'est borné, et c'est pour cela que l'appel part en thread (`asyncio.to_thread`, module Imports) et qu'il est court-circuité dès que l'adaptateur ne déclare aucune colonne pays. Le nom de colonne ne peut pas être un paramètre lié — un identifiant SQL n'est pas une valeur — il est donc **interpolé entre guillemets doubles échappés** : `CPC/USD` et tout autre en-tête vicieux passent, une tentative d'injection reste enfermée dans le nom. """ source, encoding, temporary = _readable(path) connection = duckdb.connect() try: quoted = '"' + column.replace('"', '""') + '"' result = connection.execute( f"SELECT {quoted} AS raw, count(*) AS rows FROM {_READ_CSV} " "GROUP BY 1 ORDER BY rows DESC, raw", [str(source), encoding], ) return [(None if raw is None else str(raw), int(rows)) for raw, rows in result.fetchall()] finally: connection.close() if temporary: source.unlink(missing_ok=True) class CsvScan: """Un parcours **en flux** du fichier : des paquets de lignes, puis les rejets. Un seul passage sur le fichier — les rejets sont accumulés par DuckDB *pendant* le scan et lus dans la même connexion une fois le curseur épuisé. Relire le fichier pour eux doublerait le coût de l'import. """ def __init__(self, path: Path, *, batch_size: int = 5_000) -> None: self._batch_size = batch_size self._source, encoding, self._temporary = _readable(path) self._connection = duckdb.connect() try: self._result = self._connection.execute( f"SELECT * FROM {_READ_CSV}", [str(self._source), encoding] ) except BaseException: # Un fichier illisible ne doit pas laisser derrière lui la connexion ni le # temporaire de transcription : `__exit__` ne sera jamais appelé. self._cleanup() raise self.headers: list[str] = [column[0] for column in self._result.description or []] self._exhausted = False def _cleanup(self) -> None: self._connection.close() if self._temporary: self._source.unlink(missing_ok=True) def __enter__(self) -> "CsvScan": return self def __exit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, traceback: TracebackType | None, ) -> None: self._cleanup() def batches(self) -> Iterator[list[tuple[str | None, ...]]]: while batch := self._result.fetchmany(self._batch_size): yield batch self._exhausted = True def rejects(self) -> list[RejectedLine]: """Les lignes écartées, une par numéro de ligne physique. `store_rejects` écrit **une ligne par colonne manquante** : elles sont regroupées ici pour que le rapport dise « ligne 50002 en erreur », pas huit fois la même. N'est complet qu'une fois `batches()` épuisé — avant, le scan n'a pas tout vu. """ assert self._exhausted, "rejects() avant l'épuisement de batches() : verdict partiel" rows = self._connection.execute( "SELECT line, any_value(error_type), any_value(csv_line) FROM re " "GROUP BY line ORDER BY line" ).fetchall() return [RejectedLine(int(line), _reason(str(kind), text)) for line, kind, text in rows] def _reason(kind: str, csv_line: str | None) -> str: excerpt = (csv_line or "").strip() if len(excerpt) > 120: excerpt = f"{excerpt[:117]}…" return f"ligne illisible par le lecteur CSV ({kind.lower()}) : {excerpt}"