【Python】圧縮解凍入門：zipfile・gzipを使ったZIP/gz操作完全解説

import zipfile
from pathlib import Path

# 圧縮したいファイルを準備
files_to_zip = [
    Path("docs/readme.txt"),
    Path("images/logo.png"),
    Path("data/sample.csv"),
]

# 新規にZIPファイルを作成する例
zip_path = Path("archive/example.zip")

# mode='w' は新規作成(既存があれば上書き)を意味します
with zipfile.ZipFile(zip_path, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
    for file_path in files_to_zip:
        # ZIP内での名前(アーカイブ名)を指定
        # ここでは元のパスのファイル名だけを使います
        arcname = file_path.name
        # ファイルをZIPに追加
        zf.write(file_path, arcname=arcname)
        print(f"Added {file_path} as {arcname}")

print(f"Created ZIP: {zip_path}")

import zipfile
from pathlib import Path

base_dir = Path("project")  # 圧縮したいフォルダ

zip_path = Path("archive/project_backup.zip")

with zipfile.ZipFile(zip_path, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
    for file_path in base_dir.rglob("*"):
        if file_path.is_file():
            # base_dir からの相対パスをZIP内のパスとして使う
            arcname = file_path.relative_to(base_dir)
            zf.write(file_path, arcname=arcname)
            print(f"Added {file_path} as {arcname}")

import zipfile
from pathlib import Path

zip_path = Path("archive/example.zip")
new_files = [
    Path("extra/info.txt"),
    Path("extra/config.json"),
]

# mode='a' で既存のZIPに追記します
with zipfile.ZipFile(zip_path, mode="a", compression=zipfile.ZIP_DEFLATED) as zf:
    for file_path in new_files:
        arcname = f"extra/{file_path.name}"
        zf.write(file_path, arcname=arcname)
        print(f"Appended {file_path} as {arcname}")

print(f"Updated ZIP: {zip_path}")

import zipfile
from pathlib import Path

zip_path = Path("archive/example.zip")
extract_dir = Path("extracted")

# ZIPファイルを読み込みモードで開く
with zipfile.ZipFile(zip_path, mode="r") as zf:
    # 全ファイルをextract_dir配下に展開
    zf.extractall(path=extract_dir)

print(f"Extracted all files to: {extract_dir}")

import zipfile
from pathlib import Path

zip_path = Path("archive/example.zip")
extract_dir = Path("extracted_single")

with zipfile.ZipFile(zip_path, mode="r") as zf:
    # ZIP内の特定ファイル名を指定して抽出
    member_name = "docs/readme.txt"
    zf.extract(member_name, path=extract_dir)
    print(f"Extracted {member_name} to {extract_dir}")

import zipfile
from pathlib import Path

zip_path = Path("protected.zip")
extract_dir = Path("protected_extracted")
password = "secret123"  # 実際には環境変数などから安全に取得する

with zipfile.ZipFile(zip_path, "r") as zf:
    # パスワードはbytesで渡す必要があります
    zf.setpassword(password.encode("utf-8"))
    zf.extractall(path=extract_dir)

print(f"Extracted encrypted ZIP to: {extract_dir}")

import zipfile
from pathlib import Path

files_to_zip = [Path("large_data.bin")]

# 高圧縮(時間はかかるがサイズを優先)
with zipfile.ZipFile(
    "high_compress.zip",
    mode="w",
    compression=zipfile.ZIP_DEFLATED,
    compresslevel=9,  # 最大圧縮
) as zf:
    for path in files_to_zip:
        zf.write(path, arcname=path.name)

# 低圧縮(処理速度を優先)
with zipfile.ZipFile(
    "fast_compress.zip",
    mode="w",
    compression=zipfile.ZIP_DEFLATED,
    compresslevel=1,  # 最小圧縮
) as zf:
    for path in files_to_zip:
        zf.write(path, arcname=path.name)

import zipfile
from datetime import datetime

zip_path = "archive/example.zip"

with zipfile.ZipFile(zip_path, "r") as zf:
    print("=== namelist() ===")
    for name in zf.namelist():
        print(name)

    print("\n=== infolist() ===")
    for info in zf.infolist():
        # info.filename, info.file_size, info.compress_size などが使えます
        dt = datetime(*info.date_time)
        ratio = (
            100 * (1 - info.compress_size / info.file_size)
            if info.file_size
            else 0
        )
        print(
            f"{info.filename:30} "
            f"size={info.file_size:8} "
            f"compressed={info.compress_size:8} "
            f"saved={ratio:5.1f}% "
            f"modified={dt}"
        )

# 悪い例: with文を使わずに明示的にcloseしている
import zipfile

zf = zipfile.ZipFile("example.zip", "w", compression=zipfile.ZIP_DEFLATED)
try:
    zf.write("file1.txt")
    # ここでエラーが起きると、zf.close() が呼ばれない可能性がある
    zf.write("file2.txt")
finally:
    zf.close()

# 良い例: with文を使うことで自動的にcloseされる
import zipfile

with zipfile.ZipFile("example.zip", "w", compression=zipfile.ZIP_DEFLATED) as zf:
    zf.write("file1.txt")
    # ここでエラーが起きても、ブロック終了時に必ずcloseされる
    zf.write("file2.txt")

import gzip

text = "これはgzipで圧縮されたテキストのサンプルです。\n" * 10

# テキストモード('wt')で書き込み
with gzip.open("sample.txt.gz", mode="wt", encoding="utf-8") as f:
    f.write(text)

print("Created sample.txt.gz")

import gzip
from pathlib import Path

src_path = Path("image.png")
dst_path = Path("image.png.gz")

with src_path.open("rb") as src, gzip.open(dst_path, mode="wb") as dst:
    # そのままバイナリコピーすればOK
    dst.write(src.read())

print(f"Compressed to: {dst_path}")

import gzip

text_path = "message.txt.gz"

# 書き込み(圧縮)
with gzip.open(text_path, mode="wt", encoding="utf-8") as f:
    f.write("1行目のテキストです。\n")
    f.write("2行目のテキストです。\n")

# 読み込み(解凍)
with gzip.open(text_path, mode="rt", encoding="utf-8") as f:
    content = f.read()

print("=== content ===")
print(content)

import gzip
from pathlib import Path

src_path = Path("backup.bin")
gz_path = Path("backup.bin.gz")
restored_path = Path("backup_restored.bin")

# 圧縮
with src_path.open("rb") as src, gzip.open(gz_path, mode="wb") as dst:
    # 大きなファイルにも対応できるよう、チャンク単位でコピー
    while True:
        chunk = src.read(1024 * 1024)  # 1MBずつ読む
        if not chunk:
            break
        dst.write(chunk)

# 解凍
with gzip.open(gz_path, mode="rb") as src, restored_path.open("wb") as dst:
    while True:
        chunk = src.read(1024 * 1024)
        if not chunk:
            break
        dst.write(chunk)

print("Compressed and restored backup.bin")

import gzip

def generate_lines():
    # 逐次データを生成するジェネレータの例
    for i in range(1, 6):
        yield f"{i}行目のログです。\n"

gz_path = "stream.log.gz"

with gzip.open(gz_path, mode="wt", encoding="utf-8") as f:
    for line in generate_lines():
        f.write(line)

print(f"Wrote streaming log to {gz_path}")

import zipfile
from pathlib import Path

def zip_directory(src_dir: Path, zip_path: Path, compresslevel: int = 6) -> None:
    """ディレクトリを再帰的に走査してZIPにまとめるユーティリティ関数"""
    with zipfile.ZipFile(
        zip_path,
        mode="w",
        compression=zipfile.ZIP_DEFLATED,
        compresslevel=compresslevel,
    ) as zf:
        for path in src_dir.rglob("*"):
            if path.is_file():
                # src_dirからの相対パスをそのままZIP内パスにする
                arcname = path.relative_to(src_dir)
                zf.write(path, arcname=arcname)

if __name__ == "__main__":
    src = Path("project")
    dst = Path("backup/project_backup.zip")
    dst.parent.mkdir(parents=True, exist_ok=True)
    zip_directory(src, dst, compresslevel=6)
    print(f"Backed up {src} to {dst}")

import gzip
from pathlib import Path
from datetime import datetime, timedelta

LOG_DIR = Path("logs")
DAYS_TO_KEEP_PLAIN = 7  # この日数より古いログを圧縮する

def compress_old_logs():
    cutoff = datetime.now() - timedelta(days=DAYS_TO_KEEP_PLAIN)

    for log_file in LOG_DIR.glob("*.log"):
        # すでにgz済みのファイルは対象外
        gz_file = log_file.with_suffix(log_file.suffix + ".gz")
        if gz_file.exists():
            continue

        # 更新日時が古いものだけ圧縮
        mtime = datetime.fromtimestamp(log_file.stat().st_mtime)
        if mtime > cutoff:
            continue

        print(f"Compressing {log_file} to {gz_file}")

        # 圧縮
        with log_file.open("rb") as src, gzip.open(gz_file, "wb") as dst:
            while True:
                chunk = src.read(1024 * 1024)
                if not chunk:
                    break
                dst.write(chunk)

        # 元のログファイルを削除
        log_file.unlink()

if __name__ == "__main__":
    LOG_DIR.mkdir(exist_ok=True)
    compress_old_logs()

import gzip
from pathlib import Path

CHUNK_SIZE = 50 * 1024 * 1024  # 50MBごとに分割

def split_and_compress(src_path: Path, out_dir: Path):
    out_dir.mkdir(parents=True, exist_ok=True)
    part_index = 1

    with src_path.open("rb") as src:
        while True:
            chunk = src.read(CHUNK_SIZE)
            if not chunk:
                break

            part_path = out_dir / f"{src_path.name}.part{part_index:03d}.gz"
            with gzip.open(part_path, "wb") as dst:
                dst.write(chunk)

            print(f"Wrote {part_path} ({len(chunk)} bytes before compress)")
            part_index += 1

def restore_from_parts(out_dir: Path, base_name: str, dst_path: Path):
    with dst_path.open("wb") as dst:
        part_index = 1
        while True:
            part_path = out_dir / f"{base_name}.part{part_index:03d}.gz"
            if not part_path.exists():
                break

            with gzip.open(part_path, "rb") as src:
                while True:
                    chunk = src.read(1024 * 1024)
                    if not chunk:
                        break
                    dst.write(chunk)

            print(f"Restored from {part_path}")
            part_index += 1

if __name__ == "__main__":
    source = Path("bigfile.dat")
    parts_dir = Path("parts")
    restored = Path("bigfile_restored.dat")

    split_and_compress(source, parts_dir)
    restore_from_parts(parts_dir, source.name, restored)

import io
import zipfile

def create_in_memory_zip(files: dict[str, bytes]) -> bytes:
    """メモリ上にZIPファイルを生成し、そのバイト列を返す"""
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
        for name, data in files.items():
            # writestrでメモリ上のバイト列を直接ZIPに書き込む
            zf.writestr(name, data)
    # バッファの内容をbytesとして取り出す
    return buf.getvalue()

if __name__ == "__main__":
    files = {
        "hello.txt": "こんにちはZIP\n".encode("utf-8"),
        "data.json": b'{"value": 123}',
    }
    zip_bytes = create_in_memory_zip(files)
    print(f"ZIP size in memory: {len(zip_bytes)} bytes")

import io
import gzip

def compress_to_gzip_bytes(data: bytes, compresslevel: int = 6) -> bytes:
    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="wb", compresslevel=compresslevel) as gz:
        gz.write(data)
    return buf.getvalue()

if __name__ == "__main__":
    original = b"A" * 1000
    gz_bytes = compress_to_gzip_bytes(original)
    print(f"Original: {len(original)} bytes, Gzipped: {len(gz_bytes)} bytes")

import zipfile
from pathlib import Path

SRC_DIR = Path("texts")
ZIP_PATH = Path("texts_bundle.zip")

def batch_zip_texts():
    with zipfile.ZipFile(ZIP_PATH, "w", compression=zipfile.ZIP_DEFLATED) as zf:
        for path in SRC_DIR.glob("*.txt"):
            zf.write(path, arcname=path.name)
            print(f"Added {path}")

if __name__ == "__main__":
    SRC_DIR.mkdir(exist_ok=True)
    batch_zip_texts()
    print(f"Created {ZIP_PATH}")

import gzip
from pathlib import Path

SRC_DIR = Path("gz_logs")
DST_DIR = Path("logs_unzipped")

def batch_gunzip():
    DST_DIR.mkdir(parents=True, exist_ok=True)
    for gz_path in SRC_DIR.glob("*.gz"):
        dst_path = DST_DIR / gz_path.with_suffix("").name  # .gzを取り除いた名前
        print(f"Decompressing {gz_path} -> {dst_path}")
        with gzip.open(gz_path, "rb") as src, dst_path.open("wb") as dst:
            while True:
                chunk = src.read(1024 * 1024)
                if not chunk:
                    break
                dst.write(chunk)

if __name__ == "__main__":
    SRC_DIR.mkdir(exist_ok=True)
    batch_gunzip()