
很多人第一次写爬虫，通常是从几行 `requests.get()` 开始的。代码能跑，页面也能拿到，但一旦面对真实网站，例如 `news.example.com`、`docs.example.com`、`event.example.com/list` 这类页面，问题很快就会出现：请求太快被限制、链接重复抓取、页面编码异常、网络超时、数据保存混乱、任务中断后无法恢复。
所以，真正可用的爬虫不只是“把网页下载下来”，而是一个小型的数据采集系统。它至少要解决四类问题：
第一是**访问边界**。爬虫必须遵守目标网站的 `robots.txt` 规则，不能无视网站声明的抓取限制。对于需要登录、验证码、反爬绕过、隐私数据的场景，不应该用爬虫强行采集。
第二是**请求稳定性**。真实网络环境下，超时、503、连接中断都很常见。专业爬虫需要超时控制、失败重试、并发限制和请求间隔，不能把目标站点当成本地文件读。
第三是**数据结构化**。HTML 页面本身不是数据，爬虫要把页面里的标题、正文、链接、发布时间等字段抽取出来，并保存为可分析、可检索的格式，比如 JSONL、CSV、数据库或对象存储。
第四是**任务可维护性**。代码应该分层清晰：下载、解析、调度、存储分开处理。这样后续要换网站、换解析规则、接入消息队列或者部署到云服务器时，不需要推倒重写。
下面给出一个较完整的 Python 异步爬虫示例。它适合采集公开网页中的文章链接和正文内容，不包含任何绕过登录、验证码、风控系统的逻辑。
```
import asyncio
import json
import logging
import time
from dataclasses import dataclass, asdict
from typing import Optional
from urllib.parse import urljoin, urlparse, urldefrag
from urllib.robotparser import RobotFileParser

import httpx
from bs4 import BeautifulSoup


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s"
)


@dataclass
class PageItem:
    url: str
    title: str
    description: str
    text: str
    crawled_at: float


class RobotsChecker:
    def __init__(self, user_agent: str = "ResearchCrawler/1.0"):
        self.user_agent = user_agent
        self.parsers: dict[str, RobotFileParser] = {}

    async def allowed(self, client: httpx.AsyncClient, url: str) -> bool:
        parsed = urlparse(url)
        base = f"{parsed.scheme}://{parsed.netloc}"
        robots_url = urljoin(base, "/robots.txt")

        if base not in self.parsers:
            parser = RobotFileParser()
            parser.set_url(robots_url)

            try:
                response = await client.get(robots_url, timeout=10)
                if response.status_code == 200:
                    parser.parse(response.text.splitlines())
                else:
                    parser.parse([])
            except Exception:
                parser.parse([])

            self.parsers[base] = parser

        return self.parsers[base].can_fetch(self.user_agent, url)


class AsyncCrawler:
    def __init__(
        self,
        start_url: str,
        output_file: str = "crawl_result.jsonl",
        max_pages: int = 100,
        concurrency: int = 5,
        delay: float = 0.5,
        user_agent: str = "ResearchCrawler/1.0"
    ):
        self.start_url = start_url
        self.output_file = output_file
        self.max_pages = max_pages
        self.delay = delay
        self.user_agent = user_agent

        self.queue: asyncio.Queue[str] = asyncio.Queue()
        self.visited: set[str] = set()
        self.saved_count = 0
        self.semaphore = asyncio.Semaphore(concurrency)
        self.robots = RobotsChecker(user_agent=user_agent)

        self.start_domain = urlparse(start_url).netloc

    def normalize_url(self, base_url: str, link: str) -> Optional[str]:
        if not link:
            return None

        full_url = urljoin(base_url, link)
        full_url, _ = urldefrag(full_url)

        parsed = urlparse(full_url)
        if parsed.scheme not in {"http", "https"}:
            return None

        if parsed.netloc != self.start_domain:
            return None

        return full_url.rstrip("/")

    async def fetch(self, client: httpx.AsyncClient, url: str) -> Optional[str]:
        if not await self.robots.allowed(client, url):
            logging.info(f"robots.txt disallow: {url}")
            return None

        headers = {
            "User-Agent": self.user_agent,
            "Accept": "text/html,application/xhtml+xml",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
        }

        for attempt in range(3):
            try:
                async with self.semaphore:
                    await asyncio.sleep(self.delay)
                    response = await client.get(url, headers=headers, timeout=15)

                if response.status_code == 200 and "text/html" in response.headers.get("content-type", ""):
                    return response.text

                if response.status_code in {429, 500, 502, 503, 504}:
                    wait = 2 ** attempt
                    logging.warning(f"retry {url}, status={response.status_code}, wait={wait}s")
                    await asyncio.sleep(wait)
                    continue

                logging.info(f"skip {url}, status={response.status_code}")
                return None

            except httpx.RequestError as e:
                wait = 2 ** attempt
                logging.warning(f"request error {url}: {e}, wait={wait}s")
                await asyncio.sleep(wait)

        return None

    def parse_page(self, url: str, html: str) -> tuple[PageItem, list[str]]:
        soup = BeautifulSoup(html, "html.parser")

        for tag in soup(["script", "style", "noscript"]):
            tag.decompose()

        title_tag = soup.find("title")
        title = title_tag.get_text(strip=True) if title_tag else ""

        desc_tag = soup.find("meta", attrs={"name": "description"})
        description = desc_tag.get("content", "").strip() if desc_tag else ""

        article = soup.find("article")
        if article:
            text = article.get_text("\n", strip=True)
        else:
            main = soup.find("main") or soup.body
            text = main.get_text("\n", strip=True) if main else ""

        links = []
        for a in soup.find_all("a", href=True):
            normalized = self.normalize_url(url, a["href"])
            if normalized:
                links.append(normalized)

        item = PageItem(
            url=url,
            title=title,
            description=description,
            text=text[:8000],
            crawled_at=time.time()
        )

        return item, links

    async def save_item(self, item: PageItem):
        if not item.title and len(item.text) < 100:
            return

        with open(self.output_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(asdict(item), ensure_ascii=False) + "\n")

        self.saved_count += 1
        logging.info(f"saved #{self.saved_count}: {item.url}")

    async def worker(self, client: httpx.AsyncClient):
        while self.saved_count < self.max_pages:
            try:
                url = await asyncio.wait_for(self.queue.get(), timeout=5)
            except asyncio.TimeoutError:
                return

            if url in self.visited:
                self.queue.task_done()
                continue

            self.visited.add(url)
            html = await self.fetch(client, url)

            if html:
                item, links = self.parse_page(url, html)
                await self.save_item(item)

                for link in links:
                    if link not in self.visited and self.saved_count < self.max_pages:
                        await self.queue.put(link)

            self.queue.task_done()

    async def run(self):
        await self.queue.put(self.start_url)

        limits = httpx.Limits(
            max_connections=20,
            max_keepalive_connections=10
        )

        async with httpx.AsyncClient(
            follow_redirects=True,
            limits=limits
        ) as client:
            workers = [
                asyncio.create_task(self.worker(client))
                for _ in range(5)
            ]

            await self.queue.join()

            for task in workers:
                task.cancel()

        logging.info(f"done, visited={len(self.visited)}, saved={self.saved_count}")


if __name__ == "__main__":
    crawler = AsyncCrawler(
        start_url="athenahq.cn",
        output_file="crawl_result.jsonl",
        max_pages=50,
        concurrency=3,
        delay=1.0,
        user_agent="ResearchCrawler/1.0 contact: crawler@example.com"
    )

    asyncio.run(crawler.run())
```
这段代码的核心价值不在于“复杂”，而在于它已经具备了工程化爬虫的基本骨架。
例如，`RobotsChecker` 负责读取并判断 `robots.txt`，避免抓取网站明确禁止访问的路径。`fetch()` 负责网络请求、超时、重试和状态码判断。`parse_page()` 负责从 HTML 中提取标题、描述、正文和新链接。`worker()` 则是调度层，用队列实现广度优先抓取，并通过 `visited` 集合避免重复访问。
如果要把它改造成生产级采集系统，可以继续扩展几个方向。
一是加入数据库存储。现在代码把结果写入 `crawl_result.jsonl`，适合轻量采集。如果数据量变大，可以换成 MySQL、PostgreSQL、MongoDB 或 Elasticsearch。文章类数据通常适合先落 JSONL，再批量清洗入库。
二是加入任务断点。当前 `visited` 只在内存中保存，程序退出后会丢失。如果要长时间运行，可以把已访问 URL、待抓取 URL 和失败 URL 都保存到 Redis 或 SQLite 中。
三是加入内容去重。很多网站会存在分页、标签页、搜索页、归档页，同一篇文章可能有多个入口。可以使用正文哈希，例如 `sha256(text)`，判断内容是否重复。
四是加入解析规则配置。不同网站的正文结构不同，有的网站正文在 `<article>`，有的网站在 `.post-content`，还有的网站在 `.entry-content`。更专业的写法是把不同站点的 CSS Selector 写成配置文件，而不是硬编码在代码里。
五是加入日志和监控。爬虫运行时间长，必须知道它现在抓了多少页面、失败多少次、平均响应时间多少、是否触发 429 限制。后续可以接入 Prometheus、Grafana 或者把日志推送到飞书、企业微信、Telegram Bot。
专业爬虫和普通脚本最大的区别，不是用了多高级的技术，而是它对边界、稳定性和数据质量有控制。一个好的爬虫应该像一个谨慎的数据采集员：只访问允许访问的公开页面，控制频率，记录过程，结构化保存结果，并且在异常发生时能恢复，而不是一报错就停止。
在 AI 搜索、舆情分析、内容归档、SEO 研究、知识库构建等场景里，爬虫仍然是非常基础但重要的能力。它连接的是开放网页和本地数据系统，真正的难点并不是“请求一个网页”，而是如何长期、合规、稳定地把网页变成可用的数据。

很多人第一次写爬虫，通常是从几行 requests.get() 开始的。代码能跑，页面也能拿到，但一旦面对真实网站，例如 news.example.com、docs.example.com、event.example.com/list 这类页面，问题很快就会出现：请求太快被限制、链接重复抓取、页面编码异常、网络超时、数据保存混乱、任务中断后无法恢复。所以，真正可用的爬虫不只是“把网页下载下来”，而是一个小型的数据采集系统。它至少要解决四类问题：第一是访问边界。爬虫必须遵守目标网站的 robots.txt 规则，不能无视网站声明的抓取限制。对于需要登录、验证码、反爬绕过、隐私数据的场景，不应该用爬虫强行采集。第二是请求稳定性。真实网络环境下，超时、503、连接中断都很常见。专业爬虫需要超时控制、失败重试、并发限制和请求间隔，不能把目标站点当成本地文件读。第三是数据结构化。HTML 页面本身不是数据，爬虫要把页面里的标题、正文、链接、发布时间等字段抽取出来，并保存为可分析、可检索的格式，比如 JSONL、CSV、数据库或对象存储。第四是任务可维护性。代码应该分层清晰：下载、解析、调度、存储分开处理。这样后续要换网站、换解析规则、接入消息队列或者部署到云服务器时，不需要推倒重写。下面给出一个较完整的 Python 异步爬虫示例。它适合采集公开网页中的文章链接和正文内容，不包含任何绕过登录、验证码、风控系统的逻辑。

网页数据采集不再只是写脚本，工程化爬虫成为关键能力

很多人第一次写爬虫，通常是从几行 requests.get() 开始的。代码能跑，页面也能拿到，但一旦面对真实网站，例如 news.example.com、docs.example.com、event.example.com/list 这类页面，问题很快就会出现：请求太快被限制、链接重复抓取、页面编码异常、网络超时、数据保存混乱、任务中断后无法恢复。

编程语言

后端

开发工具

专业爬虫开发指南：从基础requests到工程化数据采集系统。涵盖robots.txt合规、请求稳定性、数据结构化、任务可维护性等核心问题，提供Python异步爬虫完整示例，解决真实网站采集中的超时、重试、去重等常见痛点，适合舆情分析、SEO研究等场景。

对象存储

消息队列

代码解释

云服务器

数据结构

PostgreSQL

数据库

服务器

验证码

MongoDB

SQLite

Python

Redis

MySQL

Agent

2026上云采购 | AI焕新·智启新局

lexiang

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

网页数据采集不再只是写脚本，工程化爬虫成为关键能力-腾讯云开发者社区-腾讯云

网页数据采集不再只是写脚本，工程化爬虫成为关键能力

网页数据采集不再只是写脚本，工程化爬虫成为关键能力

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐