117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
"""Pinterest 参考模式共享辅助：种子词加载、已用搜索词持久化、搜索词全局过滤。
|
|
|
|
独立于 Google Trends 采集链路，供 pinterest_search / scrape / analyze 节点复用。
|
|
"""
|
|
import json
|
|
import random
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
from graph.paths import project_root, runtime_root
|
|
|
|
# 不适合 T 恤印花的类目关键词（复用 product_batch 的兜底清单）
|
|
_UNSUITABLE = re.compile(
|
|
r"\b(nails?|manicure|pedicure|recipes?|cooking|lottery|jackpot|results?|score|scores?|"
|
|
r"fixtures?|forecast|weather|temperature|map|directions?|parking|opening hours?|"
|
|
r"prices?|price|reviews?|jobs?|salary|mortgage|council tax|election|referendum|"
|
|
r"stock market|exchange rate|gas prices?)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def pinterest_seed_path(country: str) -> Path:
|
|
for root in (runtime_root(), project_root()):
|
|
p = root / "configs" / "pinterest" / f"{country}.yaml"
|
|
if p.exists():
|
|
return p
|
|
return Path("configs") / "pinterest" / f"{country}.yaml"
|
|
|
|
|
|
def load_pinterest_seeds(country: str) -> List[str]:
|
|
"""读国家 Pinterest 种子词池（configs/pinterest/<CC>.yaml 的 seeds）。"""
|
|
try:
|
|
import yaml
|
|
p = pinterest_seed_path(country)
|
|
if not p.exists():
|
|
print(f"[pinterest] 未找到种子词配置: {p}")
|
|
return []
|
|
data = yaml.safe_load(p.read_text(encoding="utf-8")) or {}
|
|
seeds = [str(s).strip() for s in (data.get("seeds") or []) if str(s).strip()]
|
|
return seeds
|
|
except Exception as e: # noqa: BLE001
|
|
print(f"[pinterest] 种子词加载失败: {e}")
|
|
return []
|
|
|
|
|
|
def sample_seeds(country: str, n: int) -> List[str]:
|
|
"""从国家种子池随机抽取 n 个种子词（不足则全取）。"""
|
|
seeds = load_pinterest_seeds(country)
|
|
if not seeds:
|
|
return []
|
|
if len(seeds) <= n:
|
|
return list(seeds)
|
|
return random.sample(seeds, n)
|
|
|
|
|
|
def used_terms_path(output_dir: str, country: str) -> Path:
|
|
return Path(output_dir) / "pinterest_ref" / country / "used_search_terms.json"
|
|
|
|
|
|
def load_used_terms(output_dir: str, country: str) -> List[str]:
|
|
"""读已用搜索词（跨多次运行持久化，供动态注入防重复）。"""
|
|
try:
|
|
p = used_terms_path(output_dir, country)
|
|
if p.exists():
|
|
data = json.loads(p.read_text(encoding="utf-8"))
|
|
return [str(t).strip() for t in (data.get("terms") or []) if str(t).strip()]
|
|
except Exception as e: # noqa: BLE001
|
|
print(f"[pinterest] 已用搜索词读取失败: {e}")
|
|
return []
|
|
|
|
|
|
def save_used_terms(output_dir: str, country: str, terms: List[str]) -> None:
|
|
"""持久化已用搜索词（去重保序）。"""
|
|
try:
|
|
p = used_terms_path(output_dir, country)
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
seen, out = set(), []
|
|
for t in terms:
|
|
k = t.strip().lower()
|
|
if k and k not in seen:
|
|
seen.add(k)
|
|
out.append(t.strip())
|
|
p.write_text(json.dumps({"terms": out}, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
except Exception as e: # noqa: BLE001
|
|
print(f"[pinterest] 已用搜索词保存失败: {e}")
|
|
|
|
|
|
def filter_search_terms(terms: List[str], used: List[str], blacklist: List[str]) -> List[str]:
|
|
"""全局搜索词过滤：剔除已用、黑名单、不适合 T 恤类目、去重（大小写不敏感）。"""
|
|
used_set = {str(u).strip().lower() for u in used if str(u).strip()}
|
|
black = [str(b).strip().lower() for b in (blacklist or []) if str(b).strip()]
|
|
seen, out = set(), []
|
|
for t in terms:
|
|
s = str(t).strip()
|
|
low = s.lower()
|
|
if not s or low in seen or low in used_set:
|
|
continue
|
|
if any(b and b in low for b in black):
|
|
continue
|
|
if _UNSUITABLE.search(low):
|
|
continue
|
|
seen.add(low)
|
|
out.append(s)
|
|
return out
|
|
|
|
|
|
def merge_used(existing: List[str], new_terms: List[str]) -> List[str]:
|
|
"""合并已用搜索词（新词追加到末尾，去重保序）。"""
|
|
seen, out = set(), []
|
|
for t in list(existing) + list(new_terms):
|
|
k = str(t).strip().lower()
|
|
if k and k not in seen:
|
|
seen.add(k)
|
|
out.append(str(t).strip())
|
|
return out
|