将采集脚本封装成任务 API：提交、进度、结果与取消
用 FastAPI 和 asyncio 建立单进程采集任务接口，返回任务 ID，查询进度，下载结果，处理幂等提交和取消，并明确持久化边界。
把采集脚本直接放在 HTTP 请求里等待完成，调用方容易超时，也无法知道任务进行到哪里。任务 API 应先返回身份，再让调用方查询进度、取得结果或请求取消。
本文用一个受限的文档标题采集器演示完整协议，重点是任务行为而不是提取复杂度。它是本地单进程实现：任务与结果保存在内存，重启不恢复，不用于声称生产级持久调度。
准备输入地址
先在浏览器找到下列页面并确认内容，再从地址栏复制完整地址。服务或仓库地址从相应账户设置、官方项目的接入说明或克隆菜单取得。不要从示例输出中反推真实地址。
DOC_URL_1：Python 教程 introduction.html 页地址。DOC_URL_2：同目录 controlflow.html 页地址。
在执行后续命令的同一个 Bash/WSL 终端中设置：
export DOC_URL_1='<paste-the-actual-doc-url-1>'
export DOC_URL_2='<paste-the-actual-doc-url-2>'下文 JSON 或任务参数中的 <变量名> 是待替换标记，提交前填写刚复制的实际值；JSON 不会自动展开环境变量。结果示例中的 example.16yun.cn 只表示来源字段的形状，不是本文部署的演示站。
安装并限定允许采集的来源
使用 Python 3.12：
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install fastapi uvicorn httpx beautifulsoup4
export CRAWL_API_TOKEN='replace-with-your-local-api-token'示例只允许 Python 教程路径的 HTTPS URL，并关闭自动重定向，避免任意输入把服务变成访问内部网络的入口。真实服务应按业务来源设计范围，不仅检查 URL 能解析。
完整服务代码
创建 app.py：
import asyncio
import hashlib
import json
import logging
import os
import secrets
import uuid
from contextlib import asynccontextmanager
from urllib.parse import urlsplit
import httpx
from bs4 import BeautifulSoup
from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, HttpUrl
TOKEN = os.environ["CRAWL_API_TOKEN"]
jobs, tasks, keys = {}, {}, {}
slots = asyncio.Semaphore(2)
logger = logging.getLogger(__name__)
async def authorize(x_api_key: str = Header()):
if not secrets.compare_digest(x_api_key, TOKEN):
raise HTTPException(401, "invalid token")
@asynccontextmanager
async def lifespan(app):
yield
for task in tasks.values():
if not task.done():
task.cancel()
await asyncio.gather(*tasks.values(), return_exceptions=True)
app = FastAPI(lifespan=lifespan, dependencies=[Depends(authorize)])
class Submission(BaseModel):
urls: list[HttpUrl] = Field(min_length=1, max_length=10)
def get_job(job_id):
if job_id not in jobs:
raise HTTPException(404, "unknown job")
return jobs[job_id]
async def run_job(job_id, urls):
job = jobs[job_id]
try:
async with slots:
job["state"] = "running"
async with httpx.AsyncClient(timeout=20, follow_redirects=False) as client:
for url in urls:
try:
response = await client.get(url)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
title = soup.title.get_text(strip=True) if soup.title else ""
if not title:
raise ValueError("missing title")
job["records"].append({"source_url": url, "title": title})
except (httpx.HTTPError, ValueError) as error:
job["errors"].append({"source_url": url, "error_type": type(error).__name__})
job["processed"] += 1
job["state"] = "completed_with_errors" if job["errors"] else "completed"
except asyncio.CancelledError:
job["state"] = "cancelled"
raise
except Exception:
job["state"] = "failed"
logger.exception("unexpected job failure")
raise
@app.post("/jobs", status_code=202)
async def submit(body: Submission, idempotency_key: str = Header(min_length=1, max_length=80)):
urls = list(dict.fromkeys(str(value) for value in body.urls))
for url in urls:
parsed = urlsplit(url)
if (parsed.scheme != "https" or parsed.hostname != "docs.python.org"
or parsed.port not in {None, 443}
or not parsed.path.startswith("/3/tutorial/")
or parsed.username is not None):
raise HTTPException(400, "source is outside allowed scope")
digest = hashlib.sha256(json.dumps(urls).encode()).hexdigest()
if idempotency_key in keys:
job_id, old_digest = keys[idempotency_key]
if old_digest != digest:
raise HTTPException(409, "key already belongs to different inputs")
return {"job_id": job_id, "state": jobs[job_id]["state"]}
if len(jobs) >= 100:
raise HTTPException(503, "teaching instance capacity reached")
job_id = uuid.uuid4().hex
jobs[job_id] = {"state": "queued", "total": len(urls), "processed": 0,
"records": [], "errors": []}
keys[idempotency_key] = (job_id, digest)
tasks[job_id] = asyncio.create_task(run_job(job_id, urls))
return {"job_id": job_id, "state": "queued"}
@app.get("/jobs/{job_id}")
async def progress(job_id: str):
job = get_job(job_id)
return {**{key: job[key] for key in ("state", "total", "processed")},
"accepted": len(job["records"]), "failed": len(job["errors"])}
@app.get("/jobs/{job_id}/result")
async def result(job_id: str):
job = get_job(job_id)
if job["state"] in {"queued", "running", "cancelling"}:
raise HTTPException(409, "job is not terminal")
return JSONResponse(job, headers={
"Content-Disposition": f'attachment; filename="{job_id}.json"'
})
@app.post("/jobs/{job_id}/cancel")
async def cancel(job_id: str):
job = get_job(job_id)
task = tasks[job_id]
if not task.done():
job["state"] = "cancelling"
task.cancel()
try:
await task
except asyncio.CancelledError:
job["state"] = "cancelled"
return {"job_id": job_id, "state": job["state"]}启动：
uvicorn app:app --host 127.0.0.1 --port 8008 --workers 1不要加多个 worker 来扩展这个内存字典实现。不同进程看不到对方任务，查询可能返回未知 ID。跨进程与重启恢复需要独立任务存储和队列，不是把 workers 改大就完成。
提交与查询完整流程
在另一个终端配置相同 token，并重新设置前面的 DOC_URL_1、DOC_URL_2，再提交两页：
python -c 'import json,os; print(json.dumps({"urls":[os.environ["DOC_URL_1"],os.environ["DOC_URL_2"]]}))' > submission.json
curl --fail-with-body -X POST http://localhost:8008/jobs -H "X-API-Key: $CRAWL_API_TOKEN" -H 'Idempotency-Key: tutorial-batch-1' -H 'Content-Type: application/json' -d @submission.json > submitted.json从 submitted.json 取 job_id，再查询：
CRAWL_JOB_ID=$(python -c 'import json;print(json.load(open("submitted.json"))["job_id"])')
curl --fail-with-body -H "X-API-Key: $CRAWL_API_TOKEN" "http://localhost:8008/jobs/$CRAWL_JOB_ID"
curl --fail-with-body -H "X-API-Key: $CRAWL_API_TOKEN" "http://localhost:8008/jobs/$CRAWL_JOB_ID/result" -o result.json任务未进入终态时 result 返回 409，稍后根据进度再取。这里不是让客户端无间隔忙轮询，应设置合理间隔与有限等待策略，保留任务 ID 以便继续。
completed_with_errors 表示所有输入已尝试，但部分失败；不能按 completed 前缀把它当作全成功。processed 是已经得到成功或失败结论的输入数，不等于合格条数。
幂等提交与取消怎么解释
同一个 Idempotency-Key 和同样输入返回原任务，不重复创建；同一 key 对应不同输入返回 409。该约定只在当前进程存活期间有效，重启后字典丢失，不是持久幂等保证。
需要取消时：
curl --fail-with-body -X POST -H "X-API-Key: $CRAWL_API_TOKEN" "http://localhost:8008/jobs/$CRAWL_JOB_ID/cancel"短任务可能已经完成，此时返回实际终态，不强行改成 cancelled。排队中的任务也能取消；代码等待取消处理，并覆盖尚未开始执行的情况。
取消保留已经完成的内存记录，result 会包含 cancelled 状态与部分数据。它不撤销之前的远程副作用；本文只做 GET，接入会写入外部系统的任务时需另定取消与补偿规则。asyncio 取消语义见 官方任务文档。
换成已有采集器时保持什么
把 run_job 中的一次 URL 处理替换成你已有的业务采集，但保持每条输入最终去向、进度定义和异常传播。阻塞库不能直接塞进异步事件循环而不考虑其执行方式；浏览器进程也需要任务级结束逻辑。
FastAPI 的普通后台任务不是自动提供状态、取消和持久队列的系统，相关定位见 官方后台任务说明。本例用显式 Task 管理，是为了展示协议行为，不代替完整的生产调度平台。
常见问题与修复
返回 202 后马上查到完成。 小任务可能很快结束，这不是异常。客户端应支持所有合法状态，不依赖一定观察到 running。
取消后仍有记录。 这些是已经完成的数据，状态明确标为部分结果；不要为了让数量归零删除有效结果。
重启后找不到 ID。 本例内存状态已丢失。需要持久恢复时将任务和结果迁到独立存储，不声称旧 ID 自动恢复。
同一个 key 拿到旧任务。 检查是否有意复用提交身份。新业务批次使用新的 key，重试同一提交才复用。
来源被拒绝。 检查允许域名与路径，不为方便测试直接开放任意地址。重新设计范围后再接入其他站点。
进度卡住。 区分请求超时、任务异常与阻塞代码；检查服务器日志和状态，不只延长 HTTP 客户端等待。
一个可用的任务接口应让调用方知道任务是谁、做到哪里、哪些结果可取以及取消意味着什么。把这些语义先落实，再考虑持久化和多 Worker。
需要企业代理方案？
我们可根据目标站点、并发规模与稳定性目标提供定制方案。