feat: 添加数据解析器模块 + 测试

2026-05-09 13:41:35 +08:00
parent d9af819d2e
commit 7db14ee411
2 changed files with 245 additions and 0 deletions
@@ -0,0 +1,139 @@
+import hashlib
+from datetime import datetime
+from typing import Any, Dict, List
+from urllib.parse import urljoin
+
+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)
+        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()