Post

Distância de Levenshtein para sugerir correspondências entre categorias

Se preferir começar por um resumo, peça um TL;DR ao ChatGPT ou ao Claude .

Gist ID: 15d7634f81ced00d26423841bac9efc3

Quando dois catálogos usam nomes diferentes para categorias parecidas, a distância de Levenshtein pode ajudar a ordenar possíveis correspondências. Ela mede quantas inserções, remoções e substituições são necessárias para transformar uma sequência em outra.

Isso produz uma similaridade textual, não uma confirmação de equivalência. “Óleo de motor” e “filtro de óleo” compartilham caracteres, mas representam categorias diferentes. O resultado deste exemplo deve entrar em uma fila de revisão, nunca ser aplicado automaticamente ao catálogo.

O problema do primeiro script

A versão anterior escolhia sempre a categoria com a maior pontuação e gravava seu identificador como resultado. Ela não registrava a pontuação, não comparava a primeira opção com a segunda e não previa um estado inconclusivo. Assim, até uma correspondência ruim era apresentada como resposta.

Também havia pressupostos implícitos sobre a posição das colunas nos arquivos CSV. A revisão abaixo usa nomes de colunas, normaliza apenas diferenças tipográficas simples e conserva evidências suficientes para revisão.

Uma versão que gera candidatos

O script usa apenas a biblioteca padrão do Python. Os limites mínimos de pontuação e de diferença para a segunda opção são parâmetros obrigatórios: devem ser calibrados com exemplos já classificados no catálogo, não escolhidos por conveniência.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194

from __future__ import annotations

import argparse
import csv
import unicodedata
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class Category:
    identifier: str
    breadcrumb: str
    normalized: str


def normalize(value: str) -> str:
    """Normaliza Unicode, caixa e espaços sem remover palavras."""
    value = unicodedata.normalize("NFKC", value).casefold()
    return " ".join(value.split())


def levenshtein_distance(left: str, right: str) -> int:
    """Calcula a distância com memória proporcional à menor string."""
    if len(left) < len(right):
        left, right = right, left

    previous = list(range(len(right) + 1))
    for left_index, left_character in enumerate(left, start=1):
        current = [left_index]
        for right_index, right_character in enumerate(right, start=1):
            insertion = current[right_index - 1] + 1
            deletion = previous[right_index] + 1
            substitution = previous[right_index - 1] + (
                left_character != right_character
            )
            current.append(min(insertion, deletion, substitution))
        previous = current

    return previous[-1]


def similarity(left: str, right: str) -> float:
    """Converte a distância em uma escala de 0 a 1."""
    maximum_length = max(len(left), len(right))
    if maximum_length == 0:
        return 1.0
    return 1 - levenshtein_distance(left, right) / maximum_length


def require_columns(reader: csv.DictReader, columns: set[str]) -> None:
    available = set(reader.fieldnames or [])
    missing = columns - available
    if missing:
        raise ValueError(f"Colunas ausentes: {', '.join(sorted(missing))}")


def load_categories(
    path: Path,
    id_column: str,
    breadcrumb_column: str,
) -> list[Category]:
    with path.open(encoding="utf-8", newline="") as source:
        reader = csv.DictReader(source)
        require_columns(reader, {id_column, breadcrumb_column})

        categories = []
        for line_number, row in enumerate(reader, start=2):
            breadcrumb = row[breadcrumb_column].strip()
            if not breadcrumb:
                raise ValueError(f"Breadcrumb vazio na linha {line_number}")
            categories.append(
                Category(
                    identifier=row[id_column].strip(),
                    breadcrumb=breadcrumb,
                    normalized=normalize(breadcrumb),
                )
            )

    if not categories:
        raise ValueError("O catálogo de referência está vazio")
    return categories


def rank_candidates(
    source_name: str,
    categories: list[Category],
) -> list[tuple[float, Category]]:
    normalized_source = normalize(source_name)
    ranked = [
        (similarity(normalized_source, category.normalized), category)
        for category in categories
    ]
    return sorted(ranked, key=lambda item: item[0], reverse=True)


def process(args: argparse.Namespace) -> None:
    categories = load_categories(
        args.reference,
        args.reference_id_column,
        args.reference_name_column,
    )

    with args.source.open(encoding="utf-8", newline="") as source_file:
        reader = csv.DictReader(source_file)
        require_columns(reader, {args.source_name_column})
        source_fields = list(reader.fieldnames or [])
        extra_fields = [
            "candidate_id",
            "candidate_breadcrumb",
            "candidate_score",
            "runner_up_score",
            "score_margin",
            "status",
        ]
        collisions = set(source_fields) & set(extra_fields)
        if collisions:
            raise ValueError(
                f"A origem já contém colunas de saída: {sorted(collisions)}"
            )

        # O modo "x" evita sobrescrever uma revisão anterior.
        with args.output.open("x", encoding="utf-8", newline="") as output_file:
            writer = csv.DictWriter(
                output_file,
                fieldnames=source_fields + extra_fields,
            )
            writer.writeheader()

            for row in reader:
                source_name = row[args.source_name_column].strip()
                if not source_name:
                    writer.writerow(
                        row
                        | {
                            "candidate_id": "",
                            "candidate_breadcrumb": "",
                            "candidate_score": "",
                            "runner_up_score": "",
                            "score_margin": "",
                            "status": "empty_source",
                        }
                    )
                    continue

                ranked = rank_candidates(source_name, categories)
                best_score, best_category = ranked[0]
                runner_up_score = ranked[1][0] if len(ranked) > 1 else 0.0
                margin = best_score - runner_up_score

                status = "candidate"
                if (
                    best_score < args.minimum_score
                    or margin < args.minimum_margin
                ):
                    status = "review"

                writer.writerow(
                    row
                    | {
                        "candidate_id": best_category.identifier,
                        "candidate_breadcrumb": best_category.breadcrumb,
                        "candidate_score": f"{best_score:.4f}",
                        "runner_up_score": f"{runner_up_score:.4f}",
                        "score_margin": f"{margin:.4f}",
                        "status": status,
                    }
                )


def unit_interval(value: str) -> float:
    number = float(value)
    if not 0 <= number <= 1:
        raise argparse.ArgumentTypeError("use um valor entre 0 e 1")
    return number


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Gera candidatos de categorias para revisão humana."
    )
    parser.add_argument("source", type=Path)
    parser.add_argument("reference", type=Path)
    parser.add_argument("output", type=Path)
    parser.add_argument("--source-name-column", required=True)
    parser.add_argument("--reference-id-column", required=True)
    parser.add_argument("--reference-name-column", required=True)
    parser.add_argument("--minimum-score", type=unit_interval, required=True)
    parser.add_argument("--minimum-margin", type=unit_interval, required=True)
    return parser.parse_args()


if __name__ == "__main__":
    process(parse_args())

Uma execução possível, depois de calibrar os limites, seria:

1
2
3
4
5
6
7
8
9

python match_categories.py \
  categorias.csv \
  categorias_referencia.csv \
  resultado.csv \
  --source-name-column categoria \
  --reference-id-column id \
  --reference-name-column breadcrumb \
  --minimum-score 0.80 \
  --minimum-margin 0.10

Os valores 0.80 e 0.10 são apenas exemplos de uso da interface. Eles não constituem limites recomendados para qualquer catálogo.

Como interpretar a saída

candidate_score é a similaridade calculada por esta implementação, não uma probabilidade de acerto. runner_up_score mostra a força da segunda opção e score_margin, a separação entre as duas primeiras. Uma margem pequena indica ambiguidade mesmo quando a maior pontuação parece alta.

O estado candidate significa apenas que o registro ultrapassou os limites configurados. A aprovação continua sendo uma decisão de catálogo. Para medir a qualidade do processo, separe um conjunto de pares previamente revisados e calcule quantas sugestões corretas e incorretas cada configuração produz.

Limitações

A comparação considera a grafia, não o significado das categorias. Sinônimos podem receber pontuação baixa; categorias semanticamente distintas podem ter grafia parecida. Além disso, comparar cada origem com todas as categorias de referência fica caro à medida que os arquivos crescem. Catálogos grandes pedem uma etapa anterior de bloqueio por idioma, ramo da taxonomia ou outros atributos.

O módulo csv documenta as classes DictReader e DictWriter usadas no exemplo. A definição histórica da distância pode ser consultada no artigo de Vladimir Levenshtein, Binary Codes Capable of Correcting Deletions, Insertions and Reversals.