Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 16 additions & 6 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -259,12 +259,22 @@ ruff check .
каталоге) — в блоке про `collector.py` выше.

- **`csv_io.py`** — парсинг только что скачанных CSV. Кодировка
автоопределяется перебором (`utf-8-sig`, `utf-8`, `cp1251` — Wordstat
экспортирует в cp1251), диалект — через `csv.Sniffer` с фолбэком на
`excel`. Заголовки валидируются (непустые, уникальные), но их конкретные
названия/язык — нет: они локализованы Wordstat и сохраняются как есть.
Пустые строки (все значения — пробелы) отбрасываются. Значения на этом
слое остаются строками.
автоопределяется перебором (`utf-8-sig`, `utf-8`, `cp1251`) — на живых
прогонах Вордстат отдаёт **UTF-8 с BOM**, `cp1251` остаётся фолбэком для
более старых/нестандартных экспортов, а не основным форматом. Терминатор
строк у некоторых видов — одиночный `\r` (CR), не `\r\n`; `Path.read_text`
нормализует это сам через universal newlines, отдельной обработки не
требуется. Диалект определяется через `csv.Sniffer` (разделители
`;`, `,`, `\t`), а на однострочном (только заголовок, без данных) CSV,
где `Sniffer` не может ничего вывести из образца, и при `csv.Error` —
фолбэк не на `excel`/запятую, а на реально встречающийся у Вордстата
разделитель (`;` или `\t`, выбирается по тому, какой чаще встречается в
строке заголовка). Заголовки валидируются (непустые, уникальные, без
оставшегося неразобранного `;`/`\t` — иначе `CsvFormatError`, а не молча
испорченная схема колонок), но их конкретные названия/язык — нет: они
локализованы Wordstat и сохраняются как есть. Пустые строки (все
значения — пробелы) отбрасываются. Значения на этом слое остаются
строками.

- **`dtypes.py`** — вывод типов колонок. Типизация идёт **по значениям, а не
по именам заголовков**: Wordstat локализует заголовки и переименовывает их
Expand Down
40 changes: 38 additions & 2 deletions src/wordstat/csv_io.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
"""CSV decoding for Wordstat downloads."""
"""CSV decoding for Wordstat downloads.

Wordstat exports are commonly UTF-8 with a BOM and may use a lone CR as the
line terminator. ``cp1251`` remains a supported fallback for older exports.
"""

import csv
from collections.abc import Iterable
Expand All @@ -10,6 +14,15 @@
_ENCODINGS = ("utf-8-sig", "utf-8", "cp1251")


class _FallbackDialect(csv.excel):
"""Fallback dialect for a Wordstat separator Sniffer couldn't confirm."""

delimiter = ";"


_FALLBACK_DELIMITERS = (";", "\t")


def parse_wordstat_csv(path: Path, view: WordstatView) -> CsvDataset:
"""Decode a Wordstat CSV while preserving its localized column names.

Expand Down Expand Up @@ -40,11 +53,32 @@ def _read_text(path: Path) -> str:


def _detect_dialect(text: str) -> csv.Dialect:
# Sniffer cannot establish a delimiter from a header-only export. Do not
# let its Excel fallback turn commas inside a localized header into
# columns; Wordstat uses either semicolon or tab depending on the view
# (see the "Fix the CSV sniffer's tab delimiter" fix), so pick whichever
# of the two known separators actually appears in the header line rather
# than hardcoding one.
lines = text.splitlines()
if len(lines) < 2:
return _fallback_dialect(lines[0] if lines else "")
sample = text[:4096]
try:
return csv.Sniffer().sniff(sample, delimiters=";,\t")
except csv.Error:
return csv.excel
return _fallback_dialect(lines[0])


def _fallback_dialect(header_line: str) -> csv.Dialect:
delimiter = max(_FALLBACK_DELIMITERS, key=header_line.count)
if header_line.count(delimiter) == 0:
delimiter = ";"

class _Dialect(_FallbackDialect):
pass

_Dialect.delimiter = delimiter
return _Dialect()


def _validate_headers(fieldnames: Iterable[str | None] | None, path: Path) -> list[str]:
Expand All @@ -53,6 +87,8 @@ def _validate_headers(fieldnames: Iterable[str | None] | None, path: Path) -> li
headers = [header.strip() if header else "" for header in fieldnames]
if not all(headers):
raise CsvFormatError(f"CSV {path.name} contains an empty header")
if any(delimiter in header for header in headers for delimiter in _FALLBACK_DELIMITERS):
raise CsvFormatError(f"CSV {path.name} contains an unparsed delimiter in a header")
if len(set(headers)) != len(headers):
raise CsvFormatError(f"CSV {path.name} contains duplicate headers")
return headers
Expand Down
83 changes: 82 additions & 1 deletion tests/test_csv_io.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import pytest

from wordstat.csv_io import parse_wordstat_csv
from wordstat.csv_io import _validate_headers, parse_wordstat_csv
from wordstat.errors import CsvFormatError
from wordstat.models import WordstatView

Expand All @@ -27,6 +27,87 @@ def test_parse_wordstat_csv_reads_a_tab_delimited_export(tmp_path: Path) -> None
assert dataset.rows == [{"Запрос": "чай", "Число запросов": "5228679"}]


def test_parse_wordstat_csv_reads_header_only_semicolon_export(tmp_path: Path) -> None:
source = tmp_path / "report.csv"
source.write_text(
"Запросы со словами;Число запросов;"
"Топ частотных запросов «новогодние подарки», 20.07.2026 — 20.08.2026, Россия, все устройства",
encoding="utf-8",
)

dataset = parse_wordstat_csv(source, WordstatView.TOP_POPULAR)

assert dataset.headers == [
"Запросы со словами",
"Число запросов",
"Топ частотных запросов «новогодние подарки», 20.07.2026 — 20.08.2026, Россия, все устройства",
]
assert all(";" not in header for header in dataset.headers)
assert dataset.rows == []


def test_parse_wordstat_csv_reads_bom_and_cr_header_only_export(tmp_path: Path) -> None:
source = tmp_path / "report.csv"
source.write_bytes(
"Запросы со словами;Число запросов;Топ частотных запросов «новогодние подарки», Россия".encode(
"utf-8-sig"
)
+ b"\r"
)

dataset = parse_wordstat_csv(source, WordstatView.TOP_RELATED)

assert dataset.headers == [
"Запросы со словами",
"Число запросов",
"Топ частотных запросов «новогодние подарки», Россия",
]
assert dataset.rows == []


def test_validate_headers_rejects_unparsed_semicolon(tmp_path: Path) -> None:
source = tmp_path / "report.csv"

with pytest.raises(CsvFormatError, match="unparsed delimiter"):
_validate_headers(["first;broken header"], source)


def test_validate_headers_rejects_unparsed_tab(tmp_path: Path) -> None:
source = tmp_path / "report.csv"

with pytest.raises(CsvFormatError, match="unparsed delimiter"):
_validate_headers(["first\tbroken header"], source)


def test_parse_wordstat_csv_reads_header_only_tab_export(tmp_path: Path) -> None:
source = tmp_path / "report.csv"
source.write_text(
"Запросы со словами\tЧисло запросов\t"
"Топ частотных запросов «новогодние подарки», 20.07.2026 — 20.08.2026, Россия, все устройства",
encoding="utf-8",
)

dataset = parse_wordstat_csv(source, WordstatView.TOP_POPULAR)

assert dataset.headers == [
"Запросы со словами",
"Число запросов",
"Топ частотных запросов «новогодние подарки», 20.07.2026 — 20.08.2026, Россия, все устройства",
]
assert all("\t" not in header for header in dataset.headers)
assert dataset.rows == []


def test_parse_wordstat_csv_keeps_normal_multiline_export_working(tmp_path: Path) -> None:
source = tmp_path / "report.csv"
source.write_text("Запрос;Число запросов\nчай;5\nкофе;3\n", encoding="utf-8")

dataset = parse_wordstat_csv(source, WordstatView.TOP_POPULAR)

assert dataset.headers == ["Запрос", "Число запросов"]
assert dataset.rows == [{"Запрос": "чай", "Число запросов": "5"}, {"Запрос": "кофе", "Число запросов": "3"}]


def test_parse_wordstat_csv_rejects_duplicate_headers(tmp_path: Path) -> None:
source = tmp_path / "report.csv"
source.write_text("query,query\none,two\n", encoding="utf-8")
Expand Down
Loading