"""Tier 1 主爬虫 - 增量爬取政府公告 工作流程： 1. 读取所有Tier 1站点的列表页URL 2. 解析列表页，提取标题+详情链接+日期 3. 对新的详情页发起请求，提取正文 4. 自动发现翻页 5. URL去重实现增量 """ import scrapy import re from datetime import datetime from gov_crawler.items import AnnouncementItem from utils.classifier import load_sources from utils.dedup import UrlDedup from utils.page_parser import PageParser class GovListSpider(scrapy.Spider): name = 'gov_list' allowed_domains = [] custom_settings = { 'DOWNLOAD_TIMEOUT': 60, 'RETRY_TIMES': 3, 'RETRY_HTTP_CODES': [500, 502, 503, 504, 408, 429, 400, 403, 302, 301], 'CONCURRENT_REQUESTS': 4, 'CONCURRENT_REQUESTS_PER_DOMAIN': 1, 'DOWNLOAD_DELAY': 2.0, 'AUTOTHROTTLE_ENABLED': False, 'ROBOTSTXT_OBEY': False, 'COOKIES_ENABLED': True, 'REDIRECT_ENABLED': True, 'REDIRECT_MAX_TIMES': 3, 'LOG_LEVEL': 'INFO', } def __init__(self, tier=1, limit=0, *args, **kwargs): """初始化 Args: tier: 爬取等级 (1/2/3) limit: 限制爬取数量 (0=不限制) """ super().__init__(*args, **kwargs) self.tier_num = int(tier) self.limit = int(limit) # 加载站点 tiers = load_sources('data/sources.csv') self.sites = tiers[self.tier_num] if self.limit > 0: self.sites = self.sites[:self.limit] self.dedup = UrlDedup() self.total = len(self.sites) self.crawled_count = 0 self.logger.info(f'=' * 50) self.logger.info(f'爬虫启动: Tier {self.tier_num} | {self.total} 个站点') self.logger.info(f'已爬取详情页: {self.dedup.stats()["detail_urls_crawled"]} 个') self.logger.info(f'=' * 50) def start_requests(self): for site in self.sites: self.crawled_count += 1 yield scrapy.Request( url=site['url'], callback=self.parse_list, errback=self.handle_list_error, meta={ 'site_name': site['site_name'], 'list_url': site['url'], 'domain': site['domain'], 'page_num': 1, }, dont_filter=True, ) def parse_list(self, response): """解析列表页""" meta = response.meta parser = PageParser(response.url) self.logger.info(f'[{self.crawled_count}/{self.total}] 📋 列表页: {meta["site_name"]} ({response.status})') # 解析列表项 items = parser.parse_list_page(response) self.logger.info(f' → 发现 {len(items)} 条记录') # 如果列表为空，尝试查找iframe（政府网站常见结构） if not items: iframes = response.xpath('//iframe/@src').extract() for iframe_src in iframes: iframe_url = response.urljoin(iframe_src) if iframe_url != response.url: self.logger.info(f' → 发现iframe: {iframe_url}') yield scrapy.Request( url=iframe_url, callback=self.parse_list, errback=self.handle_list_error, meta=meta, dont_filter=True, ) return # 只跟第一个有效iframe # 处理每条记录 for item_data in items: detail_url = item_data['url'] title = item_data['title'] date = item_data['date'] # 去重检查 if self.dedup.is_detail_crawled(detail_url): continue # 爬取详情页 yield scrapy.Request( url=detail_url, callback=self.parse_detail, errback=self.handle_detail_error, meta={ 'site_name': meta['site_name'], 'list_url': meta['list_url'], 'title': title, 'date': date, 'detail_url': detail_url, 'domain': meta['domain'], }, dont_filter=True, ) # 翻页处理 - 只爬第一页 if items: max_pages = 1 if hasattr(parser, 'ljobx_pagination'): _, ljobx_max = parser.ljobx_pagination max_pages = min(max_pages, ljobx_max) next_page = parser.find_next_page(response) if next_page and meta.get('page_num', 1) < max_pages: self.logger.info(f' → 翻到第 {meta["page_num"] + 1} 页') yield scrapy.Request( url=next_page, callback=self.parse_list, errback=self.handle_list_error, meta={ **meta, 'page_num': meta.get('page_num', 1) + 1, }, dont_filter=True, ) # 标记列表页已爬取 self.dedup.mark_list_crawled(meta['list_url']) def parse_detail(self, response): """解析详情页""" meta = response.meta parser = PageParser(response.url) # 提取正文和附件 content = parser.parse_detail_page(response) attachments = parser.extract_attachments(response) # Try ljobx title extraction (more precise than list-page title) ljobx_title, _ = parser.extract_by_ljobx_rules(response) item_title = ljobx_title if ljobx_title else meta['title'] item = AnnouncementItem() item['site_name'] = meta['site_name'] item['source_url'] = meta['list_url'] item['page_url'] = meta['detail_url'] item['domain'] = meta['domain'] item['title'] = item_title item['publish_date'] = meta['date'] # Strip any remaining HTML tags from content if content: content_stripped = re.sub(r'<[^>]+>', '', content) content_stripped = re.sub(r'\s+', ' ', content_stripped).strip() else: content_stripped = '' # 质量过滤：正文太短视为无效，不污染数据库 if content_stripped and len(content_stripped) < 80: content_stripped = '' item['content'] = content_stripped[:5000] item['content_text'] = content_stripped[:2000] item['crawl_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') item['status'] = 'success' if content_stripped else 'no_content' item['error_msg'] = f'attachments:{len(attachments)}' if attachments else '' # 标记已爬取 self.dedup.mark_detail_crawled(meta['detail_url']) # 每50条保存一次去重记录 if self.dedup.stats()['detail_urls_crawled'] % 50 == 0: self.dedup.save() self.logger.info(f' ✅ 详情: {meta["title"][:40]} | {meta["date"]} | {meta["domain"]}') yield item def handle_list_error(self, failure): """列表页请求失败""" request = failure.request meta = request.meta self.logger.warning(f'[{self.crawled_count}/{self.total}] ❌ 列表页失败: {meta["site_name"]} - {str(failure.value)[:60]}') item = AnnouncementItem() item['site_name'] = meta['site_name'] item['source_url'] = meta['list_url'] item['page_url'] = request.url item['domain'] = meta['domain'] item['title'] = '' item['publish_date'] = '' item['content'] = '' item['content_text'] = '' item['crawl_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') item['status'] = 'failed' item['error_msg'] = f'列表页: {str(failure.value)[:200]}' yield item def handle_detail_error(self, failure): """详情页请求失败""" request = failure.request meta = request.meta item = AnnouncementItem() item['site_name'] = meta['site_name'] item['source_url'] = meta['list_url'] item['page_url'] = meta['detail_url'] item['domain'] = meta['domain'] item['title'] = meta.get('title', '') item['publish_date'] = meta.get('date', '') item['content'] = '' item['content_text'] = '' item['crawl_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') item['status'] = 'failed' item['error_msg'] = f'详情页: {str(failure.value)[:200]}' self.logger.info(f' ❌ 详情页失败: {meta["title"][:30]} - {str(failure.value)[:40]}') yield item def closed(self, reason): """爬虫关闭时保存去重数据""" self.dedup.save() stats = self.dedup.stats() self.logger.info(f'=' * 50) self.logger.info(f'爬虫完成: {reason}') self.logger.info(f'共爬取详情页: {stats["detail_urls_crawled"]} 个') self.logger.info(f'共爬取列表页: {stats["list_urls_crawled"]} 个') self.logger.info(f'=' * 50)