使用 Exa API 搭建每日搜索监控：SQLite 去重与 UTC+8 简报

想每天追踪某个主题的新文章，又不想重复看到同一篇。 这篇做 exa_watch.py：每次从上次运行时间往前 1 天开始搜索，用 SQLite 记录见过的网址，只把新结果写进当天的简报；时间按 UTC+8 显示。

先看清成品

• 替身上运行 3 次：第一次往前 7 天，之后从上次运行往前 1 天；第二天只出现新文章。
• UTC 发布时间正确换算为 UTC+8（2026-10-21T20:30Z → 2026-10-22 04:30）。
• 发现并修复一个会丢数据的问题：同一天第二次运行覆盖了上午的简报。pytest 3 项通过。

合规提醒：Exa 返回的是第三方网页内容，转载或商用前要遵守原网站的版权和使用条款；不要用 Exa 采集个人信息。本文只使用 Exa 官方 API 和官方 Python SDK（exa-py）。API key 在 Exa 控制台（dashboard.exa.ai）创建，放在环境变量 EXA_API_KEY 中。调用费用以 Exa 官方价格页为准（本文不列价格）。

运行说明：box 上没有 Exa API key。我用无效 key 真实调用了 POST https://api.exa.ai/search，得到 HTTP 401 {"error":"Invalid API key. …","tag":"INVALID_API_KEY"}；通过官方 SDK（exa-py 2.26.0）调用时，SDK 抛出的是 ValueError（2026-10-21 20:06，UTC+8）。其余响应来自我写的本地替身：两条结果照抄官方快速入门页面的示例响应，另外的结果和网页正文是我编的（example.com）。SDK 通过官方的 base_url 参数指向替身。所有脚本和测试都在 box 上真实运行过。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

74-exa-B-src/
├── exa_watch.py
├── fake_exa.py
├── demo_local.py
└── test_watch.py

第 1 步：脚本

"""每天运行一次：用 Exa 搜索若干主题中新发布的网页，只把以前没见过的结果写进当天的简报。
- 用 SQLite 记录见过的网址（去掉 utm_*、www.、末尾斜杠后比较）
- 搜索起点 = 上次运行时间往前 1 天（发布日期常常只精确到天，留出重叠，再靠去重过滤）
- 时间按 UTC+8 显示；Exa 返回的 publishedDate 是 UTC
调用频率和额度以 Exa 控制台为准；不要把间隔设得过短。"""
import argparse, datetime as dt, os, sqlite3, sys
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
from exa_py import Exa

CST = dt.timezone(dt.timedelta(hours=8))

def norm_url(u):
    p = urlsplit(u); q = urlencode([(k, v) for k, v in parse_qsl(p.query) if not k.lower().startswith("utm_")])
    return urlunsplit((p.scheme.lower(), p.netloc.lower().removeprefix("www."), p.path.rstrip("/") or "/", q, ""))

def strip_tracking(u):
    """简报中的链接去掉 utm_* 跟踪参数（与 74A 相同）。"""
    p = urlsplit(u); q = urlencode([(k, v) for k, v in parse_qsl(p.query) if not k.lower().startswith("utm_")])
    return urlunsplit((p.scheme, p.netloc, p.path, q, p.fragment))

def db(path):
    c = sqlite3.connect(path)
    c.execute("create table if not exists seen(key text primary key, url text, title text, topic text, published text, first_seen text)")
    c.execute("create table if not exists runs(topic text primary key, last_run text)")
    return c

def run_topic(c, exa, topic, now, n=20):
    row = c.execute("select last_run from runs where topic=?", (topic,)).fetchone()
    since = (dt.datetime.fromisoformat(row[0]) - dt.timedelta(days=1)) if row else now - dt.timedelta(days=7)
    r = exa.search(topic, type="auto", num_results=n, start_published_date=since.astimezone(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z"),
                   contents={"highlights": True})
    new = []
    for x in r.results:
        k = norm_url(x.url)
        if c.execute("select 1 from seen where key=?", (k,)).fetchone(): continue
        c.execute("insert into seen values(?,?,?,?,?,?)", (k, x.url, x.title, topic, x.published_date, now.isoformat())); new.append(x)
    c.execute("insert or replace into runs values(?,?)", (topic, now.isoformat())); c.commit()
    return new

def cst(ts):
    if not ts: return "日期未知"
    return dt.datetime.fromisoformat(ts.replace("Z", "+00:00")).astimezone(CST).strftime("%Y-%m-%d %H:%M")

def digest(now, found, header=True):
    L = ([f"# 每日追踪 {now.astimezone(CST):%Y-%m-%d}（UTC+8）", ""] if header else []) + [f"## {now.astimezone(CST):%H:%M} 运行", ""]
    for topic, items in found.items():
        L += [f"### {topic}（新增 {len(items)}）", ""] + ([f"- [{x.title}]({strip_tracking(x.url)}) · 发布 {cst(x.published_date)} (UTC+8)" for x in items] or ["- 无新结果"]) + [""]
    return "\n".join(L)

def main(argv=None, now=None):
    ap = argparse.ArgumentParser(); ap.add_argument("topics", nargs="+"); ap.add_argument("--db", default="watch.db"); a = ap.parse_args(argv)
    now = now or dt.datetime.now(CST)
    exa = Exa(api_key=os.environ.get("EXA_API_KEY"), base_url=os.environ.get("EXA_BASE_URL", "https://api.exa.ai"), user_agent="binbho-tutorial/1.0")
    c = db(a.db); found = {t: run_topic(c, exa, t, now) for t in a.topics}
    path = f"digest-{now.astimezone(CST):%Y%m%d}.md"
    # 同一天多次运行时追加，不能覆盖：结果已记入数据库，覆盖后上午的新结果就永远丢了（调试时实际发生过）
    exists = os.path.exists(path)
    with open(path, "a", encoding="utf-8") as f: f.write(("\n" if exists else "") + digest(now, found, header=not exists))
    print(f"新增 {sum(map(len, found.values()))} 条 → {path}"); return 0

if __name__ == "__main__":
    sys.exit(main())

第 2 步：演示

"""用本地替身演示三次运行（同一天两次、第二天一次），不连接 Exa。"""
import datetime as dt, os, fake_exa as F
st = F.State(); srv, url = F.make(st); os.environ["EXA_BASE_URL"] = url; os.environ.setdefault("EXA_API_KEY", "good")
import exa_watch as W
try:
    W.main(["RAG retrieval"], now=dt.datetime(2026, 10, 21, 8, 0, tzinfo=W.CST))
    W.main(["RAG retrieval"], now=dt.datetime(2026, 10, 21, 20, 0, tzinfo=W.CST))
    st.results.append({"id": "https://example.com/new", "title": "示例：新发布的文章", "url": "https://example.com/new?utm_medium=rss",
                       "publishedDate": "2026-10-21T20:30:00.000Z", "highlights": ["x"]})
    W.main(["RAG retrieval"], now=dt.datetime(2026, 10, 22, 8, 0, tzinfo=W.CST))
finally: srv.shutdown()
print("请求起点：", [b["startPublishedDate"] for _, b in st.bodies])

第 3 步：测试

import datetime as dt, pytest, fake_exa as F, exa_watch as W

CST = W.CST

@pytest.fixture
def exa(monkeypatch, tmp_path):
    st = F.State(); srv, url = F.make(st)
    monkeypatch.setenv("EXA_BASE_URL", url); monkeypatch.setenv("EXA_API_KEY", "good"); monkeypatch.chdir(tmp_path)
    yield st; srv.shutdown()

def test_two_runs(exa):
    d1 = dt.datetime(2026, 10, 21, 8, 0, tzinfo=CST)
    W.main(["RAG retrieval"], now=d1)
    assert exa.bodies[0][1]["startPublishedDate"] == "2026-10-14T00:00:00.000Z"     # 首次：往前 7 天，换成 UTC
    first = open("digest-20261021.md", encoding="utf-8").read()
    assert "新增 1" in first and "发布 2026-10-19 16:00 (UTC+8)" in first           # 只有发布日期 ≥ 起点的结果；两个网址版本合并
    exa.results.append({"id": "https://example.com/new", "title": "新文章", "url": "https://example.com/new", "publishedDate": "2026-10-21T20:30:00.000Z", "highlights": ["x"]})
    d2 = dt.datetime(2026, 10, 22, 8, 0, tzinfo=CST)
    W.main(["RAG retrieval"], now=d2)
    assert exa.bodies[1][1]["startPublishedDate"] == "2026-10-20T00:00:00.000Z"     # 上次运行往前 1 天
    second = open("digest-20261022.md", encoding="utf-8").read()
    assert "新增 1" in second and "新文章" in second and "示例：RAG" not in second
    assert "发布 2026-10-22 04:30 (UTC+8)" in second

def test_utc_to_cst():
    assert W.cst("2026-10-21T20:30:00.000Z") == "2026-10-22 04:30" and W.cst(None) == "日期未知"

def test_same_day_appends(exa):
    W.main(["RAG retrieval"], now=dt.datetime(2026, 10, 21, 8, 0, tzinfo=CST))
    W.main(["RAG retrieval"], now=dt.datetime(2026, 10, 21, 20, 0, tzinfo=CST))
    d = open("digest-20261021.md", encoding="utf-8").read()
    assert d.count("# 每日追踪") == 1 and "## 08:00 运行" in d and "## 20:00 运行" in d and "示例：RAG 检索调优指南" in d
    assert "utm_" not in d

第 4 步：定时运行

# crontab -e，每天 08:00（服务器时区为 UTC+8 时）
0 8 * * * cd ~/exa-watch && EXA_API_KEY=… python exa_watch.py "RAG retrieval" "vector database release" >> watch.log 2>&1

第 5 步：运行

$ python demo_local.py
新增 1 条 → digest-20261021.md
新增 0 条 → digest-20261021.md
新增 1 条 → digest-20261022.md
请求起点： ['2026-10-14T00:00:00.000Z', '2026-10-20T00:00:00.000Z', '2026-10-20T12:00:00.000Z']
--- digest-20261021.md
# 每日追踪 2026-10-21（UTC+8）

## 08:00 运行

### RAG retrieval（新增 1）

- [示例：RAG 检索调优指南](https://example.com/rag-guide) · 发布 2026-10-19 16:00 (UTC+8)

## 20:00 运行

### RAG retrieval（新增 0）

- 无新结果
--- digest-20261022.md
# 每日追踪 2026-10-22（UTC+8）

## 08:00 运行

### RAG retrieval（新增 1）

- [示例：新发布的文章](https://example.com/new) · 发布 2026-10-22 04:30 (UTC+8)
$ python -m pytest -q
3 passed

怎么检查结果？

• watch.db 中 seen 表的数量只增不减。
• 同一篇文章的不同网址版本只出现一次。
• 每个主题每次运行调用一次搜索；主题多时注意额度。

调试示例

同一天运行两次，上午的结果丢了。 我在同一天的 08:00 和 20:00 各运行一次。上午找到 1 篇新文章，晚上没有新结果，结果 digest-20261021.md 变成：

## RAG retrieval（新增 0）

- 无新结果

上午那篇文章已经写进数据库的 seen 表，以后也不会再出现，等于永远丢了。原因是简报文件用 "w" 模式打开。修复：文件已存在时追加，每次运行用“## HH:MM 运行”分节。

常见问题

模板规则

• 记录见过的网址，不重复推送。
• 起始时间留出重叠。
• 文件追加，不覆盖。
• 时间统一换算成 UTC+8 显示。

项目资源

每天只看新东西，旧的交给数据库记住。