关键词研究工作流

竞品域名列表
    ↓ keywords_for_site（每个域名）
合并词库 + 去重
    ↓ search_volume（批量 1000 词/次）
过滤：搜索量 ≥ 100，竞争度 ≤ MEDIUM
    ↓ suggestions（高价值种子词）
扩长尾 + 再次 search_volume 验证
    ↓
输出：内容计划 CSV

import requests
 
API_KEY = "smt_live_YOUR_KEY"
BASE = "https://api.seermartech.cn/v3"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}
 
COMPETITORS = ["competitor-a.com", "competitor-b.com", "competitor-c.com"]
 
def keywords_for_site(domain: str) -> list[str]:
    resp = requests.post(
        f"{BASE}/keywords_data/google_ads/keywords_for_site/live",
        headers={**HEADERS, "Content-Type": "application/json"},
        json=[{"target": domain, "location_code": 2156, "language_code": "zh", "limit": 200}],
        timeout=120,
    )
    resp.raise_for_status()
    return [r["keyword"] for r in resp.json()["tasks"][0]["result"]]
 
all_keywords = set()
for domain in COMPETITORS:
    all_keywords.update(keywords_for_site(domain))
print(f"collected {len(all_keywords)} unique keywords")

def search_volume(keywords: list[str]) -> list[dict]:
    # 每批最多 1000 词
    results = []
    for i in range(0, len(keywords), 1000):
        batch = keywords[i:i + 1000]
        resp = requests.post(
            f"{BASE}/keywords_data/google_ads/search_volume/live",
            headers={**HEADERS, "Content-Type": "application/json"},
            json=[{"keywords": batch, "location_code": 2156, "language_code": "zh"}],
            timeout=120,
        )
        resp.raise_for_status()
        results.extend(resp.json()["tasks"][0]["result"])
    return results
 
volumes = search_volume(list(all_keywords))
qualified = [
    v for v in volumes
    if (v.get("search_volume") or 0) >= 100
    and v.get("competition") in ("LOW", "MEDIUM", None)
]
print(f"qualified: {len(qualified)} / {len(volumes)}")

def expand_keyword(seed: str, limit: int = 30) -> list[str]:
    resp = requests.post(
        f"{BASE}/keywords_data/google_ads/suggestions/live",
        headers={**HEADERS, "Content-Type": "application/json"},
        json=[{"keyword": seed, "location_code": 2156, "language_code": "zh", "limit": limit}],
        timeout=120,
    )
    resp.raise_for_status()
    return [r["keyword"] for r in resp.json()["tasks"][0]["result"]]
 
# 取搜索量最高的 10 个种子词扩词
seeds = sorted(qualified, key=lambda x: x.get("search_volume", 0), reverse=True)[:10]
long_tail = set()
for seed in seeds:
    long_tail.update(expand_keyword(seed["keyword"]))
 
final_list = search_volume(list(long_tail))

const API_KEY = "smt_live_YOUR_KEY";
const BASE = "https://api.seermartech.cn/v3";
 
async function apiPost<T>(path: string, body: unknown): Promise<T> {
  const resp = await fetch(`${BASE}${path}`, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${API_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify([body]),
  });
  return resp.json();
}
 
// Step 1 → 2 → 3 可封装为 pipeline 函数
async function researchPipeline(competitors: string[]) {
  const keywords = new Set<string>();
  for (const domain of competitors) {
    const data = await apiPost<{ tasks: [{ result: { keyword: string }[] }] }>(
      "/keywords_data/google_ads/keywords_for_site/live",
      { target: domain, location_code: 2156, language_code: "zh", limit: 200 }
    );
    data.tasks[0].result.forEach((r) => keywords.add(r.keyword));
  }
  return [...keywords];
}