1. 项目背景与需求分析数字报纸的刊期目录是新闻机构发布内容的重要组织形式通常以分页形式展示历史内容。作为一名数据采集工程师我最近接到一个需求需要完整抓取某主流数字报纸过去5年的刊期目录数据包含每期的发布日期、版面名称、文章标题等关键信息并将这些数据规范存储到数据库中供后续分析使用。这个需求看似简单但在实际开发中会遇到几个典型难题归档翻页问题数字报纸的刊期目录通常采用分页加载部分老报纸的翻页机制特殊需要模拟真实用户操作才能获取完整数据。日期格式混乱不同时期的报纸可能使用不同的日期格式如2023-01-01、2023年1月1日、Jan 1, 2023等需要统一处理。反爬机制应对新闻网站通常对目录页有访问频率限制直接暴力抓取容易被封IP。数据存储一致性需要确保抓取的数据完整、准确并在入库时处理好重复数据和异常情况。2. 技术选型与环境准备2.1 核心工具选择经过对比测试我选择了以下技术栈请求库Requests Session维持会话解析库BeautifulSoup4HTML解析 lxml解析引擎浏览器模拟Selenium仅用于特殊翻页场景日期处理dateutil.parser智能日期解析数据库PostgreSQL支持JSON字段存储原始数据任务调度APScheduler定时增量抓取# 基础环境安装 pip install requests beautifulsoup4 python-dateutil selenium psycopg2 apscheduler2.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;q0.9 } DELAY random.uniform(1, 3) # 随机延迟1-3秒 MAX_RETRY 3 # 最大重试次数提示实际项目中建议将配置参数提取到单独的文件中方便后期调整和维护。3. 分页抓取实现方案3.1 基础分页模式识别数字报纸的分页通常有两种实现方式传统分页URL中包含page参数如?page2动态加载通过AJAX请求获取数据查看XHR请求通过分析目标网站的翻页机制我发现了以下规律def generate_page_urls(base_url, total_pages): 生成分页URL列表 return [f{base_url}?page{i} for i in range(1, total_pages1)]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 items3.3 断点续爬设计为防止意外中断导致重复抓取需要记录已抓取的分页状态import json from pathlib import Path class CrawlState: def __init__(self, state_filecrawl_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, successTrue): 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 常见日期格式分析在抓取过程中我遇到了以下日期格式变体2023-01-012023年1月1日01/01/2023Jan 1, 20231st January 20234.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, fuzzyTrue) return dt.strftime(%Y-%m-%d) except Exception as e: print(f日期解析失败: {date_str}, 错误: {e}) return None4.3 日期验证与修正对于历史报纸还需要验证日期的合理性def validate_newspaper_date(date_str, start_year2018): 验证报纸日期是否合理 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_str5. 数据存储与去重设计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/xhtmlxml, Accept-Language: zh-CN,zh;q0.9,en;q0.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 session6.3 请求频率控制class RequestThrottler: def __init__(self, base_delay1.0, max_delay5.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( levellogging.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_delay2.0) self.state CrawlState() def crawl_issues(self, start_page1, end_pageNone): 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, headersget_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, successTrue) 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, successFalse) 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], headersget_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(dbnamenewspaper userpostgres) spider NewspaperSpider(https://epaper.example.com, db) spider.crawl_issues() scheduler BlockingScheduler() scheduler.add_job(run_spider, cron, hour2, minute30) # 每天凌晨2:30运行 scheduler.start()8.2 日志监控与报警配置日志监控系统如Sentry捕获异常import sentry_sdk sentry_sdk.init( dsnyour_sentry_dsn, traces_sample_rate1.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动态加载的内容可以采用以下策略分析AJAX请求使用浏览器开发者工具查看XHR请求直接调用API如果网站提供公开APISelenium渲染作为最后手段def get_dynamic_content(url): from selenium.webdriver.chrome.options import Options options Options() options.headless True driver webdriver.Chrome(optionsoptions) 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 验证码破解策略遇到验证码时的处理流程降低请求频率增加延迟时间使用验证码识别服务如第三方API人工干预重要任务可以暂停等待人工处理def handle_captcha(page_source): if captcha in page_source.lower(): logging.warning(遇到验证码尝试自动处理...) # 这里可以集成验证码识别服务 raise Exception(需要人工处理验证码)9.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_exceptionsTrue)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, streamTrue) as r: r.raise_for_status() for chunk in r.iter_content(chunk_size8192): # 逐步处理数据块避免内存爆炸 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_filereport.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 trth月份/thth刊期数/thth文章数/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(statsstats)) 
长河君 · 2026/10/31 15:29:29 · 阅读 1.2k 数字报纸刊期目录爬虫实战：分页抓取与数据存储
1. 项目背景与需求分析数字报纸的刊期目录是新闻机构发布内容的重要组织形式通常以分页形式展示历史内容。作为一名数据采集工程师我最近接到一个需求需要完整抓取某主流数字报纸过去5年的刊期目录数据包含每期的发布日期、版面名称、文章标题等关键信息并将这些数据规范存储到数据库中供后续分析使用。这个需求看似简单但在实际开发中会遇到几个典型难题归档翻页问题数字报纸的刊期目录通常采用分页加载部分老报纸的翻页机制特殊需要模拟真实用户操作才能获取完整数据。日期格式混乱不同时期的报纸可能使用不同的日期格式如2023-01-01、2023年1月1日、Jan 1, 2023等需要统一处理。反爬机制应对新闻网站通常对目录页有访问频率限制直接暴力抓取容易被封IP。数据存储一致性需要确保抓取的数据完整、准确并在入库时处理好重复数据和异常情况。2. 技术选型与环境准备2.1 核心工具选择经过对比测试我选择了以下技术栈请求库Requests Session维持会话解析库BeautifulSoup4HTML解析 lxml解析引擎浏览器模拟Selenium仅用于特殊翻页场景日期处理dateutil.parser智能日期解析数据库PostgreSQL支持JSON字段存储原始数据任务调度APScheduler定时增量抓取# 基础环境安装 pip install requests beautifulsoup4 python-dateutil selenium psycopg2 apscheduler2.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;q0.9 } DELAY random.uniform(1, 3) # 随机延迟1-3秒 MAX_RETRY 3 # 最大重试次数提示实际项目中建议将配置参数提取到单独的文件中方便后期调整和维护。3. 分页抓取实现方案3.1 基础分页模式识别数字报纸的分页通常有两种实现方式传统分页URL中包含page参数如?page2动态加载通过AJAX请求获取数据查看XHR请求通过分析目标网站的翻页机制我发现了以下规律def generate_page_urls(base_url, total_pages): 生成分页URL列表 return [f{base_url}?page{i} for i in range(1, total_pages1)]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 items3.3 断点续爬设计为防止意外中断导致重复抓取需要记录已抓取的分页状态import json from pathlib import Path class CrawlState: def __init__(self, state_filecrawl_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, successTrue): 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 常见日期格式分析在抓取过程中我遇到了以下日期格式变体2023-01-012023年1月1日01/01/2023Jan 1, 20231st January 20234.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, fuzzyTrue) return dt.strftime(%Y-%m-%d) except Exception as e: print(f日期解析失败: {date_str}, 错误: {e}) return None4.3 日期验证与修正对于历史报纸还需要验证日期的合理性def validate_newspaper_date(date_str, start_year2018): 验证报纸日期是否合理 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_str5. 数据存储与去重设计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/xhtmlxml, Accept-Language: zh-CN,zh;q0.9,en;q0.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 session6.3 请求频率控制class RequestThrottler: def __init__(self, base_delay1.0, max_delay5.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( levellogging.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_delay2.0) self.state CrawlState() def crawl_issues(self, start_page1, end_pageNone): 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, headersget_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, successTrue) 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, successFalse) 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], headersget_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(dbnamenewspaper userpostgres) spider NewspaperSpider(https://epaper.example.com, db) spider.crawl_issues() scheduler BlockingScheduler() scheduler.add_job(run_spider, cron, hour2, minute30) # 每天凌晨2:30运行 scheduler.start()8.2 日志监控与报警配置日志监控系统如Sentry捕获异常import sentry_sdk sentry_sdk.init( dsnyour_sentry_dsn, traces_sample_rate1.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动态加载的内容可以采用以下策略分析AJAX请求使用浏览器开发者工具查看XHR请求直接调用API如果网站提供公开APISelenium渲染作为最后手段def get_dynamic_content(url): from selenium.webdriver.chrome.options import Options options Options() options.headless True driver webdriver.Chrome(optionsoptions) 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 验证码破解策略遇到验证码时的处理流程降低请求频率增加延迟时间使用验证码识别服务如第三方API人工干预重要任务可以暂停等待人工处理def handle_captcha(page_source): if captcha in page_source.lower(): logging.warning(遇到验证码尝试自动处理...) # 这里可以集成验证码识别服务 raise Exception(需要人工处理验证码)9.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_exceptionsTrue)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, streamTrue) as r: r.raise_for_status() for chunk in r.iter_content(chunk_size8192): # 逐步处理数据块避免内存爆炸 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_filereport.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 trth月份/thth刊期数/thth文章数/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(statsstats))