""" 数据解析器模块 负责解析广西政府采购网的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)}"