输入一个关键词，调用秘塔 AI 搜索 API 并生成调研笔记

输入一个关键词，调用秘塔 AI 搜索的 API（POST /api/v1/search），把结果整理成一份带链接、日期和检索时间的 Markdown 调研笔记。重点处理一个坑：秘塔出错时仍然返回 HTTP 200，错误放在 body 的 errCode 里，只检查状态码的话会以为调用成功了。

先看清成品

• 测试： pytest 2 项通过。
• 模拟服务器： 生成了 3 条结果的笔记。
• 真实接口： 用无效 Key 调用，程序输出“✗ 秘塔返回错误 errCode=2005：API密钥无效”。

没有在 box 上跑的部分： box 上没有可用的秘塔 API Key，唯一对真实接口做过的调用是用无效 Key 测错误返回。

• 实测结果： 错误也是 HTTP 200，body 里带 errCode：
  • search 返回 errCode 2005“API密钥无效”。
  • reader 返回 errCode 500“服务器错误”。
• 其余部分： 都在我自己写的模拟服务器 fake_metaso.py 上跑。它返回的是 3 条虚构结果。
• 字段名的来源： 响应字段（webpages 里的 title、link、snippet、score、date）参考的是开放平台 playground 页面和第三方集成的说明，没有找到正式的官方字段文档，接入前请先用真实 Key 打印一次完整响应。

合规提醒：METASO_API_KEY 只放在环境变量里。搜索结果和网页正文的版权属于原作者：笔记里保留来源链接，引用时注明出处，不要整篇转载。使用读取网页接口时，遵守对方网站的规则（服务条款、robots），不要批量抓取，也不要抓取需要登录才能看的页面。调用会消耗积分，价格和积分规则以秘塔开放平台的页面为准，本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

45-metaso-A-src/
├── metaso_search.py
├── fake_metaso.py
└── test_metaso.py

第 1 步：代码

metaso_search.py：

"""秘塔AI搜索 API → 带来源的 Markdown 调研笔记。
接口：POST https://metaso.cn/api/v1/search，Bearer 鉴权；参数 q、scope（webpage/document/scholar/podcast/video/image）、
includeSummary、size。返回里网页结果在 webpages[]，每条有 title / link / snippet / date 等字段。
注意：实测出错时 HTTP 状态码也是 200，错误放在响应体的 errCode / errMsg 里，必须检查。
环境变量：METASO_API_KEY、METASO_BASE（默认 https://metaso.cn/api/v1）
用法：python metaso_search.py "关键词" [--scope scholar] [--size 10] [--out 笔记.md]"""
import argparse, os, sys, time
from datetime import date
import requests

BASE = os.environ.get("METASO_BASE", "https://metaso.cn/api/v1")

class MetasoError(Exception): pass

def post(path, body, accept="application/json"):
    key = os.environ.get("METASO_API_KEY")
    if not key: raise MetasoError("没有设置 METASO_API_KEY（在 metaso.cn 的 API Keys 页面创建）")
    r = requests.post(f"{BASE}/{path}", json=body, timeout=60,
                      headers={"Authorization": f"Bearer {key}", "Accept": accept, "Content-Type": "application/json"})
    r.raise_for_status()
    if "json" in r.headers.get("Content-Type", ""):
        d = r.json()
        if isinstance(d, dict) and d.get("errCode") not in (None, 0):      # HTTP 200 也可能是错误
            raise MetasoError(f"秘塔返回错误 errCode={d['errCode']}：{d.get('errMsg')}")
        return d
    return r.text

def search(q, scope="webpage", size=10, summary=False):
    return post("search", {"q": q, "scope": scope, "includeSummary": summary, "size": size})

def to_markdown(q, d, scope):
    items = d.get("webpages") or d.get(f"{scope}s") or []
    lines = [f"# 调研笔记：{q}", "", f"- 检索时间：{time.strftime('%Y-%m-%d %H:%M')}（来源：秘塔AI搜索 API，scope={scope}）",
             f"- 结果数：{len(items)}" + (f"（总计约 {d['total']}）" if d.get("total") else ""), "",
             "> 摘要是搜索结果自带的片段，不代表原文全部内容；引用前请打开链接核对。", ""]
    for i, it in enumerate(items, 1):
        lines += [f"## {i}. {it.get('title', '（无标题）')}", f"- 链接：{it.get('link', '')}",
                  f"- 日期：{it.get('date') or '未知'}", "", (it.get("summary") or it.get("snippet") or "").strip(), ""]
    return "\n".join(lines)

if __name__ == "__main__":
    ap = argparse.ArgumentParser(); ap.add_argument("q"); ap.add_argument("--scope", default="webpage")
    ap.add_argument("--size", type=int, default=10); ap.add_argument("--out"); a = ap.parse_args()
    try: d = search(a.q, a.scope, a.size)
    except MetasoError as e: sys.exit(f"✗ {e}")
    md = to_markdown(a.q, d, a.scope)
    if a.out: open(a.out, "w", encoding="utf-8").write(md); print(f"✓ 已写入 {a.out}，消耗积分 {d.get('credits', '?')}")
    else: print(md)

fake_metaso.py（仅用于离线测试）：

"""本地模拟秘塔 API（没有 key 时测流程用）。响应结构按官方 playground 示例和公开集成代码整理：
search → {credits, searchParameters, total, webpages:[{title, link, snippet, score, position, date}]}；
reader → text/plain 正文；key 不对时 HTTP 200 + {"errCode":2005,"errMsg":"API密钥无效"}（这个错误和真实接口实测一致）。
搜索结果内容是写死的示例，不是真实搜索。
用法：python fake_metaso.py   （监听 127.0.0.1:18400，有效 key：mk-test）"""
from flask import Flask, jsonify, request
app = Flask(__name__)
PAGES = [
    {"title": "示例：家用 NAS 选购要点", "link": "http://127.0.0.1:18400/page/1", "snippet": "选购家用 NAS 时，先确认硬盘位数量和是否支持硬件转码。", "score": "high", "position": 1, "date": "2026-09-12"},
    {"title": "示例：RAID1 和 RAID5 的区别", "link": "http://127.0.0.1:18400/page/2", "snippet": "RAID1 是镜像，两块盘存同样数据；RAID5 至少三块盘，允许坏一块。", "score": "medium", "position": 2, "date": "2026-06-30"},
    {"title": "示例：NAS 远程访问安全", "link": "http://127.0.0.1:18400/page/3", "snippet": "不要把 NAS 管理界面直接暴露到公网，优先使用 VPN 或 Tailscale。", "score": "medium", "position": 3, "date": ""}]
TEXT = {"1": "# 家用 NAS 选购要点（示例页面）\n\n1. 硬盘位：至少两盘位，方便做 RAID1。\n2. CPU：需要转码 4K 视频时选带核显的 Intel 型号。\n3. 内存：跑 Docker 的话建议 8GB 以上。\n",
        "2": "# RAID1 和 RAID5（示例页面）\n\nRAID1 把数据同时写到两块盘；RAID5 至少三块盘，可以容忍一块盘损坏。RAID 不是备份。\n",
        "3": "# NAS 远程访问安全（示例页面）\n\n不要直接把管理端口映射到公网；用 VPN 或 Tailscale；开启两步验证。\n"}

def bad_key(): return request.headers.get("Authorization") != "Bearer mk-test"

@app.post("/api/v1/search")
def search():
    if bad_key(): return jsonify(errCode=2005, errMsg="API密钥无效")
    b = request.get_json(); n = int(b.get("size", 10))
    return jsonify(credits=n, searchParameters={"q": b["q"], "scope": b["scope"], "size": n, "includeSummary": b["includeSummary"]},
                   total=len(PAGES), webpages=PAGES[:n])

@app.post("/api/v1/reader")
def reader():
    if bad_key(): return jsonify(errCode=2005, errMsg="API密钥无效")
    pid = request.get_json()["url"].rstrip("/").split("/")[-1]
    return (TEXT[pid], 200, {"Content-Type": "text/plain; charset=utf-8"}) if pid in TEXT else (jsonify(errCode=500, errMsg="服务器错误"))

@app.get("/page/<pid>")
def page(pid): return TEXT.get(pid, "not found")

if __name__ == "__main__":
    app.run(port=18400)

test_metaso.py：

import pytest, requests
import metaso_search as m

class R:
    def __init__(s, j): s._j = j; s.headers = {"Content-Type": "application/json;charset=UTF-8"}
    def raise_for_status(s): pass
    def json(s): return s._j

def test_http200_with_errcode_raises(monkeypatch):
    monkeypatch.setenv("METASO_API_KEY", "mk-x")
    monkeypatch.setattr(requests, "post", lambda *a, **k: R({"errCode": 2005, "errMsg": "API密钥无效"}))
    with pytest.raises(m.MetasoError, match="2005"): m.search("x")

def test_markdown_has_sources():
    md = m.to_markdown("NAS", {"total": 1, "webpages": [{"title": "T", "link": "https://a.b", "snippet": "S", "date": ""}]}, "webpage")
    assert "https://a.b" in md and "日期：未知" in md

第 2 步：运行

$ export METASO_API_KEY=mk-...
$ python metaso_search.py "家用 NAS 怎么选" --out nas.md
# box 上：python fake_metaso.py 后设置 METASO_BASE=http://127.0.0.1:18400/api/v1 METASO_API_KEY=mk-test
✓ 已写入 nas.md，消耗积分 10          ← 积分数是模拟服务器的假数据

nas.md 的开头：

# 调研笔记：家用 NAS 怎么选

- 检索时间：2026-10-21 17:03（来源：秘塔AI搜索 API，scope=webpage）
- 结果数：3（总计约 3）

> 摘要是搜索结果自带的片段，不代表原文全部内容；引用前请打开链接核对。

## 1. 示例：家用 NAS 选购要点
- 链接：http://127.0.0.1:18400/page/1
- 日期：2026-09-12

选购家用 NAS 时，先确认硬盘位数量和是否支持硬件转码。

调试示例：“成功”的失败

第一版只用 r.raise_for_status() 检查状态码。我用一个无效 Key 调用真实接口，状态码是 200，程序没有报错，写出来的笔记却是空的。打印 body 才看到：

{"errCode":2005,"errMsg":"API密钥无效"}

现在 post() 只要发现 body 里有 errCode，就抛出 MetasoError。修改后用真实接口复测：

✗ 秘塔返回错误 errCode=2005：API密钥无效

常见问题

模板规则

• 每次调用都检查 body 里的错误码。
• 笔记要保留链接、日期和检索时间。
• 摘要只作为线索，引用前打开原文核对。

项目资源

搜索 API 用起来很简单，难的是不被“HTTP 200”骗过。