import hashlib from datetime import datetime from typing import Any from urllib.parse import urljoin from zoneinfo import ZoneInfo _TZ = ZoneInfo("Asia/Shanghai") from bs4 import BeautifulSoup def parse_gxgp_api_response( response_data: dict[str, Any], source_code: str, source_name: str, crawled_at: datetime, category_id: int, ) -> list[dict[str, Any]]: if not response_data.get("success"): return [] data = response_data.get("result", {}).get("data", {}) records = data.get("data", []) if not records: return [] results = [] for record in records: title = str(record.get("title", "")).strip() if not title: continue timestamp = record.get("publishDate") if not timestamp: continue try: publish_date = datetime.fromtimestamp(int(timestamp) / 1000, tz=_TZ).replace(tzinfo=None) except (ValueError, TypeError): continue purchase_name = str(record.get("purchaseName", "")).strip() article_id = record.get("articleId") if not article_id: continue content_url = ( f"https://zfcg.gxzf.gov.cn/site/detail?" f"parentId={category_id}&articleId={article_id}" ) announce = { "title": title, "publish_date": publish_date, "purchase_name": purchase_name, "content_url": content_url, "source_code": source_code, "source_name": source_name, "announcement_type": "purchase", "crawl_mode": "auto", "is_new": True, "is_today": publish_date.date() == datetime.now().date(), } announce["content_hash"] = _generate_hash(announce) results.append(announce) return results def extract_pagination(response_data: dict[str, Any]) -> dict[str, Any]: data = response_data.get("result", {}).get("data", {}) return { "total": data.get("total", 0), "page_no": data.get("pageNo", 1), "page_size": data.get("pageSize", 100), "pages": data.get("pages", 0), "empty": data.get("empty", True), "has_next": data.get("hasNext", False), "has_previous": data.get("hasPrevious", False), } def parse_dahuagov_html(html: str, crawled_at: datetime) -> list[dict[str, Any]]: soup = BeautifulSoup(html, "html.parser") lists = soup.find_all("ul", class_="more-list") if not lists: return [] results = [] base_url = "http://www.gxdh.gov.cn" base_path = "/xxgk/zdlyxxgk/ggzypzly/zfcgly/cggg/" for ul in lists: for li in ul.find_all("li"): date_span = li.find("span") if not date_span: continue date_text = date_span.get_text(strip=True) try: publish_date = datetime.strptime(date_text, "%Y-%m-%d") except ValueError: continue link_tag = li.find("a") if not link_tag: continue title = link_tag.get("title", "") or link_tag.get_text(strip=True) href = link_tag.get("href", "") if not title or not href: continue if href.startswith("./") or href.startswith("../"): content_url = urljoin(base_url + base_path, href) elif href.startswith("/"): content_url = base_url + href elif href.startswith("http"): content_url = href else: content_url = urljoin(base_url + base_path, href) announce = { "title": title, "publish_date": publish_date, "purchase_name": "大化瑶族自治县", "content_url": content_url, "source_code": "dahuagov", "source_name": "大化县政府网采购公告", "announcement_type": "purchase", "crawl_mode": "auto", "is_new": True, "is_today": publish_date.date() == datetime.now().date(), } announce["content_hash"] = _generate_hash(announce) results.append(announce) return results def _generate_hash(ann: dict[str, Any]) -> str: content = ( f"{ann['title']}|{ann['publish_date'].strftime('%Y-%m-%d')}" f"|{ann['purchase_name']}|{ann['content_url']}|{ann['source_code']}" ) return hashlib.sha256(content.encode("utf-8")).hexdigest()