添加新的表情包

转化.py

@@ -0,0 +1,651 @@
+#!/usr/bin/env python3
+"""
+一键：从 原图/ 读取素材 → 输出到 原图-已处理/（默认）。
+
+- 未编号、格式不符的：转为编号 .webp / .gif，并按最长边缩放（默认 300px）。
+- 已符合 N.webp / N.gif 的：原样复制到 原图-已处理/ 对应子目录，便于整夹迁移。
+
+默认保留 原图 中源文件；需清理可加 --delete-sources。依赖 Pillow。
+"""
+from __future__ import annotations
+
+import argparse
+import io
+import re
+import sys
+import tempfile
+import threading
+import zipfile
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from typing import Callable
+
+try:
+    from PIL import Image, ImageSequence
+except ImportError as exc:  # pragma: no cover
+    raise SystemExit(
+        "需要 Pillow：python -m pip install pillow"
+    ) from exc
+
+IMAGE_EXTENSIONS = {
+    ".avif",
+    ".bmp",
+    ".gif",
+    ".jfif",
+    ".jpeg",
+    ".jpg",
+    ".jpe",
+    ".png",
+    ".tif",
+    ".tiff",
+    ".webp",
+}
+
+OUTPUT_NAME_RE = re.compile(r"^(\d+)\.(webp|gif)$", re.IGNORECASE)
+
+DEFAULT_SOURCE_DIR = Path("原图")
+DEFAULT_OUTPUT_DIR = Path("原图-已处理")
+DEFAULT_MAX_SIDE = 300
+DEFAULT_JOBS = 10
+
+_print_lock = threading.Lock()
+
+
+def _log(*args, file=sys.stdout, **kwargs) -> None:
+    with _print_lock:
+        print(*args, file=file, **kwargs)
+
+
+@dataclass(frozen=True)
+class Job:
+    source_label: str
+    target_dir: Path
+    index: int
+    loader: Callable[[], bytes]
+    source_file: Path | None = None
+    # 若设置：直接写入 target_dir / dest_name（已编号成品复制，不转码、不缩放）
+    dest_name: str | None = None
+
+
+def is_image_path(path: Path) -> bool:
+    return path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS
+
+
+def is_numbered_asset_file(path: Path) -> bool:
+    return path.is_file() and bool(OUTPUT_NAME_RE.match(path.name))
+
+
+def max_index_in_directory(directory: Path) -> int:
+    highest = 0
+    if not directory.is_dir():
+        return 0
+    for p in directory.iterdir():
+        if not p.is_file():
+            continue
+        m = OUTPUT_NAME_RE.match(p.name)
+        if m:
+            highest = max(highest, int(m.group(1)))
+    return highest
+
+
+def encode_to_webp_bytes(data: bytes) -> tuple[bytes, bool]:
+    with Image.open(io.BytesIO(data)) as im:
+        is_animated = bool(
+            getattr(im, "is_animated", False) or getattr(im, "n_frames", 1) > 1
+        )
+
+        if is_animated:
+            frames = []
+            durations = []
+            loop = im.info.get("loop", 0)
+            for frame in ImageSequence.Iterator(im):
+                frames.append(frame.convert("RGBA"))
+                durations.append(
+                    frame.info.get("duration", im.info.get("duration", 100))
+                )
+
+            output = io.BytesIO()
+            frames[0].save(
+                output,
+                format="WEBP",
+                save_all=True,
+                append_images=frames[1:],
+                duration=durations,
+                loop=loop,
+                lossless=True,
+                method=4,
+            )
+            return output.getvalue(), True
+
+        output = io.BytesIO()
+        im.convert("RGBA").save(output, format="WEBP", lossless=True, method=4)
+        return output.getvalue(), False
+
+
+def is_animated_gif(data: bytes) -> bool:
+    with Image.open(io.BytesIO(data)) as im:
+        if im.format != "GIF":
+            return False
+        n_frames = getattr(im, "n_frames", 1)
+        return bool(getattr(im, "is_animated", False) or n_frames > 1)
+
+
+def is_animated_webp_bytes(data: bytes) -> bool:
+    with Image.open(io.BytesIO(data)) as im:
+        if im.format != "WEBP":
+            return False
+        return bool(
+            getattr(im, "is_animated", False) or getattr(im, "n_frames", 1) > 1
+        )
+
+
+def save_animated_webp_image_to_gif(im: Image.Image, dst: Path) -> None:
+    if im.format != "WEBP":
+        raise ValueError("expected WebP image")
+    loop = im.info.get("loop", 0)
+    frames: list[Image.Image] = []
+    durations: list[int] = []
+    for frame in ImageSequence.Iterator(im):
+        rgba = frame.convert("RGBA")
+        frames.append(rgba)
+        d = frame.info.get(
+            "duration", im.info.get("duration", im.info.get("gif_duration", 100))
+        )
+        try:
+            durations.append(int(d) if d is not None else 100)
+        except (TypeError, ValueError):
+            durations.append(100)
+
+    if len(frames) == 0:
+        raise ValueError("no frames")
+    if len(durations) < len(frames):
+        durations.extend([durations[-1]] * (len(frames) - len(durations)))
+
+    dst.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(
+        suffix=".gif", delete=False, dir=dst.parent
+    ) as tmp:
+        tmp_path = Path(tmp.name)
+    try:
+        if len(frames) == 1:
+            frames[0].save(
+                tmp_path,
+                format="GIF",
+                duration=durations[0],
+                loop=loop,
+            )
+        else:
+            frames[0].save(
+                tmp_path,
+                format="GIF",
+                save_all=True,
+                append_images=frames[1:],
+                duration=durations,
+                loop=loop,
+                disposal=2,
+            )
+        tmp_path.replace(dst)
+    finally:
+        if tmp_path.exists():
+            try:
+                tmp_path.unlink()
+            except OSError:
+                pass
+
+
+def webp_bytes_to_gif(data: bytes, dst: Path) -> None:
+    with Image.open(io.BytesIO(data)) as im:
+        save_animated_webp_image_to_gif(im, dst)
+
+
+def _read_zip_member(zip_path: Path, member_name: str) -> bytes:
+    with zipfile.ZipFile(zip_path) as zf:
+        return zf.read(member_name)
+
+
+def collect_directory_jobs(
+    source_root: Path, output_root: Path, *, delete_sources: bool
+) -> list[Job]:
+    grouped: dict[Path, list[Path]] = {}
+    for path in sorted(source_root.rglob("*")):
+        if not is_image_path(path):
+            continue
+        grouped.setdefault(path.parent, []).append(path)
+
+    jobs: list[Job] = []
+    for directory in sorted(grouped, key=lambda p: str(p)):
+        files = sorted(grouped[directory], key=lambda p: p.name)
+        target_dir = output_root / directory.relative_to(source_root)
+
+        conforming = [p for p in files if is_numbered_asset_file(p)]
+        non_conforming = [p for p in files if not is_numbered_asset_file(p)]
+
+        max_src = 0
+        for p in conforming:
+            m = OUTPUT_NAME_RE.match(p.name)
+            if m:
+                max_src = max(max_src, int(m.group(1)))
+
+        max_t = max_index_in_directory(target_dir)
+        next_i = max(max_src, max_t) + 1
+
+        for path in conforming:
+            jobs.append(
+                Job(
+                    source_label=path.relative_to(source_root).as_posix(),
+                    target_dir=target_dir,
+                    index=0,
+                    loader=lambda p=path: p.read_bytes(),
+                    source_file=path if delete_sources else None,
+                    dest_name=path.name,
+                )
+            )
+
+        for path in non_conforming:
+            jobs.append(
+                Job(
+                    source_label=path.relative_to(source_root).as_posix(),
+                    target_dir=target_dir,
+                    index=next_i,
+                    loader=lambda p=path: p.read_bytes(),
+                    source_file=path if delete_sources else None,
+                    dest_name=None,
+                )
+            )
+            next_i += 1
+    return jobs
+
+
+def collect_zip_jobs(zip_path: Path, output_root: Path) -> list[Job]:
+    grouped: dict[PurePosixPath, list[PurePosixPath]] = {}
+    with zipfile.ZipFile(zip_path) as zf:
+        for info in zf.infolist():
+            if info.is_dir():
+                continue
+            member_path = PurePosixPath(info.filename)
+            suf = member_path.suffix.lower()
+            if suf not in IMAGE_EXTENSIONS:
+                continue
+            if OUTPUT_NAME_RE.match(member_path.name):
+                continue
+            grouped.setdefault(member_path.parent, []).append(member_path)
+
+    jobs: list[Job] = []
+    for directory in sorted(grouped, key=lambda p: str(p)):
+        files = sorted(grouped[directory], key=lambda p: str(p.name))
+        target_dir = output_root / Path(str(directory))
+        next_i = max_index_in_directory(target_dir) + 1
+        for member_path in files:
+            jobs.append(
+                Job(
+                    source_label=member_path.as_posix(),
+                    target_dir=target_dir,
+                    index=next_i,
+                    loader=lambda name=member_path.as_posix(), zp=zip_path: _read_zip_member(
+                        zp, name
+                    ),
+                    source_file=None,
+                )
+            )
+            next_i += 1
+    return jobs
+
+
+def _remove_source_if_any(job: Job, out: Path, kind: str) -> str:
+    if job.source_file is None:
+        return kind
+    try:
+        src = job.source_file.resolve()
+        if src == out.resolve():
+            return kind
+        src.unlink()
+        return f"{kind}; removed-source"
+    except OSError as exc:
+        _log(
+            f"[warn] ok: {out} ({kind}) but could not delete {job.source_file}: {exc}",
+            file=sys.stderr,
+        )
+        return kind
+
+
+# --- 最长边缩放（多帧仅 GIF/WebP）---
+
+
+def _is_multi_frame(im: Image.Image) -> bool:
+    return bool(
+        getattr(im, "is_animated", False) or getattr(im, "n_frames", 1) > 1
+    )
+
+
+def _target_size(w: int, h: int, max_side: int) -> tuple[int, int] | None:
+    if max(w, h) <= max_side:
+        return None
+    if w >= h:
+        nw = max_side
+        nh = max(round(h * max_side / w), 1)
+    else:
+        nh = max_side
+        nw = max(round(w * max_side / h), 1)
+    return nw, nh
+
+
+def _resize_rgba(im: Image.Image, size: tuple[int, int]) -> Image.Image:
+    return im.convert("RGBA").resize(size, Image.Resampling.LANCZOS)
+
+
+def _static_save_format(path: Path, im: Image.Image) -> tuple[str, dict]:
+    ext = path.suffix.lower()
+    fmt = (im.format or "").upper()
+    if ext in (".jpg", ".jpeg", ".jpe") or fmt in ("JPEG", "MPO"):
+        return "JPEG", {"quality": 90, "optimize": True}
+    if ext == ".png" or fmt == "PNG":
+        return "PNG", {"optimize": True}
+    if ext == ".webp" or fmt == "WEBP":
+        return "WEBP", {"quality": 85, "method": 6}
+    if ext == ".gif" or fmt == "GIF":
+        return "GIF", {}
+    if ext in (".bmp",) or fmt == "BMP":
+        return "BMP", {}
+    if ext in (".tif", ".tiff") or fmt == "TIFF":
+        return "TIFF", {"compression": "tiff_lzw"}
+    if ext == ".avif" or fmt == "AVIF":
+        return "AVIF", {"quality": 80}
+    if ext == ".webp":
+        return "WEBP", {"quality": 85, "method": 6}
+    return "PNG", {"optimize": True}
+
+
+def _atomic_save(path: Path, saver) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(
+        suffix=path.suffix, delete=False, dir=path.parent
+    ) as tmp:
+        tmp_path = Path(tmp.name)
+    try:
+        saver(tmp_path)
+        tmp_path.replace(path)
+    finally:
+        if tmp_path.exists():
+            try:
+                tmp_path.unlink()
+            except OSError:
+                pass
+
+
+def _resize_static(im: Image.Image, path: Path, max_side: int) -> bool:
+    w, h = im.size
+    th = _target_size(w, h, max_side)
+    if th is None:
+        return False
+    out = im.resize(th, Image.Resampling.LANCZOS)
+    fmt, kwargs = _static_save_format(path, im)
+    if fmt == "JPEG" and out.mode in ("RGBA", "P"):
+        out = out.convert("RGB")
+    elif fmt == "JPEG" and out.mode not in ("RGB", "L"):
+        out = out.convert("RGB")
+    _atomic_save(path, lambda p: out.save(p, format=fmt, **kwargs))
+    return True
+
+
+def _resize_animated(im: Image.Image, path: Path, max_side: int) -> bool:
+    w, h = im.size
+    th = _target_size(w, h, max_side)
+    if th is None:
+        return False
+
+    loop = int(im.info.get("loop", 0) or 0)
+    frames: list[Image.Image] = []
+    durations: list[int] = []
+    src_format = (im.format or "").upper()
+
+    for frame in ImageSequence.Iterator(im):
+        rgba = _resize_rgba(frame, th)
+        frames.append(rgba)
+        d = frame.info.get("duration", im.info.get("duration", 100))
+        try:
+            durations.append(int(d) if d is not None else 100)
+        except (TypeError, ValueError):
+            durations.append(100)
+
+    if len(frames) == 0:
+        return False
+    if len(durations) < len(frames):
+        durations.extend([durations[-1]] * (len(frames) - len(durations)))
+
+    def save_gif(p: Path) -> None:
+        if len(frames) == 1:
+            frames[0].save(p, format="GIF", duration=durations[0], loop=loop)
+        else:
+            frames[0].save(
+                p,
+                format="GIF",
+                save_all=True,
+                append_images=frames[1:],
+                duration=durations,
+                loop=loop,
+                disposal=2,
+            )
+
+    def save_webp(p: Path) -> None:
+        frames[0].save(
+            p,
+            format="WEBP",
+            save_all=True,
+            append_images=frames[1:],
+            duration=durations,
+            loop=loop,
+            quality=85,
+            method=6,
+        )
+
+    if src_format == "GIF" or path.suffix.lower() == ".gif":
+        _atomic_save(path, save_gif)
+    else:
+        _atomic_save(path, save_webp)
+
+    return True
+
+
+def resize_output_if_needed(path: Path, max_side: int) -> None:
+    if max_side < 1:
+        return
+    try:
+        with Image.open(path) as im:
+            im.load()
+            if _is_multi_frame(im):
+                fmt = (im.format or "").upper()
+                ext = path.suffix.lower()
+                if fmt not in ("GIF", "WEBP") and ext not in (".gif", ".webp"):
+                    _log(
+                        f"skip resize animated (only GIF/WebP): {path}",
+                        file=sys.stderr,
+                    )
+                    return
+                changed = _resize_animated(im, path, max_side)
+            else:
+                changed = _resize_static(im, path, max_side)
+            if changed:
+                _log(f"resized {path} (max side {max_side})")
+    except OSError as e:
+        _log(f"skip resize {path}: {e}", file=sys.stderr)
+
+
+def run_job(job: Job, max_side: int) -> tuple[bool, str, Path, str | None, str | None]:
+    try:
+        data = job.loader()
+        job.target_dir.mkdir(parents=True, exist_ok=True)
+
+        if job.dest_name is not None:
+            out = job.target_dir / job.dest_name
+            out.write_bytes(data)
+            kind = _remove_source_if_any(job, out, "copy")
+            return True, job.source_label, out, kind, None
+
+        if is_animated_gif(data):
+            out = job.target_dir / f"{job.index}.gif"
+            out.write_bytes(data)
+            kind = _remove_source_if_any(job, out, "gif-animated")
+            resize_output_if_needed(out, max_side)
+            return True, job.source_label, out, kind, None
+
+        if is_animated_webp_bytes(data):
+            out = job.target_dir / f"{job.index}.gif"
+            webp_bytes_to_gif(data, out)
+            kind = _remove_source_if_any(job, out, "animated-webp->gif")
+            resize_output_if_needed(out, max_side)
+            return True, job.source_label, out, kind, None
+
+        webp_data, animated = encode_to_webp_bytes(data)
+        out = job.target_dir / f"{job.index}.webp"
+        out.write_bytes(webp_data)
+        base = "animated-webp" if animated else "static-webp"
+        kind = _remove_source_if_any(job, out, base)
+        resize_output_if_needed(out, max_side)
+        return True, job.source_label, out, kind, None
+    except Exception as exc:
+        return False, job.source_label, job.target_dir / f"{job.index}.?", None, str(exc)
+
+
+def main(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "原图/ 下：未编号素材转为编号 .webp/.gif 并限最长边；"
+            "已符合 N.webp/N.gif 的复制到 原图-已处理/ 同结构。"
+            "默认读取 原图/、保留源文件。"
+        )
+    )
+    parser.add_argument(
+        "source",
+        nargs="?",
+        default=None,
+        type=Path,
+        help=f"源：目录、单图或 .zip（默认: {DEFAULT_SOURCE_DIR.as_posix()}/）",
+    )
+    parser.add_argument(
+        "-o",
+        "--output",
+        default=None,
+        dest="output_root",
+        type=Path,
+        help=f"输出根目录（默认: 目录模式为 {DEFAULT_OUTPUT_DIR.as_posix()}/；"
+        "zip 为 zip 同目录下以压缩包名为文件夹；单图为该目录）",
+    )
+    parser.add_argument(
+        "--max",
+        type=int,
+        default=DEFAULT_MAX_SIDE,
+        metavar="PX",
+        help=f"最长边像素上限，超过则缩小（默认 {DEFAULT_MAX_SIDE}；0 表示不缩放）",
+    )
+    parser.add_argument(
+        "-j",
+        "--jobs",
+        type=int,
+        default=DEFAULT_JOBS,
+        help=f"线程数（默认 {DEFAULT_JOBS}）",
+    )
+    parser.add_argument(
+        "--delete-sources",
+        action="store_true",
+        help="成功后删除 原图 中已处理的源文件（含已编号复制项；默认保留）",
+    )
+    args = parser.parse_args(argv)
+
+    cwd = Path.cwd()
+    source = (
+        args.source.expanduser().resolve()
+        if args.source is not None
+        else (cwd / DEFAULT_SOURCE_DIR).resolve()
+    )
+    jobs: list[Job] = []
+    max_side: int = args.max
+
+    if source.is_dir():
+        out_root = (
+            args.output_root.expanduser().resolve()
+            if args.output_root is not None
+            else (cwd / DEFAULT_OUTPUT_DIR).resolve()
+        )
+        out_root.mkdir(parents=True, exist_ok=True)
+        jobs = collect_directory_jobs(
+            source, out_root, delete_sources=args.delete_sources
+        )
+    elif source.is_file() and source.suffix.lower() == ".zip":
+        out_root = (
+            args.output_root.expanduser().resolve()
+            if args.output_root is not None
+            else (source.parent / source.stem).resolve()
+        )
+        out_root.mkdir(parents=True, exist_ok=True)
+        jobs = collect_zip_jobs(source, out_root)
+    elif is_image_path(source):
+        target_dir = (
+            args.output_root.expanduser().resolve()
+            if args.output_root is not None
+            else (cwd / DEFAULT_OUTPUT_DIR).resolve()
+        )
+        target_dir.mkdir(parents=True, exist_ok=True)
+        if is_numbered_asset_file(source):
+            jobs = [
+                Job(
+                    source_label=source.name,
+                    target_dir=target_dir,
+                    index=0,
+                    loader=lambda p=source: p.read_bytes(),
+                    source_file=source if args.delete_sources else None,
+                    dest_name=source.name,
+                )
+            ]
+        else:
+            idx = max_index_in_directory(target_dir) + 1
+            jobs = [
+                Job(
+                    source_label=source.name,
+                    target_dir=target_dir,
+                    index=idx,
+                    loader=lambda p=source: p.read_bytes(),
+                    source_file=source if args.delete_sources else None,
+                )
+            ]
+    else:
+        print(f"不是目录、图片或 zip: {source}", file=sys.stderr)
+        return 1
+
+    jobs.sort(
+        key=lambda job: (
+            str(job.target_dir),
+            (job.dest_name or "").lower(),
+            f"{job.index:08d}",
+        )
+    )
+
+    if not jobs:
+        print(f"没有待处理文件: {source}")
+        return 0
+
+    converted = 0
+    copied = 0
+    skipped = 0
+
+    with ThreadPoolExecutor(max_workers=max(1, args.jobs)) as pool:
+        for ok, label, target, kind, error in pool.map(
+            lambda j: run_job(j, max_side), jobs
+        ):
+            if ok:
+                print(f"{label} -> {target} ({kind})")
+                base_kind = (kind or "").split(";")[0].strip()
+                if base_kind == "copy":
+                    copied += 1
+                else:
+                    converted += 1
+            else:
+                print(f"[skip] {label}: {error}", file=sys.stderr)
+                skipped += 1
+
+    print(f"完成：转换={converted}, 复制={copied}, 失败={skipped}")
+    return 0 if (converted + copied) > 0 else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))