Статья Merge File Sort или быстрая сортировка строк и удаление дублей интерпретатором в файле от 400 GB за счет ПЗУ

Nokia3310 · 04.11.2024

m00r сказал(а):

товарищи незнаю где такое выставить попробую выставить тут.

https://nohide[.]space/threads/luch...raboty-s-tekstom-text-utils-pack-by-lays.250/

самый шикарный софт был в свое время, но как я понял сейчас всю дорогу зараженный...
мб у кого то есть такой чистенький?

а так же предлагаю тсу спрятать софт в пкм и будет просто пушка!
или может развить софт до такого же уровня как у Lays?

за удобство люди будут готовы платить деньги

сколько людей качало этот софт by Lays не счесть. прям народный
*может быть что удаление дублей там работало не оч

Контекстное меню сделать не сложно
Но нужно четкое тз, что подразумевается под каждой функцией, что должно входить и выходить, куда, формат баз и тд

m00r · 04.11.2024

разбив очень полезная штука !! для сервачков особенно!
ну и конечно же антидубли
может быть какая то принудительная формовка в базу(u:l:p u:p ; | и так далее )
вытаскивание паролей
удаление строк содержащих не латиницу

к примеру

Nokia3310 · 05.11.2024

m00r сказал(а):

разбив очень полезная штука !! для сервачков особенно!
ну и конечно же антидубли
может быть какая то принудительная формовка в базу(u:l:p u:p ; | и так далее )
вытаскивание паролей
удаление строк содержащих не латиницу

к примеру

Да, звучит интересно
Нужен C++ или C#. Может, питон с Qt
Я могу этим заняться, но не сейчас...

rand · 05.11.2024

m00r сказал(а):

а так же предлагаю тсу спрятать софт в пкм и будет просто пушка!

Предлагали уже сделать комбайн, реализовать можно, но я пока ушел из ветки (да из хобби проектов в целом) до нового года, очень занят по основной работе. =(

baba_na_lune · 23.11.2024

Код:

use std::collections::HashSet;
use std::fs::{self, File};
use std::io::{self, BufRead, BufReader, Write};
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Instant;
use uuid::Uuid;

/// Обрабатывает чанк, удаляет дубликаты, сортирует и записывает в временный файл.
fn process_chunk(chunk: Vec<String>, temp_dir: &str) -> io::Result<(String, u128, usize, usize)> {
    let start_time = Instant::now(); // запускаем таймер для измерения времени обработки
    let original_count = chunk.len(); // сохраняем исходное количество строк

    let unique_items: HashSet<_> = chunk.into_iter().collect(); // убираем дубликаты, оставляя только уникальные строки в хэшсете
    let mut sorted_chunk: Vec<_> = unique_items.into_iter().collect(); // превращаем уникальные строки из хэшсета в вектор
    sorted_chunk.sort(); // сортируем вектор строк

    let duration = start_time.elapsed().as_millis(); // вычисляем время, затраченное на обработку
    let removed_count = original_count - sorted_chunk.len(); // считаем, сколько строк было удалено
    let unique_count = sorted_chunk.len(); // считаем количество уникальных строк

    let temp_file_path = format!("{}/temp_{}.txt", temp_dir, Uuid::new_v4()); // создаем путь для временного файла в папке
    let mut temp_file = File::create(&temp_file_path)?; // создаем временный файл

    for line in sorted_chunk { // записываем каждую строку в файл
        writeln!(temp_file, "{}", line)?; // записываем строку в файл
    }

    Ok((temp_file_path, duration, removed_count, unique_count)) // возвращаем путь к файлу и метрики
}

/// Объединяет временные файлы в один итоговый файл.
fn merge_files(temp_files: Vec<String>, output_file: &str) -> io::Result<()> { // функция объединяет временные файлы в один
    let mut outfile = File::create(output_file)?; // создаем итоговый файл
    let mut unique_lines = HashSet::new(); // создаем хэшсет для уникальных строк

    for temp_file in temp_files { // проходим по каждому временному файлу
        let file = File::open(temp_file)?; // читаем файл
        let reader = BufReader::new(file); // читаем с буфера

        for line in reader.lines() { // читаем каждую строку
            let line = line?; // обрабатываем возможные ошибки чтения
            if unique_lines.insert(line.clone()) { // добавляем строку в хэшсет, если она уникальна
                writeln!(outfile, "{}", line)?; // записываем уникальную строку в итоговый файл
            }
        }
    }

    Ok(())
}
/// Удаляет временные файлы после завершения работы программы.
fn delete_temp_files(temp_files: &Vec<String>) -> io::Result<()> { // функция удаляет временные файлы
    for temp_file in temp_files { // проходим по каждому временному файлу
        fs::remove_file(temp_file)?; // удаляем файл
    }
    Ok(()) // возвращаем успешный результат
}

/// Главная функция программы.
fn main() -> io::Result<()> { // главная функция программы
    let input_file = "/home/debi/Desktop/test/test_data.txt"; // путь к исходному файлу
    let output_file = "/home/debi/Desktop/test/output-sorted-unique.txt"; // путь к итоговому файлу
    let temp_dir = "/home/debi/Desktop/test"; // путь к папке для временных файлов

    if !Path::new(temp_dir).exists() { // проверяем, существует ли папка
        fs::create_dir(temp_dir)?; // создаем папку, если её нет
    }
    let file = File::open(input_file)?; // открываем исходный файл
    let reader = BufReader::new(file); // читаем с буфера
    let mut chunk = Vec::new(); // создаем вектор для хранения текущего чанка
    let chunk_size = 20000; // задаем размер чанка

    let arc_temp_files = Arc::new(Mutex::new(Vec::new())); // создаем вектор для временных файлов
    let mut handles = Vec::new(); // создаем вектор для хранения дескрипторов потоков

    let total_duration = Arc::new(Mutex::new(0)); // создаем переменную для общего времени
    let total_removed = Arc::new(Mutex::new(0)); // создаем переменную для общего количества удаленных строк
    let total_unique = Arc::new(Mutex::new(0)); // создаем переменную для общего количества уникальных строк

    for line in reader.lines() { // читаем каждую строку из файла
        let line = line?; // обрабатываем возможные ошибки чтения
        chunk.push(line); // добавляем строку в текущий чанк

        if chunk.len() >= chunk_size { // проверяем, достиг ли чанк нужного размера
            let temp_dir = temp_dir.to_string(); // копируем путь к папке  в поток
            let chunk_clone = chunk.clone(); // клонируем чанк  в поток
            let arc_temp_files_clone = Arc::clone(&arc_temp_files); // клонируем указатель на временные файлы для  в поток
            let total_duration_clone = Arc::clone(&total_duration); // клонируем указатель на общее время для  в поток
            let total_removed_clone = Arc::clone(&total_removed); // клонируем указатель на общее количество удаленных строк в поток
            let total_unique_clone = Arc::clone(&total_unique); // клонируем указатель на общее количество уникальных строк  в поток

            let handle = thread::spawn(move || { // новый поток для обработки чанка
                if let Ok((temp_file, duration, removed_count, unique_count)) = process_chunk(chunk_clone, &temp_dir) { // обрабатываем чанк
                    let mut temp_files = arc_temp_files_clone.lock().unwrap(); // блокируем доступ к временным файлам
                    temp_files.push(temp_file); // добавляем временный файл в список

                    *total_duration_clone.lock().unwrap() += duration; // добавляем время обработки к общему времени
                    *total_removed_clone.lock().unwrap() += removed_count; // добавляем количество удаленных строк к общему количеству
                    *total_unique_clone.lock().unwrap() += unique_count; // добавляем количество уникальных строк к общему количеству
                }
            });
            handles.push(handle); // добавляем дескриптор потока в список

            chunk.clear(); // очищаем текущий чанк
        }
    }

    if !chunk.is_empty() { // проверяем, остались ли необработанные строки в чанке
        if let Ok((temp_file, duration, removed_count, unique_count)) = process_chunk(chunk, temp_dir) { // обрабатываем оставшиеся строки в чанке
            let mut temp_files = arc_temp_files.lock().unwrap(); // блокируем доступ к временным файлам
            temp_files.push(temp_file); // добавляем временный файл в список

            *total_duration.lock().unwrap() += duration; // добавляем время обработки к общему времени
            *total_removed.lock().unwrap() += removed_count; // добавляем количество удаленных строк к общему количеству
            *total_unique.lock().unwrap() += unique_count; // добавляем количество уникальных строк к общему количеству
        }
    }

    for handle in handles { // проходим по каждому дескриптору потока
        handle.join().unwrap(); // ждем завершения потока
    }

    let temp_files = Arc::try_unwrap(arc_temp_files).unwrap().into_inner().unwrap(); // извлекаем список временных файлов
    merge_files(temp_files.clone(), output_file)?; // объединяем временные файлы в итоговый

    delete_temp_files(&temp_files)?; // удаляем временные файлы

    println!("Общее время выполнения сортировки: {} мс", *total_duration.lock().unwrap()); // выводим общее время
    println!("Общее количество удаленных строк: {}", *total_removed.lock().unwrap()); // выводим общее количество удаленных строк
    println!("Общее количество оставшихся уникальных строк: {}", *total_unique.lock().unwrap()); // выводим общее количество уникальных строк

    Ok(()) // возвращаем успешный результат
}

до 10 гб тестил, будет где потестить потестю еще, пока так, работает, если что надо доработать ПМ

creator07 · 24.11.2024

baba_na_lune сказал(а):

Код:

use std::collections::HashSet;
use std::fs::{self, File};
use std::io::{self, BufRead, BufReader, Write};
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Instant;
use uuid::Uuid;

/// Обрабатывает чанк, удаляет дубликаты, сортирует и записывает в временный файл.
fn process_chunk(chunk: Vec<String>, temp_dir: &str) -> io::Result<(String, u128, usize, usize)> {
    let start_time = Instant::now(); // запускаем таймер для измерения времени обработки
    let original_count = chunk.len(); // сохраняем исходное количество строк

    let unique_items: HashSet<_> = chunk.into_iter().collect(); // убираем дубликаты, оставляя только уникальные строки в хэшсете
    let mut sorted_chunk: Vec<_> = unique_items.into_iter().collect(); // превращаем уникальные строки из хэшсета в вектор
    sorted_chunk.sort(); // сортируем вектор строк

    let duration = start_time.elapsed().as_millis(); // вычисляем время, затраченное на обработку
    let removed_count = original_count - sorted_chunk.len(); // считаем, сколько строк было удалено
    let unique_count = sorted_chunk.len(); // считаем количество уникальных строк

    let temp_file_path = format!("{}/temp_{}.txt", temp_dir, Uuid::new_v4()); // создаем путь для временного файла в папке
    let mut temp_file = File::create(&temp_file_path)?; // создаем временный файл

    for line in sorted_chunk { // записываем каждую строку в файл
        writeln!(temp_file, "{}", line)?; // записываем строку в файл
    }

    Ok((temp_file_path, duration, removed_count, unique_count)) // возвращаем путь к файлу и метрики
}

/// Объединяет временные файлы в один итоговый файл.
fn merge_files(temp_files: Vec<String>, output_file: &str) -> io::Result<()> { // функция объединяет временные файлы в один
    let mut outfile = File::create(output_file)?; // создаем итоговый файл
    let mut unique_lines = HashSet::new(); // создаем хэшсет для уникальных строк

    for temp_file in temp_files { // проходим по каждому временному файлу
        let file = File::open(temp_file)?; // читаем файл
        let reader = BufReader::new(file); // читаем с буфера

        for line in reader.lines() { // читаем каждую строку
            let line = line?; // обрабатываем возможные ошибки чтения
            if unique_lines.insert(line.clone()) { // добавляем строку в хэшсет, если она уникальна
                writeln!(outfile, "{}", line)?; // записываем уникальную строку в итоговый файл
            }
        }
    }

    Ok(())
}
/// Удаляет временные файлы после завершения работы программы.
fn delete_temp_files(temp_files: &Vec<String>) -> io::Result<()> { // функция удаляет временные файлы
    for temp_file in temp_files { // проходим по каждому временному файлу
        fs::remove_file(temp_file)?; // удаляем файл
    }
    Ok(()) // возвращаем успешный результат
}

/// Главная функция программы.
fn main() -> io::Result<()> { // главная функция программы
    let input_file = "/home/debi/Desktop/test/test_data.txt"; // путь к исходному файлу
    let output_file = "/home/debi/Desktop/test/output-sorted-unique.txt"; // путь к итоговому файлу
    let temp_dir = "/home/debi/Desktop/test"; // путь к папке для временных файлов

    if !Path::new(temp_dir).exists() { // проверяем, существует ли папка
        fs::create_dir(temp_dir)?; // создаем папку, если её нет
    }
    let file = File::open(input_file)?; // открываем исходный файл
    let reader = BufReader::new(file); // читаем с буфера
    let mut chunk = Vec::new(); // создаем вектор для хранения текущего чанка
    let chunk_size = 20000; // задаем размер чанка

    let arc_temp_files = Arc::new(Mutex::new(Vec::new())); // создаем вектор для временных файлов
    let mut handles = Vec::new(); // создаем вектор для хранения дескрипторов потоков

    let total_duration = Arc::new(Mutex::new(0)); // создаем переменную для общего времени
    let total_removed = Arc::new(Mutex::new(0)); // создаем переменную для общего количества удаленных строк
    let total_unique = Arc::new(Mutex::new(0)); // создаем переменную для общего количества уникальных строк

    for line in reader.lines() { // читаем каждую строку из файла
        let line = line?; // обрабатываем возможные ошибки чтения
        chunk.push(line); // добавляем строку в текущий чанк

        if chunk.len() >= chunk_size { // проверяем, достиг ли чанк нужного размера
            let temp_dir = temp_dir.to_string(); // копируем путь к папке  в поток
            let chunk_clone = chunk.clone(); // клонируем чанк  в поток
            let arc_temp_files_clone = Arc::clone(&arc_temp_files); // клонируем указатель на временные файлы для  в поток
            let total_duration_clone = Arc::clone(&total_duration); // клонируем указатель на общее время для  в поток
            let total_removed_clone = Arc::clone(&total_removed); // клонируем указатель на общее количество удаленных строк в поток
            let total_unique_clone = Arc::clone(&total_unique); // клонируем указатель на общее количество уникальных строк  в поток

            let handle = thread::spawn(move || { // новый поток для обработки чанка
                if let Ok((temp_file, duration, removed_count, unique_count)) = process_chunk(chunk_clone, &temp_dir) { // обрабатываем чанк
                    let mut temp_files = arc_temp_files_clone.lock().unwrap(); // блокируем доступ к временным файлам
                    temp_files.push(temp_file); // добавляем временный файл в список

                    *total_duration_clone.lock().unwrap() += duration; // добавляем время обработки к общему времени
                    *total_removed_clone.lock().unwrap() += removed_count; // добавляем количество удаленных строк к общему количеству
                    *total_unique_clone.lock().unwrap() += unique_count; // добавляем количество уникальных строк к общему количеству
                }
            });
            handles.push(handle); // добавляем дескриптор потока в список

            chunk.clear(); // очищаем текущий чанк
        }
    }

    if !chunk.is_empty() { // проверяем, остались ли необработанные строки в чанке
        if let Ok((temp_file, duration, removed_count, unique_count)) = process_chunk(chunk, temp_dir) { // обрабатываем оставшиеся строки в чанке
            let mut temp_files = arc_temp_files.lock().unwrap(); // блокируем доступ к временным файлам
            temp_files.push(temp_file); // добавляем временный файл в список

            *total_duration.lock().unwrap() += duration; // добавляем время обработки к общему времени
            *total_removed.lock().unwrap() += removed_count; // добавляем количество удаленных строк к общему количеству
            *total_unique.lock().unwrap() += unique_count; // добавляем количество уникальных строк к общему количеству
        }
    }

    for handle in handles { // проходим по каждому дескриптору потока
        handle.join().unwrap(); // ждем завершения потока
    }

    let temp_files = Arc::try_unwrap(arc_temp_files).unwrap().into_inner().unwrap(); // извлекаем список временных файлов
    merge_files(temp_files.clone(), output_file)?; // объединяем временные файлы в итоговый

    delete_temp_files(&temp_files)?; // удаляем временные файлы

    println!("Общее время выполнения сортировки: {} мс", *total_duration.lock().unwrap()); // выводим общее время
    println!("Общее количество удаленных строк: {}", *total_removed.lock().unwrap()); // выводим общее количество удаленных строк
    println!("Общее количество оставшихся уникальных строк: {}", *total_unique.lock().unwrap()); // выводим общее количество уникальных строк

    Ok(()) // возвращаем успешный результат
}

до 10 гб тестил, будет где потестить потестю еще, пока так, работает, если что надо доработать ПМ

1) Работает только с кодировками UTF-8, если иная кодировка - краш
2) Не работоспособен со строками ULP
файлы одинаковые = сверху скрипт rand снизу твой

rand · 25.11.2024

baba_na_lune сказал(а):

до 10 гб тестил, будет где потестить потестю еще, пока так, работает, если что надо доработать ПМ

Ты конечно молодец что переписал этот алгоритм на Rust, но тут раздел по питону.

baba_na_lune · 25.11.2024

creator07 сказал(а):

1) Работает только с кодировками UTF-8, если иная кодировка - краш

Код:

use std::collections::HashSet;
use std::fs::{self, File};
use std::io::{self, BufReader, Write, Read, BufWriter};
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Instant;
use uuid::Uuid;
use chardetng::EncodingDetector;

/// Обрабатывает чанк, удаляет дубликаты, сортирует и записывает в временный файл.
fn read_file_with_encoding<P: AsRef<Path>>(path: P) -> io::Result<Vec<String>> {
    let file = File::open(path)?; // открываем файл
    let mut reader = BufReader::new(file);
    let mut buffer = Vec::new();
    reader.read_to_end(&mut buffer)?;

    let mut detector = EncodingDetector::new(); // создаем детектор кодировки
    detector.feed(&buffer, true); // передаем данные детектору
    let encoding = detector.guess(None, true); // определяем кодировку

    let (decoded, _, _) = encoding.decode(&buffer); // декодируем данные из буфера
    Ok(decoded.lines().map(|line| line.to_string()).collect()) // возвращаем вектор строк
}

fn write_file_with_encoding<P: AsRef<Path>>(path: P, lines: &[String]) -> io::Result<()> {
    let file = File::create(path)?;
    let mut writer = BufWriter::new(file);

    for line in lines {
        writeln!(writer, "{}", line)?;
    }

    Ok(())
}

fn process_chunk(chunk: Vec<String>, temp_dir: &str) -> io::Result<(String, u128, usize, usize)> {
    let start_time = Instant::now();
    let original_count = chunk.len();

    let unique_items: HashSet<_> = chunk.into_iter().collect();
    let mut sorted_chunk: Vec<_> = unique_items.into_iter().collect();
    sorted_chunk.sort();

    let duration = start_time.elapsed().as_millis();
    let removed_count = original_count - sorted_chunk.len();
    let unique_count = sorted_chunk.len();

    let temp_file_path = format!("{}/temp_{}.txt", temp_dir, Uuid::new_v4());
    write_file_with_encoding(&temp_file_path, &sorted_chunk)?;

    Ok((temp_file_path, duration, removed_count, unique_count))
}

/// Объединяет временные файлы в один итоговый файл.
fn merge_files(temp_files: Vec<String>, output_file: &str) -> io::Result<()> {
    let mut unique_lines = HashSet::new();

    for temp_file in temp_files {
        let lines = read_file_with_encoding(&temp_file)?;
        unique_lines.extend(lines);
    }

    let mut sorted_lines: Vec<_> = unique_lines.into_iter().collect();
    sorted_lines.sort();

    write_file_with_encoding(output_file, &sorted_lines)
}

/// Удаляет временные файлы после завершения работы программы.
fn delete_temp_files(temp_files: &Vec<String>) -> io::Result<()> {
    for temp_file in temp_files {
        fs::remove_file(temp_file)?;
    }
    Ok(())
}

fn main() -> io::Result<()> {
    let input_file = "/home/debi/Desktop/test/test_data.txt";
    let output_file = "/home/debi/Desktop/test/output-sorted-unique.txt";
    let temp_dir = "/home/debi/Desktop/test";

    if !Path::new(temp_dir).exists() {
        fs::create_dir(temp_dir)?;
    }
    let lines = read_file_with_encoding(input_file)?;
    let mut chunk = Vec::new();
    let chunk_size = 20000;

    let arc_temp_files = Arc::new(Mutex::new(Vec::new()));
    let mut handles = Vec::new();

    let total_duration = Arc::new(Mutex::new(0));
    let total_removed = Arc::new(Mutex::new(0));
    let total_unique = Arc::new(Mutex::new(0));

    for line in lines {
        chunk.push(line);

        if chunk.len() >= chunk_size { // проверяем, достиг ли чанк нужного размера
            let temp_dir = temp_dir.to_string(); // копируем путь к директории для использования в потоке
            let chunk_clone = chunk.clone(); // клонируем чанк для использования в потоке
            let arc_temp_files_clone = Arc::clone(&arc_temp_files); // клонируем указатель на временные файлы для использования в потоке
            let total_duration_clone = Arc::clone(&total_duration); // клонируем указатель на общее время для использования в потоке
            let total_removed_clone = Arc::clone(&total_removed); // клонируем указатель на общее количество удаленных строк для использования в потоке
            let total_unique_clone = Arc::clone(&total_unique); // клонируем указатель на общее количество уникальных строк для использования в потоке

            let handle = thread::spawn(move || { // создаем новый поток для обработки чанка
                if let Ok((temp_file, duration, removed_count, unique_count)) = process_chunk(chunk_clone, &temp_dir) { // обрабатываем чанк и получаем результаты
                    let mut temp_files = arc_temp_files_clone.lock().unwrap(); // блокируем доступ к временным файлам
                    temp_files.push(temp_file); // добавляем временный файл в список

                    *total_duration_clone.lock().unwrap() += duration; // добавляем время обработки к общему времени
                    *total_removed_clone.lock().unwrap() += removed_count; // добавляем количество удаленных строк к общему количеству
                    *total_unique_clone.lock().unwrap() += unique_count; // добавляем количество уникальных строк к общему количеству
                }
            });
            handles.push(handle); // добавляем дескриптор потока в список

            chunk.clear(); // очищаем текущий чанк
        }
    }

    if !chunk.is_empty() { // проверяем, остались ли необработанные строки в чанке
        if let Ok((temp_file, duration, removed_count, unique_count)) = process_chunk(chunk, temp_dir) { // обрабатываем оставшиеся строки в чанке
            let mut temp_files = arc_temp_files.lock().unwrap(); // блокируем доступ к временным файлам
            temp_files.push(temp_file); // добавляем временный файл в список

            *total_duration.lock().unwrap() += duration; // добавляем время обработки к общему времени
            *total_removed.lock().unwrap() += removed_count; // добавляем количество удаленных строк к общему количеству
            *total_unique.lock().unwrap() += unique_count; // добавляем количество уникальных строк к общему количеству
        }
    }

    for handle in handles { // проходим по каждому дескриптору потока
        handle.join().unwrap(); // ждем завершения потока
    }

    let temp_files = Arc::try_unwrap(arc_temp_files).unwrap().into_inner().unwrap(); // извлекаем список временных файлов
    merge_files(temp_files.clone(), output_file)?; // объединяем временные файлы в итоговый

    delete_temp_files(&temp_files)?; // удаляем временные файлы

    println!("Общее время выполнения сортировки: {} мс", *total_duration.lock().unwrap()); // выводим общее время
    println!("Общее количество удаленных строк: {}", *total_removed.lock().unwrap()); // выводим общее количество удаленных строк
    println!("Общее количество оставшихся уникальных строк: {}", *total_unique.lock().unwrap()); // выводим общее количество уникальных строк

    Ok(()) // возвращаем успешный результат
}

теперь должен кушать все форматы, с ютф 8 справляется так же
по ULP: кинь пару строк в пример, не пойму что у тебя не получается

rand сказал(а):

но тут раздел по питону.

пунк II.4 правил указывает не создавать тем которые уже обсуждаются
если модератор выделит в новую тему я буду ему благодарен. не думал что так принципиально. возможно, необходимо перенести тему в раздел администрирование, или иной, чтобы на разных языках можно было писать, чтобы найти самый оптимальный вариант.

rand · 25.11.2024

baba_na_lune сказал(а):

самый оптимальный вариант

Уже давно реализовано на компилируемых языках и быстрее, здесь же раздел по интерпретируемому питону. Можешь поискать ветку по c#. Что по замерам и почему под копирку?
Посмотри лучше в эту сторону:

GitHub - nil0x42/duplicut: Remove duplicates from MASSIVE wordlist, without sorting it (for dictionary-based password cracking)

Remove duplicates from MASSIVE wordlist, without sorting it (for dictionary-based password cracking) - nil0x42/duplicut

github.com

HUGE word-lists duplicate remover and merge tool

OK, so I've started writing my own word-list merge and duplicate remover :) already done 2 word-lists which are nearly 2 GB each. It obviously trades the use of your local HD instead of memory but it does chunk the data in memory before flushing it to files on local HD which improves...

forum.hashkiller.io

baba_na_lune · 25.11.2024

rand сказал(а):

Уже давно реализовано на компилируемых языках и быстрее, здесь же раздел по интерпретируемому питону. Можешь поискать ветку по c#. Что по замерам и почему под копирку?
Посмотри лучше в эту сторону:

GitHub - nil0x42/duplicut: Remove duplicates from MASSIVE wordlist, without sorting it (for dictionary-based password cracking)

Remove duplicates from MASSIVE wordlist, without sorting it (for dictionary-based password cracking) - nil0x42/duplicut

github.com

HUGE word-lists duplicate remover and merge tool

OK, so I've started writing my own word-list merge and duplicate remover :) already done 2 word-lists which are nearly 2 GB each. It obviously trades the use of your local HD instead of memory but it does chunk the data in memory before flushing it to files on local HD which improves...

forum.hashkiller.io

а я думал что то пилим, тогда я ушел с темы, за ссылочки лайк поставил, сейчас глянем

rand · 25.11.2024

baba_na_lune сказал(а):

а я думал что то пилим, тогда я ушел с темы, за ссылочки лайк поставил, сейчас глянем

Да уже вроде все запилили. И пока остановились на достигнутом, есть идея из этого продолжить и сделать комбайн для работы со строками, накинуть гуи, через реестр можно тем же батником как и просили добавить работу с контекстным меню в винде. Попробовать вычисления на gpu.

rand · 26.11.2024

m00r сказал(а):

искал буквально на неделе и вас нашел вот

https://nohide[.]space/threads/luch...raboty-s-tekstom-text-utils-pack-by-lays.250/

самый шикарный софт был в свое время, но как я понял сейчас всю дорогу зараженный...
мб у кого то есть такой чистенький?

а так же предлагаю тсу спрятать софт в пкм и будет просто пушка!
или может развить софт до такого же уровня как у Lays?

за удобство люди будут готовы платить деньги

сколько людей качало этот софт by Lays не счесть. прям народный
*может быть что удаление дублей там работало не оч

Я берусь за реализацию такого функционала на петухоне. К новому году постараюсь сделать бета версию с учетом всех пожеланий.

Начало уже положено =)

rand · 13.12.2024

DrSleep сказал(а):

Можно добавить функционал проверки длины строки?Строки меньше 6 символов и больше 32 в выходном файле ненужны.Готов все словари с weakpass.com прогнать через ваш софт и выложить в этом топике

Реализовано тут: /threads/127780/

WANNAKILL · 03.01.2025

Это эффективней чем?
cat file.txt | sort -uf > file_sorted.txt

ckat_soft · 04.01.2025

WANNAKILL сказал(а):

Это эффективней чем?
cat file.txt | sort -uf > file_sorted.txt

думаю не эфективнее

Bash:

sort -u --parallel=4 --buffer-size=1G -T /tmp -o sorted_data.txt data.txt

rand · 03.05.2025

Дедупликатор Version 2.0 BETA.

Специально для: xss.pro
Написал: rand

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

Обязательно к установке:

Bash:

pip install colorlog

main.py

Python:

# Импорт необходимых библиотек. Ничего лишнего.
import os
import heapq
import time
import logging
import shutil
import uuid
import math  # Для динамического определения числа воркеров, если надо
from concurrent.futures import ProcessPoolExecutor, as_completed
from colorlog import ColoredFormatter

# --- Настройка Логирования ---
formatter = ColoredFormatter(
    "%(log_color)s%(asctime)s - %(levelname)s - %(message)s",
    datefmt=None,
    reset=True,
    log_colors={
        'DEBUG': 'cyan',
        'INFO': 'green',
        'WARNING': 'yellow',
        'ERROR': 'red',
        'CRITICAL': 'bold_red',
    }
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.INFO)  # Или logging.DEBUG для полной картины

# --- Создание Временной Директории (Стандартная процедура) ---
# Пусть лежит рядом, чтобы было видно, сколько мусора мы генерируем
temp_dir = os.path.join(os.getcwd(), 'TEMP_RAND')
if not os.path.exists(temp_dir):
    os.makedirs(temp_dir)
    logger.info(f"Создана временная директория: {temp_dir}")


def create_temp_file_path(temp_dir):
    """
    Генерирует уникальный путь для временного файла.
    Чисто чтобы не париться с именами.
    """
    return os.path.join(temp_dir, f"zc_tempfile_{uuid.uuid4().hex}.tmp")


def process_chunk_and_write(chunk, temp_dir):
    """
    Обрабатывает чанк: убирает дубли, сортирует и пишет в temp-файл.
    Вот здесь мы делаем грязную работу с данными.
    """
    try:
        logger.debug(f"Обработка чанка размером {len(chunk)} строк...")
        # Никаких тупых вычислений! Только хардкор: set() и sorted()
        unique_items = set(chunk)
        sorted_chunk = sorted(unique_items)

        temp_file_path = create_temp_file_path(temp_dir)
        with open(temp_file_path, 'w', encoding='utf-8') as temp_file:
            # Добавляем '\n' обратно и пишем
            temp_file.write('\n'.join(sorted_chunk) + '\n')

        logger.debug(f"Чанк обработан, записан в {temp_file_path}")
        return temp_file_path, len(chunk), len(unique_items)  # Возвращаем путь и инфу для статистики
    except Exception as e:
        logger.error(f"Ошибка при обработке чанка: {e}")
        # Возвращаем None и нули, чтобы знать, что что-то пошло не так
        return None, len(chunk), 0


def read_and_process_chunks(input_file, temp_dir, chunk_size=2000000, num_workers=None):
    """
    Читает большой файл по чанкам, отправляет их на параллельную обработку.
    Это наш конвейер данных.
    """
    temp_files = []
    chunk = []
    original_count = 0
    initial_unique_count_in_chunks = 0  # Суммируем уникальные в каждом чанке

    # Определяем количество процессов. Если не задано, используем ядра ЦПУ
    if num_workers is None:
        num_workers = os.cpu_count() or 1  # Берем доступные ядра, но минимум 1
        logger.info(f"Количество воркеров не задано, использую {num_workers} по числу ядер ЦПУ.")

    logger.info(f"Начало чтения и обработки файла {input_file} с {num_workers} воркерами...")

    try:
        # Открываем файл с обработкой возможных ошибок кодировки (errors='replace')
        with open(input_file, 'r', encoding='utf-8', errors='replace') as infile:
            # Используем ProcessPoolExecutor для распараллеливания
            with ProcessPoolExecutor(max_workers=num_workers) as executor:
                futures = []

                for line in infile:
                    # Убираем лишние пробелы и переводы строк
                    chunk.append(line.strip())
                    original_count += 1

                    # Отправляем чанк на обработку, когда он достиг нужного размера
                    if len(chunk) >= chunk_size:
                        logger.debug(f"Отправка чанка из {len(chunk)} строк на обработку...")
                        # Отправляем копию чанка, чтобы основной буфер можно было очистить
                        futures.append(executor.submit(process_chunk_and_write, list(chunk), temp_dir))
                        chunk = []  # Очищаем буфер

                # Обрабатываем оставшийся неполный чанк после цикла
                if chunk:
                    logger.debug(f"Отправка последнего чанка из {len(chunk)} строк на обработку...")
                    futures.append(executor.submit(process_chunk_and_write, list(chunk), temp_dir))

                # Собираем результаты по мере готовности
                logger.info("Ожидание завершения обработки чанков...")
                for future in as_completed(futures):
                    temp_file_path, chunk_size_processed, unique_in_chunk = future.result()
                    if temp_file_path:
                        temp_files.append(temp_file_path)
                        initial_unique_count_in_chunks += unique_in_chunk
                    else:
                        # Можно добавить логику повторной обработки или уведомление об ошибке
                        logger.error(f"Обработка одного из чанков провалилась.")

        logger.info(f"Завершено чтение и первичная обработка {original_count} строк.")
        # Возвращаем список временных файлов и общее количество исходных строк
        return temp_files, original_count

    except FileNotFoundError:
        logger.critical(f"Входной файл не найден: {input_file}")
        return [], 0
    except Exception as e:
        logger.critical(f"Критическая ошибка при чтении/обработке файла: {e}")
        # В случае критической ошибки, попытаться очистить созданные temp-файлы
        cleanup_temp_dir(temp_dir)
        return [], 0


def merge_sorted_files(input_files, output_file_path):
    """
    Сливает список отсортированных временных файлов в один отсортированный файл.
    Это сердце алгоритма слияния. Используем heapq.merge - он эффективен.
    """
    logger.info(f"Начало слияния {len(input_files)} файлов в {output_file_path}")

    file_handles = []
    try:
        # Открываем все файлы для чтения
        for file_path in input_files:
            file_handles.append(open(file_path, 'r', encoding='utf-8', errors='replace'))

        # Создаем итератор слияния
        merged_iterator = heapq.merge(*file_handles)

        unique_count = 0
        duplicate_count_in_merge = 0  # Дубли, найденные на этом этапе слияния

        # Открываем выходной файл для записи
        with open(output_file_path, 'w', encoding='utf-8', errors='replace') as outfile:
            prev_line = None
            # Идем по слитому потоку строк
            for line in merged_iterator:
                line = line.strip()  # Снова убираем пробелы/переводы строк

                # Проверяем на дубликат с предыдущей строкой (работает, т.к. поток отсортирован)
                if line and line != prev_line:  # Проверяем, что строка не пустая после strip()
                    outfile.write(line + '\n')  # Пишем уникальную строку
                    prev_line = line
                    unique_count += 1
                elif line:  # Это дубликат, но не пустая строка
                    duplicate_count_in_merge += 1

        logger.info(
            f"Слияние завершено. Уникальных строк записано: {unique_count}, дублей найдено при слиянии: {duplicate_count_in_merge}")
        return output_file_path, unique_count, duplicate_count_in_merge

    except Exception as e:
        logger.error(f"Ошибка при слиянии файлов: {e}")
        # В случае ошибки при слиянии, возвращаем None
        return None, 0, 0
    finally:
        # Всегда закрываем файловые дескрипторы! Это важно!
        for fh in file_handles:
            if not fh.closed:
                fh.close()
        # Удаляем входные файлы после успешного слияния
        for file_path in input_files:
            try:
                if os.path.exists(file_path):
                    os.remove(file_path)
                    logger.debug(f"Удален временный файл после слияния: {file_path}")
            except Exception as e:
                logger.warning(f"Не удалось удалить временный файл {file_path}: {e}")


def iterative_merge_stages(temp_files, temp_dir, batch_size=10, num_merge_workers=None):
    """
    Выполняет многостадийное слияние временных файлов.
    Берет батчи файлов, сливает их в новые, пока не останется один файл.
    Это для случаев, когда количество temp-файлов слишком велико для одного слияния.
    """
    logger.info(f"Начало многостадийного слияния с размером батча {batch_size}")

    current_files_to_merge = list(temp_files)  # Работаем со списком файлов, который будет уменьшаться
    total_duplicates_removed_in_merges = 0  # Суммируем дубликаты, найденные на этапах слияния

    # Определяем количество процессов для слияния
    if num_merge_workers is None:
        num_merge_workers = os.cpu_count() or 1
        logger.info(f"Количество воркеров слияния не задано, использую {num_merge_workers}.")

    # Пока файлов больше, чем может обработать один шаг слияния (т.е. больше одного)
    stage = 1
    while len(current_files_to_merge) > 1:
        logger.info(f"Стадия слияния {stage}: {len(current_files_to_merge)} файлов для обработки.")
        next_stage_files = []  # Список файлов, полученных на этой стадии, для следующей стадии

        # Используем ProcessPoolExecutor для параллельного слияния батчей
        with ProcessPoolExecutor(max_workers=num_merge_workers) as merge_executor:
            futures = []

            # Делим текущий список файлов на батчи
            for i in range(0, len(current_files_to_merge), batch_size):
                batch = current_files_to_merge[i:i + batch_size]
                if not batch:  # Пустой батч? Пропускаем.
                    continue

                output_path = create_temp_file_path(temp_dir)
                logger.debug(f"Отправка батча из {len(batch)} файлов на слияние в {output_path}...")
                # Отправляем задачу слияния батча в пул
                futures.append(merge_executor.submit(merge_sorted_files, batch, output_path))

            # Собираем результаты слияния батчей
            logger.info("Ожидание завершения слияния батчей...")
            for future in as_completed(futures):
                merged_file_path, unique_count, duplicates_in_batch_merge = future.result()
                if merged_file_path:
                    next_stage_files.append(merged_file_path)
                    total_duplicates_removed_in_merges += duplicates_in_batch_merge
                else:
                    # Логика обработки ошибки слияния батча
                    logger.error("Слияние одного из батчей провалилось.")
                    # Потом постараюсь придумать.

        # Список файлов для следующей стадии слияния
        current_files_to_merge = next_stage_files
        stage += 1

    # После всех стадий слияния должен остаться ровно один файл (если все прошло успешно)
    if len(current_files_to_merge) == 1:
        logger.info(f"Многостадийное слияние завершено. Остался один файл: {current_files_to_merge[0]}")
        return current_files_to_merge[0], total_duplicates_removed_in_merges
    elif not current_files_to_merge:
        logger.error("Многостадийное слияние завершилось, но не осталось файлов.")
        return None, total_duplicates_removed_in_merges
    else:
        # Это не должно произойти при batch_size > 1, но всякое бывает
        logger.error(
            f"Многостадийное слияние завершилось с неожиданным количеством файлов: {len(current_files_to_merge)}")
        return None, total_duplicates_removed_in_merges


def cleanup_temp_dir(temp_dir_path):
    """
    Удаляет временную директорию со всем содержимым.
    Очистка следов - признак хорошего тона (и истинного хэкера).
    """
    logger.info(f"Начало очистки временной директории: {temp_dir_path}")
    try:
        if os.path.exists(temp_dir_path):
            shutil.rmtree(temp_dir_path)
            logger.info(f"Временная директория {temp_dir_path} успешно удалена.")
        else:
            logger.warning(f"Временная директория {temp_dir_path} не найдена для удаления.")
    except Exception as e:
        logger.error(f"Ошибка при удалении временной директории {temp_dir_path}: {e}")


def sort_and_uniq_large_file(input_file, output_file, chunk_size=2000000, batch_size=10, num_read_workers=None,
                             num_merge_workers=None):
    """
    Основная функция: координирует чтение, обработку чанков и многостадийное слияние.
    Это дирижер нашего оркестра дедупликации.
    """
    logger.info(f"--- СТАРТ ОБРАБОТКИ ФАЙЛА: {input_file} ---")
    start_time = time.perf_counter()  # Время пошло!

    # Шаг 1: Чтение файла по чанкам, обработка и запись временных файлов
    # Получаем список путей к временным файлам и общее количество строк в оригинале
    temp_files_list, original_line_count = read_and_process_chunks(
        input_file, temp_dir, chunk_size, num_read_workers
    )

    if not temp_files_list:
        logger.critical("Не удалось создать временные файлы обработки. Выполнение прервано.")
        # Попытка очистки на всякий случай
        cleanup_temp_dir(temp_dir)
        return 0, 0  # Возвращаем нули
    logger.info(f"Создано {len(temp_files_list)} временных файлов после первичной обработки.")

    # Шаг 2: Многостадийное слияние временных файлов
    final_merged_file_path, duplicates_removed_in_merges = iterative_merge_stages(
        temp_files_list, temp_dir, batch_size, num_merge_workers
    )

    # Шаг 3: Финализация - переименование последнего файла и подсчет уникальных
    unique_line_count = 0
    if final_merged_file_path and os.path.exists(final_merged_file_path):
        logger.info(f"Финальный файл с уникальными строками готов: {final_merged_file_path}")
        try:
            # Считаем количество уникальных строк в финальном файле
            # Это самый точный способ получить финальное число уникальных
            with open(final_merged_file_path, 'r', encoding='utf-8', errors='replace') as f:
                unique_line_count = sum(1 for _ in f)
            logger.info(f"Подсчитано уникальных строк в финальном файле: {unique_line_count}")

            # Переименовываем финальный временный файл в выходной файл
            if os.path.exists(output_file):
                # Если выходной файл уже существует, удаляем его перед переименованием
                try:
                    os.remove(output_file)
                    logger.warning(f"Удален существующий выходной файл: {output_file}")
                except Exception as e:
                    logger.error(f"Ошибка при удалении существующего выходного файла {output_file}: {e}")


            os.rename(final_merged_file_path, output_file)
            logger.info(f"Финальный результат сохранен в {output_file}")

        except Exception as e:
            logger.critical(f"Ошибка при финализации (подсчете уникальных или переименовании): {e}")
            # В случае ошибки финализации, оставляем временный файл для отладки
            logger.warning(f"Ошибка финализации. Оставляю временный файл {final_merged_file_path} для анализа.")
            # Но удаляем все остальные temp-файлы, которые могли остаться
            # (хотя iterative_merge_stages должна была их удалить)
            remaining_temp_files = [os.path.join(temp_dir, f) for f in os.listdir(temp_dir) if
                                    os.path.isfile(os.path.join(temp_dir, f)) and os.path.join(temp_dir,
                                                                                               f) != final_merged_file_path]
            for rf in remaining_temp_files:
                try:
                    os.remove(rf)
                except Exception as ie:
                    logger.error(f"Ошибка при очистке оставшегося temp-файла {rf}: {ie}")

            # Оставляем пустую временную папку, если она осталась
            if os.path.exists(temp_dir) and not os.listdir(temp_dir):
                try:
                    os.rmdir(temp_dir)  # Удалить только пустую папку
                    logger.info(f"Удалена пустая временная папка {temp_dir}")
                except OSError:  # Не пустая или ошибка
                    pass  # Оставляем

            # Возвращаем нули, так как результат невалиден
            return original_line_count, 0  # Оригинальное количество известно, уникальное - нет

    else:
        logger.critical("Финальный файл слияния не был создан или найден. Что-то пошло очень не так.")
        unique_line_count = 0  # Нет финального файла, нет уникальных строк

    # Шаг 4: Очистка временной директории (если все прошло хорошо)
    # Если финализация прошла успешно, удаляем всю временную директорию
    if final_merged_file_path and os.path.exists(output_file):  # Проверяем, что выходной файл успешно создан
        cleanup_temp_dir(temp_dir)
    else:
        logger.warning("Пропуск автоматической очистки временной директории из-за ошибки финализации.")

    end_time = time.perf_counter()  # Время остановилось!

    # Выводим статистику для пафоса
    logger.info(f"--- ОБРАБОТКА ЗАВЕРШЕНА ---")
    logger.info(f"Исходных строк: {original_line_count}")
    logger.info(f"Уникальных строк (в финальном файле): {unique_line_count}")
    logger.info(f"Дубликатов удалено: {original_line_count - unique_line_count}")  # Самый честный подсчет
    # Можно добавить total_duplicates_removed_in_merges для дебага, но original - unique точнее
    logger.info(f"Общее время выполнения: {end_time - start_time:.2f} секунд")
    logger.info(f"Результат сохранен в: {output_file}")

    return original_line_count, unique_line_count


# --- Главная точка входа (То, что ты будешь вызывать) ---
def remove_duplicates(data, output_filename="output_deduplicated.txt", chunk_size=2000000, batch_size=10,
                      num_read_workers=None, num_merge_workers=None):
    """
    Запускает процесс удаления дубликатов из большого файла.
    Принимает путь к файлу, остальные параметры опциональны.
    """
    input_file = data  # Твой входной файл
    output_file = output_filename  # Куда сохранить результат

    # Проверка существования входного файла перед началом
    if not os.path.exists(input_file):
        logger.critical(f"Ошибка: Входной файл не найден по пути {input_file}")
        return 0, 0  # Возвращаем нули, если файла нет

    return sort_and_uniq_large_file(
        input_file, output_file, chunk_size, batch_size, num_read_workers, num_merge_workers
    )


if __name__ == "__main__":
# Замени 'your_large_file.txt' на путь к твоему 400GB файлу
     input_file_path = 'input.txt'
     output_file_path = 'cleaned_output.txt'

#     # Можешь задать параметры явно, или оставить None для автоопределения
#     # num_processes_read = 16 # Например, если у тебя 16 ядер
#     # num_processes_merge = 8 # Например, если хочешь меньше процессов для слияния

#     # Запускаем маховик дедупликации!
     original_count, unique_count = remove_duplicates(
          input_file_path,
          output_file_path,
          chunk_size=5000000, # Увеличь или уменьши по необходимости
          batch_size=20, # Сколько файлов сливать за раз
          num_read_workers=None, # None = автоопределение
          num_merge_workers=None) # None = автоопределение

     print(f"\n--- Результаты ---")
     print(f"Обработано исходных строк: {original_count}")
     print(f"Найдено уникальных строк: {unique_count}")
#     # print(f"Удалено дубликатов: {original_count - unique_count}")

Zeta · 04.05.2025

SSD, файл 34 гб

rand · 16.05.2025

Zeta сказал(а):

Посмотреть вложение 106698

SSD, файл 34 гб

Приветствую Zeta, хотелось бы видеть сравнение на одном скриншоте двух алгоритмов. Первой и второй версии. =)

Zeta · 16.05.2025

SSD, файл 34 гб

1 версия

2 версия
https://xss.pro/attachments/106698/

rand · 18.05.2025

Zeta сказал(а):

SSD, файл 34 гб
1 версия.
2 версия.

Изображения прогрузились под клир версией, ща вижу что показатели счетчиков разнятся. Файлы разные? Или это у меня аномалия в алгоритме? =\
Походу вторая версия то поточнее должна рассчитывать. Потом проверю.

Статья Merge File Sort или быстрая сортировка строк и удаление дублей интерпретатором в файле от 400 GB за счет ПЗУ

(L3) cache

CPU register

(L3) cache

CooL-Lamer

floppy-диск

HDD-drive

CooL-Lamer

floppy-диск

CooL-Lamer

floppy-диск

CooL-Lamer

CooL-Lamer

CooL-Lamer

HDD-drive

(L3) cache

CooL-Lamer

(L1) cache

CooL-Lamer

(L1) cache

CooL-Lamer