예제 소스

chart_png.py

"""차트 PNG 경로: ``chart/{bucket:04d}/{bar:07d}.png``.

한 폴더에 파일이 너무 많으면 macOS fseventsd/Spotlight 가 CPU 를 먹는다.
bucket = bar_index // 10000 이라 폴더당 최대 10000장이다.
"""

from __future__ import annotations

import os
from collections.abc import Iterator
from pathlib import Path

CHART_BUCKET_SIZE = 10_000


def chart_bucket(bar: int) -> int:
    return int(bar) // CHART_BUCKET_SIZE


def chart_png_path(chart_dir: Path | str, bar: int) -> Path:
    bar = int(bar)
    return Path(chart_dir) / f"{chart_bucket(bar):04d}" / f"{bar:07d}.png"


def iter_chart_pngs(chart_dir: Path | str) -> Iterator[Path]:
    root = Path(chart_dir)
    if not root.is_dir():
        return
    with os.scandir(root) as it:
        for entry in it:
            if entry.is_file() and entry.name.endswith(".png"):
                stem = entry.name[:-4]
                if stem.isdigit():
                    yield Path(entry.path)
                continue
            if not entry.is_dir() or not entry.name.isdigit():
                continue
            with os.scandir(entry.path) as nested:
                for child in nested:
                    if not child.is_file() or not child.name.endswith(".png"):
                        continue
                    stem = child.name[:-4]
                    if stem.isdigit():
                        yield Path(child.path)


def existing_chart_bars(chart_dir: Path | str) -> set[int]:
    have: set[int] = set()
    for path in iter_chart_pngs(chart_dir):
        try:
            if path.stat().st_size > 0:
                have.add(int(path.stem))
        except OSError:
            continue
    return have


def migrate_flat_chart_dir(chart_dir: Path | str, *, log=None) -> tuple[int, int]:
    """루트의 ``NNNNNNN.png`` 를 버킷 폴더로 옮긴다. 이미 버킷에 있으면 건너뛴다."""
    root = Path(chart_dir)
    if not root.is_dir():
        return 0, 0
    names: list[str] = []
    with os.scandir(root) as it:
        for entry in it:
            if entry.is_file() and entry.name.endswith(".png") and entry.name[:-4].isdigit():
                names.append(entry.name)
    if log is not None:
        log("flat png %s장 목록 확보", len(names))
    moved = 0
    skipped = 0
    made: set[str] = set()
    for n, name in enumerate(names, start=1):
        dest = chart_png_path(root, int(name[:-4]))
        parent = str(dest.parent)
        if parent not in made:
            dest.parent.mkdir(parents=True, exist_ok=True)
            made.add(parent)
        src = os.path.join(str(root), name)
        if os.path.lexists(dest):
            skipped += 1
            continue
        os.rename(src, dest)
        moved += 1
        if log is not None and (moved == 1 or moved % 1000 == 0):
            log("moved %s / scanned %s → %s", moved, n, dest)
    return moved, skipped