陈世睿
74caa9d112
- crawler/:豆果(主)/下厨房/Bing 爬虫 + dishes.jsonl(3056 菜→9167 图映射)+ verify_repair 按 URL 重下 - retrieval/:三路检索(BM25 + 本地 BGE-M3 向量 + RRF 融合),FastAPI + 前端;写死图片目录(默认 crawler/images,可 IMAGE_DIR 覆盖)、绑 0.0.0.0 局域网访问、启动自动建索引、服务器 serve 图片 - 图片(1.7G)与向量模型(2.3G)不进 git Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
376 lines
16 KiB
Python
376 lines
16 KiB
Python
#!/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 块:<a href="/recipe/ID/">...内含 <img data-src=chuimg alt=标题>...</a>
|
|
for m in re.finditer(r'<a href="(/recipe/(\d+)/)"[^>]*>(.*?)</a>', 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/<safe>_<idx>.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()
|