v6ole
c548a8b5bd
feat(core): 添加大化县政府网采购公告数据表和相关功能 - 创建 dahuagov_announcements 表用于存储大化县政府网采购公告 - 添加相关索引以提高查询性能 - 实现 save_dahuagov_announcements、get_new_dahuagov_announcements 和 mark_dahuagov_announcements_sent 方法 - 修改统计查询以包含大化县公告数据 - 更新内容哈希检查逻辑以支持新表 feat(cron): 集成大化县政府网采购公告爬取功能 - 导入大化县政府网爬虫模块 - 修改定时任务流程以同时爬取广西政府采购网和大化县政府网 - 对不同来源公告采用不同处理策略： - 广西政府采购网：关键词筛选后推送 - 大化县政府网：全部推送，不过滤关键词 - 分别处理和统计两个来源的公告数据 - 实现独立的通知发送和状态更新机制 feat(notification): 优化企业微信通知显示大化县来源标识 - 为不同来源公告添加前缀标识（【大化县政府网】或【广西政府采购网】） - 根据公告来源动态调整通知标题： - 单一来源显示具体来源 - 双来源显示"双源监控"标识 - 改进通知卡片的来源区分度，便于用户识别公告来源 ```
356 lines
12 KiB
Python
356 lines
12 KiB
Python
"""
|
|
大化瑶族自治县政府采购公告爬虫
|
|
爬取大化县政府网站的采购公告页面
|
|
"""
|
|
|
|
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()
|