1. 项目背景与需求分析

数字报纸的刊期目录是新闻机构发布内容的重要组织形式，通常以分页形式展示历史内容。作为一名数据采集工程师，我最近接到一个需求：需要完整抓取某主流数字报纸过去5年的刊期目录数据，包含每期的发布日期、版面名称、文章标题等关键信息，并将这些数据规范存储到数据库中供后续分析使用。

这个需求看似简单，但在实际开发中会遇到几个典型难题：

1. 归档翻页问题：数字报纸的刊期目录通常采用分页加载，部分老报纸的翻页机制特殊，需要模拟真实用户操作才能获取完整数据。

2. 日期格式混乱：不同时期的报纸可能使用不同的日期格式（如"2023-01-01"、"2023年1月1日"、"Jan 1, 2023"等），需要统一处理。

3. 反爬机制应对：新闻网站通常对目录页有访问频率限制，直接暴力抓取容易被封IP。

4. 数据存储一致性：需要确保抓取的数据完整、准确，并在入库时处理好重复数据和异常情况。

2. 技术选型与环境准备

2.1 核心工具选择

经过对比测试，我选择了以下技术栈：

• 请求库：Requests + Session（维持会话）
• 解析库：BeautifulSoup4（HTML解析） + lxml（解析引擎）
• 浏览器模拟：Selenium（仅用于特殊翻页场景）
• 日期处理：dateutil.parser（智能日期解析）
• 数据库：PostgreSQL（支持JSON字段存储原始数据）
• 任务调度：APScheduler（定时增量抓取）

# 基础环境安装 pip install requests beautifulsoup4 python-dateutil selenium psycopg2 apscheduler

2.2 爬虫道德配置

在开始前，必须检查目标网站的robots.txt并设置合理的爬取间隔。以人民日报网站为例：

import time import random # 遵守爬虫道德的基本配置 HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Accept-Language': 'zh-CN,zh;q=0.9' } DELAY = random.uniform(1, 3) # 随机延迟1-3秒 MAX_RETRY = 3 # 最大重试次数

提示：实际项目中建议将配置参数提取到单独的文件中，方便后期调整和维护。

3. 分页抓取实现方案

3.1 基础分页模式识别

数字报纸的分页通常有两种实现方式：

1. 传统分页：URL中包含page参数（如?page=2）
2. 动态加载：通过AJAX请求获取数据（查看XHR请求）

通过分析目标网站的翻页机制，我发现了以下规律：

def generate_page_urls(base_url, total_pages): """生成分页URL列表""" return [f"{base_url}?page={i}" for i in range(1, total_pages+1)]

3.2 特殊翻页处理

对于需要交互触发的分页（如"点击加载更多"），使用Selenium模拟：

from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC def selenium_pagination(driver, max_pages): """使用Selenium处理动态分页""" items = [] for _ in range(max_pages): try: # 定位并点击"加载更多"按钮 button = WebDriverWait(driver, 10).until( EC.element_to_be_clickable((By.CSS_SELECTOR, ".load-more")) ) button.click() time.sleep(2) # 等待数据加载 items.extend(parse_items(driver.page_source)) except: break return items

3.3 断点续爬设计

为防止意外中断导致重复抓取，需要记录已抓取的分页状态：

import json from pathlib import Path class CrawlState: def __init__(self, state_file='crawl_state.json'): self.state_file = Path(state_file) self.state = self._load_state() def _load_state(self): if self.state_file.exists(): with open(self.state_file, 'r') as f: return json.load(f) return {'last_page': 0, 'success_pages': []} def update(self, page_num, success=True): self.state['last_page'] = page_num if success: self.state['success_pages'].append(page_num) self._save_state() def _save_state(self): with open(self.state_file, 'w') as f: json.dump(self.state, f)

4. 日期归一化处理实战

4.1 常见日期格式分析

在抓取过程中，我遇到了以下日期格式变体：

1. "2023-01-01"
2. "2023年1月1日"
3. "01/01/2023"
4. "Jan 1, 2023"
5. "1st January 2023"

4.2 智能日期解析方案

使用dateutil.parser配合自定义规则处理：

from dateutil import parser from datetime import datetime def normalize_date(date_str): """将各种日期格式统一为YYYY-MM-DD格式""" try: # 预处理中文日期 if '年' in date_str: date_str = date_str.replace('年', '-').replace('月', '-').replace('日', '') # 智能解析日期 dt = parser.parse(date_str, fuzzy=True) return dt.strftime('%Y-%m-%d') except Exception as e: print(f"日期解析失败: {date_str}, 错误: {e}") return None

4.3 日期验证与修正

对于历史报纸，还需要验证日期的合理性：

def validate_newspaper_date(date_str, start_year=2018): """验证报纸日期是否合理""" dt = datetime.strptime(date_str, '%Y-%m-%d') current_year = datetime.now().year if not (start_year <= dt.year <= current_year): raise ValueError(f"异常年份: {dt.year}") if dt > datetime.now(): raise ValueError("未来日期") return date_str

5. 数据存储与去重设计

5.1 数据库表结构设计

CREATE TABLE newspaper_issues ( id SERIAL PRIMARY KEY, publish_date DATE NOT NULL, issue_number VARCHAR(50), page_count INTEGER, sections JSONB, raw_data JSONB, created_at TIMESTAMP DEFAULT NOW(), UNIQUE (publish_date, issue_number) ); CREATE TABLE newspaper_articles ( id SERIAL PRIMARY KEY, issue_id INTEGER REFERENCES newspaper_issues(id), section_name VARCHAR(100), title VARCHAR(255) NOT NULL, author VARCHAR(100), page_number INTEGER, url VARCHAR(255), created_at TIMESTAMP DEFAULT NOW() );

5.2 Python入库实现

使用psycopg2实现高效批量插入：

import psycopg2 from psycopg2.extras import execute_batch class DatabaseManager: def __init__(self, conn_string): self.conn = psycopg2.connect(conn_string) def save_issue(self, issue_data): with self.conn.cursor() as cur: # 使用ON CONFLICT处理重复数据 query = """ INSERT INTO newspaper_issues (publish_date, issue_number, page_count, sections, raw_data) VALUES (%s, %s, %s, %s, %s) ON CONFLICT (publish_date, issue_number) DO NOTHING RETURNING id """ cur.execute(query, ( issue_data['publish_date'], issue_data['issue_number'], issue_data.get('page_count'), issue_data.get('sections'), issue_data.get('raw_data') )) return cur.fetchone()[0] if cur.rowcount > 0 else None def save_articles_batch(self, articles): with self.conn.cursor() as cur: query = """ INSERT INTO newspaper_articles (issue_id, section_name, title, author, page_number, url) VALUES (%s, %s, %s, %s, %s, %s) """ execute_batch(cur, query, articles) self.conn.commit()

5.3 断点续传与数据校验

def check_existing_issue(db, publish_date, issue_number): """检查是否已存在相同刊期""" with db.conn.cursor() as cur: cur.execute( "SELECT id FROM newspaper_issues WHERE publish_date=%s AND issue_number=%s", (publish_date, issue_number) ) return cur.fetchone() is not None def validate_data_completeness(db, issue_id, expected_article_count): """验证数据完整性""" with db.conn.cursor() as cur: cur.execute( "SELECT COUNT(*) FROM newspaper_articles WHERE issue_id=%s", (issue_id,) ) actual_count = cur.fetchone()[0] return actual_count >= expected_article_count * 0.9 # 允许10%的误差

6. 反爬策略应对方案

6.1 请求头随机化

from fake_useragent import UserAgent def get_random_headers(): ua = UserAgent() return { 'User-Agent': ua.random, 'Accept': 'text/html,application/xhtml+xml', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Referer': 'https://www.example.com/' }

6.2 IP轮换与代理池

import random PROXY_LIST = [ 'http://proxy1.example.com:8080', 'http://proxy2.example.com:8080', # 添加更多代理... ] def get_session_with_proxy(): session = requests.Session() proxy = random.choice(PROXY_LIST) session.proxies = {'http': proxy, 'https': proxy} return session

6.3 请求频率控制

class RequestThrottler: def __init__(self, base_delay=1.0, max_delay=5.0): self.base_delay = base_delay self.max_delay = max_delay self.last_request_time = 0 def wait(self): elapsed = time.time() - self.last_request_time wait_time = max(0, self.base_delay - elapsed) if wait_time > 0: time.sleep(wait_time + random.uniform(0, 0.5)) self.last_request_time = time.time()

7. 完整爬虫实现示例

import logging from urllib.parse import urljoin logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) class NewspaperSpider: def __init__(self, base_url, db_manager): self.base_url = base_url self.db = db_manager self.session = requests.Session() self.throttler = RequestThrottler(base_delay=2.0) self.state = CrawlState() def crawl_issues(self, start_page=1, end_page=None): current_page = start_page while True: if end_page and current_page > end_page: break self.throttler.wait() url = self._get_page_url(current_page) try: response = self.session.get(url, headers=get_random_headers()) response.raise_for_status() issues = self.parse_issues(response.text) for issue in issues: self.process_issue(issue) self.state.update(current_page, success=True) current_page += 1 if not self._has_next_page(response.text): break except Exception as e: logging.error(f"抓取第{current_page}页失败: {str(e)}") self.state.update(current_page, success=False) break def _get_page_url(self, page_num): return f"{self.base_url}?page={page_num}" def parse_issues(self, html): soup = BeautifulSoup(html, 'lxml') issues = [] for item in soup.select('.issue-item'): date_str = item.select_one('.date').text.strip() normalized_date = normalize_date(date_str) issue = { 'publish_date': normalized_date, 'issue_number': item.select_one('.issue-num').text.strip(), 'page_count': int(item.select_one('.pages').text), 'url': urljoin(self.base_url, item.select_one('a')['href']) } issues.append(issue) return issues def process_issue(self, issue): if check_existing_issue(self.db, issue['publish_date'], issue['issue_number']): return self.throttler.wait() try: response = self.session.get(issue['url'], headers=get_random_headers()) response.raise_for_status() articles = self.parse_articles(response.text) issue_id = self.db.save_issue(issue) if issue_id: self.db.save_articles_batch([ (issue_id, a['section'], a['title'], a.get('author'), a.get('page'), a['url']) for a in articles ]) logging.info(f"成功保存刊期: {issue['publish_date']} {issue['issue_number']}") except Exception as e: logging.error(f"处理刊期失败: {issue['publish_date']}, 错误: {str(e)}") def parse_articles(self, html): # 实现文章解析逻辑 pass def _has_next_page(self, html): soup = BeautifulSoup(html, 'lxml') return bool(soup.select_one('.next-page'))

8. 项目部署与监控

8.1 定时任务配置

from apscheduler.schedulers.blocking import BlockingScheduler def run_spider(): db = DatabaseManager("dbname=newspaper user=postgres") spider = NewspaperSpider("https://epaper.example.com", db) spider.crawl_issues() scheduler = BlockingScheduler() scheduler.add_job(run_spider, 'cron', hour=2, minute=30) # 每天凌晨2:30运行 scheduler.start()

8.2 日志监控与报警

配置日志监控系统（如Sentry）捕获异常：

import sentry_sdk sentry_sdk.init( dsn="your_sentry_dsn", traces_sample_rate=1.0 ) try: run_spider() except Exception as e: logging.exception("爬虫运行失败") sentry_sdk.capture_exception(e)

8.3 数据质量检查

定期运行数据校验脚本：

def check_data_quality(db): """检查数据完整性和一致性""" with db.conn.cursor() as cur: # 检查缺失的刊期 cur.execute(""" SELECT DISTINCT date_trunc('month', publish_date) AS month FROM generate_series( '2018-01-01'::date, CURRENT_DATE, '1 month'::interval ) month LEFT JOIN newspaper_issues ON date_trunc('month', publish_date) = month WHERE newspaper_issues.id IS NULL """) missing_months = [row[0] for row in cur.fetchall()] # 检查文章数量异常的刊期 cur.execute(""" SELECT issue_id, COUNT(*) FROM newspaper_articles GROUP BY issue_id HAVING COUNT(*) < 5 # 假设每期至少有5篇文章 """) sparse_issues = cur.fetchall() return { 'missing_months': missing_months, 'sparse_issues': sparse_issues }

9. 常见问题与解决方案

9.1 动态加载内容处理

对于通过JavaScript动态加载的内容，可以采用以下策略：

1. 分析AJAX请求：使用浏览器开发者工具查看XHR请求
2. 直接调用API：如果网站提供公开API
3. Selenium渲染：作为最后手段

def get_dynamic_content(url): from selenium.webdriver.chrome.options import Options options = Options() options.headless = True driver = webdriver.Chrome(options=options) try: driver.get(url) WebDriverWait(driver, 10).until( EC.presence_of_element_located((By.CSS_SELECTOR, ".dynamic-content")) ) return driver.page_source finally: driver.quit()

9.2 验证码破解策略

遇到验证码时的处理流程：

1. 降低请求频率：增加延迟时间
2. 使用验证码识别服务：如第三方API
3. 人工干预：重要任务可以暂停等待人工处理

def handle_captcha(page_source): if "captcha" in page_source.lower(): logging.warning("遇到验证码，尝试自动处理...") # 这里可以集成验证码识别服务 raise Exception("需要人工处理验证码")

9.3 数据更新策略

保持数据新鲜的几种方案：

1. 增量抓取：只抓取新增的刊期
2. 校验更新：检查已有刊期是否有内容变更
3. 全量刷新：定期（如每年）全量更新一次

def get_latest_issue_date(db): """获取数据库中最新刊期的日期""" with db.conn.cursor() as cur: cur.execute("SELECT MAX(publish_date) FROM newspaper_issues") return cur.fetchone()[0]

10. 性能优化技巧

10.1 异步请求加速

使用aiohttp实现异步抓取：

import aiohttp import asyncio async def fetch_async(session, url): async with session.get(url) as response: return await response.text() async def crawl_async(urls): async with aiohttp.ClientSession() as session: tasks = [fetch_async(session, url) for url in urls] return await asyncio.gather(*tasks, return_exceptions=True)

10.2 数据库批量操作优化

使用COPY命令实现高效批量导入：

def bulk_insert_articles(db, articles): """使用COPY命令批量导入文章数据""" from io import StringIO buffer = StringIO() for article in articles: buffer.write("\t".join([ str(article['issue_id']), article['section_name'], article['title'], article.get('author', ''), str(article.get('page_number', 0)), article['url'] ]) + "\n") buffer.seek(0) with db.conn.cursor() as cur: cur.copy_from( buffer, 'newspaper_articles', columns=('issue_id', 'section_name', 'title', 'author', 'page_number', 'url') ) db.conn.commit()

10.3 内存管理技巧

处理大规模数据时的内存优化：

def process_large_issue(issue_url): """流式处理大刊期数据""" with requests.get(issue_url, stream=True) as r: r.raise_for_status() for chunk in r.iter_content(chunk_size=8192): # 逐步处理数据块，避免内存爆炸 process_chunk(chunk)

11. 项目扩展方向

11.1 内容全文抓取

在目录基础上扩展文章正文抓取：

class ContentSpider: def __init__(self, db_manager): self.db = db_manager def crawl_missing_contents(self): """抓取缺失正文的文章""" with self.db.conn.cursor() as cur: cur.execute(""" SELECT id, url FROM newspaper_articles WHERE content IS NULL LIMIT 1000 """) for article_id, url in cur.fetchall(): try: content = self.fetch_article_content(url) self.save_content(article_id, content) except Exception as e: logging.error(f"抓取文章失败: {url}, 错误: {str(e)}") def fetch_article_content(self, url): # 实现文章正文抓取逻辑 pass def save_content(self, article_id, content): with self.db.conn.cursor() as cur: cur.execute( "UPDATE newspaper_articles SET content=%s WHERE id=%s", (content, article_id) ) self.db.conn.commit()

11.2 数据分析应用

基于抓取数据构建分析功能：

def analyze_section_trends(db, start_date, end_date): """分析各版面的文章数量趋势""" with db.conn.cursor() as cur: cur.execute(""" SELECT section_name, COUNT(*) as article_count, date_trunc('month', publish_date) as month FROM newspaper_articles JOIN newspaper_issues ON issue_id = newspaper_issues.id WHERE publish_date BETWEEN %s AND %s GROUP BY section_name, month ORDER BY month, article_count DESC """, (start_date, end_date)) return cur.fetchall()

11.3 自动化报告生成

定期生成数据质量报告：

def generate_monthly_report(db, output_file='report.html'): """生成月度抓取报告""" from jinja2 import Template with db.conn.cursor() as cur: cur.execute(""" SELECT date_trunc('month', publish_date) as month, COUNT(DISTINCT newspaper_issues.id) as issue_count, COUNT(newspaper_articles.id) as article_count FROM newspaper_issues LEFT JOIN newspaper_articles ON issue_id = newspaper_issues.id GROUP BY month ORDER BY month DESC LIMIT 12 """) stats = cur.fetchall() template = Template(''' <html> <body> <h1>月度抓取报告</h1> <table> <tr><th>月份</th><th>刊期数</th><th>文章数</th></tr> {% for row in stats %} <tr> <td>{{ row[0].strftime('%Y-%m') }}</td> <td>{{ row[1] }}</td> <td>{{ row[2] }}</td> </tr> {% endfor %} </table> </body> </html> ''') with open(output_file, 'w') as f: f.write(template.render(stats=stats))