Home » Реализация расстояния Левенштейна — автозаполнение и автокоррекция слов

Реализация расстояния Левенштейна — автозаполнение и автокоррекция слов

Кто из нас не сталкивался с багом в команде и не получал “command not found”? Или не мучился с поиском файлов, когда забыл точное название? Расстояние Левенштейна — это как раз тот алгоритм, который стоит за автокоррекцией в bash, автозаполнением в поисковых системах и умными подсказками в IDE. Сегодня разберём, как самому реализовать эту штуку и внедрить в свои скрипты для администрирования серверов. Особенно полезно будет тем, кто работает с большими конфигами, базами данных логов или просто хочет сделать свои утилиты чуточку умнее.

Что такое расстояние Левенштейна и зачем оно нужно

Расстояние Левенштейна (edit distance) — это минимальное количество операций вставки, удаления или замены символов, необходимых для превращения одной строки в другую. Простыми словами: насколько две строки похожи друг на друга.

Примеры использования в серверном администрировании:

Автокоррекция команд в bash-скриптах
Поиск похожих файлов в логах
Умные подсказки в custom CLI-утилитах
Дедупликация данных в базах
Поиск по конфигурационным файлам

Как работает алгоритм

Алгоритм строит матрицу, где каждая ячейка содержит минимальное расстояние между префиксами двух строк. Звучит сложно, но на практике всё довольно просто.

Пример для строк “nginx” и “apache”:

		a	p	a	c	h	e
	0	1	2	3	4	5	6
n	1	1	2	3	4	5	6
g	2	2	2	3	4	5	6
i	3	3	3	3	4	5	6
n	4	4	4	4	4	5	6
x	5	5	5	5	5	5	6

Результат: расстояние = 6. Это означает, что “nginx” и “apache” совсем не похожи.

Реализация на Python

Базовая реализация алгоритма:

def levenshtein_distance(s1, s2):
    """Вычисляет расстояние Левенштейна между двумя строками"""
    m, n = len(s1), len(s2)
    
    # Создаём матрицу (m+1) x (n+1)
    dp = [[0] * (n + 1) for _ in range(m + 1)]
    
    # Инициализация первой строки и столбца
    for i in range(m + 1):
        dp[i][0] = i
    for j in range(n + 1):
        dp[0][j] = j
    
    # Заполняем матрицу
    for i in range(1, m + 1):
        for j in range(1, n + 1):
            if s1[i-1] == s2[j-1]:
                dp[i][j] = dp[i-1][j-1]
            else:
                dp[i][j] = 1 + min(
                    dp[i-1][j],     # удаление
                    dp[i][j-1],     # вставка
                    dp[i-1][j-1]    # замена
                )
    
    return dp[m][n]

# Тестируем
print(levenshtein_distance("nginx", "apache"))  # 6
print(levenshtein_distance("systemctl", "systemd"))  # 4

Оптимизированная версия для больших строк

Для серверных задач часто нужна более быстрая реализация. Вот версия с оптимизацией по памяти:

def levenshtein_optimized(s1, s2):
    """Оптимизированная версия — O(min(m,n)) по памяти"""
    if len(s1) < len(s2):
        s1, s2 = s2, s1
    
    if len(s2) == 0:
        return len(s1)
    
    previous_row = list(range(len(s2) + 1))
    for i, c1 in enumerate(s1):
        current_row = [i + 1]
        for j, c2 in enumerate(s2):
            insertions = previous_row[j + 1] + 1
            deletions = current_row[j] + 1
            substitutions = previous_row[j] + (c1 != c2)
            current_row.append(min(insertions, deletions, substitutions))
        previous_row = current_row
    
    return previous_row[-1]

Практическое применение: автокоррекция команд

Создадим скрипт для автокоррекции bash-команд:

#!/usr/bin/env python3
import os
import sys
from difflib import get_close_matches

def get_available_commands():
    """Получает список доступных команд из PATH"""
    commands = set()
    for path in os.environ.get('PATH', '').split(':'):
        if os.path.isdir(path):
            try:
                for cmd in os.listdir(path):
                    if os.access(os.path.join(path, cmd), os.X_OK):
                        commands.add(cmd)
            except PermissionError:
                continue
    return list(commands)

def suggest_command(wrong_cmd, threshold=0.6):
    """Предлагает похожие команды"""
    commands = get_available_commands()
    suggestions = []
    
    for cmd in commands:
        distance = levenshtein_distance(wrong_cmd, cmd)
        similarity = 1 - distance / max(len(wrong_cmd), len(cmd))
        
        if similarity >= threshold:
            suggestions.append((cmd, similarity))
    
    # Сортируем по убыванию схожести
    suggestions.sort(key=lambda x: x[1], reverse=True)
    return [cmd for cmd, _ in suggestions[:5]]

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: ./autocorrect.py ")
        sys.exit(1)
    
    wrong_cmd = sys.argv[1]
    suggestions = suggest_command(wrong_cmd)
    
    if suggestions:
        print(f"Command '{wrong_cmd}' not found. Did you mean:")
        for i, cmd in enumerate(suggestions, 1):
            print(f"  {i}. {cmd}")
    else:
        print(f"No similar commands found for '{wrong_cmd}'")

Использование:

$ ./autocorrect.py systmctl
Command 'systmctl' not found. Did you mean:
  1. systemctl
  2. systemd
  3. system

Автозаполнение для конфигурационных файлов

Часто нужно искать параметры в конфигах. Вот утилита для nginx:

#!/usr/bin/env python3
import re

class NginxConfigHelper:
    def __init__(self):
        self.common_directives = [
            'server_name', 'listen', 'root', 'index', 'location',
            'proxy_pass', 'proxy_set_header', 'ssl_certificate',
            'ssl_certificate_key', 'access_log', 'error_log',
            'worker_processes', 'worker_connections', 'keepalive_timeout',
            'client_max_body_size', 'gzip', 'upstream'
        ]
    
    def suggest_directive(self, partial, max_distance=2):
        """Предлагает директивы nginx по частичному вводу"""
        suggestions = []
        
        for directive in self.common_directives:
            distance = levenshtein_distance(partial.lower(), directive.lower())
            if distance <= max_distance: suggestions.append((directive, distance)) return [directive for directive, _ in sorted(suggestions, key=lambda x: x[1])] def validate_config_line(self, line): """Проверяет строку конфига и предлагает исправления""" line = line.strip() if not line or line.startswith('#'): return None # Извлекаем директиву match = re.match(r'^(\w+)', line) if not match: return None directive = match.group(1) suggestions = self.suggest_directive(directive) if directive not in self.common_directives and suggestions: return f"Unknown directive '{directive}'. Did you mean: {', '.join(suggestions[:3])}?" return None # Пример использования helper = NginxConfigHelper() # Тестируем test_lines = [ "server_nam example.com;", "liste 80;", "proxy_pas http://backend;", "ssl_certifcate /path/to/cert.pem;" ] for line in test_lines: suggestion = helper.validate_config_line(line) if suggestion: print(f"'{line}' -> {suggestion}")

Поиск по логам с опечатками

Для поиска в логах nginx или apache:

#!/usr/bin/env python3
import re
import sys
from collections import defaultdict

class LogSearcher:
    def __init__(self, log_file):
        self.log_file = log_file
        self.entries = self._parse_log()
    
    def _parse_log(self):
        """Парсит лог-файл"""
        entries = []
        try:
            with open(self.log_file, 'r') as f:
                for line_num, line in enumerate(f, 1):
                    # Простой парсинг nginx access.log
                    match = re.match(r'^(\S+) .* "(\w+) ([^"]+)" (\d+)', line)
                    if match:
                        ip, method, path, status = match.groups()
                        entries.append({
                            'line': line_num,
                            'ip': ip,
                            'method': method,
                            'path': path,
                            'status': status,
                            'raw': line.strip()
                        })
        except FileNotFoundError:
            print(f"Log file {self.log_file} not found")
            sys.exit(1)
        return entries
    
    def fuzzy_search(self, query, field='path', max_distance=2):
        """Нечёткий поиск в логах"""
        results = []
        
        for entry in self.entries:
            distance = levenshtein_distance(query.lower(), entry[field].lower())
            if distance <= max_distance:
                results.append((entry, distance))
        
        return sorted(results, key=lambda x: x[1])

# Использование
searcher = LogSearcher('/var/log/nginx/access.log')
results = searcher.fuzzy_search('/api/usres', field='path', max_distance=2)

for entry, distance in results[:10]:
    print(f"Distance {distance}: {entry['path']} (line {entry['line']})")

Интеграция с bash

Создадим bash-функцию для автокоррекции:

#!/bin/bash

# Добавить в ~/.bashrc
command_not_found_handle() {
    local cmd="$1"
    local suggestions
    
    # Используем наш Python-скрипт
    suggestions=$(python3 /usr/local/bin/autocorrect.py "$cmd" 2>/dev/null)
    
    if [ -n "$suggestions" ]; then
        echo "$suggestions"
        echo ""
        read -p "Run suggested command? (y/N): " -n 1 -r
        echo
        if [[ $REPLY =~ ^[Yy]$ ]]; then
            # Выполняем первую предложенную команду
            local first_suggestion=$(echo "$suggestions" | grep "1\." | sed 's/.*1\. //')
            if [ -n "$first_suggestion" ]; then
                eval "$first_suggestion ${@:2}"
            fi
        fi
    else
        echo "bash: $cmd: command not found"
    fi
}

# Функция для нечёткого поиска файлов
fzf_find() {
    local query="$1"
    find . -type f -name "*$query*" 2>/dev/null | head -20
}

# Алиас для умного поиска
alias ff='fzf_find'

Сравнение с другими решениями

Решение	Скорость	Точность	Память	Простота
Расстояние Левенштейна	O(mn)	Высокая	O(mn)	Средняя
Расстояние Хэмминга	O(n)	Низкая	O(1)	Высокая
Jaro-Winkler	O(nm)	Средняя	O(n+m)	Низкая
Soundex	O(n)	Низкая	O(1)	Высокая

Оптимизация для production

Для серверной нагрузки нужны дополнительные оптимизации:

#!/usr/bin/env python3
import sqlite3
import hashlib
from functools import lru_cache

class FastLevenshtein:
    def __init__(self, db_path='/tmp/levenshtein_cache.db'):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS cache (
                hash TEXT PRIMARY KEY,
                s1 TEXT,
                s2 TEXT,
                distance INTEGER
            )
        ''')
        self.conn.commit()
    
    def _hash_pair(self, s1, s2):
        """Создаёт хеш для пары строк"""
        return hashlib.md5(f"{s1}:{s2}".encode()).hexdigest()
    
    @lru_cache(maxsize=1000)
    def cached_distance(self, s1, s2):
        """Вычисляет расстояние с кешированием"""
        cache_key = self._hash_pair(s1, s2)
        
        # Проверяем кеш в БД
        result = self.conn.execute(
            'SELECT distance FROM cache WHERE hash = ?',
            (cache_key,)
        ).fetchone()
        
        if result:
            return result[0]
        
        # Вычисляем и сохраняем в кеш
        distance = levenshtein_optimized(s1, s2)
        self.conn.execute(
            'INSERT OR REPLACE INTO cache (hash, s1, s2, distance) VALUES (?, ?, ?, ?)',
            (cache_key, s1, s2, distance)
        )
        self.conn.commit()
        
        return distance

# Использование
fast_lev = FastLevenshtein()
print(fast_lev.cached_distance("nginx", "apache"))  # Вычисляется
print(fast_lev.cached_distance("nginx", "apache"))  # Из кеша

Интеграция с мониторингом

Для мониторинга ошибок в логах:

#!/usr/bin/env python3
import time
import json
from datetime import datetime

class ErrorMonitor:
    def __init__(self):
        self.known_errors = [
            "Connection refused",
            "Permission denied",
            "File not found",
            "Timeout",
            "Memory allocation failed",
            "Disk full",
            "Network unreachable"
        ]
    
    def classify_error(self, error_msg, threshold=0.7):
        """Классифицирует ошибку по известным паттернам"""
        best_match = None
        best_score = 0
        
        for known_error in self.known_errors:
            distance = levenshtein_distance(error_msg.lower(), known_error.lower())
            score = 1 - distance / max(len(error_msg), len(known_error))
            
            if score > best_score and score >= threshold:
                best_score = score
                best_match = known_error
        
        return best_match, best_score
    
    def process_log_line(self, line):
        """Обрабатывает строку лога"""
        if 'ERROR' in line or 'WARN' in line:
            match, score = self.classify_error(line)
            if match:
                return {
                    'timestamp': datetime.now().isoformat(),
                    'original': line.strip(),
                    'classified_as': match,
                    'confidence': score
                }
        return None

# Пример использования
monitor = ErrorMonitor()
test_errors = [
    "ERROR: Conection refused to server",
    "WARN: Permision denied for user",
    "ERROR: Fil not found at /path"
]

for error in test_errors:
    result = monitor.process_log_line(error)
    if result:
        print(json.dumps(result, indent=2))

Деплой на VPS

Для разворачивания на сервере понадобится VPS или выделенный сервер. Создаём systemd-сервис:

# /etc/systemd/system/autocorrect.service
[Unit]
Description=Autocorrect Service
After=network.target

[Service]
Type=simple
User=autocorrect
Group=autocorrect
WorkingDirectory=/opt/autocorrect
ExecStart=/usr/bin/python3 /opt/autocorrect/server.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

# Установка
sudo systemctl daemon-reload
sudo systemctl enable autocorrect
sudo systemctl start autocorrect

Полезные библиотеки и утилиты

Готовые решения для использования:

python-Levenshtein — C-реализация, очень быстрая
fuzzywuzzy — удобная обёртка для нечёткого поиска
jellyfish — множество алгоритмов похожести строк
difflib — встроенная в Python библиотека
fzf — консольный fuzzy finder
zsh-autosuggestions — автодополнение для zsh

Установка оптимизированной версии:

pip install python-Levenshtein fuzzywuzzy[speedup]

# Пример использования
from fuzzywuzzy import fuzz, process

commands = ['systemctl', 'systemd', 'service', 'nginx', 'apache']
result = process.extract('systmctl', commands, limit=3)
print(result)  # [('systemctl', 89), ('systemd', 62), ('service', 44)]

Интересные кейсы и хаки

Несколько нестандартных применений:

Дедупликация логов: группировка похожих записей для анализа
Детекция инъекций: поиск подозрительных SQL-запросов
Анализ конфигов: поиск опечаток в параметрах
Умный grep: поиск с учётом опечаток
Автокоррекция путей: исправление неточных путей к файлам

Пример умного grep:

#!/usr/bin/env python3
import sys
import re

def smart_grep(pattern, text, max_distance=2):
    """Grep с поддержкой нечёткого поиска"""
    lines = text.split('\n')
    matches = []
    
    for line_num, line in enumerate(lines, 1):
        words = re.findall(r'\w+', line.lower())
        for word in words:
            if levenshtein_distance(pattern.lower(), word) <= max_distance: matches.append((line_num, line)) break return matches # Использование if len(sys.argv) >= 2:
    pattern = sys.argv[1]
    text = sys.stdin.read()
    
    for line_num, line in smart_grep(pattern, text):
        print(f"{line_num}: {line}")

Статистика и бенчмарки

Тестирование производительности на реальных данных:

#!/usr/bin/env python3
import time
import random
import string

def benchmark_levenshtein():
    """Бенчмарк различных реализаций"""
    
    def generate_random_string(length):
        return ''.join(random.choices(string.ascii_lowercase, k=length))
    
    # Тестовые данные
    test_cases = [
        (10, 100),    # длина строк, количество тестов
        (50, 100),
        (100, 50),
        (200, 20)
    ]
    
    for str_len, num_tests in test_cases:
        print(f"\nТест: строки длиной {str_len}, {num_tests} итераций")
        
        # Генерируем тестовые строки
        test_strings = [
            (generate_random_string(str_len), generate_random_string(str_len))
            for _ in range(num_tests)
        ]
        
        # Базовая реализация
        start = time.time()
        for s1, s2 in test_strings:
            levenshtein_distance(s1, s2)
        basic_time = time.time() - start
        
        # Оптимизированная реализация
        start = time.time()
        for s1, s2 in test_strings:
            levenshtein_optimized(s1, s2)
        optimized_time = time.time() - start
        
        print(f"Базовая: {basic_time:.4f}s")
        print(f"Оптимизированная: {optimized_time:.4f}s")
        print(f"Ускорение: {basic_time/optimized_time:.2f}x")

if __name__ == "__main__":
    benchmark_levenshtein()

Автоматизация и интеграция в CI/CD

Проверка конфигурационных файлов в CI:

#!/usr/bin/env python3
# validate_configs.py
import os
import sys
import yaml

class ConfigValidator:
    def __init__(self):
        self.known_keys = {
            'nginx': ['server_name', 'listen', 'root', 'location', 'proxy_pass'],
            'apache': ['ServerName', 'Listen', 'DocumentRoot', 'VirtualHost'],
            'docker': ['image', 'ports', 'volumes', 'environment', 'command']
        }
    
    def validate_file(self, filepath, config_type):
        """Валидация файла конфигурации"""
        errors = []
        
        try:
            with open(filepath, 'r') as f:
                if config_type == 'yaml':
                    config = yaml.safe_load(f)
                    errors.extend(self._validate_yaml_keys(config, filepath))
                else:
                    lines = f.readlines()
                    errors.extend(self._validate_text_config(lines, filepath, config_type))
        except Exception as e:
            errors.append(f"Error reading {filepath}: {e}")
        
        return errors
    
    def _validate_yaml_keys(self, config, filepath):
        """Валидация ключей YAML"""
        errors = []
        if isinstance(config, dict):
            for key in config.keys():
                suggestions = self._get_suggestions(key, 'docker')
                if suggestions:
                    errors.append(f"{filepath}: Unknown key '{key}', did you mean: {suggestions[0]}?")
        return errors
    
    def _get_suggestions(self, key, config_type):
        """Получение предложений для ключа"""
        if config_type not in self.known_keys:
            return []
        
        suggestions = []
        for known_key in self.known_keys[config_type]:
            distance = levenshtein_distance(key.lower(), known_key.lower())
            if distance <= 2: suggestions.append((known_key, distance)) return [key for key, _ in sorted(suggestions, key=lambda x: x[1])] # Использование в CI validator = ConfigValidator() config_files = [ ('nginx.conf', 'nginx'), ('docker-compose.yml', 'yaml'), ('httpd.conf', 'apache') ] total_errors = 0 for filepath, config_type in config_files: if os.path.exists(filepath): errors = validator.validate_file(filepath, config_type) if errors: total_errors += len(errors) for error in errors: print(f"ERROR: {error}") sys.exit(1 if total_errors > 0 else 0)

Заключение и рекомендации

Расстояние Левенштейна — это мощный инструмент для решения многих задач в серверном администрировании. Вот основные рекомендации по использованию:

Для интерактивных утилит: используйте небольшой threshold (1-2) для точности
Для автоматизации: кешируйте результаты и используйте оптимизированные алгоритмы
Для production: обязательно тестируйте производительность на реальных данных
Для больших объёмов: рассмотрите использование C-реализаций (python-Levenshtein)

Алгоритм отлично подходит для:

Создания user-friendly CLI-утилит
Автоматизации рутинных задач
Анализа и мониторинга логов
Валидации конфигурационных файлов

Не стоит использовать для:

Очень длинных строк (>1000 символов) без оптимизации
Задач, требующих 100% точности
Сравнения семантически разных данных

Начните с простой реализации, протестируйте на ваших данных, а затем оптимизируйте под конкретные нужды. Помните: хорошая автокоррекция может значительно улучшить UX ваших инструментов и сэкономить время коллег.

В этой статье собрана информация и материалы из различных интернет-источников. Мы признаем и ценим работу всех оригинальных авторов, издателей и веб-сайтов. Несмотря на то, что были приложены все усилия для надлежащего указания исходного материала, любая непреднамеренная оплошность или упущение не являются нарушением авторских прав. Все упомянутые товарные знаки, логотипы и изображения являются собственностью соответствующих владельцев. Если вы считаете, что какой-либо контент, использованный в этой статье, нарушает ваши авторские права, немедленно свяжитесь с нами для рассмотрения и принятия оперативных мер.

Данная статья предназначена исключительно для ознакомительных и образовательных целей и не ущемляет права правообладателей. Если какой-либо материал, защищенный авторским правом, был использован без должного упоминания или с нарушением законов об авторском праве, это непреднамеренно, и мы исправим это незамедлительно после уведомления. Обратите внимание, что переиздание, распространение или воспроизведение части или всего содержимого в любой форме запрещено без письменного разрешения автора и владельца веб-сайта. Для получения разрешений или дополнительных запросов, пожалуйста, свяжитесь с нами.