>

用 API 监控中国科技创投新闻：Python 自动聚合与关键词追踪

Reading Time: < 1 minute

用 API 监控中国科技创投新闻：Python 自动聚合与关键词追踪实战

一、问题：科技创投的信息更新太快，人力盯不过来

如果你在做行业研究、竞品调研，或者只是想第一时间知道某个赛道有没有新的融资、新产品、新政策，那你一定体会过一件事：中国科技创投领域的新闻更新速度极快，而且分散在无数个来源里。今天这家公司拿了融资，明天那个赛道出了新玩家，信息一多，人力根本盯不过来。

具体到工程实现上，痛点主要有几个：

  • 信息源太分散，没有一个统一入口；
  • 你只关心某几个关键词（比如某个行业、某家公司、某种技术），但大多数资讯流是「全量推送」，噪声很大；
  • 想做归档、做统计、做时间线，就必须先把这些新闻变成结构化数据，而不是一堆网页。

二、手动方式的痛点

最原始的做法，就是每天早上打开一堆网站和公众号，逐条看、逐条记。这种方式的问题非常直接：

  1. 重复劳动：每天做一样的事，还容易漏；
  2. 无法沉淀：看过就过了，想回头做「过去三个月某关键词出现频率」的分析，根本没有数据底子；
  3. 不可扩展：今天你盯 3 个关键词还行，等你要盯 30 个，人就废了。

稍微进阶一点的同学会自己写爬虫。但正如上一篇讲大宗交易时说的——自研爬虫维护成本高，页面一改版就得返工，还要处理反爬、编码、去重等一堆杂活。对于「我只是想拿到干净的新闻列表」这个朴素需求来说，这有点重了。

三、方案：用 Actor 按关键词拉取创投新闻

Apify 上的 china-tech-startup-news 这个 Actor，正好解决这个需求。它的输入非常克制，只有两个参数：

  • keyword：你关心的关键词；
  • maxResults：最多返回多少条。

Actor 页面：https://apify.com/nexgendata/china-tech-startup-news?fpr=2ayu9b

3.1 最小可运行示例

同样用 Apify 的 run-sync-get-dataset-items 端点，一次请求同步拿数据：

import requests
APIFY_TOKEN = "<你的-apify-token>"
ACTOR_SLUG = "nexgendata~china-tech-startup-news"
ENDPOINT = (
    f"https://api.apify.com/v2/acts/{ACTOR_SLUG}"
    f"/run-sync-get-dataset-items?token={APIFY_TOKEN}"
)
payload = {
    "keyword": "人工智能",   # 你关心的关键词
    "maxResults": 50,
}
resp = requests.post(ENDPOINT, json=payload, timeout=180)
resp.raise_for_status()
items = resp.json()
print(f"关键词『{payload['keyword']}』共抓到 {len(items)} 条新闻")
for row in items[:3]:
    print(row)

3.2 诚实说明输出结构

这里要老实讲：china-tech-startup-news 没有对外声明固定的输出字段 schema。所以别去假设「一定有 title / url / date」这些字段——正确的姿势是先跑一次，把实际返回的键打印出来，再基于真实结构去写后续处理。这样脚本更健壮，不会因为你猜错字段名而崩。

# 探测实际返回了哪些字段
if items:
    sample = items[0]
    print("单条记录包含的字段：", list(sample.keys()))
else:
    print("没有返回数据，换个关键词或调大 maxResults 试试")

3.3 多关键词批量监控

真正实用的场景，是一次盯多个关键词。把上面的调用封装成函数，循环跑一遍即可：

import time
def fetch_news(keyword, max_results=50):
    payload = {"keyword": keyword, "maxResults": max_results}
    r = requests.post(ENDPOINT, json=payload, timeout=180)
    r.raise_for_status()
    return r.json()
watchlist = ["人工智能", "新能源", "半导体", "机器人"]
all_news = {}
for kw in watchlist:
    rows = fetch_news(kw)
    all_news[kw] = rows
    print(f"{kw}: {len(rows)} 条")
    time.sleep(2)  # 温柔一点，别把请求打太密

拿到 all_news 之后，你可以按关键词分组落库、去重、做时间线，或者简单统计每个关键词今天出现了多少条新闻，快速判断哪个赛道最近最热闹。

四、定时与自动化

监控类需求的价值，八成来自「自动、持续」。两条路：

  • 平台侧定时：在 Apify 上给 Actor 配 Schedule，比如每天早上 8 点自动跑，跑完的结果直接存进 Apify 的数据集，你随时拉取。
  • 自己这边定时：用 cron 或者任意任务调度器触发上面的脚本。

一个每日早晨 8 点执行的 crontab 示例：

0 8 * * * /usr/bin/python3 /home/you/jobs/monitor_vc_news.py >> /home/you/logs/vc.log 2>&1

再加一层去重逻辑，就能做到「只推送今天新出现的、我还没见过的新闻」。思路很简单：把已经处理过的新闻做一个唯一标识（比如根据探测到的字段拼一个 key），存进本地集合或数据库，每次跑完先比对：

seen = set()  # 实际项目里换成持久化存储
def dedupe(rows):
    fresh = []
    for r in rows:
        # 用探测到的真实字段拼唯一 key（示意）
        key = str(sorted(r.items()))
        if key not in seen:
            seen.add(key)
            fresh.append(r)
    return fresh

五、小结

监控中国科技创投新闻，难点不在「分析」，而在「持续、稳定地把分散信息变成结构化数据」。china-tech-startup-news 这个 Actor 把这件苦活收敛成 keyword + maxResults 两个参数，你写几十行 Python 就能搭起一套多关键词的自动监控管线，把精力留给判断本身。

想了解更多面向中文与 APAC 场景的数据采集 Actor，可以到 https://thenextgennexus.com 逛逛。有踩坑心得欢迎评论区交流。