dish-image/crawler/scrape_douguo.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""豆果美食 菜品图爬虫(每菜多图版,最高质量源)。

走豆果 App JSON 接口 POST api.douguo.net/recipe/search/0/20(无需签名),每菜取前 N 个
菜谱的 photo_path(960px 高清真实菜品图),下载到 images/,每菜一行 JSON 到 jsonl。
与 scrape_xiachufang/scrape_bing 共用 dishes.jsonl + images/:续抓只跳 found:true。
--replace: 下载前先删该菜已有的 <safe>_*.jpg(用于把旧 Bing 图替换成豆果图)。

用法:
  python scrape_douguo.py --out dishes.jsonl --download --per-dish 3 --workers 8 --replace --input todo.txt
"""
from __future__ import annotations

import argparse
import glob
import json
import os
import random
import re
import sys
import threading
import time
import urllib.parse
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone

import requests

for _s in (sys.stdout, sys.stderr):
    try:
        _s.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass

API = "https://api.douguo.net/recipe/search/0/20"

UA_POOL = [
    "Mozilla/5.0 (Linux; Android 9; MI 8 Build/PKQ1.181121.001) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.136 Mobile Safari/537.36",
    "Mozilla/5.0 (Linux; Android 10; Redmi K30) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.106 Mobile Safari/537.36",
    "Mozilla/5.0 (Linux; Android 11; Pixel 5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.91 Mobile Safari/537.36",
]


def build_headers() -> dict:
    return {"User-Agent": random.choice(UA_POOL),
            "Content-Type": "application/x-www-form-urlencoded; charset=utf-8"}


_tls = threading.local()


def session() -> requests.Session:
    s = getattr(_tls, "s", None)
    if s is None:
        s = _tls.s = requests.Session()
    return s


# ---------- 全局限速治理(同前:节流 + 全员冷却闸) ----------
_pace_lock = threading.Lock()
_next_search = 0.0
_resume_at = 0.0
_penalty = 0
SEARCH_SPACING = 0.3  # 豆果接口快且耐用,默认间隔小


def search_pace() -> None:
    global _next_search
    with _pace_lock:
        now = time.time()
        slot = max(now, _next_search)
        _next_search = slot + SEARCH_SPACING + random.uniform(0, 0.2)
        wait = slot - now
    if wait > 0:
        time.sleep(wait)


def gate_wait() -> None:
    while True:
        with _pace_lock:
            wait = _resume_at - time.time()
        if wait <= 0:
            return
        time.sleep(min(wait, 3) + random.uniform(0, 0.4))


def trip_limit() -> None:
    global _resume_at, _penalty
    with _pace_lock:
        _penalty += 1
        cool = min(45 * (1.5 ** (_penalty - 1)), 600)
        nr = time.time() + cool
        if nr > _resume_at:
            _resume_at = nr
            print(f"  [!] 限速#{_penalty} 全局冷却 {cool:.0f}s", file=sys.stderr, flush=True)


def ease_limit() -> None:
    global _penalty
    if _penalty:
        with _pace_lock:
            _penalty = max(0, _penalty - 1)


def fetch_search(dish: str, proxy: str | None, retries: int = 5) -> dict | None:
    proxies = {"http": proxy, "https": proxy} if proxy else None
    body = urllib.parse.urlencode({"client": 4, "keyword": dish, "order": 0, "_vs": 400}).encode()
    for attempt in range(retries):
        gate_wait()
        search_pace()
        try:
            r = session().post(API, data=body, headers=build_headers(), proxies=proxies, timeout=20)
            if r.status_code == 200:
                ease_limit()
                return r.json()
            if r.status_code in (403, 429, 503):
                trip_limit()
            else:
                print(f"  [!] 状态 {r.status_code}", file=sys.stderr)
                time.sleep(3 * (attempt + 1))
        except (requests.RequestException, ValueError) as e:
            print(f"  [!] 请求异常 {type(e).__name__}", file=sys.stderr)
            time.sleep(5 + random.uniform(0, 5))
    return None


def parse_images(data: dict | None, n: int) -> list[dict]:
    recs = ((data or {}).get("result") or {}).get("recipes") or []
    out: list[dict] = []
    seen: set[str] = set()
    for rc in recs:
        url = rc.get("photo_path") or rc.get("thumb_path") or rc.get("image")
        if not url:
            continue
        key = str(rc.get("cook_id") or url)
        if key in seen:
            continue
        seen.add(key)
        # .com 图床有 hotlink/限流(HTTP 493),.net 镜像同图可用,但 .net 证书不匹配→走 http
        net_url = url.replace("douguo.com", "douguo.net").replace("https://", "http://")
        out.append({
            "image_url": net_url,
            "src_com": url,
            "title": (rc.get("title") or "").strip() or None,
            "cook_id": rc.get("cook_id"),
        })
        if len(out) >= n:
            break
    return out


def is_image(b: bytes) -> bool:
    return (b[:3] == b"\xff\xd8\xff" or b[:8] == b"\x89PNG\r\n\x1a\n"
            or b[:6] in (b"GIF87a", b"GIF89a")
            or (b[:4] == b"RIFF" and b[8:12] == b"WEBP") or b[:2] == b"BM")


def safe_name(dish: str) -> str:
    return re.sub(r"[^\w一-鿿]+", "_", dish).strip("_") or "dish"


def download(url: str, dish: str, idx: int, img_dir: str,
             proxy: str | None, retries: int = 4) -> tuple[str | None, int]:
    proxies = {"http": proxy, "https": proxy} if proxy else None
    headers = {"User-Agent": random.choice(UA_POOL), "Referer": "https://www.douguo.com/",
               "Accept": "image/avif,image/webp,image/png,image/*;q=0.8,*/*;q=0.5"}
    for attempt in range(retries):
        gate_wait()  # 图床被限(493/429)时也走全局冷却
        try:
            r = session().get(url, headers=headers, proxies=proxies, timeout=25)
            if r.status_code == 200 and len(r.content) >= 800 and is_image(r.content):
                ease_limit()
                path = os.path.join(img_dir, f"{safe_name(dish)}_{idx:02d}.jpg")
                with open(path, "wb") as f:
                    f.write(r.content)
                rel = os.path.relpath(path, os.path.dirname(img_dir) or ".").replace(os.sep, "/")
                return rel, len(r.content)
            if r.status_code in (403, 429, 493, 503):
                trip_limit()
            else:
                time.sleep(1.5 * (attempt + 1))
        except requests.RequestException as e:
            print(f"  [!] 下图失败 {e}", file=sys.stderr)
            time.sleep(2)
    return None, 0


def scrape_one(dish: str, proxy: str | None, do_download: bool, img_dir: str,
               per_dish: int, jitter: float = 0.0, replace: bool = False) -> dict:
    if jitter:
        time.sleep(random.uniform(0, jitter))
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    base = {"dish": dish, "source": "douguo", "fetched_at": now}
    data = fetch_search(dish, proxy)
    if data is None:
        return {**base, "found": False, "count": 0, "images": [], "reason": "fetch_failed"}
    recs = parse_images(data, per_dish)
    if not recs:
        return {**base, "found": False, "count": 0, "images": [], "reason": "no_result"}
    if do_download and replace:  # 先删旧图(把该菜的 Bing 图替换掉)
        for old in glob.glob(os.path.join(img_dir, safe_name(dish) + "_*.jpg")):
            try:
                os.remove(old)
            except OSError:
                pass
    images = []
    for i, rec in enumerate(recs, 1):
        e = dict(rec)
        if do_download:
            rel, size = download(rec["image_url"], dish, i, img_dir, proxy)
            e["image_file"] = rel
            e["bytes"] = size
            if i < len(recs):
                time.sleep(random.uniform(0.05, 0.2))
        images.append(e)
    saved = sum(1 for e in images if not do_download or e.get("image_file"))
    if do_download and saved == 0:
        return {**base, "found": False, "count": 0, "images": images, "reason": "download_failed"}
    return {**base, "found": True, "count": saved, "images": images}


def load_done(out_path: str) -> set[str]:
    done = set()
    if os.path.exists(out_path):
        with open(out_path, encoding="utf-8") as f:
            for line in f:
                try:
                    rec = json.loads(line)
                    if rec.get("found"):
                        done.add(rec["dish"])
                except Exception:
                    pass
    return done


def main():
    global SEARCH_SPACING
    ap = argparse.ArgumentParser()
    ap.add_argument("dishes", nargs="*")
    ap.add_argument("--input")
    ap.add_argument("--out", default="dishes.jsonl")
    ap.add_argument("--download", action="store_true")
    ap.add_argument("--img-dir", default="images")
    ap.add_argument("--per-dish", type=int, default=3)
    ap.add_argument("--proxy", default=os.environ.get("CRAWL_PROXY"))
    ap.add_argument("--workers", type=int, default=8)
    ap.add_argument("--jitter", type=float, default=0.8)
    ap.add_argument("--spacing", type=float, default=SEARCH_SPACING)
    ap.add_argument("--replace", action="store_true", help="下载前删该菜已有图(替换旧源)")
    args = ap.parse_args()
    SEARCH_SPACING = args.spacing

    dishes = list(args.dishes)
    if args.input:
        with open(args.input, encoding="utf-8") as f:
            dishes += [ln.strip() for ln in f if ln.strip()]
    if not dishes:
        ap.error("没给菜名")
    dishes = list(dict.fromkeys(dishes))

    os.makedirs(args.img_dir, exist_ok=True)
    done = load_done(args.out)
    todo = [d for d in dishes if d not in done]
    print(f"共 {len(dishes)} 道,已完成 {len(done)},本次待抓 {len(todo)},并行 {args.workers} 线程,源 豆果",
          flush=True)
    if not todo:
        print("没有待抓的菜。")
        return

    ok = miss = 0
    t0 = time.time()
    lock = threading.Lock()
    with open(args.out, "a", encoding="utf-8") as fout, \
            ThreadPoolExecutor(max_workers=args.workers) as ex:
        futs = {ex.submit(scrape_one, d, args.proxy, args.download, args.img_dir,
                          args.per_dish, args.jitter, args.replace): d for d in todo}
        for i, fut in enumerate(as_completed(futs), 1):
            dish = futs[fut]
            try:
                rec = fut.result()
            except Exception as e:
                now = datetime.now(timezone.utc).isoformat(timespec="seconds")
                rec = {"dish": dish, "source": "douguo", "found": False, "count": 0,
                       "images": [], "reason": f"error:{type(e).__name__}", "fetched_at": now}
            with lock:
                fout.write(json.dumps(rec, ensure_ascii=False) + "\n")
                fout.flush()
            if rec["found"]:
                ok += 1
            else:
                miss += 1
            if i % 25 == 0 or i == len(todo):
                rate = i / max(time.time() - t0, 1e-6)
                eta = (len(todo) - i) / rate if rate else 0
                print(f"[{i}/{len(todo)}] ✓{ok} ✗{miss}  {rate*60:.0f}道/分  ETA {eta/60:.0f}分",
                      flush=True)
    print(f"\n完成:成功 {ok}  缺图 {miss}  → {args.out}", flush=True)


if __name__ == "__main__":
    main()