3218485270
2a96ec0870
- 产品持久化：每完成一个产品立即追加写入 products_pending.jsonl，崩溃不丢已完成产品，finish 读盘合并后统一写模板 - 503 致命错误提前终止：compose/product/seed_shot 端到端识别，提前终止搜索分析，丢弃未完成简报，保留已完成落盘产品直接合成模板 - 模特分配：material_library 合格模特图按任务序号独立随机，同 SPU 多款不再共用同一模特 - 图像网关适配：execution_mode/background 默认不再传入 yunfei 等标准网关，base_url 需带 /v1；429/5xx/空响应退避重试 - Pinterest 分析：删除 term 注入与纯文本降级，失败直接放弃；图片上传前 PIL 完整性校验；suitable_for_print=False 过滤丢弃 - 模板导出：不再产生空白 xlsx，文件名=模板原文件名_已填写；写入前按货号末 3 位升序排序 - 删除对接文档.md，更新 README，gitignore 排除测试产物
66 lines
3.0 KiB
Python
66 lines
3.0 KiB
Python
"""节点 1/6：抓取（fetch）。
|
||
|
||
按 config.sources 启用各可插拔数据源，汇总统一格式行。
|
||
单源失败不影响其它源（内部逐个 try），整体再套 with_fallback 兜底。
|
||
|
||
缓存策略（用户要求：成功采集就不用缓存）：
|
||
1. 先用种子词走数据源抓取（Google Trends 等），成功即用新数据；
|
||
2. 抓取失败/无结果才回退 output/<国>/collected_keywords.json 旧缓存，保证流水线不中断。
|
||
"""
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List
|
||
|
||
from graph.sources import get_source
|
||
from graph.validate import validate_rows, with_fallback
|
||
|
||
|
||
@with_fallback("fetch")
|
||
def fetch_node(state: Dict[str, Any]) -> Dict[str, Any]:
|
||
country = state["country"]
|
||
config = state["config"]
|
||
cc = state["country_config"]
|
||
enabled = config.get("sources") or ["google_trends"]
|
||
rows: List[Dict[str, Any]] = []
|
||
errors = list(state.get("errors") or [])
|
||
|
||
# 1) 先尝试数据源抓取（用种子词），成功即用新数据
|
||
for name in enabled:
|
||
try:
|
||
src = get_source(name)
|
||
rows.extend(src.fetch(country, cc, config))
|
||
except Exception as e: # noqa: BLE001
|
||
errors.append({
|
||
"node": "fetch", "type": type(e).__name__,
|
||
"message": f"source[{name}]: {e}", "trace": "",
|
||
})
|
||
print(f"[fetch] 数据源 {name} 失败（跳过）: {e}")
|
||
|
||
if rows:
|
||
rows = validate_rows(rows, "fetch")
|
||
stats = dict(state.get("stats") or {})
|
||
stats["fetch"] = {"raw_rows": len(rows), "sources": enabled, "errors": len(errors)}
|
||
return {"raw_rows": rows, "errors": errors, "stats": stats}
|
||
|
||
# 2) 抓取失败/无结果 → 回退采集缓存（filter 上次写入的 collected_keywords.json）
|
||
use_collected = (config.get("fetch") or {}).get("use_collected", True)
|
||
if use_collected:
|
||
try:
|
||
p = Path(state.get("cache_dir") or state.get("output_dir", "")) / "collected_keywords.json"
|
||
if p.exists():
|
||
data = json.loads(p.read_text(encoding="utf-8"))
|
||
cached_rows = data.get("keywords") or []
|
||
if cached_rows:
|
||
rows = [dict(r) for r in cached_rows] # 已过滤去重的关键词
|
||
print(f"[fetch] 数据源抓取失败，回退采集缓存 {len(rows)} 条（{country}）")
|
||
stats = dict(state.get("stats") or {})
|
||
stats["fetch"] = {"raw_rows": len(rows), "sources": ["collected_cache"], "errors": len(errors)}
|
||
return {"raw_rows": rows, "errors": errors, "stats": stats}
|
||
except Exception as e: # noqa: BLE001
|
||
print(f"[fetch] 读取采集缓存失败（回退数据源）: {e}")
|
||
|
||
rows = validate_rows(rows, "fetch")
|
||
stats = dict(state.get("stats") or {})
|
||
stats["fetch"] = {"raw_rows": len(rows), "sources": enabled, "errors": len(errors)}
|
||
return {"raw_rows": rows, "errors": errors, "stats": stats}
|