#!/usr/bin/env python3
"""下厨房菜品图爬虫(每菜多图版)。
给一批菜名,逐个去下厨房搜索,取**前 N 个 recipe 的首图**(真图在 data-src,不是 src,
N 由 --per-dish 控制,默认 3 → 即同名菜的不同做法各一张),每道菜输出一行 JSON 到 jsonl,
行内 images:[...] 记录该菜对应的所有图(文件名 + 原图 URL + 做法标题/链接)。
反爬策略(用户要求"悠着点"):
- 每请求随机换 User-Agent + Accept-Language + 轻微变化的 header 组合
- 请求间随机延迟(默认 4~9s),每 N 条额外长歇一次
- 非 200 / 异常 → 指数退避重试;疑似被封(403/503/429)歇更久
- 不持久化 cookie(每次请求像"新访客",避免同 cookie 配不同 UA 的机器人特征)
- 代理可选(--proxy / 文件轮换);国内站默认直连
断点续抓:已在输出 jsonl 里的菜名自动跳过。
用法:
python3 scrape_xiachufang.py --out dishes.jsonl --download --per-dish 3 鸡尾酒 布朗尼 牛肉拉面
python3 scrape_xiachufang.py --out dishes.jsonl --download --per-dish 3 --input dishes.txt
"""
from __future__ import annotations
import argparse
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
# Windows 控制台默认 GBK,无法输出 ✓/✗/… 等字符 → 强制 UTF-8(长跑任务必须稳)
for _s in (sys.stdout, sys.stderr):
try:
_s.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
SEARCH_URL = "https://www.xiachufang.com/search/"
# 真实浏览器 UA 池(桌面 + 移动混合),每请求随机取一条
UA_POOL = [
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4.1 Safari/605.1.15",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:126.0) Gecko/20100101 Firefox/126.0",
"Mozilla/5.0 (iPhone; CPU iPhone OS 17_4 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Mobile/15E148 Safari/604.1",
"Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Mobile Safari/536.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Edg/123.0.0.0",
]
ACCEPT_LANGS = [
"zh-CN,zh;q=0.9,en;q=0.8",
"zh-CN,zh;q=0.9",
"zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3",
]
def build_headers(referer: str = "https://www.xiachufang.com/") -> dict:
"""每次随机组装 header,制造"不同访客"特征。"""
h = {
"User-Agent": random.choice(UA_POOL),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language": random.choice(ACCEPT_LANGS),
"Referer": referer,
"Upgrade-Insecure-Requests": "1",
}
# 随机带或不带这些可选头,进一步打散指纹
if random.random() < 0.6:
h["DNT"] = "1"
if random.random() < 0.5:
h["Sec-Fetch-Dest"] = "document"
h["Sec-Fetch-Mode"] = "navigate"
h["Sec-Fetch-Site"] = "same-origin"
return h
# 每线程独立 Session(连接复用更快,并行更稳;requests.Session 非线程安全→线程本地)
_tls = threading.local()
def session() -> requests.Session:
s = getattr(_tls, "s", None)
if s is None:
s = _tls.s = requests.Session()
return s
# ---------- 全局限速治理(根治多线程退避雪崩) ----------
# 1) 搜索节流:全局把"搜索请求"按最小间隔排队(限速的是搜索域名,不是图片CDN)
# 2) 冷却闸:任一线程被 429/503 → 设全局冷却,所有线程一起等,避免罚时还有线程乱戳
_pace_lock = threading.Lock()
_next_search = 0.0 # 下一次允许搜索的时刻
_resume_at = 0.0 # 全局冷却到期时刻
_penalty = 0 # 连续被限速次数(自适应升级/消解)
SEARCH_SPACING = 2.0 # 全局两次搜索最小间隔(秒),可被 --spacing 覆盖
def search_pace() -> None:
"""全局给搜索请求排队,保证两次搜索至少间隔 SEARCH_SPACING 秒。"""
global _next_search
with _pace_lock:
now = time.time()
slot = max(now, _next_search)
_next_search = slot + SEARCH_SPACING + random.uniform(0, 0.5)
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:
"""被 429/503 时触发全局冷却,连续触发自适应升级(60s→封顶600s)。"""
global _resume_at, _penalty
with _pace_lock:
_penalty += 1
cool = min(60 * (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(url: str, proxy: str | None, retries: int = 6, timeout: int = 25) -> str | None:
"""GET 搜索页:全局节流 + 全局冷却闸 + 自适应重试。"""
proxies = {"http": proxy, "https": proxy} if proxy else None
for attempt in range(retries):
gate_wait() # 冷却期一起等
search_pace() # 全局排队,控制搜索速率
try:
resp = session().get(
url, headers=build_headers(), proxies=proxies, timeout=timeout
)
if resp.status_code == 200:
ease_limit()
return resp.text
if resp.status_code in (403, 429, 503):
trip_limit() # 设全局冷却,下轮 gate_wait 一起等
else:
print(f" [!] 状态 {resp.status_code}", file=sys.stderr)
time.sleep(5 * (attempt + 1))
except requests.RequestException as e:
print(f" [!] 请求异常 {e}", file=sys.stderr)
time.sleep(8 + random.uniform(0, 5))
return None
def normalize_img(url: str, width: int = 600) -> str:
"""把 215px 缩略图参数换成更大尺寸。chuimg 走七牛 imageView2:
base.jpg?imageView2/2/w/600 = 等比缩到宽 600。"""
base = url.split("?", 1)[0]
return f"{base}?imageView2/2/w/{width}/interlace/1/q/85"
def parse_top_recipes(html: str, n: int) -> list[dict]:
"""从搜索结果页抽**前 n 个** recipe 的 {image_url, recipe_title, recipe_url}。
一个搜索页通常有 ~15 个 recipe,各带一张真图 → 即每菜多图(同名菜的不同做法)。
无结果 / 结构变化 → []。"""
# 先定位结果列表区,避免误抓页面装饰图(IE 提示图等)
start = html.find('class="normal-recipe-list"')
if start == -1:
return []
region = html[start:]
out: list[dict] = []
seen: set[str] = set()
# 逐个 recipe 块:...内含 ![标题]()
...
for m in re.finditer(r']*>(.*?)', region, re.S):
block = m.group(0)
img = re.search(r'data-src="([^"]+chuimg[^"]*)"', block)
if not img:
continue
rid = m.group(2)
if rid in seen: # 同一 recipe 在页面可能出现两次,去重
continue
seen.add(rid)
image_raw = img.group(1)
alt = re.search(r'alt="([^"]*)"', block)
out.append({
"image_url": normalize_img(image_raw),
"image_url_raw": image_raw,
"recipe_title": (alt.group(1).strip() if alt else None),
"recipe_url": "https://www.xiachufang.com" + m.group(1),
})
if len(out) >= n:
break
return out
def safe_name(dish: str) -> str:
return re.sub(r"[^\w一-鿿]+", "_", dish).strip("_") or "dish"
def download_image(url: str, dish: str, idx: int, img_dir: str,
proxy: str | None) -> tuple[str | None, int]:
"""下图到 img_dir/_.jpg。chuimg 可能校验 Referer,带上。
返回 (相对路径, 字节数);失败 (None, 0)。"""
proxies = {"http": proxy, "https": proxy} if proxy else None
headers = build_headers()
headers["Referer"] = "https://www.xiachufang.com/"
headers["Accept"] = "image/avif,image/webp,image/png,image/*;q=0.8,*/*;q=0.5"
try:
r = session().get(url, headers=headers, proxies=proxies, timeout=25)
if r.status_code == 200 and r.content:
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)
except requests.RequestException as e:
print(f" [!] 下图失败 {e}", file=sys.stderr)
return None, 0
def scrape_one(dish: str, proxy: str | None, do_download: bool, img_dir: str,
per_dish: int, jitter: float = 0.0) -> dict:
if jitter: # 并行时每个任务起点随机错开,避免同时齐刷刷打过去
time.sleep(random.uniform(0, jitter))
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
url = SEARCH_URL + "?" + urllib.parse.urlencode({"keyword": dish, "cat": 1001})
html = fetch(url, proxy)
if html is None:
return {"dish": dish, "found": False, "count": 0, "images": [],
"reason": "fetch_failed", "fetched_at": now}
recs = parse_top_recipes(html, per_dish)
if not recs:
return {"dish": dish, "found": False, "count": 0, "images": [],
"reason": "no_result", "fetched_at": now}
images = []
for i, rec in enumerate(recs, 1):
entry = dict(rec)
if do_download:
rel, size = download_image(rec["image_url"], dish, i, img_dir, proxy)
entry["image_file"] = rel
entry["bytes"] = size
if i < len(recs): # 同菜多图之间轻微间隔,别猛打 CDN
time.sleep(random.uniform(0.1, 0.3))
images.append(entry)
saved = sum(1 for e in images if not do_download or e.get("image_file"))
return {"dish": dish, "found": True, "count": saved, "images": images,
"fetched_at": now}
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="*", help="菜名(也可用 --input 从文件读)")
ap.add_argument("--input", help="菜名文件,一行一道")
ap.add_argument("--out", default="dishes.jsonl", help="输出 jsonl")
ap.add_argument("--download", action="store_true", help="同时把图片下载到 images/")
ap.add_argument("--img-dir", default="images")
ap.add_argument("--per-dish", type=int, default=3, help="每道菜抓取的图片张数(取搜索前 N 个做法)")
ap.add_argument("--proxy", default=os.environ.get("CRAWL_PROXY"),
help="如 http://127.0.0.1:7897;默认直连(国内站)")
ap.add_argument("--min-delay", type=float, default=4.0)
ap.add_argument("--max-delay", type=float, default=9.0)
ap.add_argument("--rest-every", type=int, default=40, help="(串行)每抓 N 条额外长歇")
ap.add_argument("--workers", type=int, default=1,
help="并行线程数;>1 启用并行(自动去掉串行长延迟)")
ap.add_argument("--jitter", type=float, default=0.0,
help="每任务起点随机错开秒数(并行未设时默认 1.5)")
ap.add_argument("--spacing", type=float, default=SEARCH_SPACING,
help="全局两次搜索最小间隔秒(控制搜索速率,防限速;默认 2.0)")
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()
if args.workers > 1:
jitter = args.jitter if args.jitter > 0 else 1.5
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, jitter): 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, "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 % 20 == 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} "
f"{rate*60:.0f}道/分 ETA {eta/60:.0f}分", flush=True)
else:
with open(args.out, "a", encoding="utf-8") as fout:
for i, dish in enumerate(todo, 1):
print(f"[{i}/{len(todo)}] {dish} ...", flush=True)
rec = scrape_one(dish, args.proxy, args.download, args.img_dir, args.per_dish)
fout.write(json.dumps(rec, ensure_ascii=False) + "\n")
fout.flush()
if rec["found"]:
ok += 1
titles = "、".join(filter(None, (e.get("recipe_title") for e in rec["images"][:3])))
print(f" ✓ {rec['count']} 张 {titles}")
else:
miss += 1
print(f" ✗ {rec['reason']}")
if i < len(todo):
delay = random.uniform(args.min_delay, args.max_delay)
if i % args.rest_every == 0:
delay += random.uniform(20, 40)
print(f" …长歇 {delay:.0f}s")
time.sleep(delay)
print(f"\n完成:成功 {ok} 缺图 {miss} → {args.out}", flush=True)
if __name__ == "__main__":
main()