Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import json
- from collections import defaultdict
- import sys
- def compress_data(data: bytes):
- """
- Рекурсивно сжимает входные данные, заменяя часто встречающиеся пары.
- Возвращает:
- - tokens: финальная последовательность токенов (целые числа)
- - dictionary: словарь замен вида {новый_токен: (токен1, токен2)}
- """
- # Представляем данные как список целых чисел (0..255)
- tokens = list(data)
- dictionary = {}
- next_token = 256 # новые токены начинаем с 256
- tokens, dictionary, _ = _compress_tokens(tokens, dictionary, next_token)
- return tokens, dictionary
- def _compress_tokens(tokens, dictionary, next_token):
- """
- Рекурсивная функция, которая:
- 1. Сканирует список токенов для подсчёта пар (неперекрывающих).
- 2. Если найдена пара, встречающаяся ≥2 раза, заменяет все её вхождения новым токеном,
- обновляет словарь и продолжает рекурсию.
- 3. Если замена не даёт эффекта, возвращает текущую последовательность.
- """
- pair_counts = defaultdict(int)
- # Подсчёт всех (неперекрывающих) пар
- for i in range(len(tokens) - 1):
- pair = (tokens[i], tokens[i+1])
- pair_counts[pair] += 1
- if not pair_counts:
- return tokens, dictionary, next_token
- # Выбираем пару с максимальной частотой
- best_pair, count = max(pair_counts.items(), key=lambda x: x[1])
- # Если ни одна пара не встречается дважды – сжатие закончено
- if count < 2:
- return tokens, dictionary, next_token
- # Новый токен для замены найденной пары
- new_token = next_token
- next_token += 1
- dictionary[new_token] = best_pair
- # Замена всех неперекрывающих вхождений best_pair
- new_tokens = []
- i = 0
- while i < len(tokens):
- if i < len(tokens) - 1 and (tokens[i], tokens[i+1]) == best_pair:
- new_tokens.append(new_token)
- i += 2 # пропускаем оба элемента пары
- else:
- new_tokens.append(tokens[i])
- i += 1
- # Рекурсивно продолжаем сжатие с обновлённой последовательностью токенов
- return _compress_tokens(new_tokens, dictionary, next_token)
- def decompress_data(tokens, dictionary):
- """
- Рекурсивно восстанавливает исходные данные из сжатой последовательности токенов.
- Каждый токен, который не является исходным байтом (0..255), разворачивается по словарю.
- """
- def expand(token):
- if token < 256:
- return [token]
- else:
- left, right = dictionary[token]
- return expand(left) + expand(right)
- result = []
- for token in tokens:
- result.extend(expand(token))
- return bytes(result)
- def save_compressed(filename, tokens, dictionary):
- """
- Сохраняет сжатые данные в файл в формате JSON.
- Словарь преобразуется так, чтобы ключи стали строками.
- """
- dict_to_save = {str(k): v for k, v in dictionary.items()}
- data = {
- "tokens": tokens,
- "dictionary": dict_to_save
- }
- with open(filename, 'w') as f:
- json.dump(data, f)
- def load_compressed(filename):
- """
- Загружает сжатые данные из JSON-файла.
- Преобразует ключи словаря обратно в числа.
- """
- with open(filename, 'r') as f:
- data = json.load(f)
- dictionary = {int(k): tuple(v) for k, v in data["dictionary"].items()}
- tokens = data["tokens"]
- return tokens, dictionary
- def print_usage():
- print("Usage:")
- print(" python recursive_compress.py c input_file # для сжатия")
- print(" python recursive_compress.py d input_file # для разжатия")
- sys.exit(1)
- if __name__ == "__main__":
- if len(sys.argv) < 3:
- print_usage()
- mode = sys.argv[1]
- filename = sys.argv[2]
- if mode == "c":
- # Сжатие файла
- try:
- with open(filename, 'rb') as f:
- data = f.read()
- except Exception as e:
- print(f"Ошибка при чтении файла: {e}")
- sys.exit(1)
- tokens, dictionary = compress_data(data)
- out_filename = filename + ".rc" # расширение для сжатых файлов
- try:
- save_compressed(out_filename, tokens, dictionary)
- except Exception as e:
- print(f"Ошибка при сохранении сжатых данных: {e}")
- sys.exit(1)
- print(f"Файл сжат и сохранён как {out_filename}")
- elif mode == "d":
- # Декомпрессия файла
- try:
- tokens, dictionary = load_compressed(filename)
- except Exception as e:
- print(f"Ошибка при загрузке сжатых данных: {e}")
- sys.exit(1)
- try:
- data = decompress_data(tokens, dictionary)
- except Exception as e:
- print(f"Ошибка при декомпрессии: {e}")
- sys.exit(1)
- # Если расширение .rc присутствует – убираем его
- if filename.endswith(".rc"):
- out_filename = filename[:-3]
- else:
- out_filename = filename + ".dec"
- try:
- with open(out_filename, 'wb') as f:
- f.write(data)
- except Exception as e:
- print(f"Ошибка при записи декомпрессированного файла: {e}")
- sys.exit(1)
- print(f"Файл декомпрессирован и сохранён как {out_filename}")
- else:
- print_usage()
Advertisement
Add Comment
Please, Sign In to add comment