爬虫系统建设实战

爬虫系统建设实战文档

—— 从零搭建一套可维护、可扩展、可自愈的招聘信息爬虫

一、先建立正确认知：爬虫不是"抓网页"，而是"建立数据生产线"

很多人学爬虫，停留在：

  • requests 请求网页
  • BeautifulSoup / XPath 抽内容
  • 存 CSV

这只能算**"脚本"**。

真正能长期跑、能维护、能适应网站变化的爬虫，应该是一条完整的数据生产线：

目标发现
→ 站点分析
→ 采集策略选择
→ 页面获取
→ 数据解析
→ 清洗标准化
→ 去重校验
→ 存储落盘
→ 上传入库
→ 监控报警
→ 规则迭代
→ 自动修复

flowchart LR
    A[目标发现] --> B[站点分析]
    B --> C[采集策略选择]
    C --> D[页面获取]
    D --> E[数据解析]
    E --> F[清洗标准化]
    F --> G[去重校验]
    G --> H[存储落盘]
    H --> I[上传入库]
    I --> J[监控报警]
    J --> K[规则迭代]
    K --> L[自动修复]
    L -.-> C

你给的这套系统可取之处就在于：它已经不是单点爬虫，而是**"配置驱动 + 多策略抓取 + AI解析 + 自动修复 + 数据闭环"**。

所以如果你未来要自己做爬虫，应该按下面这些步骤来建设。

二、做一套爬虫，建议分成 10 个阶段

阶段 1：先定义目标，而不是先写代码

你先要回答 6 个核心问题

序号 问题 典型选项
❶ 我要抓什么 招聘职位？招聘公告？商品价格？新闻正文？评论？
❷ 数据最终要长什么样 标题、发布时间、地点、岗位名称、招聘类型、正文、图片、附件、投递链接
❸ 数据源是什么类型 静态网页、动态渲染网页、接口 JSON、小程序接口、图片海报、PDF、视频字幕
❹ 抓一次就行，还是要长期增量更新 一次性采集、每日更新、每小时更新、实时监控
❺ 后续怎么用 搜索、展示、推荐、数据分析、自动通知
❻ 允许多大维护成本 只抓几个站可以写死；上百站点必须配置化；长期运行必须自动监控和自动修复

这套代码给你的经验

它不是"抓几个公司页面"，而是一开始就明确了：

维度 设计
目标 招聘公告 / 职位
输出 结构化 JSON
场景 公司官网、学校就业网、图片海报、接口数据
运行方式 定时任务 + 分片
后续 上传正式环境数据库

核心经验：你以后做任何爬虫，第一步都要先设计统一数据模型。不要先抓，抓完再想怎么存——那样后面一定会返工。

统一数据模型设计示例

{
  "announcement": {
    "title": "2026年春季校园招聘公告",
    "company_name": "某科技有限公司",
    "publish_time": "2026-03-15",
    "source_url": "https://example.com/jobs/123",
    "content_html": "<div>...</div>",
    "content_text": "纯文本正文...",
    "category": "校招",
    "graduation_year": "2026届",
    "apply_method": "在线投递",
    "images": ["https://obs.xxx/img1.jpg"],
    "attachments": ["https://obs.xxx/岗位表.pdf"]
  },
  "positions": [
    {
      "job_name": "Java开发工程师",
      "city": "北京",
      "diploma": "本科及以上",
      "salary": "15k-25k",
      "headcount": 5,
      "deadline": "2026-04-30",
      "apply_link": "https://example.com/apply/456"
    }
  ]
}

三、阶段 2：站点分析——决定用什么抓法

这是最关键的一步。不要一上来就 requests + XPath。

你应该先判断目标站属于哪种类型，然后选抓取策略。

flowchart TD
    START[目标站点] --> CHECK1{view-source 里有数据？}
    CHECK1 -->|是| A[场景A：静态HTML]
    CHECK1 -->|否| CHECK2{Network 面板有 JSON 接口？}
    CHECK2 -->|是| C[场景C：前后端分离/API]
    CHECK2 -->|否| CHECK3{数据藏在 JS 文件/变量里？}
    CHECK3 -->|是| D[场景D：JS嵌入数据]
    CHECK3 -->|否| CHECK4{需要滚动/点击/等待？}
    CHECK4 -->|是| B[场景B：动态渲染SPA]
    CHECK4 -->|否| CHECK5{主要内容在图片/PDF里？}
    CHECK5 -->|是| E[场景E：图片/PDF]
    CHECK5 -->|否| F[回到场景B或组合方案]

    A --> MA[requests + lxml/BS4]
    B --> MB[Playwright/Selenium]
    C --> MC[直调API 或 拦截响应]
    D --> MD[请求JS文件 + 正则抽JSON]
    E --> ME[OCR + PDF解析]

场景 A：页面源码里就有数据

特征：

  • view-source: 里能看到职位名称、正文、链接
  • 页面刷新后内容不变
  • 没有复杂 JS 动态请求

方案：

  • requests + lxml + BeautifulSoup + 正则辅助

优点：

  • 快、成本低、稳定、并发高

风险：

  • 网站稍微改 DOM 就失效

适合：

  • 新闻站、公告页、简单企业官网

代码示例：

import requests
from lxml import etree

def fetch_static_page(url, selectors):
    """静态页面抓取标准流程"""
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    resp = requests.get(url, headers=headers, timeout=30)
    resp.encoding = resp.apparent_encoding  # 自动检测编码

    tree = etree.HTML(resp.text)
    items = []

    for node in tree.xpath(selectors['list_xpath']):
        item = {
            'title': ''.join(node.xpath(selectors['title_xpath'])).strip(),
            'url': ''.join(node.xpath(selectors['url_xpath'])).strip(),
            'publish_time': ''.join(node.xpath(selectors['time_xpath'])).strip(),
        }
        # 补全相对链接
        if item['url'] and not item['url'].startswith('http'):
            from urllib.parse import urljoin
            item['url'] = urljoin(url, item['url'])
        items.append(item)

    return items

场景 B：页面是 JS 动态渲染

特征：

  • requests 抓源码只有框架，没有正文
  • F12 的 Elements 面板有内容，但 page source 里没有
  • 需要滚动、点击、切换 tab 才出现内容

方案：

  • Playwright / Selenium 模拟浏览器
  • 等待 DOM 渲染、处理滚动加载、点击 tab、翻页

这套系统怎么做的：

它在 spider_com.py 里用 Playwright 打开列表页和详情页，并且支持：

  • tab 点击
  • 分页点击
  • 当前页操作
  • 页面等待
  • 动态内容加载

代码示例：

from playwright.sync_api import sync_playwright

def fetch_dynamic_page(url, wait_selector=None, scroll=False, click_selectors=None):
    """动态页面抓取标准流程"""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context(
            viewport={'width': 1920, 'height': 1080},
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        )
        page = context.new_page()

        try:
            page.goto(url, timeout=60000, wait_until='networkidle')

            # 等待关键元素出现
            if wait_selector:
                page.wait_for_selector(wait_selector, timeout=15000)

            # 执行点击操作（如切换tab）
            if click_selectors:
                for selector in click_selectors:
                    try:
                        page.click(selector, timeout=5000)
                        page.wait_for_timeout(2000)  # 等待内容加载
                    except Exception:
                        pass

            # 处理滚动加载
            if scroll:
                for _ in range(5):
                    page.evaluate('window.scrollBy(0, window.innerHeight)')
                    page.wait_for_timeout(1500)

            html_content = page.content()
            return html_content

        finally:
            browser.close()

经验：能不用浏览器就不用，因为浏览器贵。但遇到动态站点，不要硬抠 requests，直接上 Playwright。

场景 C：前后端分离，网页本身只是壳，真实数据在接口里

特征：

  • 页面上的数据是 Ajax 请求回来的
  • F12 Network 里能看到 JSON
  • HTML 本身没数据

方案优先级：

  1. 直接调用接口
  2. 不行再用浏览器拦截响应
  3. 最后才解析 HTML

这套系统的两种成熟做法：

做法 1：API 直连

比如百度、京东、软通、金蝶这类，直接请求接口拿 JSON。

import requests

def fetch_api_direct(api_url, params=None, headers=None):
    """API直连抓取"""
    default_headers = {
        'User-Agent': 'Mozilla/5.0',
        'Accept': 'application/json',
        'Referer': 'https://example.com/careers'
    }
    if headers:
        default_headers.update(headers)

    resp = requests.get(api_url, params=params, headers=default_headers, timeout=30)
    data = resp.json()

    # 标准化处理
    items = []
    for job in data.get('data', {}).get('list', []):
        items.append({
            'title': job.get('positionName', ''),
            'city': job.get('cityName', ''),
            'publish_time': job.get('publishDate', ''),
            'url': f"https://example.com/job/{job.get('id', '')}",
        })
    return items

做法 2：拦截浏览器响应

比如兴业银行，用 page.on("response", handler) 监听浏览器收到的 Ajax 响应，然后直接解析 JSON。

from playwright.sync_api import sync_playwright
import json

def fetch_by_intercept(page_url, api_pattern):
    """浏览器响应拦截抓取"""
    captured_data = []

    def handle_response(response):
        if api_pattern in response.url:
            try:
                body = response.json()
                captured_data.append(body)
            except Exception:
                pass

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.on("response", handle_response)

        page.goto(page_url, wait_until='networkidle')
        page.wait_for_timeout(5000)

        browser.close()

    return captured_data

为什么这很值得学：因为很多人明明可以拿 JSON，还在那里费劲解析 HTML。这是低效的。

你以后应该怎么判断：

1. 先看接口能不能直接调
→ 2. 不行看能不能在浏览器里拦截响应
→ 3. 再不行才解析页面 DOM

场景 D：数据藏在 JS 文件或 script 变量里

特征：

  • 页面加载一个 jobs.js
  • `` 标签里有 var jobs = [...]
  • 页面本身只是把 JS 里的数据渲染出来

方案：

  • 请求 JS 文件 → 正则抽 JSON 片段 → json.loads

代码示例：

import requests
import re
import json

def fetch_js_embedded(js_url):
    """从JS文件中提取嵌入数据"""
    resp = requests.get(js_url, timeout=30)
    text = resp.text

    # 方法1：匹配 var xxx = [...] 或 var xxx = {...}
    pattern = r'var\s+\w+\s*=\s*(\[[\s\S]*?\]);'
    matches = re.findall(pattern, text)

    results = []
    for match in matches:
        try:
            data = json.loads(match)
            if isinstance(data, list):
                results.extend(data)
        except json.JSONDecodeError:
            # 尝试修复常见的 JS→JSON 问题
            fixed = re.sub(r"'", '"', match)  # 单引号转双引号
            fixed = re.sub(r',\s*}', '}', fixed)  # 移除尾逗号
            fixed = re.sub(r',\s*]', ']', fixed)
            try:
                data = json.loads(fixed)
                if isinstance(data, list):
                    results.extend(data)
            except:
                pass

    return results

经验总结：数据在哪里，就去哪儿拿。不要被页面表象迷惑。

场景 E：数据主要存在图片 / 长图 / 海报 / PDF 中

特征：

  • 招聘公告是海报
  • 微信图文里长图为主
  • PDF 扫描件
  • 页面正文少，主要信息在图片中

方案：

  • OCR、图像下载、切图、二维码检测、图片去重

这套代码的亮点——完整的图片处理体系：

flowchart TD
    IMG[图片输入] --> DL[图片下载]
    DL --> MD5[MD5去重]
    MD5 --> PHASH[感知哈希去重]
    PHASH --> CLASS{图片分类}
    CLASS -->|二维码| QR[二维码识别]
    CLASS -->|长图| SPLIT[智能切图]
    CLASS -->|普通图| FILTER{有效性过滤}
    SPLIT --> OCR[PaddleOCR识别]
    FILTER -->|有效| OCR
    FILTER -->|无效| DROP[丢弃]
    QR --> STORE[上传OBS]
    OCR --> STORE
    STORE --> REPLACE[替换HTML图片地址]

这说明一个成熟系统会把图片当成正式数据源，而不是附属品。

站点分析速查决策表

判断条件 场景类型 推荐方案 成本 稳定性
view-source 能看到数据 A: 静态HTML requests + lxml ⭐ ⭐⭐⭐⭐
Elements 有、source 没有 B: 动态渲染 Playwright ⭐⭐⭐ ⭐⭐⭐
Network 有 JSON 接口 C: API分离 直连/拦截 ⭐⭐ ⭐⭐⭐⭐⭐
数据在 .js 文件/变量中 D: JS嵌入 正则+JSON ⭐ ⭐⭐⭐
主要信息在图/PDF里 E: 非文本 OCR/PDF解析 ⭐⭐⭐⭐ ⭐⭐

四、阶段 3：采集策略设计——你必须有"多武器库"

未来你做爬虫，千万不要只会一种方案。这套代码最值得学习的地方之一，就是它不是一个爬法打天下，而是按场景切换武器。

你至少应该有 5 种抓取武器

mindmap
  root((抓取武器库))
    1. requests 抓静态页面
      静态HTML
      简单公告站
      详情页正文
    2. Playwright 抓动态页面
      JS渲染
      登录态
      点击/滚动加载
    3. API 直连
      找到接口
      JSON完整
      反爬不强
    4. 浏览器响应拦截
      接口参数复杂
      token难还原
      页面可正常触发
    5. OCR / PDF 解析
      图片海报
      扫描公告
      嵌入式信息

为什么要做"策略优先级"

建议你以后任何站点都按这个优先级试：

1. 直接接口         → 最轻最快最稳
2. 浏览器拦截接口    → 省去参数还原
3. 静态 HTML 解析    → 简单直接
4. 浏览器渲染后解析 DOM → 成本较高
5. OCR / PDF         → 最后手段

原因：

  • 越靠前越轻
  • 越靠前越稳
  • 越靠前越容易维护
  • 越靠后越贵、越慢、越容易出错

各武器详细对比

武器 速度 并发能力 维护成本 资源消耗 适用场景
requests ⭐⭐⭐⭐⭐ 高（100+并发） 低 极低 静态页面、简单API
Playwright ⭐⭐ 低（5-10并发） 中 高（1-2GB/实例） SPA、登录态、复杂交互
API 直连 ⭐⭐⭐⭐⭐ 高 低 极低 前后端分离站点
响应拦截 ⭐⭐⭐ 低 中 高 加密参数、复杂token
OCR/PDF ⭐ 低 高 极高（GPU加速） 图片/扫描件

五、阶段 4：不要把规则写死，要做"配置驱动"

这是这套系统非常值得学的工程思想。

问题：如果你每个网站都硬编码，会怎样？

# ❌ 错误做法：硬编码
if company == "腾讯":
    ...
elif company == "阿里":
    ...
elif company == "百度":
    ...
# 等做到 50 个公司、100 个学校，你就会崩

因为你会遇到：

  • URL 改了
  • 站点增加校招/社招/实习三套入口
  • 一个站点要换模板
  • 某公司有特殊逻辑
  • 想并行跑不同批次任务

这套系统的做法：配置驱动

用 .ini 把站点元信息抽出来：

[company_00321]
company_name = 某科技公司
logo = https://example.com/logo.png
social_url = https://example.com/social-recruitment
campus_url = https://example.com/campus-recruitment
intern_url = https://example.com/intern
template = template_A
func_name = gen_00321
json_domain = https://api.example.com
city_mapping = {"BJ": "北京", "SH": "上海", "GZ": "广州"}
click_strategy = tab_click
page_strategy = click_next

好处：

  • 新增站点不一定改主程序
  • 规则变更只动配置
  • 方便任务分片
  • 便于运维和排查
  • 便于 AI 自动生成代码后接入

你以后应该怎么设计配置——三层结构

classDiagram
    class SiteConfig {
        +site_id: str
        +site_name: str
        +base_url: str
        +list_urls: list
        +source_type: enum
    }

    class CrawlConfig {
        +request_method: str
        +headers: dict
        +cookies: dict
        +pre_open_url: str
        +api_domain: str
        +paging_strategy: str
        +click_strategy: str
        +rate_limit: float
        +retry_count: int
        +proxy_pool: str
    }

    class ParseConfig {
        +list_parser: str
        +detail_parser: str
        +detail_selector: str
        +publish_time_selector: str
        +title_selector: str
        +content_selector: str
        +field_mapping: dict
    }

    SiteConfig "1" --> "1" CrawlConfig : 包含
    SiteConfig "1" --> "1" ParseConfig : 包含

第一层：站点级

字段 说明 示例
site_id 唯一标识 com_00321
site_name 站点名称 某科技有限公司
base_url 站点根地址 https://careers.example.com
list_urls 列表页地址(可多个) ["https://...campus", "https://...social"]
source_type 数据源类型 html_static / api_json / spa_intercept

第二层：抓取级

字段 说明 示例
request_method 请求方式 GET / POST
headers 自定义请求头 {"Referer": "..."}
cookies 预置cookie {"session": "..."}
pre_open_url 需要先访问的页面(拿cookie) https://example.com/home
api_domain API域名 https://api.example.com
paging_strategy 翻页策略 url_param / click_next / scroll / api_offset
click_strategy 点击策略 tab_click / none
rate_limit 请求间隔(秒) 2.0
retry_count 重试次数 3
proxy_pool 代理池标识 pool_A

第三层：解析级

字段 说明 示例
list_parser 列表解析器名 gen_00321
detail_parser 详情解析器名 detail_00321
title_selector 标题选择器 //h1[@class="title"]/text()
time_selector 时间选择器 .publish-date::text
content_selector 正文选择器 div.article-body
field_mapping 字段映射 {"posName": "job_name", "cityName": "city"}

配置驱动的完整代码示例

import configparser
import importlib

class ConfigDrivenCrawler:
    """配置驱动的爬虫调度器"""

    def __init__(self, config_path):
        self.config = configparser.ConfigParser()
        self.config.read(config_path, encoding='utf-8')

    def get_site_config(self, site_id):
        """读取站点配置"""
        section = self.config[site_id]
        return {
            'site_name': section.get('site_name'),
            'source_type': section.get('source_type', 'html_static'),
            'list_urls': section.get('list_urls', '').split(','),
            'func_name': section.get('func_name'),
            'template': section.get('template'),
            'rate_limit': section.getfloat('rate_limit', 1.0),
            'retry_count': section.getint('retry_count', 3),
            'paging_strategy': section.get('paging_strategy', 'none'),
            'click_strategy': section.get('click_strategy', 'none'),
        }

    def load_parser(self, func_name):
        """动态加载解析器"""
        try:
            module = importlib.import_module(f'parsers.{func_name}')
            return module.extract_table_from_html
        except ImportError:
            return None

    def run_site(self, site_id):
        """执行单个站点的采集"""
        config = self.get_site_config(site_id)
        parser = self.load_parser(config['func_name'])

        if parser is None:
            print(f"[WARN] {site_id} 无可用解析器，尝试AI生成...")
            self.trigger_ai_generation(site_id, config)
            return

        # 根据 source_type 选择抓取策略
        fetcher = self.select_fetcher(config['source_type'])

        for url in config['list_urls']:
            html = fetcher(url, config)
            results = parser(html, site_id)

            if not results:
                print(f"[WARN] {site_id} 解析结果为空，触发AI修复...")
                self.trigger_ai_generation(site_id, config, html_sample=html)
            else:
                self.save_results(site_id, results)

    def select_fetcher(self, source_type):
        """根据数据源类型选择抓取器"""
        fetcher_map = {
            'html_static': self.fetch_static,
            'html_dynamic': self.fetch_dynamic,
            'api_json': self.fetch_api,
            'spa_intercept': self.fetch_intercept,
            'js_embed': self.fetch_js,
        }
        return fetcher_map.get(source_type, self.fetch_static)

六、阶段 5：解析逻辑要"插件化"，不要"主程序越写越大"

这是你未来做大型爬虫必须掌握的。

典型错误做法

主程序一个文件 3000 行：

  • 这个网站这么解析
  • 那个网站那么解析
  • 这里点击一下
  • 那里翻一页

最后没人能维护。

这套系统的可取之处：解析函数插件化

flowchart TB
    MAIN[主调度程序<br/>spider_main.py] --> CONFIG[配置文件<br/>sites.ini]
    MAIN --> |动态加载| GEN[列表解析插件<br/>gen_000xx.py]
    MAIN --> |动态加载| CLICK[点击插件<br/>click_500xx.py]
    MAIN --> |动态加载| PAGE[翻页插件<br/>page_000xx.py]
    MAIN --> |动态加载| DETAIL[详情插件<br/>detail_xxx.py]

    subgraph 插件目录
        GEN
        CLICK
        PAGE
        DETAIL
    end

    style MAIN fill:#e1f5fe
    style CONFIG fill:#fff3e0

这就是插件架构。主程序只负责：

  1. 读取配置
  2. 找到该用哪个插件
  3. 动态加载
  4. 执行
  5. 收集结果

插件化的核心实现

import importlib
import os
import traceback

class PluginManager:
    """插件管理器"""

    PLUGIN_DIRS = {
        'list_parser': 'parsers/gen/',       # 列表解析插件
        'click_handler': 'parsers/click/',    # 点击操作插件
        'page_handler': 'parsers/page/',      # 翻页操作插件
        'detail_parser': 'parsers/detail/',   # 详情解析插件
    }

    def __init__(self):
        self._cache = {}  # 模块缓存

    def load_plugin(self, plugin_type, plugin_name):
        """
        动态加载插件

        Args:
            plugin_type: 插件类型 ('list_parser', 'click_handler', ...)
            plugin_name: 插件名称 (如 'gen_00321')

        Returns:
            callable: 插件函数
        """
        cache_key = f"{plugin_type}:{plugin_name}"

        if cache_key in self._cache:
            return self._cache[cache_key]

        plugin_dir = self.PLUGIN_DIRS.get(plugin_type)
        if not plugin_dir:
            raise ValueError(f"未知插件类型: {plugin_type}")

        module_path = f"{plugin_dir.replace('/', '.')}{plugin_name}"

        try:
            module = importlib.import_module(module_path)

            # 每种插件类型有统一的入口函数名
            entry_points = {
                'list_parser': 'extract_table_from_html',
                'click_handler': 'perform_click',
                'page_handler': 'perform_paging',
                'detail_parser': 'parse_detail',
            }

            func = getattr(module, entry_points[plugin_type])
            self._cache[cache_key] = func
            return func

        except (ImportError, AttributeError) as e:
            print(f"[ERROR] 加载插件失败: {cache_key} -> {e}")
            return None

    def reload_plugin(self, plugin_type, plugin_name):
        """热重载插件（用于AI生成新代码后）"""
        cache_key = f"{plugin_type}:{plugin_name}"
        self._cache.pop(cache_key, None)

        module_path = f"{self.PLUGIN_DIRS[plugin_type].replace('/', '.')}{plugin_name}"
        if module_path in importlib.sys.modules:
            importlib.reload(importlib.sys.modules[module_path])

        return self.load_plugin(plugin_type, plugin_name)

插件统一接口规范

# ============================================================
# 列表解析插件接口规范
# 文件位置: parsers/gen/gen_00321.py
# ============================================================

def extract_table_from_html(htmlcontext: str, tempfile: str) -> list:
    """
    从列表页HTML中提取职位/公告列表

    Args:
        htmlcontext: 列表页HTML源码
        tempfile: 临时文件路径前缀（用于存储中间数据）

    Returns:
        list[dict]: 每个dict包含:
            - announcement_name: str  公告/职位标题
            - publish_time: str       发布时间 (YYYY-MM-DD)
            - link: str               详情页绝对URL

    示例:
        [
            {
                "announcement_name": "2026年春季校园招聘",
                "publish_time": "2026-03-15",
                "link": "https://example.com/job/123"
            }
        ]
    """
    pass

# ============================================================
# 点击操作插件接口规范
# 文件位置: parsers/click/click_50321.py
# ============================================================

def perform_click(page, config: dict) -> None:
    """
    在Playwright page对象上执行点击操作

    Args:
        page: Playwright Page对象
        config: 站点配置字典
    """
    pass

# ============================================================
# 翻页操作插件接口规范
# 文件位置: parsers/page/page_00321.py
# ============================================================

def perform_paging(page, config: dict, current_page: int) -> bool:
    """
    执行翻页操作

    Args:
        page: Playwright Page对象
        config: 站点配置字典
        current_page: 当前页码

    Returns:
        bool: 是否还有下一页
    """
    pass

# ============================================================
# 详情解析插件接口规范
# 文件位置: parsers/detail/detail_00321.py
# ============================================================

def parse_detail(html: str, url: str, config: dict) -> dict:
    """
    解析详情页

    Args:
        html: 详情页HTML源码
        url: 详情页URL
        config: 站点配置字典

    Returns:
        dict: 包含 title, content_html, content_text, images,
              publish_time, attachments 等字段
    """
    pass

这样做的价值：

  • 每个站点互不干扰
  • 一个站点坏了不影响其他站点
  • 易于人工修复
  • 易于自动生成代码替换

七、阶段 6：当页面结构变化时，怎么办？

这才是爬虫真正的核心问题。不是"怎么抓"，而是**"网站变了你怎么办"**。

传统做法是：

  1. 人工发现失效
  2. 人工改 XPath
  3. 人工发版

这很慢。

你这套代码最先进的地方，就是：它已经开始用大模型自动生成解析代码来解决这个问题。

八、大模型如何自动生成爬虫代码

1. 本质是什么？

不是"让大模型凭空写整个爬虫系统"。而是：

当某个网页的结构解析失败时，把该网页 HTML 样本交给大模型，让它只生成一个"局部解析函数"。

也就是说，大模型不是在接管整套系统，而是在补一个"网站专属解析插件"。

这就很合理、很可控。

2. 自动修复的完整触发流程

flowchart TD
    START[开始执行站点采集] --> LOAD[读取配置里的 func_name]
    LOAD --> IMPORT[动态加载对应 gen_xxx.py]
    IMPORT --> EXEC["执行 extract_table_from_html(html, tempfile)"]
    EXEC --> CHECK{结果是否为空？}

    CHECK -->|不为空| OK[✅ 正常使用结果]
    CHECK -->|为空| TRIGGER[🔴 触发AI自动生成]

    TRIGGER --> PREP[准备发送给模型的材料]
    PREP --> CALL[调用大模型API]
    CALL --> RECEIVE[接收生成的代码]
    RECEIVE --> SAVE["保存为 gen_func_code_xxx_tmp.py"]
    SAVE --> VALIDATE{文件是否已存在？}
    VALIDATE -->|已存在| SKIP[跳过，防止重复生成]
    VALIDATE -->|不存在| WRITE[写入新文件]
    WRITE --> REVIEW[等待人工审核]
    REVIEW --> RENAME["审核通过 → 改名为正式 gen_000xx.py"]
    RENAME --> OK2[✅ 下次执行使用新代码]

    style TRIGGER fill:#ffcdd2
    style OK fill:#c8e6c9
    style OK2 fill:#c8e6c9

3. 为什么用"结果为空"作为触发条件？

因为它简单、明确、工程上好实现。

你不需要先判断：

  • 是不是 XPath 失效
  • 是不是 class 改了
  • 是不是标签层级变了

只要最终提取不到结果，就说明解析失败。 这个判断很稳定。

当然，未来你可以升级成更丰富的失败判定：

判定条件 严重程度 说明
结果为空 🔴 严重 解析完全失效
结果字段缺失过多 🟡 中等 部分选择器失效
链接全为空 🟡 中等 URL拼接逻辑失败
时间解析失败率高 🟡 中等 时间格式变化
标题长度异常 🟢 轻微 可能抓到了无关内容
解析结果数量比历史均值低很多 🟡 中等 可能只解析了部分

但"结果为空"是一个很好的第一步。

4. 发给大模型的是什么？

def build_ai_prompt(html_sample, site_config):
    """构建发给大模型的Prompt"""

    prompt = f"""
你是一个专业的网页解析工程师。
请根据以下HTML样本，生成一个Python函数，从中提取招聘信息列表。

## 严格要求

### 函数签名（必须完全一致）
```python
def extract_table_from_html(htmlcontext, tempfile):
```

### 返回格式

返回 `list[dict]`，每个dict必须包含：

- `announcement_name`: str，公告/职位标题
- `publish_time`: str，发布时间（格式 YYYY-MM-DD）
- `link`: str，详情页绝对URL

### 编码约束

- 只能使用标准库 + lxml + re
- 不要使用 BeautifulSoup
- 不要使用 print 输出
- 链接必须补全为绝对路径，base\_url 为: {site\_config['base\_url']}
- 时间格式统一为 YYYY-MM-DD
- 如果取不到某字段，返回空字符串，不要抛异常
- 函数内部要有 try-except 保护

### 返回示例

```
[
    {{
        "announcement_name": "2026年春季校园招聘公告",
        "publish_time": "2026-03-15",
        "link": "https://example.com/job/123"
    }}
]
```

## HTML样本（已截取前15000字符）

```
{html_sample[:15000]}
```

请直接输出完整的Python函数代码，不要输出任何解释。
"""
return prompt

这里最关键的经验：

你不是让模型自由发挥，而是让它在严格约束下输出一个固定接口的函数。 这样主程序才能无感接入。

5. 为什么这种方式靠谱？

因为网页解析这件事，本质上就是：

  1. 找到列表项容器
  2. 找到标题节点
  3. 找到时间节点
  4. 找到链接节点
  5. 做少量清洗

对于大模型来说，这种**"给 HTML，找重复结构，生成 CSS/XPath 提取逻辑"**是很擅长的。

它比人工快在：

  • 能快速看懂陌生 DOM
  • 能给出备选选择器
  • 能自动补容错逻辑

6. 生成后的代码怎么管理？

这套系统的经验是：

  • 生成到临时文件，如 gen_func_code_xxx_tmp.py
  • 若文件已存在则跳过，防止重复生成
  • 人工审核后再改名成正式 gen_000xx.py

这一步很重要，因为自动生成代码不应直接无审核进生产。

最稳妥方式是：


自动生成 → 自动测试 → 人工确认 → 正式纳管

完整的代码生成与管理流程：

import os
import importlib
import hashlib

class AICodeGenerator:
    """AI代码生成管理器"""

    TEMP_DIR = 'parsers/gen/tmp/'
    PROD_DIR = 'parsers/gen/'
    ARCHIVE_DIR = 'parsers/gen/archive/'

    def __init__(self, llm_client):
        self.llm = llm_client
        os.makedirs(self.TEMP_DIR, exist_ok=True)
        os.makedirs(self.ARCHIVE_DIR, exist_ok=True)

    def generate(self, site_id, html_sample, site_config):
        """生成新的解析代码"""

        # 1. 检查是否已有临时文件（防止重复生成）
        tmp_path = os.path.join(self.TEMP_DIR, f'gen_func_code_{site_id}_tmp.py')
        if os.path.exists(tmp_path):
            print(f"[INFO] {site_id} 已有临时代码，跳过生成")
            return tmp_path

        # 2. 构建Prompt并调用模型
        prompt = build_ai_prompt(html_sample, site_config)
        code = self.llm.generate(prompt)

        # 3. 清理模型输出（去掉markdown代码块标记等）
        code = self._clean_code_output(code)

        # 4. 基础语法验证
        if not self._validate_syntax(code):
            print(f"[ERROR] {site_id} 生成代码语法错误")
            return None

        # 5. 写入临时文件
        with open(tmp_path, 'w', encoding='utf-8') as f:
            f.write(code)

        print(f"[INFO] {site_id} 代码已生成: {tmp_path}")
        return tmp_path

    def auto_test(self, site_id, html_sample):
        """自动测试生成的代码"""
        tmp_path = os.path.join(self.TEMP_DIR, f'gen_func_code_{site_id}_tmp.py')

        if not os.path.exists(tmp_path):
            return False, "临时文件不存在"

        try:
            # 动态导入临时模块
            spec = importlib.util.spec_from_file_location(f"tmp_{site_id}", tmp_path)
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)

            # 调用函数
            func = getattr(module, 'extract_table_from_html')
            results = func(html_sample, f'/tmp/test_{site_id}')

            # 验证结果
            checks = []
            checks.append(('是否为list', isinstance(results, list)))
            checks.append(('是否非空', len(results) > 0))

            if results:
                first = results[0]
                checks.append(('包含announcement_name', 'announcement_name' in first))
                checks.append(('包含publish_time', 'publish_time' in first))
                checks.append(('包含link', 'link' in first))
                checks.append(('标题非空', bool(first.get('announcement_name', '').strip())))
                checks.append(('链接像URL', first.get('link', '').startswith('http')))

            passed = all(ok for _, ok in checks)
            report = '\n'.join(f"  {'✅' if ok else '❌'} {name}" for name, ok in checks)

            return passed, report

        except Exception as e:
            return False, f"执行异常: {e}"

    def promote_to_production(self, site_id):
        """将临时代码提升为正式代码"""
        tmp_path = os.path.join(self.TEMP_DIR, f'gen_func_code_{site_id}_tmp.py')
        prod_path = os.path.join(self.PROD_DIR, f'gen_{site_id}.py')

        # 归档旧版本
        if os.path.exists(prod_path):
            content_hash = hashlib.md5(open(prod_path, 'rb').read()).hexdigest()[:8]
            archive_path = os.path.join(
                self.ARCHIVE_DIR,
                f'gen_{site_id}_v_{content_hash}.py'
            )
            os.rename(prod_path, archive_path)
            print(f"[INFO] 旧版本已归档: {archive_path}")

        # 提升新版本
        os.rename(tmp_path, prod_path)
        print(f"[INFO] {site_id} 代码已上线: {prod_path}")

    def rollback(self, site_id):
        """回退到上一个版本"""
        archive_files = sorted([
            f for f in os.listdir(self.ARCHIVE_DIR)
            if f.startswith(f'gen_{site_id}_v_')
        ])

        if not archive_files:
            print(f"[ERROR] {site_id} 没有可回退的版本")
            return False

        latest_archive = os.path.join(self.ARCHIVE_DIR, archive_files[-1])
        prod_path = os.path.join(self.PROD_DIR, f'gen_{site_id}.py')

        os.replace(latest_archive, prod_path)
        print(f"[INFO] {site_id} 已回退到: {archive_files[-1]}")
        return True

    def _clean_code_output(self, code):
        """清理模型输出"""
        # 去掉 ```python ... ``` 包裹
        if '```python' in code:
            code = code.split('```python')[1]
            if '```' in code:
                code = code.split('```')[0]
        elif '```' in code:
            parts = code.split('```')
            if len(parts) >= 3:
                code = parts[1]
        return code.strip()

    def _validate_syntax(self, code):
        """验证Python语法"""
        try:
            compile(code, '<generated>', 'exec')
            return True
        except SyntaxError:
            return False

九、大模型怎么"自动选择正确的执行"？

这里要区分两层意思。

第一层：系统如何找到该执行哪个解析器？

这不是 AI 在"理解"，而是系统在**"路由"**。

路由依据——配置文件里有：

  • template
  • func_name
  • company / site_id

主程序根据这些字段：

  1. 找到该站点对应的解析模块名
  2. 用 importlib 动态加载模块
  3. 调用统一入口函数

所以它"自动选择正确执行"，本质上是：配置驱动 + 动态加载 + 统一接口。不是神秘智能，而是工程设计。

第二层：当原解析器失效时，怎么自动切换到新代码？

先执行旧 gen_xxx.py
→ 提取为空
→ 触发生成新 gen_func_code_xxx_tmp.py
→ 后续调度优先尝试新生成的代码 or 人工纳管后替换正式代码

进阶设计：解析器自动竞赛与择优

如果你未来要把它做得更高级，可以设计成：

对同一个站点，同时维护多个解析器版本：

  • gen_00321_v1.py
  • gen_00321_v2.py
  • gen_00321_gpt_tmp.py
flowchart TD
    HTML[HTML样本输入] --> V1[当前正式版本<br/>gen_00321_v2.py]
    HTML --> V2[最近AI生成版本<br/>gen_00321_gpt_tmp.py]
    HTML --> V3[通用模板版本<br/>gen_template_A.py]
    HTML --> V4[回退老版本<br/>gen_00321_v1.py]

    V1 --> R1["结果A: 8条, 完整率95%"]
    V2 --> R2["结果B: 10条, 完整率88%"]
    V3 --> R3["结果C: 5条, 完整率70%"]
    V4 --> R4["结果D: 0条, 失败"]

    R1 --> JUDGE{质量评估}
    R2 --> JUDGE
    R3 --> JUDGE
    R4 --> JUDGE

    JUDGE --> BEST["选择最优: 结果A<br/>（数量×完整率最高）"]

    style BEST fill:#c8e6c9
    style R4 fill:#ffcdd2

质量比较维度：

  • 提取数量
  • 字段完整率
  • 时间有效率
  • 链接可访问率

这就是**"解析器自动竞赛"和"自动择优"**。

class ParserCompetitor:
    """解析器竞赛评估器"""

    def compete(self, html_sample, candidates: list):
        """
        让多个解析器竞争，选择最优

        Args:
            html_sample: HTML文本
            candidates: [(parser_name, parser_func), ...]

        Returns:
            (best_name, best_results)
        """
        scores = []

        for name, func in candidates:
            try:
                results = func(html_sample, '/tmp/test')
                score = self._evaluate(results)
                scores.append((name, results, score))
            except Exception:
                scores.append((name, [], 0))

        scores.sort(key=lambda x: x[2], reverse=True)

        if scores and scores[0][2] > 0:
            return scores[0][0], scores[0][1]
        return None, []

    def _evaluate(self, results):
        """评估解析结果质量，返回0-100分"""
        if not results or not isinstance(results, list):
            return 0

        score = 0
        n = len(results)

        # 数量分：有结果就有基础分
        score += min(n * 5, 30)

        # 字段完整率
        required = ['announcement_name', 'publish_time', 'link']
        completeness = []
        for item in results:
            filled = sum(1 for f in required if item.get(f, '').strip())
            completeness.append(filled / len(required))
        avg_completeness = sum(completeness) / len(completeness) if completeness else 0
        score += avg_completeness * 30

        # 链接有效性
        valid_links = sum(1 for item in results
                         if item.get('link', '').startswith('http'))
        link_rate = valid_links / n if n else 0
        score += link_rate * 20

        # 标题合理性（长度在2-200之间）
        valid_titles = sum(1 for item in results
                          if 2 <= len(item.get('announcement_name', '')) <= 200)
        title_rate = valid_titles / n if n else 0
        score += title_rate * 20

        return score

十、未来你自己做"AI 自动生成爬虫代码"时，建议这样设计

1. 只让 AI 生成"小模块"，不要生成"大系统"

不要让 AI 生成：

  • 调度框架
  • 存储系统
  • 整个 Playwright 流程

只让 AI 生成：

  • 某个列表页解析函数
  • 某个详情页正文提取函数
  • 某个 click/page 辅助函数

因为小模块：边界清晰、更容易验证、更容易回退、不会污染主系统。

2. 强制统一接口

# 列表解析——统一接口
def extract_table_from_html(htmlcontext: str, tempfile: str) -> list[dict]:
    ...

# 详情解析——统一接口
def parse_detail(html: str, url: str, config: dict) -> dict:
    ...

# 点击操作——统一接口
def perform_click(page, config: dict) -> None:
    ...

为什么： 因为调度器要"盲调"。它不应该知道每个模块的细节。

3. 给模型足够具体的约束

提示词里要明确：

约束类型 具体内容
依赖限制 只能使用 lxml, re, json, urllib
函数签名 必须叫 extract_table_from_html，参数、返回值固定
必须字段 announcement_name, publish_time, link
链接处理 补全绝对路径
时间格式 YYYY-MM-DD
异常处理 取不到字段返回空字符串，不报错
输出限制 不要打印无关内容

原则：你给 AI 的约束越清晰，它生成的代码越可用。

4. 生成后必须做自动验收

def auto_validate(generated_code: str, html_sample: str) -> dict:
    """自动验收生成的代码"""

    checks = {
        'can_import': False,
        'can_run': False,
        'returns_list': False,
        'has_required_fields': False,
        'non_empty': False,
        'links_valid': False,
        'titles_reasonable': False,
    }

    # 1. 能否 import
    try:
        compiled = compile(generated_code, '<test>', 'exec')
        namespace = {}
        exec(compiled, namespace)
        checks['can_import'] = True
    except Exception:
        return checks

    # 2. 能否运行
    try:
        func = namespace['extract_table_from_html']
        results = func(html_sample, '/tmp/test')
        checks['can_run'] = True
    except Exception:
        return checks

    # 3. 返回是否是 list
    checks['returns_list'] = isinstance(results, list)

    if not results:
        return checks

    checks['non_empty'] = True

    # 4. 每个 item 是否包含必须字段
    required = ['announcement_name', 'publish_time', 'link']
    checks['has_required_fields'] = all(
        all(f in item for f in required)
        for item in results
    )

    # 5. 链接是否像 URL
    checks['links_valid'] = all(
        item.get('link', '').startswith('http')
        for item in results if item.get('link')
    )

    # 6. 标题长度是否合理
    checks['titles_reasonable'] = all(
        2 <= len(item.get('announcement_name', '')) <= 500
        for item in results
    )

    return checks

不要生成完直接上线。

5. 留回退机制

永远保留：

  • 上一个稳定版本
  • 最近一次生成版本
  • 当前正式版本
parsers/gen/
├── gen_00321.py              ← 当前正式版本
├── tmp/
│   └── gen_func_code_00321_tmp.py  ← AI最近生成的
└── archive/
    ├── gen_00321_v_a1b2c3d4.py     ← 历史版本1
    └── gen_00321_v_e5f6g7h8.py     ← 历史版本2

因为 AI 生成代码一定会偶尔出错。

十一、阶段 7：数据清洗与标准化，比"抓到"更重要

很多人觉得抓到网页就完成了。不是。抓到只是开始。

真正难的是：把不同网站、不同时代、不同风格的脏数据，统一成标准结构。

你未来一定会遇到的数据脏问题

1. 时间格式混乱

# 你会遇到的各种时间格式
time_samples = [
    "2025-03-01",
    "2025/3/1",
    "03-01",
    "昨天",
    "3天前",
    "2025年3月1日",
    "Mar 1, 2025",
    "20250301",
    "2025.03.01",
    "1小时前",
    "刚刚",
]

标准化处理器：

import re
from datetime import datetime, timedelta

class TimeNormalizer:
    """时间格式标准化"""

    PATTERNS = [
        (r'(\d{4})-(\d{1,2})-(\d{1,2})', lambda m: f"{m[1]}-{int(m[2]):02d}-{int(m[3]):02d}"),
        (r'(\d{4})/(\d{1,2})/(\d{1,2})', lambda m: f"{m[1]}-{int(m[2]):02d}-{int(m[3]):02d}"),
        (r'(\d{4})\.(\d{1,2})\.(\d{1,2})', lambda m: f"{m[1]}-{int(m[2]):02d}-{int(m[3]):02d}"),
        (r'(\d{4})年(\d{1,2})月(\d{1,2})日', lambda m: f"{m[1]}-{int(m[2]):02d}-{int(m[3]):02d}"),
        (r'(\d{8})', lambda m: f"{m[1][:4]}-{m[1][4:6]}-{m[1][6:8]}"),
    ]

    RELATIVE_PATTERNS = [
        (r'刚刚', lambda: datetime.now()),
        (r'(\d+)\s*分钟前', lambda m: datetime.now() - timedelta(minutes=int(m[1]))),
        (r'(\d+)\s*小时前', lambda m: datetime.now() - timedelta(hours=int(m[1]))),
        (r'(\d+)\s*天前', lambda m: datetime.now() - timedelta(days=int(m[1]))),
        (r'昨天', lambda: datetime.now() - timedelta(days=1)),
        (r'前天', lambda: datetime.now() - timedelta(days=2)),
    ]

    def normalize(self, raw_time: str) -> str:
        """统一输出 YYYY-MM-DD 格式"""
        if not raw_time:
            return ''

        raw_time = raw_time.strip()

        # 尝试相对时间
        for pattern, func in self.RELATIVE_PATTERNS:
            match = re.search(pattern, raw_time)
            if match:
                try:
                    dt = func(match) if 'match' in func.__code__.co_varnames else func()
                    return dt.strftime('%Y-%m-%d')
                except:
                    pass

        # 尝试绝对时间
        for pattern, formatter in self.PATTERNS:
            match = re.search(pattern, raw_time)
            if match:
                try:
                    return formatter(match)
                except:
                    pass

        # 仅有月日，补当前年
        m = re.search(r'(\d{1,2})-(\d{1,2})', raw_time)
        if m:
            year = datetime.now().year
            return f"{year}-{int(m[1]):02d}-{int(m[2]):02d}"

        return raw_time  # 无法识别，返回原文

2. 地点不统一

class LocationNormalizer:
    """地点标准化"""

    CITY_MAPPING = {
        # 简称 → 标准名
        '京': '北京', 'BJ': '北京', '北京市': '北京',
        '沪': '上海', 'SH': '上海', '上海市': '上海',
        '深': '深圳', 'SZ': '深圳', '深圳市': '深圳',
        '广': '广州', 'GZ': '广州', '广州市': '广州',
        '杭': '杭州', 'HZ': '杭州', '杭州市': '杭州',
        '蓉': '成都', 'CD': '成都', '成都市': '成都',
        '宁': '南京', 'NJ': '南京', '南京市': '南京',
        '汉': '武汉', 'WH': '武汉', '武汉市': '武汉',
    }

    def normalize(self, raw_location: str) -> list:
        """标准化地点，支持多地点拆分"""
        if not raw_location:
            return []

        # 去掉常见修饰词
        raw = re.sub(r'(等地|地区|及其他)', '', raw_location)

        # 按分隔符拆分
        cities = re.split(r'[,，/、|+]', raw)

        result = []
        for city in cities:
            city = city.strip()
            # 处理"北京-海淀"格式
            city = city.split('-')[0].split('·')[0].strip()
            # 查映射表
            city = self.CITY_MAPPING.get(city, city)
            if city and len(city) >= 2:
                result.append(city)

        return list(dict.fromkeys(result))  # 去重保序

3. 岗位类型标准化

class JobTypeNormalizer:
    """岗位类型标准化"""

    RULES = [
        (r'校招|校园招聘|应届', '校招'),
        (r'社招|社会招聘|经验|资深', '社招'),
        (r'实习|intern', '实习'),
        (r'兼职|part.?time', '兼职'),
    ]

    def normalize(self, raw_type: str, title: str = '') -> str:
        combined = f"{raw_type} {title}"
        for pattern, label in self.RULES:
            if re.search(pattern, combined, re.IGNORECASE):
                return label
        return '社招'  # 默认

4. 噪声清洗

class HTMLCleaner:
    """HTML预清洗器"""

    # 需要移除的标签/选择器
    REMOVE_SELECTORS = [
        'nav', 'header', 'footer',
        '.breadcrumb', '.sidebar', '.advertisement',
        '.recommend', '.related', '.share-bar',
        '#comment', '.comment',
        'script', 'style', 'noscript',
        'iframe',
    ]

    def clean(self, html: str, base_url: str = '') -> str:
        """清洗HTML，保留正文"""
        from lxml import etree
        from lxml.html.clean import Cleaner

        try:
            tree = etree.HTML(html)
        except:
            return html

        # 移除噪声元素
        for selector in self.REMOVE_SELECTORS:
            for elem in tree.cssselect(selector) if '.' in selector or '#' in selector \
                    else tree.xpath(f'//{selector}'):
                elem.getparent().remove(elem)

        # 修复相对链接
        if base_url:
            from urllib.parse import urljoin
            for tag in tree.xpath('//a[@href]'):
                href = tag.get('href', '')
                if href and not href.startswith(('http', 'mailto:', '', '#')):
                    tag.set('href', urljoin(base_url, href))

            for tag in tree.xpath('//img[@src]'):
                src = tag.get('src', '')
                if src and not src.startswith(('http', 'data:')):
                    tag.set('src', urljoin(base_url, src))

        return etree.tostring(tree, encoding='unicode', method='html')

这套系统的完整后处理流程

flowchart LR
    RAW[原始HTML] --> CLEAN[HTML预清洗<br/>ann_md.py]
    CLEAN --> FIX_LINK[修复相对链接]
    FIX_LINK --> FIX_IMG[修复图片地址]
    FIX_IMG --> BASE64[处理base64图片]
    BASE64 --> TABLE[表格结构保留]
    TABLE --> OCR_SUP[图片OCR补充]
    OCR_SUP --> LLM[LLM结构化提取]
    LLM --> OUTPUT[标准JSON输出]

    style RAW fill:#ffcdd2
    style OUTPUT fill:#c8e6c9

经验：解析器负责"拿到原材料"，标准化器负责"做成可用数据"。这两个职责必须分开。

十二、阶段 8：图片、长图、二维码场景怎么处理

这是很多人忽略的，但招聘场景里非常常见。

你会碰到三类图

类型 示例 处理方式
纯装饰图 banner、logo、校徽、分隔线 过滤丢弃
信息图 招聘海报、岗位一览图、校招流程图 OCR提取
二维码图 投递二维码、报名二维码、公众号二维码 二维码识别

这套系统为什么值得学

它不是所有图都 OCR，而是先分类：

flowchart TD
    IMG[图片输入] --> SIZE{尺寸检查}
    SIZE -->|太小/太窄| DROP1[丢弃：装饰图]
    SIZE -->|尺寸合理| BLACK{黑名单检查}
    BLACK -->|命中| DROP2[丢弃：已知无效图]
    BLACK -->|未命中| QR{二维码检测}
    QR -->|是二维码| QR_PROCESS[二维码识别<br/>提取链接]
    QR -->|不是| MIXED{是否混合图？}
    MIXED -->|是| SPLIT[智能切图]
    MIXED -->|否| OCR_CHECK{OCR字数检查}
    OCR_CHECK -->|字数够| OCR[PaddleOCR识别]
    OCR_CHECK -->|字数不够| DROP3[丢弃：无文字图]
    SPLIT --> OCR

好处： 减少无效 OCR，节省算力。

智能切图 OCR 算法

问题背景：

  • 长图非常大，直接 OCR 会慢、爆内存、漏识别
  • 简单按固定高度切图，会把文字从中间切断
  • 一旦切断，OCR 结果就会很差

解决思路：用 OpenCV 做"连通区域检测"

import cv2
import numpy as np

class SmartImageSplitter:
    """智能切图器——避免切断文字"""

    def __init__(self, max_height=2000, min_gap=20):
        self.max_height = max_height  # 单片最大高度
        self.min_gap = min_gap        # 最小空白间距

    def split(self, image_path: str) -> list:
        """
        智能切图

        Returns:
            list[np.ndarray]: 切片列表
        """
        img = cv2.imread(image_path)
        if img is None:
            return []

        h, w = img.shape[:2]

        # 图片不够高，不需要切
        if h <= self.max_height:
            return [img]

        # 转灰度 + 二值化
        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        _, binary = cv2.threshold(gray, 200, 255, cv2.THRESH_BINARY_INV)

        # 计算每行的文字像素密度
        row_density = np.sum(binary > 0, axis=1) / w

        # 找安全切割线
        slices = []
        start = 0

        while start < h:
            end = min(start + self.max_height, h)

            if end >= h:
                slices.append(img[start:end])
                break

            # 在 end 附近寻找空白行（文字密度最低的位置）
            search_start = max(start + self.max_height - 200, start)
            search_end = min(start + self.max_height + 200, h)

            search_region = row_density[search_start:search_end]

            # 找连续空白区域
            best_cut = self._find_best_cut(search_region, search_start)

            if best_cut is not None:
                slices.append(img[start:best_cut])
                start = best_cut
            else:
                # 找不到好的切割点，强制切
                slices.append(img[start:end])
                start = end

        return slices

    def _find_best_cut(self, density_region, offset):
        """在密度区域中找最佳切割点"""
        # 找连续空白行
        blank_runs = []
        run_start = None

        for i, d in enumerate(density_region):
            if d < 0.01:  # 几乎没有文字
                if run_start is None:
                    run_start = i
            else:
                if run_start is not None:
                    run_len = i - run_start
                    if run_len >= self.min_gap:
                        blank_runs.append((run_start, i, run_len))
                    run_start = None

        if not blank_runs:
            return None

        # 选最长的空白区域的中间位置
        best = max(blank_runs, key=lambda x: x[2])
        cut_point = offset + (best[0] + best[1]) // 2
        return cut_point

完整图片处理Pipeline

import hashlib
import os
from PIL import Image
from pyzbar import pyzbar

class ImageProcessor:
    """图片处理完整流程"""

    def __init__(self, ocr_engine, obs_client, output_dir):
        self.ocr = ocr_engine        # PaddleOCR实例
        self.obs = obs_client         # OBS上传客户端
        self.output_dir = output_dir
        self.seen_md5 = set()         # MD5去重集合
        self.seen_phash = set()       # 感知哈希去重集合
        self.splitter = SmartImageSplitter()

    def process(self, image_path: str) -> dict:
        """处理单张图片"""
        result = {
            'type': 'unknown',      # decoration/info/qrcode
            'ocr_text': '',
            'qr_content': '',
            'obs_url': '',
            'is_valid': False,
        }

        # 1. MD5去重
        md5 = self._file_md5(image_path)
        if md5 in self.seen_md5:
            result['type'] = 'duplicate'
            return result
        self.seen_md5.add(md5)

        # 2. 感知哈希去重
        phash = self._perceptual_hash(image_path)
        if self._is_similar(phash):
            result['type'] = 'duplicate'
            return result
        self.seen_phash.add(phash)

        # 3. 尺寸过滤
        img = Image.open(image_path)
        w, h = img.size
        if w < 50 or h < 50:
            result['type'] = 'decoration'
            return result

        # 4. 二维码检测
        qr_codes = pyzbar.decode(img)
        if qr_codes:
            result['type'] = 'qrcode'
            result['qr_content'] = qr_codes[0].data.decode('utf-8', errors='ignore')
            result['is_valid'] = True
            result['obs_url'] = self.obs.upload(image_path)
            return result

        # 5. 长图切图 + OCR
        if h > 2000:
            slices = self.splitter.split(image_path)
            all_text = []
            for i, s in enumerate(slices):
                slice_path = f"{image_path}_slice_{i}.png"
                cv2.imwrite(slice_path, s)
                text = self._ocr_image(slice_path)
                all_text.append(text)
                os.remove(slice_path)
            result['ocr_text'] = '\n'.join(all_text)
        else:
            result['ocr_text'] = self._ocr_image(image_path)

        # 6. 判断是否有效
        if len(result['ocr_text'].strip()) > 20:
            result['type'] = 'info'
            result['is_valid'] = True
            result['obs_url'] = self.obs.upload(image_path)
        else:
            result['type'] = 'decoration'

        return result

    def _file_md5(self, path):
        return hashlib.md5(open(path, 'rb').read()).hexdigest()

    def _perceptual_hash(self, path, hash_size=8):
        """感知哈希"""
        img = Image.open(path).convert('L').resize((hash_size + 1, hash_size))
        pixels = list(img.getdata())
        diff = [1 if pixels[i] > pixels[i + 1] else 0
                for i in range(hash_size * hash_size)]
        return int(''.join(map(str, diff)), 2)

    def _is_similar(self, phash, threshold=5):
        """判断是否与已有图片相似"""
        for existing in self.seen_phash:
            hamming = bin(phash ^ existing).count('1')
            if hamming <= threshold:
                return True
        return False

    def _ocr_image(self, path):
        """OCR识别"""
        try:
            result = self.ocr.ocr(path, cls=True)
            texts = []
            if result:
                for line in result:
                    if line:
                        for word_info in line:
                            texts.append(word_info[1][0])
            return '\n'.join(texts)
        except Exception:
            return ''

你未来碰到图片场景时的建议

场景 处理方式
普通截图类图片 直接 OCR
超长图 先切图再 OCR，切图前做文字块检测
图里有二维码 单独识别二维码，裁出二维码区域单独存储
图片太小/太透明/OCR几乎没文字 直接过滤掉

十三、阶段 9：去重、增量、时效性管理

这是商用爬虫最容易被忽视的地方。

你未来一定会碰到的问题

  • 今天抓到的职位，明天还在
  • 同一职位多个入口重复出现
  • 页面链接变了但内容一样
  • 一条数据失效了，但你本地还留着
  • 旧文件怎么知道还能不能用

1. URL 的 MD5 作为文件名

import hashlib
import os

def url_to_filepath(url: str, base_dir: str, extension: str = '.html') -> str:
    """将URL转换为安全的文件路径"""
    url_hash = hashlib.md5(url.encode()).hexdigest()
    return os.path.join(base_dir, f"{url_hash}{extension}")

作用：

  • 同一 URL 永远落到同一个文件
  • 方便天然去重
  • 文件名安全
  • 路径长度受控

2. 文件已存在时，不重下，只更新 mtime

这个点特别值得学。

import os
import time

def smart_save(filepath: str, content: str, force: bool = False):
    """
    智能存储：已存在则更新mtime，不存在则新建

    为什么不是简单跳过？
    因为"跳过"无法告诉系统：
    - 这条数据是今天又抓到了
    - 还是很久以前遗留的

    更新 mtime 的好处：
    - 本次抓到过的文件，mtime 会变新
    - 没抓到的旧文件，mtime 会变旧
    - 后续清理程序可以根据时间判断是否过期
    """
    if os.path.exists(filepath) and not force:
        # 文件已存在，只更新修改时间
        os.utime(filepath, None)  # 更新为当前时间
        return 'touched'
    else:
        # 新文件，写入内容
        os.makedirs(os.path.dirname(filepath), exist_ok=True)
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(content)
        return 'created'

def cleanup_stale_files(directory: str, max_age_days: int = 7):
    """
    清理过期文件

    mtime 超过 max_age_days 天的文件，说明最近几轮都没再抓到过，
    很可能已经下线或失效。
    """
    now = time.time()
    cutoff = now - (max_age_days * 86400)

    removed = 0
    for root, dirs, files in os.walk(directory):
        for f in files:
            fpath = os.path.join(root, f)
            if os.path.getmtime(fpath) < cutoff:
                os.remove(fpath)
                removed += 1

    return removed

这是一种非常轻量的增量管理策略——不需要很复杂的数据库状态表，也能管理"本轮仍有效的数据"。

3. 多层去重体系

flowchart TD
    DATA[新抓取数据] --> L1{第1层：URL哈希去重}
    L1 -->|重复URL| SKIP1[跳过]
    L1 -->|新URL| L2{第2层：内容哈希去重}
    L2 -->|内容一样| SKIP2[跳过<br/>URL变了但内容没变]
    L2 -->|新内容| L3{第3层：标题+时间去重}
    L3 -->|重复| SKIP3[跳过<br/>多入口同一职位]
    L3 -->|不重复| L4{第4层：云端比对}
    L4 -->|已入库| SKIP4[跳过]
    L4 -->|未入库| SAVE[✅ 入库]

    style SAVE fill:#c8e6c9

import hashlib

class MultiLayerDeduplicator:
    """多层去重器"""

    def __init__(self):
        self.url_hashes = set()      # URL哈希集
        self.content_hashes = set()  # 内容哈希集
        self.title_time_keys = set() # 标题+时间键集

    def is_duplicate(self, item: dict) -> tuple:
        """
        判断是否重复

        Returns:
            (is_dup: bool, reason: str)
        """
        url = item.get('url', '')
        title = item.get('title', '')
        content = item.get('content', '')
        pub_time = item.get('publish_time', '')

        # 第1层：URL去重
        url_hash = hashlib.md5(url.encode()).hexdigest()
        if url_hash in self.url_hashes:
            return True, 'url_duplicate'

        # 第2层：内容去重
        if content:
            content_hash = hashlib.md5(content.encode()).hexdigest()
            if content_hash in self.content_hashes:
                return True, 'content_duplicate'
            self.content_hashes.add(content_hash)

        # 第3层：标题+时间去重
        if title and pub_time:
            key = f"{title.strip()}_{pub_time.strip()}"
            if key in self.title_time_keys:
                return True, 'title_time_duplicate'
            self.title_time_keys.add(key)

        self.url_hashes.add(url_hash)
        return False, ''

    def check_cloud(self, item: dict, cloud_client) -> bool:
        """第4层：与云端/业务库比对"""
        try:
            exists = cloud_client.check_exists(
                url=item.get('url'),
                title=item.get('title'),
                company=item.get('company_name')
            )
            return exists
        except:
            return False  # 查询失败时不去重，宁多不少

经验：抓虫不是越多越好，而是要尽量少制造脏数据。

十四、阶段 10：入库前一定要做"结构化理解"

招聘信息这种场景，光抓文本没用。你最后要的是：

字段 说明
公司名 发布招聘的企业
岗位名 具体职位名称
工作地点 城市/区域
学历要求 本科/硕士/博士...
招聘人数 若干/5人/不限
薪资 15k-25k/面议
截止时间 投递截止日期
毕业届 2026届/不限
投递方式 链接/邮箱/二维码

这些字段往往散落在标题、正文、表格、图片、脚注、联系方式里。传统规则很难覆盖。

分阶段让 AI 处理

不是让模型一次性做全部，而是拆成几层：

flowchart TD
    INPUT[清洗后的内容] --> L1[第1层：公告级信息提取]
    L1 --> L2[第2层：职位级信息提取]
    L2 --> L3[第3层：展示级格式化]

    L1 --> |输出| R1["公司名、公告类型<br/>毕业届、应聘方式"]
    L2 --> |输出| R2["岗位列表<br/>地点、学历、人数、薪资"]
    L3 --> |输出| R3["Markdown整理<br/>表格转展示格式"]

第 1 层：公告级信息

PROMPT_ANNOUNCEMENT = """
你是一个招聘信息分析专家。请从以下招聘公告中提取公告级别的信息。

## 提取字段

| 字段 | 说明 | 格式 |
|------|------|------|
| company_name | 招聘公司名称 | 字符串 |
| announcement_type | 公告类型 | 校招/社招/实习 |
| graduation_year | 面向毕业届 | 如 "2026届" |
| apply_method | 应聘方式 | 如 "在线投递/邮箱投递" |
| apply_link | 投递链接 | URL |
| deadline | 截止时间 | YYYY-MM-DD |

## 注意
- 如果某字段无法提取，返回空字符串
- 只返回JSON，不要解释

## 公告内容
{content}

## 输出格式
```json
{{
    "company_name": "",
    "announcement_type": "",
    "graduation_year": "",
    "apply_method": "",
    "apply_link": "",
    "deadline": ""
}}
```

"""

第 2 层：职位级信息

PROMPT_POSITIONS = """
你是一个招聘信息分析专家。请从以下招聘公告中提取所有职位信息。

## 提取字段（每个职位）

| 字段 | 说明 |
|------|------|
| job_name | 岗位名称 |
| city | 工作城市（多个用逗号分隔） |
| diploma | 学历要求 |
| headcount | 招聘人数（数字或"若干"） |
| salary | 薪资范围 |
| department | 所属部门 |

## 注意
- 如果公告中有表格，请逐行解析
- 如果只有岗位名没有其他信息，也要提取
- 返回JSON数组

## 公告内容
{content}

## 输出格式
```json
[
    {{
        "job_name": "",
        "city": "",
        "diploma": "",
        "headcount": "",
        "salary": "",
        "department": ""
    }}
]
```

"""

第 3 层：展示级格式

PROMPT_FORMAT = """
请将以下招聘公告内容整理为清晰的Markdown格式，用于前端展示。

## 要求
- 保留所有实质信息
- 表格保持表格格式
- 去除导航、广告等噪声
- 图片保留原始地址
- 联系方式放在最后

## 原始内容
{content}
"""

为什么这样拆：

  • 任务边界清晰
  • 提示词更容易写
  • 每一层更容易校验
  • 模型出错更容易定位

十五、你未来做爬虫时，推荐的整体开发步骤

下面给你一套可直接落地的方法。

完整系统架构

flowchart TB
    subgraph 配置层
        CFG[站点配置 sites.ini]
        MODEL[数据模型定义]
    end

    subgraph 调度层
        SCHED[主调度器]
        TASK[任务分片]
        RETRY[重试队列]
    end

    subgraph 抓取层
        F1[requests 抓取器]
        F2[Playwright 抓取器]
        F3[API 直连器]
        F4[响应拦截器]
    end

    subgraph 解析层
        P1[列表解析插件]
        P2[详情解析插件]
        P3[点击/翻页插件]
        P4[AI生成解析器]
    end

    subgraph 处理层
        CL[HTML清洗器]
        NM[标准化器]
        OCR_P[图片处理器]
        LLM[LLM结构化]
    end

    subgraph 存储层
        RAW[原始数据落盘]
        DEDUP[去重器]
        DB[数据库入库]
        OBS[OBS对象存储]
    end

    subgraph 监控层
        MON[运行监控]
        ALERT[报警通知]
        STATUS[状态文件 .ok/.err]
        LOG[结构化日志]
    end

    CFG --> SCHED
    MODEL --> SCHED
    SCHED --> TASK
    TASK --> F1 & F2 & F3 & F4
    F1 & F2 & F3 & F4 --> RAW
    RAW --> P1 & P2 & P3
    P1 --> |失败| P4
    P1 & P2 --> CL
    CL --> NM
    CL --> OCR_P
    NM --> LLM
    OCR_P --> LLM
    LLM --> DEDUP
    DEDUP --> DB
    OCR_P --> OBS

    SCHED --> MON
    MON --> ALERT
    MON --> STATUS
    MON --> LOG

    RETRY --> SCHED

第一步：先做统一数据模型

from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class ListItem:
    """列表页最少字段"""
    title: str = ''
    url: str = ''
    publish_time: str = ''
    source: str = ''
    category: str = ''

@dataclass
class DetailPage:
    """详情页最少字段"""
    title: str = ''
    content_html: str = ''
    content_text: str = ''
    images: List[str] = field(default_factory=list)
    attachments: List[str] = field(default_factory=list)
    publish_time: str = ''
    location: str = ''
    tags: List[str] = field(default_factory=list)

@dataclass
class StructuredJob:
    """结构化字段"""
    company_name: str = ''
    job_name: str = ''
    city: str = ''
    diploma: str = ''
    salary: str = ''
    headcount: str = ''
    job_type: str = ''       # 校招/社招/实习
    deadline: str = ''
    apply_link: str = ''
    department: str = ''
    graduation_year: str = ''

@dataclass
class CrawlResult:
    """完整的爬取结果"""
    site_id: str = ''
    list_item: ListItem = field(default_factory=ListItem)
    detail: DetailPage = field(default_factory=DetailPage)
    jobs: List[StructuredJob] = field(default_factory=list)
    raw_html_path: str = ''
    status: str = 'pending'   # pending/success/failed
    error_msg: str = ''
    crawl_time: str = ''

第二步：做站点分类

给每个站点打标签：

class SourceType:
    HTML_STATIC = 'html_static'         # 静态HTML
    HTML_DYNAMIC = 'html_dynamic'       # 动态渲染
    API_JSON = 'api_json'               # JSON接口
    SPA_INTERCEPT = 'spa_intercept'     # SPA响应拦截
    JS_EMBED = 'js_embed'              # JS嵌入数据
    IMAGE_OCR = 'image_ocr'            # 图片OCR
    PDF_PARSE = 'pdf_parse'            # PDF解析

然后决定走哪条抓取链路。

第三步：配置化

不要把站点信息写死，抽成配置（参见阶段4的详细设计）。

第四步：主程序只做调度

class MainScheduler:
    """主调度器——只负责调度，不写具体解析"""

    def __init__(self, config_path):
        self.config = ConfigDrivenCrawler(config_path)
        self.plugins = PluginManager()
        self.dedup = MultiLayerDeduplicator()
        self.ai_gen = AICodeGenerator(llm_client)

    def run(self, site_ids: list = None):
        """执行调度"""
        sites = site_ids or self.config.get_all_site_ids()

        for site_id in sites:
            try:
                self._run_single_site(site_id)
                self._write_status(site_id, 'ok')
            except Exception as e:
                self._write_status(site_id, 'err', str(e))
                self.logger.error(f"{site_id} 执行失败: {e}")

    def _run_single_site(self, site_id):
        """单站点执行流程"""
        cfg = self.config.get_site_config(site_id)

        # 1. 选策略
        fetcher = self._select_fetcher(cfg['source_type'])

        # 2. 调 parser
        parser = self.plugins.load_plugin('list_parser', cfg['func_name'])

        # 3. 抓取列表
        for url in cfg['list_urls']:
            html = fetcher(url, cfg)
            self._save_raw(site_id, url, html)  # 原始数据落盘

            results = parser(html, site_id) if parser else []

            # 4. 解析失败 → AI修复
            if not results:
                self.ai_gen.generate(site_id, html, cfg)
                continue

            # 5. 抓取详情 + 后处理
            for item in results:
                if not self.dedup.is_duplicate(item)[0]:
                    detail = self._fetch_detail(item, cfg)
                    structured = self._structurize(detail, cfg)
                    self._save_results(site_id, structured)

        # 6. 记日志
        self._log_stats(site_id)

第五步：插件化解析器

每类解析器统一接口（参见阶段5的详细设计）。

第六步：所有原始数据先落盘

def _save_raw(self, site_id, url, content):
    """原始数据落盘"""
    filepath = url_to_filepath(url, f'data/raw/{site_id}/')
    smart_save(filepath, content)

为什么： 因为以后规则错了，你还能回放。如果不留原始数据，后面很难排查。

第七步：做增量与去重

（参见阶段9的详细设计）

第八步：清洗与结构化分离

分开两层：

  • 原始解析层：负责尽可能完整地提取内容，不要过度加工
  • 标准化层：负责时间统一、地点统一、学历标准化、类型归一、AI 结构化

第九步：加监控和状态文件

import json
import time
from datetime import datetime

class CrawlMonitor:
    """爬虫监控器"""

    def __init__(self, status_dir='data/status/'):
        self.status_dir = status_dir
        os.makedirs(status_dir, exist_ok=True)
        self.stats = {}

    def record(self, site_id, event, detail=''):
        """记录事件"""
        if site_id not in self.stats:
            self.stats[site_id] = {
                'success_count': 0,
                'fail_count': 0,
                'empty_count': 0,
                'detail_fail_count': 0,
                'ocr_fail_count': 0,
                'api_limit_count': 0,
                'llm_fail_count': 0,
                'start_time': datetime.now().isoformat(),
            }

        s = self.stats[site_id]

        event_map = {
            'success': 'success_count',
            'fail': 'fail_count',
            'empty': 'empty_count',
            'detail_fail': 'detail_fail_count',
            'ocr_fail': 'ocr_fail_count',
            'api_limit': 'api_limit_count',
            'llm_fail': 'llm_fail_count',
        }

        if event in event_map:
            s[event_map[event]] += 1

    def write_status(self, site_id, status):
        """写状态文件 .ok / .err"""
        # 清除旧状态
        for ext in ['.ok', '.err']:
            path = os.path.join(self.status_dir, f'{site_id}{ext}')
            if os.path.exists(path):
                os.remove(path)

        # 写新状态
        ext = '.ok' if status == 'ok' else '.err'
        path = os.path.join(self.status_dir, f'{site_id}{ext}')

        with open(path, 'w') as f:
            json.dump({
                'site_id': site_id,
                'status': status,
                'time': datetime.now().isoformat(),
                'stats': self.stats.get(site_id, {}),
            }, f, ensure_ascii=False, indent=2)

    def check_alerts(self):
        """检查是否需要报警"""
        alerts = []
        for site_id, s in self.stats.items():
            total = s['success_count'] + s['fail_count']
            if total == 0:
                continue

            fail_rate = s['fail_count'] / total

            if fail_rate > 0.5:
                alerts.append(f"🔴 {site_id} 失败率过高: {fail_rate:.0%}")
            if s['empty_count'] > 3:
                alerts.append(f"🟡 {site_id} 连续空结果: {s['empty_count']}次")
            if s['api_limit_count'] > 0:
                alerts.append(f"🟡 {site_id} API限流: {s['api_limit_count']}次")

        return alerts

    def summary(self):
        """输出汇总报告"""
        total_success = sum(s['success_count'] for s in self.stats.values())
        total_fail = sum(s['fail_count'] for s in self.stats.values())
        total_sites = len(self.stats)

        return {
            'total_sites': total_sites,
            'total_success': total_success,
            'total_fail': total_fail,
            'overall_success_rate': f"{total_success/(total_success+total_fail):.1%}"
                if (total_success + total_fail) > 0 else 'N/A',
            'alerts': self.check_alerts(),
        }

第十步：最后再引入 AI 自动修复

顺序不要反。

先有：

  • ✅ 稳定调度框架
  • ✅ 配置系统
  • ✅ 插件接口
  • ✅ 原始数据回放
  • ✅ 测试机制

然后再上：

  • 🤖 大模型生成解析代码
  • 🤖 大模型结构化抽取
  • 🤖 大模型格式修复

这样 AI 才是增强器，不是风险源。

十六、未来你会碰到的问题，以及建议方案

问题速查表

问题 症状 方案
页面结构改了 XPath 全失效 插件化解析器 + AI生成新解析器 + 旧版本回退
看得到抓不到 requests 返回空壳 先查 Network → 直连接口 → Playwright + response hook
职位在图片里 正文几乎为空 下载图片 → OCR → 长图切图 → 二维码识别
多个tab/类型 只抓到一部分 click 做独立插件 + 多入口配置到 urls
列表能抓详情不行 详情页404/超时 列表详情分阶段 + 详情页单独重试 + 保留原始HTML
反爬限流 403/429/验证码 控频 + 代理池 + UA模拟 + 预热cookie + 退避重试
文本很乱 规则提不干净 HTML清洗 → 转Markdown → 再交给LLM结构化
模型成本太高 每月账单吓人 分层调模型 + 规则优先 + Prompt缓存 + 只让AI做高价值环节

问题 1：页面结构改了，XPath 全失效

# 推荐的防御性解析写法
def resilient_parse(html, selectors):
    """防御性解析——多选择器备选"""
    tree = etree.HTML(html)

    # 定义多个备选选择器，按优先级尝试
    title_selectors = [
        selectors.get('title_v2'),    # 最新版本
        selectors.get('title_v1'),    # 旧版本
        '//h1/text()',                # 通用备选
        '//title/text()',             # 最后兜底
    ]

    for sel in title_selectors:
        if not sel:
            continue
        result = tree.xpath(sel)
        if result and result[0].strip():
            return result[0].strip()

    return ''

问题 6：网站有反爬、限流

import random
import time

class RateLimiter:
    """智能限速器"""

    def __init__(self, base_interval=2.0, max_interval=60.0):
        self.base_interval = base_interval
        self.max_interval = max_interval
        self.fail_count = 0

    def wait(self):
        """等待适当时间"""
        # 基础间隔 + 随机抖动
        interval = self.base_interval * (1 + random.random() * 0.5)

        # 失败退避
        if self.fail_count > 0:
            backoff = min(self.base_interval * (2 ** self.fail_count), self.max_interval)
            interval = backoff + random.random() * 5

        time.sleep(interval)

    def on_success(self):
        self.fail_count = max(0, self.fail_count - 1)

    def on_failure(self):
        self.fail_count += 1

class ProxyRotator:
    """代理池轮换"""

    def __init__(self, proxy_list):
        self.proxies = proxy_list
        self.index = 0
        self.blacklist = set()

    def get_proxy(self):
        available = [p for p in self.proxies if p not in self.blacklist]
        if not available:
            self.blacklist.clear()  # 全黑名单时重置
            available = self.proxies

        proxy = available[self.index % len(available)]
        self.index += 1
        return {'http': proxy, 'https': proxy}

    def mark_bad(self, proxy):
        self.blacklist.add(proxy)

问题 8：模型成本太高

flowchart TD
    INPUT[待处理数据] --> R1{规则能提取？}
    R1 -->|能| RULE[✅ 规则提取<br/>成本: $0]
    R1 -->|不能| R2{简单模型能搞定？}
    R2 -->|能| SMALL["小模型处理<br/>(如 GPT-4o-mini)<br/>成本: $0.001"]
    R2 -->|不能| R3{需要复杂推理？}
    R3 -->|是| BIG["大模型处理<br/>(如 Claude Opus)<br/>成本: $0.05"]
    R3 -->|否| SMALL

    style RULE fill:#c8e6c9
    style SMALL fill:#e1f5fe
    style BIG fill:#fff3e0

Prompt 缓存策略：

import hashlib

class PromptCache:
    """提示词缓存——避免重复调用模型"""

    def __init__(self, cache_dir='data/prompt_cache/'):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)

    def get_or_call(self, prompt: str, llm_func, ttl_hours=24):
        """有缓存用缓存，没有才调模型"""
        cache_key = hashlib.md5(prompt.encode()).hexdigest()
        cache_path = os.path.join(self.cache_dir, f'{cache_key}.json')

        # 检查缓存
        if os.path.exists(cache_path):
            age = time.time() - os.path.getmtime(cache_path)
            if age < ttl_hours * 3600:
                with open(cache_path, 'r') as f:
                    return json.load(f)

        # 调用模型
        result = llm_func(prompt)

        # 写缓存
        with open(cache_path, 'w') as f:
            json.dump(result, f, ensure_ascii=False)

        return result

十七、这套代码最值得你学的 12 个核心思想

mindmap
  root((12个核心思想))
    工程架构
      1. 配置驱动，而不是硬编码
      2. 多策略抓取，按场景切换
      3. 主流程调度，站点逻辑插件化
    数据管理
      4. 原始数据必须落盘
      5. 去重要多层做
      6. 增量抓取用mtime管理
    内容处理
      7. HTML清洗非常重要
      8. 表格、图片、二维码特殊处理
      9. AI分阶段做结构化
    AI增强
      10. 大模型补规则难维护的环节
      11. AI生成代码限制输出接口
      12. 自动验收和回退比生成更重要

详细说明

# 思想 关键点
1 配置驱动 站点规则放配置，逻辑放程序
2 多策略抓取 API、拦截、HTML、浏览器、OCR 按场景切换
3 主流程调度，插件化 主程序小而稳，站点逻辑可插拔
4 原始数据必须落盘 后续排查、回放、修复全靠它
5 去重要多层做 URL、内容、云端、哈希都要考虑
6 增量用 mtime 管理 不靠"是否存在"，mtime 更新是轻量方案
7 HTML 清洗非常重要 别把脏 DOM 直接扔给模型
8 特殊内容特殊处理 表格、图片、二维码是重要信息源，不是边角料
9 AI 分阶段处理 公告级 → 职位级 → 格式级分层处理
10 AI 补规则短板 解析变化页面、结构化复杂文本是 AI 最佳场景
11 AI 生成限制接口 只生成小模块，统一函数签名，便于调度
12 验收和回退更重要 没有验收与回退，AI 代码不适合进生产

十八、给你一个未来做爬虫的推荐路线

如果你要从零学到做出一套像样的系统，按这个顺序练：

timeline
    title 爬虫学习路线图

    第一阶段 : 基础能力
        : requests / lxml / xpath
        : BeautifulSoup / 正则
        : JSON / 文件存储

    第二阶段 : 动态站
        : Playwright / Selenium
        : 登录态 / Cookie管理
        : 点击 / 翻页 / 滚动
        : Network面板分析

    第三阶段 : 接口分析
        : F12 抓包
        : 还原接口参数
        : Token / Cookie / Header
        : 分页参数推导

    第四阶段 : 工程化
        : 配置驱动
        : 插件化架构
        : 日志 / 重试 / 任务分片
        : 定时任务 / 断点续跑

    第五阶段 : 数据治理
        : 去重 / 增量
        : 标准化清洗
        : HTML清洗
        : OCR / PDF解析

    第六阶段 : AI 增强
        : LLM 结构化提取
        : Prompt 设计与缓存
        : AI 生成解析器
        : 自动测试与回退

各阶段建议练习项目

阶段 练习项目 预期成果
第一阶段 抓取3个新闻网站的最新文章列表 能输出标题+链接+时间的CSV
第二阶段 抓取一个需要滚动加载的招聘网站 能处理动态加载、翻页
第三阶段 还原一个招聘网站的搜索API 能直接调API拿到全量数据
第四阶段 把上述3个爬虫统一成配置驱动系统 加新站点只改配置不改代码
第五阶段 做一个支持增量更新+去重的招聘数据管道 每日定时运行，不产生重复数据
第六阶段 接入LLM做自动解析+自动修复 新站点能自动适配，旧站点变更能自动修复

各阶段推荐技术栈

{
  "$schema": "https://vega.github.io/schema/vega-lite/v6.json",
  "data": {
    "values": [
      {"阶段": "1.基础能力", "技术": "requests", "重要性": 95},
      {"阶段": "1.基础能力", "技术": "lxml/xpath", "重要性": 90},
      {"阶段": "1.基础能力", "技术": "BeautifulSoup", "重要性": 75},
      {"阶段": "1.基础能力", "技术": "正则表达式", "重要性": 85},
      {"阶段": "2.动态站", "技术": "Playwright", "重要性": 92},
      {"阶段": "2.动态站", "技术": "Selenium", "重要性": 60},
      {"阶段": "2.动态站", "技术": "Network分析", "重要性": 88},
      {"阶段": "3.接口分析", "技术": "F12抓包", "重要性": 95},
      {"阶段": "3.接口分析", "技术": "参数还原", "重要性": 80},
      {"阶段": "4.工程化", "技术": "配置驱动", "重要性": 90},
      {"阶段": "4.工程化", "技术": "插件架构", "重要性": 88},
      {"阶段": "4.工程化", "技术": "日志/监控", "重要性": 85},
      {"阶段": "5.数据治理", "技术": "去重/增量", "重要性": 92},
      {"阶段": "5.数据治理", "技术": "OCR/PDF", "重要性": 70},
      {"阶段": "6.AI增强", "技术": "LLM结构化", "重要性": 85},
      {"阶段": "6.AI增强", "技术": "AI生成代码", "重要性": 78},
      {"阶段": "6.AI增强", "技术": "自动验收", "重要性": 90}
    ]
  },
  "mark": "bar",
  "encoding": {
    "y": {"field": "技术", "type": "nominal", "sort": "-x", "title": "技术栈"},
    "x": {"field": "重要性", "type": "quantitative", "title": "重要性评分", "scale": {"domain": [0, 100]}},
    "color": {"field": "阶段", "type": "nominal", "title": "学习阶段"},
    "tooltip": [
      {"field": "技术", "type": "nominal"},
      {"field": "阶段", "type": "nominal"},
      {"field": "重要性", "type": "quantitative"}
    ]
  },
  "width": 500,
  "height": 400,
  "title": "爬虫技术栈重要性评估"
}

十九、附录：完整项目目录结构推荐

crawl-system/
├── config/
│   ├── sites.ini                    # 站点配置
│   ├── global.ini                   # 全局配置（代理、模型、OBS等）
│   └── city_mapping.json            # 城市映射表
│
├── core/
│   ├── scheduler.py                 # 主调度器
│   ├── fetcher/
│   │   ├── static_fetcher.py        # requests 抓取
│   │   ├── dynamic_fetcher.py       # Playwright 抓取
│   │   ├── api_fetcher.py           # API 直连
│   │   └── intercept_fetcher.py     # 响应拦截
│   ├── plugin_manager.py            # 插件管理器
│   ├── deduplicator.py              # 多层去重器
│   ├── rate_limiter.py              # 限速器
│   └── monitor.py                   # 监控与状态
│
├── parsers/
│   ├── gen/                         # 列表解析插件
│   │   ├── gen_00321.py
│   │   ├── gen_00322.py
│   │   ├── tmp/                     # AI生成的临时代码
│   │   └── archive/                 # 历史版本归档
│   ├── click/                       # 点击操作插件
│   │   └── click_50321.py
│   ├── page/                        # 翻页操作插件
│   │   └── page_00321.py
│   └── detail/                      # 详情解析插件
│       └── detail_00321.py
│
├── processors/
│   ├── html_cleaner.py              # HTML清洗
│   ├── normalizer.py                # 标准化器（时间/地点/类型）
│   ├── image_processor.py           # 图片处理（OCR/切图/二维码）
│   ├── smart_splitter.py            # 智能切图
│   └── llm_structurizer.py          # LLM结构化提取
│
├── ai/
│   ├── code_generator.py            # AI代码生成器
│   ├── auto_validator.py            # 自动验收器
│   ├── parser_competitor.py         # 解析器竞赛器
│   ├── prompt_cache.py              # Prompt缓存
│   └── prompts/                     # 提示词模板
│       ├── gen_list_parser.txt
│       ├── extract_announcement.txt
│       └── extract_positions.txt
│
├── storage/
│   ├── file_storage.py              # 文件存储（raw/JSON/状态）
│   ├── db_uploader.py               # 数据库上传
│   └── obs_client.py                # 对象存储客户端
│
├── data/
│   ├── raw/                         # 原始HTML/JSON
│   │   └── {site_id}/
│   ├── parsed/                      # 解析结果
│   │   └── {site_id}/
│   ├── images/                      # 下载的图片
│   ├── status/                      # 状态文件 (.ok/.err)
│   ├── logs/                        # 日志
│   └── prompt_cache/                # Prompt缓存
│
├── tests/
│   ├── test_parsers.py
│   ├── test_normalizer.py
│   ├── test_dedup.py
│   └── samples/                     # 测试用HTML样本
│
├── scripts/
│   ├── run_all.py                   # 全量执行
│   ├── run_single.py                # 单站点执行
│   ├── cleanup_stale.py             # 清理过期数据
│   └── promote_ai_code.py           # 提升AI代码到正式
│
├── requirements.txt
├── Dockerfile
└── README.md

二十、总结——一句话概括每个阶段

阶段 一句话
1. 定义目标 先设计数据模型，再动手写代码
2. 站点分析 数据在哪就去哪拿，不要被页面骗
3. 采集策略 至少5种武器，按场景选最轻的
4. 配置驱动 站点规则放配置，新增站点不改代码
5. 插件化 主程序只调度，解析逻辑可插拔
6. AI修复 结果为空就触发，只生成小模块
7. 数据清洗 解析器拿原材料，标准化器出成品
8. 图片处理 图片是正式数据源，不是附属品
9. 去重增量 多层去重 + mtime 管理时效性
10. 结构化 分层让 AI 处理，不要一次做全部

最后记住：爬虫不是写一个脚本抓几个页面，而是建设一条能长期稳定运行、能自我修复的数据生产线。