用 Perplexity Search API 做一个竞品新闻监控

用 Perplexity Search API 做一个竞品新闻监控：设定关键词和只看的网站，每天搜索一次最近一周的结果，新文章存进 SQLite（同一篇文章不会重复记录），随时导出成 Excel 能打开的 CSV。

先看清成品

成品是 monitor.py 加 watch.json 配置。本文已用模拟客户端实际运行测试并通过：网址去掉跟踪参数后去重，两个关键词搜到同一篇文章只记一次，第二次运行没有新结果，请求带上了网站过滤和时间过滤参数，CSV 表头正确。另外用真实 SDK 和无效密钥发了请求，服务器返回 401“Invalid API key”，SDK 能接受这些参数；但 401 发生在参数校验之前，参数本身只按官方文档核对过。本文环境没有有效的 Perplexity API 密钥，没有拿到真实搜索结果。

注意：Perplexity API 按用量计费，价格以官方价格页为准，每次响应的 usage 字段会给出实际费用。请在服务可用的地区使用，不要绕过地区限制。搜索结果和 AI 总结可能出错，发布前要点开来源核对；转载他人文章要遵守版权规定。

做完之后你会得到什么？

需要准备哪些环境？

第一步：Search API 和 Agent API 有什么区别？

官方文档说明，max_results 取值 1～20，默认 10；model、preset 这类 Agent API 参数不能用在搜索请求里。本文用到的过滤参数都列在官方 API 参考中：search_domain_filter（网站）、search_recency_filter（时间范围）。

第二步：代码怎么写？

1. watch.json

{
  "queries": ["某品牌 新品 发布", "某品牌 价格 调整"],
  "domains": ["36kr.com", "ithome.com", "sspai.com"],
  "recency": "week",
  "max_results": 10
}

把“某品牌”换成你要关注的公司或产品。

2. monitor.py

"""竞品新闻监控：用 Perplexity Search API 定期搜索，新结果存入 SQLite 并导出 CSV。
用法：python monitor.py            # 搜索并记录新结果
      python monitor.py --export   # 导出全部记录到 news.csv
"""
import csv
import json
import sqlite3
import sys
from datetime import datetime
from urllib.parse import urlsplit

DB = "news.db"


def init(db):
    db.execute("""CREATE TABLE IF NOT EXISTS news (
        url TEXT PRIMARY KEY, title TEXT, query TEXT, date TEXT, snippet TEXT, found_at TEXT)""")


def normalize(url: str) -> str:
    """去掉 #片段 和常见跟踪参数，避免同一篇文章重复入库。"""
    p = urlsplit(url)
    q = "&".join(x for x in p.query.split("&") if x and not x.startswith(("utm_", "spm=", "from=")))
    return f"{p.scheme}://{p.netloc.lower()}{p.path}" + (f"?{q}" if q else "")


def run(client, cfg, db):
    init(db)
    new = []
    for q in cfg["queries"]:
        res = client.search.create(
            query=q,
            max_results=cfg.get("max_results", 10),
            search_domain_filter=cfg.get("domains") or None,
            search_recency_filter=cfg.get("recency", "week"),
        )
        for r in res.results:
            url = normalize(r.url)
            cur = db.execute("INSERT OR IGNORE INTO news VALUES (?, ?, ?, ?, ?, ?)",
                             (url, r.title, q, getattr(r, "date", None), (getattr(r, "snippet", "") or "")[:300],
                              datetime.now().isoformat(timespec="seconds")))
            if cur.rowcount:
                new.append((q, r.title, url))
    db.commit()
    return new


def export(db, path="news.csv"):
    init(db)
    rows = db.execute("SELECT found_at, query, date, title, url FROM news ORDER BY found_at DESC").fetchall()
    with open(path, "w", newline="", encoding="utf-8-sig") as f:
        w = csv.writer(f)
        w.writerow(["发现时间", "关键词", "发布日期", "标题", "链接"])
        w.writerows(rows)
    return len(rows)


def main(argv=None, client=None):
    argv = sys.argv[1:] if argv is None else argv
    db = sqlite3.connect(DB)
    if "--export" in argv:
        print(f"已导出 {export(db)} 条到 news.csv")
        return
    if client is None:
        from perplexity import Perplexity
        client = Perplexity()
    cfg = json.load(open("watch.json", encoding="utf-8"))
    new = run(client, cfg, db)
    for q, title, url in new:
        print(f"[新] {q}｜{title}\n     {url}")
    print(f"本次新增 {len(new)} 条")


if __name__ == "__main__":
    main()

第三步：怎样测试和运行？

1. 离线测试

import sqlite3
from types import SimpleNamespace as NS

import monitor

CFG = {"queries": ["甲", "乙"], "domains": ["36kr.com"], "recency": "week", "max_results": 5}


class Fake:
    def __init__(self):
        self.calls = []
        self.search = NS(create=self._create)

    def _create(self, **kw):
        self.calls.append(kw)
        return NS(results=[NS(title="新品发布", url="https://36KR.com/p/1?utm_source=x&id=9#top", date="2026-10-20", snippet="……"),
                           NS(title="价格调整", url="https://36kr.com/p/2", date=None, snippet=None)])


def test_dedupe_and_params(tmp_path):
    db = sqlite3.connect(tmp_path / "n.db")
    c = Fake()
    first = monitor.run(c, CFG, db)
    assert len(first) == 2                       # 两个关键词返回相同文章，只记一次
    assert first[0][2] == "https://36kr.com/p/1?id=9"
    assert c.calls[0]["search_domain_filter"] == ["36kr.com"] and c.calls[0]["search_recency_filter"] == "week"
    assert monitor.run(c, CFG, db) == []          # 第二次运行没有新结果
    n = monitor.export(db, tmp_path / "n.csv")
    assert n == 2
    assert (tmp_path / "n.csv").read_text(encoding="utf-8-sig").startswith("发现时间,关键词")

python -m pytest -q

本文实际输出 1 passed。

2. 真实运行

export PERPLEXITY_API_KEY="你的密钥"
python monitor.py
python monitor.py --export

期望输出：

[新] 某品牌 新品 发布｜（文章标题）
     https://…
本次新增 N 条
已导出 N 条到 news.csv

第二天再运行，只会列出新出现的文章。

3. 每天自动运行

0 9 * * * cd /home/me/news-monitor && PERPLEXITY_API_KEY=你的密钥 .venv/bin/python monitor.py >> monitor.log 2>&1

出问题时该怎么调？

1. 结果太少

去掉网站白名单试试，或把 recency 改成 month。

2. 结果不相关

关键词写得更具体，加上产品型号；网站白名单只保留可靠来源（官方 API 参考说明 search_domain_filter 最多 20 个域名）。

3. 同一篇文章仍然重复

不同网址指向同一篇文章（例如手机版和电脑版）。在 normalize 里加入你发现的规则。

常见问题怎么处理？

项目资源

建议先放一个关键词手动运行几次，看看结果质量和网站过滤效果。

调整满意后再加关键词、设置定时任务，每周导出一次 CSV 浏览。