dsh-pet-floating/scripts/download_memes.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
从「梗鲸档案馆」(aigengtu.com) 下载 DeepSeek 娘表情包，转码进 assets_net，
并把条目（名称 + 语义标签）追加到 assets_common/config.json 的 memes 列表。

用法：
    python3 scripts/download_memes.py

依赖：requests, Pillow（图片转码用，可仅下载跳过转码时不需要 Pillow）。
数据：scripts/meme_meta.json —— 档案馆 id → 台词（含情绪描述），
      由 https://aigengtu.com/archive 抓取（见 README「素材再生成」）。

说明：
- 图片真实扩展名不固定（jpg/png/webp/gif），逐个探测；
- id < 1000 时 URL 用 3 位零填充（001），否则原样（1000）；
- 输出 assets_net/n{id:04d}.webp，最大边缩到 640px、q80。
"""
import json
import os
import re
import sys
import time
import threading

try:
    import requests
except ImportError:
    sys.exit("需要 requests：pip install requests")

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
META = os.path.join(HERE, "meme_meta.json")
OUT_DIR = os.path.join(ROOT, "app", "src", "main", "assets_net")
CONFIG = os.path.join(ROOT, "app", "src", "main", "assets_common", "config.json")
ARCHIVE = "https://img.aigengtu.com/meme"


def url_of(i: int, ext: str) -> str:
    # 1~999 用三位零填充，1000+ 原样（实测站点的 URL 规则）
    name = f"{i:03d}" if i < 1000 else f"{i}"
    return f"{ARCHIVE}/{name}.{ext}"


def sniff(data: bytes) -> str:
    if data.startswith(b"\xff\xd8"):
        return "jpg"
    if data.startswith(b"\x89PNG"):
        return "png"
    if data.startswith(b"RIFF"):
        return "webp"
    if data.startswith(b"GIF8"):
        return "gif"
    return ""


EXTS = ["jpg", "png", "webp", "gif"]
_tl = threading.local()


def sess():
    if not hasattr(_tl, "s"):
        _tl.s = requests.Session()
        _tl.s.headers.update({
            "User-Agent": "Mozilla/5.0 (Linux; Android 14)",
            "Referer": "https://aigengtu.com/",
        })
    return _tl.s


def download_one(i: int, out: str) -> bool:
    for attempt in range(3):
        for ext in EXTS:
            try:
                r = sess().get(url_of(i, ext), timeout=30)
                if r.status_code == 200 and len(r.content) > 3000:
                    typ = sniff(r.content[:8])
                    if typ:
                        with open(out, "wb") as f:
                            f.write(r.content)
                        return True
            except Exception:
                pass
        time.sleep(0.5 * (attempt + 1))
    return False


TAG_RULES = [
    (("气鼓鼓", "暴怒", "生气", "愤怒", "炸毛"), ["angry"]),
    (("崩溃", "落泪", "哭", "委屈", "破防", "伤心", "惨", "绝望"), ["sad"]),
    (("开心", "笑", "得意", "欢呼", "兴奋", "撒花", "卖萌", "可爱", "高兴"), ["happy"]),
    (("点赞", "认可", "夸奖", "棒", "厉害", "牛"), ["praise"]),
    (("疑惑", "不解", "问号", "懵", "思考", "沉思", "挠头", "好奇"), ["thinking"]),
    (("吃瓜", "窥屏", "无语", "吐槽", "整活", "玩梗", "恶搞", "沙雕", "翻白眼"), ["fun"]),
    (("困", "睡", "摸鱼", "咸鱼", "躺平", "打盹", "赖床"), ["rest"]),
    (("吃", "饭", "干饭", "零食", "外卖", "奶茶", "饿"), ["eat"]),
    (("工作", "加班", "搬砖", "写代码", "上班", "开会", "改bug", "报错"), ["work"]),
    (("失败", "宕机", "错误"), ["error"]),
    (("等", "排队", "加载", "稍等"), ["waiting"]),
    (("你好", "打招呼", "挥手", "早安", "晚安", "拜拜"), ["greet"]),
    (("贴贴", "抱抱", "亲亲", "喜欢", "心动", "爱你"), ["love"]),
    (("token", "钱", "余额", "工资", "氪金", "付费", "穷"), ["money"]),
]


def derive_meta(alt: str):
    m = re.search(r"『(.+?)』", alt or "")
    name = (m.group(1) if m else f"梗图").strip()
    name = re.sub(r"\s+", " ", name)[:16]
    tags = set()
    for kws, ts in TAG_RULES:
        if any(k in (alt or "") for k in kws):
            tags.update(ts)
    if not tags:
        tags = {"general"}
    return name, sorted(tags)


def main():
    os.makedirs(OUT_DIR, exist_ok=True)
    with open(META, encoding="utf-8") as f:
        meta = json.load(f)
    ids = sorted(int(k) for k in meta)

    # 下载（只下载缺失的）
    todo = []
    for i in ids:
        dest = os.path.join(OUT_DIR, f"n{i:04d}.webp")
        if not (os.path.exists(dest) and os.path.getsize(dest) > 1500):
            todo.append(i)
    print(f"需下载 {len(todo)}/{len(ids)}")

    from concurrent.futures import ThreadPoolExecutor
    ok, fail = 0, 0
    with ThreadPoolExecutor(max_workers=16) as ex:
        for i, done in zip(todo, ex.map(lambda i: download_one(i, os.path.join(OUT_DIR, f"n{i:04d}.webp")), todo)):
            if done:
                ok += 1
            else:
                fail += 1
            if (ok + fail) % 200 == 0:
                print(f"  进度 {ok + fail}/{len(todo)}  ok={ok} fail={fail}", flush=True)
    print(f"下载完成 ok={ok} fail={fail}")

    # 更新 config.json 的 memes 列表
    with open(CONFIG, encoding="utf-8") as f:
        cfg = json.load(f)
    exist = {m["file"] for m in cfg["memes"]}
    added = 0
    for i in ids:
        fid = f"n{i:04d}"
        if fid in exist:
            continue
        if not (os.path.exists(os.path.join(OUT_DIR, fid + ".webp")) and
                os.path.getsize(os.path.join(OUT_DIR, fid + ".webp")) > 1500):
            continue
        name, tags = derive_meta(str(meta.get(str(i), "")))
        cfg["memes"].append({"name": name, "file": fid, "tags": tags})
        added += 1
    with open(CONFIG, "w", encoding="utf-8") as f:
        json.dump(cfg, f, ensure_ascii=False, separators=(",", ":"))
    print(f"config.json 更新：新增 {added}，当前表情包总数 {len(cfg['memes'])}")


if __name__ == "__main__":
    main()