3218485270
f493bde8a9
- 缓存热点批量流程（有采集缓存不触发 Google） - 简报不足直接从采集缓存生成（轻量补齐） - 三图合成（模特/印花/底图）+ 底图压缩 <2MB - 热点去重→风格去重自动切换 + 不适合类目 review 兜底 - 透明背景（background=transparent）+ 提示词清洗（敏感词/背景描述） - 任务前 basemap 校验 + 模板国家校验 + 模特任务级分配
121 lines
5.1 KiB
Python
121 lines
5.1 KiB
Python
#!/usr/bin/env python
|
||
"""POD 热点抓取 Agent —— LangGraph 工程化版命令行入口。
|
||
|
||
用法示例：
|
||
# 跑全部国家（默认 US GB JP AU）
|
||
python cli.py
|
||
|
||
# 只跑英国，使用 Mock 兜底 LLM
|
||
python cli.py -c GB --provider mock
|
||
|
||
# 跑美国并用真实 LLM（provider 在 config.llm_screen 已配好时也可不传）
|
||
python cli.py -c US
|
||
|
||
# 提供平铺衣服底图，把设计印上去（需 config.compose.backend + api_key）
|
||
python cli.py -c US --base path/to/flatlay.png
|
||
|
||
产物按国家落在 output/<country>/（design_briefs.json/md、composite_prompts.json/md、report.md）。
|
||
每个国家的提示词规则在 prompts/<country>/（system_prompt.md + aesthetics.yaml）。
|
||
"""
|
||
import argparse
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# 把项目根加入 sys.path，确保 `import graph` 可用
|
||
ROOT = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
import yaml # noqa: E402
|
||
|
||
from graph.agent import run_country # noqa: E402
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="POD 热点抓取 Agent (LangGraph)")
|
||
parser.add_argument("-c", "--countries", nargs="*", default=None,
|
||
help="国家代码（US GB JP AU），可多个；默认取 config.countries")
|
||
parser.add_argument("--provider", default=None,
|
||
help="覆盖 LLM 后端：mock / openai_compat / openai / deepseek / qwen / moonshot")
|
||
parser.add_argument("--base", default=None,
|
||
help="平铺衣服底图路径；提供则尝试印图（需 config.compose 配置图像后端）")
|
||
parser.add_argument("--config", default="config.yaml", help="全局配置文件路径")
|
||
# —— 产品图生成（product 节点）——
|
||
parser.add_argument("--spu", default=None, help="选品款号（如 DG004；留空自动选第一个有底图的）")
|
||
parser.add_argument("--sku", default=None, help="颜色编码（如 DG004-BL01；多个用逗号分隔，如 DG015-VT01,DG015-DARK HEATHER；留空自动选）")
|
||
parser.add_argument("--single-spu", action="store_true", help="模板导出：单 SPU 下挂多颜色变体（默认每颜色一个 SPU 块）")
|
||
parser.add_argument("--product-backend", default=None, help="产品图像后端：openai / mock")
|
||
parser.add_argument("--list-spus", action="store_true", help="列出 db 可选 SPU 并退出")
|
||
parser.add_argument("--list-colors", default=None, metavar="SPU", help="列出某款号颜色并退出")
|
||
args = parser.parse_args()
|
||
|
||
cfg_path = ROOT / args.config
|
||
if not cfg_path.exists():
|
||
print(f"配置文件不存在: {cfg_path}")
|
||
sys.exit(1)
|
||
cfg = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) or {}
|
||
|
||
# 产品选品查询（--list-spus / --list-colors）
|
||
pcfg0 = cfg.get("product") or {}
|
||
dbp = Path(pcfg0.get("db_path", "db/spu_sku.db"))
|
||
if not dbp.is_absolute():
|
||
dbp = ROOT / dbp
|
||
if args.list_spus:
|
||
from graph.product import list_spus
|
||
print("=== db 可选 SPU（款号 / 品类 / 印花类型 / 国家）===")
|
||
for s in list_spus(dbp):
|
||
print(f" {s['code']:10s} {s.get('style') or '':8s} {s.get('printing_type') or '':12s} {s.get('country') or ''}")
|
||
sys.exit(0)
|
||
if args.list_colors:
|
||
from graph.product import list_colors
|
||
print(f"=== {args.list_colors} 可选颜色（SKU.code / 色名）===")
|
||
for c in list_colors(dbp, args.list_colors):
|
||
print(f" {c['sku_code']:16s} {c['color']}")
|
||
sys.exit(0)
|
||
|
||
if args.provider:
|
||
cfg.setdefault("llm_screen", {})["provider"] = args.provider
|
||
if args.spu or args.sku or args.product_backend or args.single_spu:
|
||
p = cfg.setdefault("product", {})
|
||
if args.spu:
|
||
p["spu_code"] = args.spu
|
||
if args.sku:
|
||
p["sku_code"] = args.sku
|
||
if args.product_backend:
|
||
p["backend"] = args.product_backend
|
||
if args.single_spu:
|
||
p["spu_per_color"] = False
|
||
|
||
countries = args.countries or cfg.get("countries", ["US", "GB", "JP", "AU"])
|
||
base_image = args.base
|
||
|
||
overall_errors = 0
|
||
for c in countries:
|
||
print(f"\n===== 开始处理 {c} =====")
|
||
try:
|
||
res = run_country(c, cfg, ROOT, base_image=base_image)
|
||
except Exception as e: # noqa: BLE001
|
||
print(f"[{c}] 运行失败: {e}")
|
||
overall_errors += 1
|
||
continue
|
||
|
||
stats = res.get("stats", {})
|
||
f = stats.get("fetch", {})
|
||
fl = stats.get("filter", {})
|
||
sc = stats.get("screen", {})
|
||
pr = stats.get("prompt", {})
|
||
print(f"[{c}] 抓取原始 {f.get('raw_rows')} 行 | 过滤后保留 {fl.get('kept')} | "
|
||
f"筛选保留 {sc.get('kept')} | 设计简报 {pr.get('briefs')} 条")
|
||
errs = res.get("errors") or []
|
||
if errs:
|
||
overall_errors += len(errs)
|
||
print(f"[{c}] 节点兜底捕获 {len(errs)} 条错误：")
|
||
for e in errs:
|
||
print(f" - [{e.get('node')}] {e.get('type')}: {e.get('message')}")
|
||
print(f"[{c}] 产物目录: output/{c}/")
|
||
|
||
print(f"\n全部完成。累计兜底错误 {overall_errors} 条（不影响产出）。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|