diff --git a/.gitignore b/.gitignore index bee3ba3..30516fc 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ +dist/ node_modules/ .wrangler/ .dev.vars diff --git a/animated_webp_to_gif.py b/animated_webp_to_gif.py deleted file mode 100644 index 4c5dcf2..0000000 --- a/animated_webp_to_gif.py +++ /dev/null @@ -1,212 +0,0 @@ -#!/usr/bin/env python3 -""" -Recursively find *.webp under 原图/ (by default). Animated WebP -> same-stem .gif -(then remove the .webp). Static WebP unchanged. Default 10 worker threads. -""" -from __future__ import annotations - -import argparse -import io -import re -import sys -import tempfile -import threading -from concurrent.futures import ThreadPoolExecutor -from pathlib import Path - -try: - from PIL import Image, ImageSequence -except ImportError as exc: # pragma: no cover - raise SystemExit( - "This script requires Pillow. Install it with: python3 -m pip install pillow" - ) from exc - - -# 只处理已编号的成品动图（避免误处理未编号的素材） -OUTPUT_NAME_RE = re.compile(r"^(\d+)\.webp$", re.IGNORECASE) - -DEFAULT_JOBS = 10 -_log_lock = threading.Lock() - - -def _log(*args, file=sys.stdout, **kwargs) -> None: - with _log_lock: - print(*args, file=file, **kwargs) - - -def convert_one(path: Path, root: Path, dry_run: bool) -> str: - try: - if not is_animated_webp(path): - return "skip_static" - except OSError as e: - _log(f"[skip] {path}: {e}", file=sys.stderr) - return "error" - gif_path = path.with_suffix(".gif") - rel = path.relative_to(root) - if dry_run: - _log(f"would convert: {rel} -> {gif_path.name}") - return "ok" - try: - webp_to_gif(path, gif_path) - path.unlink() - _log(f"{rel} -> {gif_path.name} (animated)") - return "ok" - except Exception as e: - _log(f"[error] {path}: {e}", file=sys.stderr) - if gif_path.exists(): - try: - gif_path.unlink() - except OSError: - pass - return "error" - - -def is_animated_webp_image(im: Image.Image) -> bool: - if im.format != "WEBP": - return False - return bool( - getattr(im, "is_animated", False) or getattr(im, "n_frames", 1) > 1 - ) - - -def is_animated_webp(path: Path) -> bool: - with Image.open(path) as im: - return is_animated_webp_image(im) - - -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 webp_to_gif(src: Path, dst: Path) -> None: - with Image.open(src) as im: - save_animated_webp_image_to_gif(im, dst) - - -def main(argv: list[str]) -> int: - parser = argparse.ArgumentParser( - description=( - "Convert animated WebP under 原图/ to GIF (in place); " - "static WebP unchanged. Only numbered N.webp files are considered." - ) - ) - parser.add_argument( - "root", - nargs="?", - default="原图", - type=Path, - help="Root directory to scan (default: 原图).", - ) - parser.add_argument( - "--dry-run", - action="store_true", - help="Only print what would be converted.", - ) - parser.add_argument( - "--all-webp", - action="store_true", - help="Process any *.webp, not only numbered N.webp.", - ) - parser.add_argument( - "-j", - "--jobs", - type=int, - default=DEFAULT_JOBS, - metavar="N", - help=f"Parallel worker threads (default: {DEFAULT_JOBS}).", - ) - args = parser.parse_args(argv) - root = args.root.expanduser().resolve() - if not root.is_dir(): - print(f"Not a directory: {root}", file=sys.stderr) - return 1 - - webp_files = sorted(root.rglob("*.webp")) - converted = 0 - skipped_static = 0 - skipped_name = 0 - errors = 0 - - candidates: list[Path] = [] - for path in webp_files: - if not args.all_webp and not OUTPUT_NAME_RE.match(path.name): - skipped_name += 1 - continue - candidates.append(path) - - workers = max(1, args.jobs) - with ThreadPoolExecutor(max_workers=workers) as pool: - for outcome in pool.map( - lambda p: convert_one(p, root, args.dry_run), candidates - ): - if outcome == "ok": - converted += 1 - elif outcome == "skip_static": - skipped_static += 1 - else: - errors += 1 - - print( - f"Done. converted={converted}, skipped_static_webp={skipped_static}, " - f"skipped_unnumbered={skipped_name}, errors={errors} ({workers} workers)" - ) - return 0 if errors == 0 else 1 - - -if __name__ == "__main__": - raise SystemExit(main(sys.argv[1:])) diff --git a/convert_to_webp.py b/convert_to_webp.py deleted file mode 100644 index 9c3b3b4..0000000 --- a/convert_to_webp.py +++ /dev/null @@ -1,348 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import argparse -import io -import re -import sys -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( - "This script requires Pillow. Install it with: python3 -m pip install pillow" - ) from exc - -from animated_webp_to_gif import webp_bytes_to_gif - -IMAGE_EXTENSIONS = { - ".avif", - ".bmp", - ".gif", - ".jfif", - ".jpeg", - ".jpg", - ".jpe", - ".png", - ".tif", - ".tiff", - ".webp", -} - -# 已编号的成品：1.webp、42.gif（不参与转换；用于各子文件夹内续号） -OUTPUT_NAME_RE = re.compile(r"^(\d+)\.(webp|gif)$", re.IGNORECASE) - -DEFAULT_SOURCE_DIR = Path("原图") - - -@dataclass(frozen=True) -class Job: - source_label: str - target_dir: Path - index: int - loader: Callable[[], bytes] - # 磁盘上的源文件；转换成功后删除（zip 成员为 None）。与输出路径相同时不会删。 - source_file: Path | 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 convert_to_webp(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 _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 - if is_numbered_asset_file(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) - next_i = max_index_in_directory(target_dir) + 1 - for path in files: - 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, - ) - ) - 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 - ), - ) - ) - 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: - print( - f"[warn] ok: {out} ({kind}) but could not delete {job.source_file}: {exc}", - file=sys.stderr, - ) - return kind - - -def run_job(job: Job) -> tuple[bool, str, Path, str | None, str | None]: - try: - data = job.loader() - job.target_dir.mkdir(parents=True, exist_ok=True) - - 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") - 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") - return True, job.source_label, out, kind, None - - webp_data, animated = convert_to_webp(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) - 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=( - "Read unnumbered images from 原图/ (by default), write numbered *.webp / *.gif " - "into the same folder tree (in-place). Subfolders mirror 原图. " - "Animated GIFs and animated WebPs become .gif; static images become lossless .webp. " - "Next index per folder = max existing N in that folder + 1. " - "By default unnumbered source files are deleted after a successful conversion." - ) - ) - parser.add_argument( - "source", - nargs="?", - default=None, - type=Path, - help=f"Source root folder, single image, or .zip (default: {DEFAULT_SOURCE_DIR.as_posix()}/).", - ) - parser.add_argument( - "-o", - "--output", - default=None, - dest="output_root", - type=Path, - help="Output root (default: same as source for folders/single file; for .zip default: /).", - ) - parser.add_argument( - "-j", - "--jobs", - type=int, - default=10, - help="Number of worker threads (default: 10).", - ) - parser.add_argument( - "--keep-sources", - action="store_true", - help="Keep unnumbered source files after conversion (default: delete them).", - ) - 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] = [] - - if source.is_dir(): - out_root = ( - args.output_root.expanduser().resolve() - if args.output_root is not None - else source - ) - out_root.mkdir(parents=True, exist_ok=True) - jobs = collect_directory_jobs( - source, out_root, delete_sources=not args.keep_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): - if is_numbered_asset_file(source): - print(f"Skip already-numbered file: {source}", file=sys.stderr) - return 0 - target_dir = ( - args.output_root.expanduser().resolve() - if args.output_root is not None - else source.parent.resolve() - ) - target_dir.mkdir(parents=True, exist_ok=True) - 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 not args.keep_sources else None, - ) - ] - else: - print(f"Not a folder, image, or zip: {source}", file=sys.stderr) - return 1 - - jobs.sort(key=lambda job: (str(job.target_dir), job.index)) - - if not jobs: - print(f"No new images to convert under: {source}") - return 0 - - converted = 0 - skipped = 0 - - with ThreadPoolExecutor(max_workers=max(1, args.jobs)) as pool: - for ok, label, target, kind, error in pool.map(run_job, jobs): - if ok: - print(f"{label} -> {target} ({kind})") - converted += 1 - else: - print(f"[skip] {label}: {error}", file=sys.stderr) - skipped += 1 - - print(f"Done. converted={converted}, skipped={skipped}") - return 0 if converted else 1 - - -if __name__ == "__main__": - raise SystemExit(main(sys.argv[1:])) diff --git a/package.json b/package.json index c533cfb..8909d5d 100644 --- a/package.json +++ b/package.json @@ -3,8 +3,11 @@ "private": true, "scripts": { "build": "node scripts/generate-manifest.mjs", - "dev": "npm run build && npx --yes serve public -p 8788", - "deploy": "npm run build && wrangler pages deploy public --project-name=meme-api" + "build:manifest-from-source": "node scripts/generate-manifest.mjs --manifest-from-source", + "dev": "npm run build && npx --yes serve dist -p 8788", + "deploy": "npm run deploy:upload", + "deploy:upload": "npm run build && wrangler pages deploy dist --project-name=meme-api", + "deploy:preview": "npm run build && wrangler pages deploy dist --project-name=meme-api --branch preview" }, "devDependencies": { "wrangler": "^4.6.0" diff --git a/public/index.html b/public/index.html index c0740ef..06686dd 100644 --- a/public/index.html +++ b/public/index.html @@ -17,7 +17,7 @@

meme-api

-

静态站点（Cloudflare Pages）。素材在仓库 原图/ 内编号生成 *.webp / *.gif；npm run build 将 原图/ 镜像到 public/meme/ 用于部署。直链可被外部引用；JSON 已配置 CORS。

+

静态站点（Cloudflare Pages）。在 原图/ 放未编号素材后运行 python 转化.py，输出到 原图-已处理/。将成品拷入 public/meme/ 后运行 npm run build：只生成 memes.json 并把 public/ 复制到 dist/，不会改写 public/meme/。清单默认按 public/meme/ 扫描；若尚未拷素材则回退扫描 原图-已处理/（无则 原图/）。需强制按源目录生成清单时用 npm run build:manifest-from-source。部署目录为 dist/。直链可外部引用；JSON 已配置 CORS。

浏览图库（分页、按文件夹分类、点击复制链接）

获取表情包列表

@@ -26,16 +26,17 @@ 
curl -sS ""

直链访问单张表情包

-

部署后文件在 public/meme/<分类>/（与仓库 原图/<分类>/ 内容一致），对应 URL：

+

站点根目录对应构建产物 dist/（本地从 public/ 同步而来）；表情文件在 /meme/<分类>/，对应 URL：

GET /meme/<分类>/<文件名>.webp 或 .gif

https://example.com/meme/%E7%99%BD%E8%89%B2%E5%B0%8F%E4%BA%BA/1.webp

中文或多段路径请使用编码后的 URL（浏览器会自动处理，图库「复制链接」亦为已编码地址）。

维护流程

    -
  1. 在 原图/<分类>/ 放入未编号素材；运行 python convert_to_webp.py（就地生成编号文件，成功后会删除这些未编号源图；若需保留源文件可加 --keep-sources）。
    2. -
  2. 若仍有「动图 WebP」编号文件，可在 原图/ 下运行 python animated_webp_to_gif.py 转为 .gif。
    3. -
  3. npm run build 将 原图/ 镜像到 public/meme/ 并生成 public/memes.json，再部署 public/。
    4. +
  4. 在 原图/<分类>/ 放入未编号素材；运行 python 转化.py，在 原图-已处理/ 生成编号 .webp/.gif 并限最长边（默认 300px）；默认保留 原图/ 源文件，清理可加 --delete-sources。
    5. +
  5. 动图 WebP 会在 python 转化.py 转换流程中转为 GIF。
    6. +
  6. npm run build：生成 dist/（含 dist/memes.json）。请先把 原图-已处理/ 拷入 public/meme/ 再构建；构建脚本不会覆盖 public/meme/。
    7. +
  7. 上线（生产）：项目在 Cloudflare 已连接 Git 时，由推送生产分支触发云端构建：npm run build → git add/commit/push；仪表盘「构建命令」填 npm run build，「构建输出目录」填 dist。wrangler pages deploy dist 在 Git 项目里常显示为预览，属正常。