""" 大化瑶族自治县政府采购公告爬虫 爬取大化县政府网站的采购公告页面 """ import time import random from typing import List, Optional, Tuple from datetime import datetime from urllib.parse import urljoin, urlparse from bs4 import BeautifulSoup import requests from fake_useragent import UserAgent try: from ..core.models import Announcement, AnnouncementSource, AnnouncementType, CrawlResult, CrawlStatus from ..core.config_manager import get_config from ..core.logger import get_logger, log_crawl_start, log_crawl_success, log_crawl_error from ..core.reliability import ( retry_on_exception, RetryConfig, session_with_retry, TimeoutConfig, safe_execute, check_system_health ) except ImportError: from core.models import Announcement, AnnouncementSource, AnnouncementType, CrawlResult, CrawlStatus from core.config_manager import get_config from core.logger import get_logger, log_crawl_start, log_crawl_success, log_crawl_error from core.reliability import ( retry_on_exception, RetryConfig, session_with_retry, TimeoutConfig, safe_execute, check_system_health ) logger = get_logger(__name__) class DahuagovSpider: """大化县政府网站爬虫""" # 大化县政府网站配置 BASE_URL = "http://www.gxdh.gov.cn" ANNOUNCEMENT_PATH = "/xxgk/zdlyxxgk/ggzypzly/zfcgly/cggg/" def __init__(self): self.config = get_config() self.ua = UserAgent() self.session = None self.request_count = 0 self.error_count = 0 def _get_random_user_agent(self) -> str: """获取随机User-Agent""" try: from fake_useragent import UserAgent return self.ua.random except: return random.choice(self.config.crawler.user_agents) def init_session(self): """初始化会话""" if self.session is None: timeout_config = TimeoutConfig( connect_timeout=self.config.crawler.timeout, read_timeout=self.config.crawler.timeout ) retry_config = RetryConfig( max_retries=self.config.crawler.max_retries, initial_delay=self.config.crawler.retry_delay, max_delay=self.config.crawler.max_retry_delay, backoff_factor=self.config.crawler.backoff_factor ) self.session = requests.Session() adapter = requests.adapters.HTTPAdapter( pool_connections=10, pool_maxsize=20, max_retries=0 ) self.session.mount('http://', adapter) self.session.mount('https://', adapter) self.session.timeout = (timeout_config.connect_timeout, timeout_config.read_timeout) return self.session def close_session(self): """关闭会话""" if self.session: self.session.close() self.session = None def _fetch_page(self, url: str) -> Tuple[Optional[str], Optional[str]]: """ 获取页面内容 Args: url: 页面URL Returns: Tuple[Optional[str], Optional[str]]: (页面内容, 错误信息) """ try: session = self.init_session() user_agent = self._get_random_user_agent() headers = { "User-Agent": user_agent, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Connection": "keep-alive", "Referer": self.BASE_URL } # 添加随机延迟 delay = random.uniform( self.config.crawler.request_delay, self.config.crawler.request_delay_max ) time.sleep(delay) response = session.get(url, headers=headers, timeout=self.session.timeout) self.request_count += 1 if response.status_code == 200: # 确保使用正确的编码 response.encoding = response.apparent_encoding or 'utf-8' return response.text, None else: return None, f"请求失败, 状态码: {response.status_code}" except Exception as e: self.error_count += 1 logger.error(f"获取页面异常: {str(e)}") return None, f"请求异常: {str(e)}" def _parse_page(self, html: str, crawled_at: datetime) -> List[Announcement]: """ 解析页面内容 Args: html: 页面HTML内容 crawled_at: 爬取时间 Returns: List[Announcement]: 解析后的公告列表 """ announcements = [] try: soup = BeautifulSoup(html, 'html.parser') # 查找公告列表 # 大化县政府网站使用 ul.more-list 结构 lists = soup.find_all('ul', class_='more-list') if not lists: logger.info("未找到公告列表") return announcements for ul in lists: lis = ul.find_all('li') for li in lis: try: announcement = self._parse_li_element(li, crawled_at) if announcement: announcements.append(announcement) except Exception as e: logger.warning(f"解析单个公告失败: {str(e)}") continue logger.info(f"成功解析 {len(announcements)} 条公告") except Exception as e: logger.error(f"解析页面失败: {str(e)}") return announcements def _parse_li_element(self, li, crawled_at: datetime) -> Optional[Announcement]: """ 解析单个li元素 Args: li: BeautifulSoup li元素 crawled_at: 爬取时间 Returns: Optional[Announcement]: 解析后的公告对象 """ try: # 查找日期 span date_span = li.find('span') if not date_span: return None date_text = date_span.get_text(strip=True) if not date_text: return None # 解析日期 try: publish_date = datetime.strptime(date_text, "%Y-%m-%d") except ValueError: logger.warning(f"日期格式无法解析: {date_text}") return None # 查找链接和标题 link_tag = li.find('a') if not link_tag: return None title = link_tag.get('title', '') or link_tag.get_text(strip=True) if not title: return None href = link_tag.get('href', '') if not href: return None # 构建完整URL if href.startswith('./') or href.startswith('../'): content_url = urljoin(self.BASE_URL + self.ANNOUNCEMENT_PATH, href) elif href.startswith('/'): content_url = self.BASE_URL + href elif href.startswith('http'): content_url = href else: content_url = urljoin(self.BASE_URL + self.ANNOUNCEMENT_PATH, href) # 创建公告对象 announcement = Announcement( title=title, publish_date=publish_date, purchase_name="大化瑶族自治县", # 默认采购单位 content_url=content_url, source_code="dahuagov", source_name="大化县政府网采购公告", announcement_type=AnnouncementType.PURCHASE, crawled_at=crawled_at, is_new=True ) # 生成内容哈希用于去重 announcement.generate_content_hash() return announcement except Exception as e: logger.warning(f"解析li元素失败: {str(e)}") return None @retry_on_exception(RetryConfig(max_retries=2)) def crawl(self) -> CrawlResult: """ 爬取公告（只爬取第一页） Returns: CrawlResult: 爬取结果 """ log_crawl_start("大化县政府网采购公告") start_time = datetime.now() result = CrawlResult( source=AnnouncementSource( code="dahuagov", category_id=0, name="大化县政府网采购公告", type=AnnouncementType.PURCHASE ), status=CrawlStatus.RUNNING, crawled_at=start_time ) try: # 构建完整URL（只爬取第一页） url = self.BASE_URL + self.ANNOUNCEMENT_PATH logger.info(f"开始爬取大化县政府网站: {url}") # 获取页面内容 html, error_msg = self._fetch_page(url) if error_msg: logger.warning(f"获取页面失败: {error_msg}") result.status = CrawlStatus.FAILED result.error_message = error_msg return result if not html: logger.info("页面内容为空") result.status = CrawlStatus.SUCCESS result.total_count = 0 result.new_count = 0 return result # 解析页面 announcements = self._parse_page(html, start_time) # 更新结果 result.announcements = announcements result.total_count = len(announcements) result.new_count = len(announcements) result.status = CrawlStatus.SUCCESS duration = (datetime.now() - start_time).total_seconds() result.duration = duration log_crawl_success("大化县政府网采购公告", len(announcements), duration) except Exception as e: duration = (datetime.now() - start_time).total_seconds() result.duration = duration result.status = CrawlStatus.FAILED result.error_message = str(e) log_crawl_error("大化县政府网采购公告", str(e)) return result def get_stats(self) -> dict: """获取爬虫统计信息""" return { "request_count": self.request_count, "error_count": self.error_count, "error_rate": self.error_count / max(self.request_count, 1), "session_active": self.session is not None } def create_dahuagov_spider() -> DahuagovSpider: """ 创建大化县爬虫实例 Returns: DahuagovSpider: 爬虫实例 """ return DahuagovSpider() def crawl_dahuagov_announcements() -> List[CrawlResult]: """ 便捷函数：爬取大化县公告 Returns: List[CrawlResult]: 爬取结果列表 """ spider = create_dahuagov_spider() try: result = spider.crawl() return [result] finally: spider.close_session()