Python: как распаковать zip-файл – все методы с примерами кода

import zipfile

# Открытие zip-архива в режиме чтения
with zipfile.ZipFile('example.zip', 'r') as zip_ref:
# Получение списка файлов в архиве
file_list = zip_ref.namelist()
print(f"Файлы в архиве: {file_list}")

# Открытие архива в бинарном режиме
with zipfile.ZipFile('example.zip', 'r', zipfile.ZIP_STORED) as zip_ref:
# Операции с архивом
pass

import zipfile

# Распаковка всех файлов из архива
with zipfile.ZipFile('example.zip', 'r') as zip_ref:
zip_ref.extractall('output_directory')

import zipfile

# Извлечение отдельного файла
with zipfile.ZipFile('example.zip', 'r') as zip_ref:
zip_ref.extract('specific_file.txt', 'output_directory')

import zipfile

files_to_extract = ['file1.txt', 'subfolder/file2.jpg']

with zipfile.ZipFile('example.zip', 'r') as zip_ref:
for file in files_to_extract:
zip_ref.extract(file, 'output_directory')

import zipfile
import os

with zipfile.ZipFile('example.zip', 'r') as zip_ref:
# Получаем список файлов
file_list = zip_ref.namelist()
total_files = len(file_list)

# Распаковываем с выводом прогресса
for index, file in enumerate(file_list, 1):
zip_ref.extract(file, 'output_directory')
print(f"Извлечено {index}/{total_files}: {file}")

import zipfile

# Для архивов с русскими именами файлов, созданных на Windows
with zipfile.ZipFile('russian_files.zip', 'r') as zip_ref:
# CP866 для DOS/Windows архивов, utf-8 для современных систем
zip_ref.extractall('output_directory', path=None, members=None, pwd=None, encoding='cp866')

import zipfile

# Пароль должен быть байтовой строкой
password = b'my_secret_password'

try:
with zipfile.ZipFile('protected.zip', 'r') as zip_ref:
zip_ref.extractall('output_directory', pwd=password)
print("Архив успешно распакован")
except zipfile.BadPasswordException:
print("Неверный пароль для архива")
except zipfile.LargeZipFile:
print("Архив слишком большой для обработки")
except zipfile.BadZipFile:
print("Файл не является zip-архивом или поврежден")

# Преобразование строкового пароля в bytes
string_password = "my_password"
bytes_password = string_password.encode('utf-8') # или другая кодировка при необходимости

import zipfile

passwords = [b'password1', b'12345', b'qwerty', b'admin']
success = False

with zipfile.ZipFile('protected.zip', 'r') as zip_ref:
for password in passwords:
try:
# Пробуем распаковать с текущим паролем
zip_ref.extractall('output_directory', pwd=password)
print(f"Успешная распаковка с паролем: {password.decode()}")
success = True
break
except zipfile.BadPasswordException:
continue

if not success:
print("Не удалось распаковать архив. Все пароли неверны.")

import zipfile
import os

def extract_zip_safely(zip_path, extract_path, password=None):
"""
Безопасно распаковывает ZIP-архив с обработкой всех возможных ошибок

Args:
zip_path (str): Путь к zip-файлу
extract_path (str): Путь для распаковки
password (bytes, optional): Пароль архива

Returns:
bool: True при успешной распаковке, False при ошибке
"""
try:
# Проверяем существование файла
if not os.path.exists(zip_path):
print(f"Ошибка: Файл {zip_path} не найден")
return False

# Проверяем, что файл – zip-архив
if not zipfile.is_zipfile(zip_path):
print(f"Ошибка: {zip_path} не является zip-архивом")
return False

# Создаем директорию для распаковки, если не существует
os.makedirs(extract_path, exist_ok=True)

# Распаковываем архив
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
# Проверяем целостность архива
test_result = zip_ref.testzip()
if test_result:
print(f"Ошибка: Архив поврежден. Проблемный файл: {test_result}")
return False

# Распаковываем файлы
zip_ref.extractall(path=extract_path, pwd=password)
print(f"Архив {zip_path} успешно распакован в {extract_path}")
return True

except zipfile.BadPasswordException:
print("Ошибка: Неверный пароль для защищенного архива")
except zipfile.LargeZipFile:
print("Ошибка: Архив слишком большой для обработки")
except PermissionError:
print(f"Ошибка: Нет прав для записи в {extract_path}")
except Exception as e:
print(f"Непредвиденная ошибка: {str(e)}")

return False

# Пример использования
extract_zip_safely('example.zip', 'output_dir', password=b'password123')

import zipfile
import json

with zipfile.ZipFile('data_archive.zip', 'r') as zip_ref:
# Чтение текстового файла
with zip_ref.open('config.txt') as file:
content = file.read().decode('utf-8')
print(content)

# Чтение и парсинг JSON
with zip_ref.open('data.json') as json_file:
data = json.load(json_file)
print(data)

import zipfile
import io
import csv

with zipfile.ZipFile('large_data.zip', 'r') as zip_ref:
# Получаем CSV файл из архива в память без распаковки на диск
with zip_ref.open('large_dataset.csv') as csv_file:
# Создаем текстовый буфер
text_io = io.TextIOWrapper(csv_file, encoding='utf-8')

# Обрабатываем данные построчно
csv_reader = csv.reader(text_io)
header = next(csv_reader) # Пропускаем заголовок

# Обработка данных
for row in csv_reader:
# Обрабатываем каждую строку без сохранения на диск
print(row[0]) # Например, печатаем первую колонку

import zipfile
import re

def filter_files(filename):
"""Фильтрация файлов по шаблону имени"""
# Например, берем только текстовые файлы из конкретной папки
return re.match(r'data/.*\.txt$', filename) is not None

with zipfile.ZipFile('project_backup.zip', 'r') as zip_ref:
# Получаем все имена файлов
all_files = zip_ref.namelist()

# Фильтруем файлы
files_to_extract = [f for f in all_files if filter_files(f)]

# Распаковываем только отфильтрованные файлы
for file in files_to_extract:
zip_ref.extract(file, 'filtered_output')

import zipfile
from datetime import datetime

with zipfile.ZipFile('archive.zip', 'r') as zip_ref:
# Получаем информацию о всех файлах
for file_info in zip_ref.infolist():
# Преобразуем дату из DOS-формата
date = datetime(*file_info.date_time)

# Выводим метаданные
print(f"Имя: {file_info.filename}")
print(f"Размер: {file_info.file_size} байт")
print(f"Сжатый размер: {file_info.compress_size} байт")
print(f"Коэффициент сжатия: {(1 – file_info.compress_size / file_info.file_size) * 100:.2f}%")
print(f"Дата модификации: {date}")
print(f"Метод сжатия: {file_info.compress_type}")
print("-" * 30)

import zipfile
import os
import concurrent.futures
import time

def extract_file(zip_path, file_name, extract_dir):
"""Извлечение одного файла из архива"""
try:
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
zip_ref.extract(file_name, extract_dir)
return f"Успешно: {file_name}"
except Exception as e:
return f"Ошибка {file_name}: {str(e)}"

def parallel_extract(zip_path, extract_dir, max_workers=4):
"""Параллельная распаковка архива"""
# Создаем директорию для извлечения
os.makedirs(extract_dir, exist_ok=True)

# Получаем список файлов в архиве
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
file_list = zip_ref.namelist()

start_time = time.time()

# Создаем пул потоков для параллельной распаковки
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# Запускаем задачи на распаковку каждого файла
futures = {
executor.submit(extract_file, zip_path, file_name, extract_dir): file_name 
for file_name in file_list
}

# Обрабатываем результаты по мере завершения
for future in concurrent.futures.as_completed(futures):
file_name = futures[future]
try:
result = future.result()
print(result)
except Exception as e:
print(f"Ошибка при распаковке {file_name}: {str(e)}")

end_time = time.time()
print(f"Распаковка завершена за {end_time – start_time:.2f} секунд")

# Пример использования
parallel_extract('large_archive.zip', 'output_directory', max_workers=8)

import os
import time
import zipfile
import logging
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(asctime)s – %(message)s')
logger = logging.getLogger()

class ZipHandler(FileSystemEventHandler):
"""Обработчик событий файловой системы для zip-файлов"""

def __init__(self, extract_path):
self.extract_path = extract_path
# Создаем папку для извлечения, если её нет
os.makedirs(extract_path, exist_ok=True)

def on_created(self, event):
"""Обработка события создания файла"""
if not event.is_directory and event.src_path.endswith('.zip'):
logger.info(f"Обнаружен новый архив: {event.src_path}")
self._extract_archive(event.src_path)

def _extract_archive(self, zip_path):
"""Распаковка архива"""
try:
# Создаем подпапку для извлечения с именем архива
archive_name = os.path.splitext(os.path.basename(zip_path))[0]
extract_dir = os.path.join(self.extract_path, archive_name)
os.makedirs(extract_dir, exist_ok=True)

# Распаковываем архив
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
zip_ref.extractall(extract_dir)

logger.info(f"Архив {zip_path} успешно распакован в {extract_dir}")

except zipfile.BadZipFile:
logger.error(f"{zip_path} не является правильным zip-файлом")
except Exception as e:
logger.error(f"Ошибка при распаковке {zip_path}: {str(e)}")

def start_monitoring(watch_dir, extract_dir):
"""Запуск мониторинга директории"""
event_handler = ZipHandler(extract_dir)
observer = Observer()
observer.schedule(event_handler, watch_dir, recursive=False)

logger.info(f"Начинаем мониторинг директории {watch_dir}")
observer.start()

try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()

observer.join()

# Пример использования
if __name__ == "__main__":
# Директория для мониторинга на новые zip-файлы
WATCH_DIRECTORY = "./incoming"
# Директория для распаковки архивов
EXTRACT_DIRECTORY = "./extracted"

# Убедимся, что директории существуют
os.makedirs(WATCH_DIRECTORY, exist_ok=True)

# Запускаем мониторинг
start_monitoring(WATCH_DIRECTORY, EXTRACT_DIRECTORY)

import zipfile
import os
import json
import csv
import logging
from datetime import datetime

# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(asctime)s – %(message)s')
logger = logging.getLogger()

class DataValidator:
"""Класс для валидации данных из архивов"""

def __init__(self):
self.valid_count = 0
self.invalid_count = 0

def process_archive(self, zip_path, output_dir):
"""Обработка архива с данными"""
try:
if not zipfile.is_zipfile(zip_path):
logger.error(f"{zip_path} не является zip-архивом")
return False

# Создаем директории для выходных данных
os.makedirs(output_dir, exist_ok=True)
valid_dir = os.path.join(output_dir, "valid")
invalid_dir = os.path.join(output_dir, "invalid")
os.makedirs(valid_dir, exist_ok=True)
os.makedirs(invalid_dir, exist_ok=True)

# Сбрасываем счетчики
self.valid_count = 0
self.invalid_count = 0

# Извлекаем и валидируем каждый файл
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
for file_info in zip_ref.infolist():
# Пропускаем директории
if file_info.filename.endswith('/'):
continue

# Определяем тип файла по расширению
file_ext = os.path.splitext(file_info.filename)[1].lower()

if file_ext == '.json':
self._process_json(zip_ref, file_info, valid_dir, invalid_dir)
elif file_ext == '.csv':
self._process_csv(zip_ref, file_info, valid_dir, invalid_dir)
else:
# Просто копируем другие файлы без валидации
zip_ref.extract(file_info, valid_dir)

logger.info(f"Обработка завершена. Валидных: {self.valid_count}, невалидных: {self.invalid_count}")
return True

except Exception as e:
logger.error(f"Ошибка при обработке архива {zip_path}: {str(e)}")
return False

def _process_json(self, zip_ref, file_info, valid_dir, invalid_dir):
"""Обработка и валидация JSON-файлов"""
try:
with zip_ref.open(file_info.filename) as f:
data = json.load(f)

# Пример правила валидации: проверяем наличие обязательных полей
required_fields = ['id', 'name', 'timestamp']

if all(field in data for field in required_fields):
# Данные валидны
output_dir = valid_dir
self.valid_count += 1
else:
# Данные не прошли валидацию
output_dir = invalid_dir
self.invalid_count += 1

# Извлекаем файл в соответствующую директорию
zip_ref.extract(file_info, output_dir)

except json.JSONDecodeError:
logger.warning(f"Невалидный JSON в файле {file_info.filename}")
zip_ref.extract(file_info, invalid_dir)
self.invalid_count += 1
except Exception as e:
logger.error(f"Ошибка при обработке {file_info.filename}: {str(e)}")
zip_ref.extract(file_info, invalid_dir)
self.invalid_count += 1

def _process_csv(self, zip_ref, file_info, valid_dir, invalid_dir):
"""Обработка и валидация CSV-файлов"""
try:
with zip_ref.open(file_info.filename) as f:
# Создаем текстовый wrapper для CSV-ридера
text_f = (line.decode('utf-8') for line in f)
csv_reader = csv.reader(text_f)

# Проверяем заголовок
header = next(csv_reader, None)
if header is None or len(header) < 3: # Минимум 3 столбца
zip_ref.extract(file_info, invalid_dir)
self.invalid_count += 1
else:
# Файл валиден
zip_ref.extract(file_info, valid_dir)
self.valid_count += 1

except Exception as e:
logger.error(f"Ошибка при обработке CSV {file_info.filename}: {str(e)}")
zip_ref.extract(file_info, invalid_dir)
self.invalid_count += 1

# Пример использования
if __name__ == "__main__":
validator = DataValidator()
validator.process_archive("data_submissions.zip", "./processed_data")

import zipfile
import os
import pandas as pd
import glob
from datetime import datetime

def process_reports(archives_dir, output_file):
"""
Обрабатывает все zip-архивы с отчетами в указанной директории
и создает объединенный Excel-отчет с результатами
"""
# Создаем временную директорию для распаковки
temp_dir = os.path.join(archives_dir, "temp_extraction")
os.makedirs(temp_dir, exist_ok=True)

# Находим все zip-архивы в директории
zip_files = glob.glob(os.path.join(archives_dir, "*.zip"))

if not zip_files:
print("Архивы с отчетами не найдены.")
return

all_data = []

for zip_path in zip_files:
archive_name = os.path.basename(zip_path)
print(f"Обработка архива: {archive_name}")

try:
# Распаковываем архив во временную директорию
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
# Ищем Excel-файлы в архиве
excel_files = [f for f in zip_ref.namelist() 
if f.endswith('.xlsx') or f.endswith('.xls')]

if not excel_files:
print(f"Архив {archive_name} не содержит Excel-файлов.")
continue

# Распаковываем только Excel-файлы
for excel_file in excel_files:
zip_ref.extract(excel_file, temp_dir)

# Читаем данные из Excel
try:
df = pd.read_excel(os.path.join(temp_dir, excel_file))

# Добавляем информацию об источнике данных
df['source_archive'] = archive_name
df['source_file'] = excel_file
df['extraction_date'] = datetime.now().strftime('%Y-%m-%d')

all_data.append(df)
except Exception as e:
print(f"Ошибка при чтении {excel_file}: {str(e)}")

# Удаляем распакованный файл
os.remove(os.path.join(temp_dir, excel_file))

except zipfile.BadZipFile:
print(f"Файл {archive_name} не является корректным zip-архивом.")
except Exception as e:
print(f"Ошибка при обработке {archive_name}: {str(e)}")

# Удаляем временную директорию
try:
os.rmdir(temp_dir)
except:
print("Не удалось удалить временную директорию.")

# Если данные найдены, объединяем их и сохраняем
if all_data:
combined_data = pd.concat(all_data, ignore_index=True)
combined_data.to_excel(output_file, index=False)
print(f"Объединенный отчет сохранен в {output_file}")
print(f"Обработано {len(all_data)} отчетов из {len(zip_files)} архивов.")
else:
print("Нет данных для сохранения.")

# Пример использования
if __name__ == "__main__":
process_reports("./monthly_reports", "./combined_report.xlsx")