---
title: "02-爬虫系统建设"
created: 2026-04-03
tags:
  - 项目
aliases:
  - 爬虫系统建设
---
# 爬虫系统建设

> **—— 从零搭建一套可维护、可扩展、可自愈的爬虫**

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

很多人学爬虫，停留在：

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

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

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

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

![[image-d7bfacd6.png]]

> “配置驱动 + 多策略抓取 + AI解析 + 自动修复 + 数据闭环”

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

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

#### **先要考虑清楚 6 个核心问题**

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

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

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

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

#### **统一数据模型设计示例**

```html
{
  "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`**。**

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

![[image-061e0185.png]]

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

**特征：**

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

**方案：**

- `requests` + `lxml` + `BeautifulSoup` + 正则辅助

**优点：**

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

**风险：**

- 网站稍微改 DOM 就失效

**适合：**

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

**代码示例：**

```python
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 点击
- 分页点击
- 当前页操作
- 页面等待
- 动态内容加载

**代码示例：**

```python
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。

```python
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。

```python
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。这是低效的。**

**应该怎么判断：**

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

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

**特征：**

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

**方案：**

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

**代码示例：**

```python
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、图像下载、切图、二维码检测、图片去重

**参考代码的亮点——完整的图片处理体系：**

![[image-4a6a5017.png]]

### **站点分析速查决策表**

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

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

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

### **至少应该有 5 种抓取武器**

![[image-421480ae.png]]

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

任何站点都按这个优先级试：

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

**原因：**

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

### **各武器详细对比**

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

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

这是参考系统**非常值得学的工程思想**。

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

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

因为会遇到：

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

### **参考系统的做法：配置驱动**

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

```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 自动生成代码后接入

### **应该怎么设计配置——三层结构**

![[image-410b67c5.png]]

**第一层：站点级**

| **字段** | **说明** | **示例** |
| --- | --- | --- |
| `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"}` |

### **配置驱动的完整代码示例**

```python
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 行：

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

最后没人能维护。

### **参考系统的可取之处：解析函数插件化**

![[image-9bd6ee21.png]]

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

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

### **插件化的核心实现**

```python
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)
```

### **插件统一接口规范**

```python
# ============================================================
# 列表解析插件接口规范
# 文件位置: 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. 自动修复的完整触发流程**

![[image-0fefb48b.png]]

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

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

不需要先判断：

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

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

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

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

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

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

````python
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]}
````

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

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

因为网页解析这件事，本质上就是：
1. 找到列表项容器
2. 找到标题节点
3. 找到时间节点
4. 找到链接节点
5. 做少量清洗

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

它比人工快在：
- 能快速看懂陌生 DOM
- 能给出备选选择器
- 能自动补容错逻辑

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

这套系统的经验是：
- 生成到临时文件，如 `gen_func_code_xxx_tmp.py`
- 若文件已存在则跳过，防止重复生成
- 人工审核后再改名成正式 `gen_000xx.py`

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

最稳妥方式是：

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

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

```python
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. 调用统一入口函数

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

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

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

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

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

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

- `gen_00321_v1.py`
- `gen_00321_v2.py`
- `gen_00321_gpt_tmp.py`

![[image-d36786c5.png]]

**质量比较维度：**

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

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

```python
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. 强制统一接口**

```python
# 列表解析——统一接口
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. 生成后必须做自动验收**

```python
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. 留回退机制**

永远保留：

- 上一个稳定版本
- 最近一次生成版本
- 当前正式版本

```text
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. 时间格式混乱**

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

**标准化处理器：**

```python
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. 地点不统一**

```python
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. 岗位类型标准化**

```python
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. 噪声清洗**

```python
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:', 'javascript:', '#')):
                    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')
```

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

![[image-144e9661.png]]

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

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

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

### **会碰到三类图**

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

### **这套系统为什么值得学**

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

```mermaid
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 做"连通区域检测"**

```python
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**

```python
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 作为文件名**

```python
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**

```python
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. 多层去重体系**

```mermaid
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
```

```python
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 处理**

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

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

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

### **第 1 层：公告级信息**

````python
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 层：职位级信息

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

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

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

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

## 公告内容
{content}

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

### **第 3 层：展示级格式**

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

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

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

**为什么这样拆：**

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

---

## **十五、做爬虫时，推荐的整体开发步骤**

### **完整系统架构**

![[image-b6372458.png]]

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

```python
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 = ''
```

### **第二步：做站点分类**

给每个站点打标签：

```python
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的详细设计）。

### **第四步：主程序只做调度**

```python
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的详细设计）。

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

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

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

### **第七步：做增量与去重**

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

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

分开两层：

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

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

```python
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 全失效**

```python
# 推荐的防御性解析写法
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：网站有反爬、限流**

```python
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：模型成本太高**

```mermaid
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 缓存策略：**

```python
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
```

## **十七、附录：完整项目目录结构推荐**

```text
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
```

---

**项目分区导航**：[[01-爬虫架构总览|爬虫架构总览]] ⬅️ | 02-爬虫系统建设 | ➡️ [[03-爬虫系统建设实战|爬虫系统建设实战]]