Files
GX-gp-notify/gx_gp_monitor/crawler/parsers.py
T
2026-01-07 17:37:09 +08:00

302 lines
9.1 KiB
Python

"""
数据解析器模块
负责解析广西政府采购网的API响应数据
"""
import json
from typing import List, Dict, Any, Optional
from datetime import datetime
try:
from ..core.models import Announcement, AnnouncementSource, AnnouncementType
from ..core.logger import get_logger
except ImportError:
from core.models import Announcement, AnnouncementSource, AnnouncementType
from core.logger import get_logger
logger = get_logger(__name__)
class AnnouncementParser:
"""公告数据解析器"""
@staticmethod
def parse_api_response(response_data: Dict[str, Any],
source: AnnouncementSource,
crawled_at: datetime) -> List[Announcement]:
"""
解析API响应数据
Args:
response_data: API响应数据
source: 公告来源
crawled_at: 爬取时间
Returns:
List[Announcement]: 解析后的公告列表
"""
if not response_data or not isinstance(response_data, dict):
logger.warning("API响应数据无效")
return []
try:
# 检查响应状态
if not response_data.get("success", False):
logger.warning(f"API响应失败: {response_data.get('message', '未知错误')}")
return []
# 获取数据部分
result = response_data.get("result", {})
data = result.get("data", {})
records = data.get("data", [])
if not records:
logger.info(f"来源 {source.name} 没有新数据")
return []
announcements = []
for record in records:
try:
announcement = AnnouncementParser._parse_single_record(
record, source, crawled_at)
if announcement:
announcements.append(announcement)
except Exception as e:
logger.warning(f"解析公告记录失败: {str(e)}, 记录: {record}")
continue
logger.info(f"成功解析 {len(announcements)}/{len(records)} 条公告记录")
return announcements
except Exception as e:
logger.error(f"解析API响应数据失败: {str(e)}")
return []
@staticmethod
def _parse_single_record(record: Dict[str, Any],
source: AnnouncementSource,
crawled_at: datetime) -> Optional[Announcement]:
"""
解析单个公告记录
Args:
record: 公告记录数据
source: 公告来源
crawled_at: 爬取时间
Returns:
Optional[Announcement]: 解析后的公告对象
"""
try:
# 提取基本字段
title_raw = record.get("title", "")
title = str(title_raw).strip() if title_raw is not None else ""
if not title:
return None
# 解析发布时间
publish_timestamp = record.get("publishDate")
if not publish_timestamp:
logger.warning(f"公告缺少发布时间: {title[:50]}...")
return None
try:
# 时间戳转换为datetime
publish_date = datetime.fromtimestamp(int(publish_timestamp) / 1000)
except (ValueError, TypeError) as e:
logger.warning(f"发布时间格式错误: {publish_timestamp}, 错误: {str(e)}")
return None
# 提取其他字段
purchase_name_raw = record.get("purchaseName", "")
purchase_name = str(purchase_name_raw).strip() if purchase_name_raw is not None else ""
article_id = record.get("articleId")
if not article_id:
logger.warning(f"公告缺少文章ID: {title[:50]}...")
return None
# 构建内容链接
content_url = AnnouncementParser._build_content_url(
source.category_id, source.code, article_id)
# 创建公告对象
announcement = Announcement(
title=title,
publish_date=publish_date,
purchase_name=purchase_name,
content_url=content_url,
source_code=source.code,
source_name=source.name,
announcement_type=source.type,
crawled_at=crawled_at,
is_new=True # 默认标记为新公告
)
# 生成内容哈希用于去重
announcement.generate_content_hash()
return announcement
except Exception as e:
logger.error(f"解析单个公告记录失败: {str(e)}")
return None
@staticmethod
def _build_content_url(category_id: int, source_code: str, article_id: int) -> str:
"""
构建公告内容链接
Args:
category_id: 分类ID
source_code: 来源代码
article_id: 文章ID
Returns:
str: 内容链接
"""
return f"https://zfcg.gxzf.gov.cn/site/detail?parentId={category_id}&articleId={article_id}"
@staticmethod
def validate_response_structure(response_data: Dict[str, Any]) -> bool:
"""
验证API响应数据结构
Args:
response_data: API响应数据
Returns:
bool: 结构是否有效
"""
try:
if not isinstance(response_data, dict):
return False
# 检查必需的字段
if "success" not in response_data:
return False
if not response_data.get("success", False):
return False
result = response_data.get("result", {})
if not isinstance(result, dict):
return False
data = result.get("data", {})
if not isinstance(data, dict):
return False
records = data.get("data", [])
if not isinstance(records, list):
return False
return True
except Exception as e:
logger.warning(f"验证响应结构失败: {str(e)}")
return False
@staticmethod
def extract_pagination_info(response_data: Dict[str, Any]) -> Dict[str, Any]:
"""
提取分页信息
Args:
response_data: API响应数据
Returns:
Dict[str, Any]: 分页信息
"""
try:
result = response_data.get("result", {})
data = result.get("data", {})
return {
"total": data.get("total", 0),
"page_no": data.get("pageNo", 1),
"page_size": data.get("pageSize", 15),
"pages": data.get("pages", 0),
"empty": data.get("empty", True),
"has_next": data.get("hasNext", False),
"has_previous": data.get("hasPrevious", False)
}
except Exception as e:
logger.warning(f"提取分页信息失败: {str(e)}")
return {
"total": 0,
"page_no": 1,
"page_size": 15,
"pages": 0,
"empty": True,
"has_next": False,
"has_previous": False
}
class SensitiveWordChecker:
"""敏感词检查器"""
@staticmethod
def parse_check_response(response_data: Dict[str, Any]) -> bool:
"""
解析敏感词检查响应
Args:
response_data: 检查响应数据
Returns:
bool: 检查是否通过
"""
try:
if not isinstance(response_data, dict):
logger.warning("敏感词检查响应格式无效")
return False
success = response_data.get("success", False)
if not success:
message = response_data.get("message", "未知错误")
logger.warning(f"敏感词检查失败: {message}")
return False
return True
except Exception as e:
logger.error(f"解析敏感词检查响应失败: {str(e)}")
return False
class ErrorResponseParser:
"""错误响应解析器"""
@staticmethod
def parse_error(response_data: Dict[str, Any]) -> str:
"""
解析错误响应
Args:
response_data: 错误响应数据
Returns:
str: 错误信息
"""
try:
if not isinstance(response_data, dict):
return "响应格式无效"
# 尝试提取错误信息
error_msg = response_data.get("message") or response_data.get("msg")
if error_msg:
return str(error_msg)
# 检查状态码
errcode = response_data.get("errcode")
if errcode:
return f"错误码: {errcode}"
return "未知错误"
except Exception as e:
return f"解析错误响应失败: {str(e)}"