#!/usr/bin/env python3 """茂名市人民政府（maoming.gov.cn）建设项目环境影响评价信息爬虫 - NFCMS""" import requests import re import sqlite3 import sys import time from bs4 import BeautifulSoup from urllib.parse import urljoin import urllib.parse BASE_URL = "http://www.maoming.gov.cn" LIST_URL = BASE_URL + "/zwgk/zwzl/zdlyxxgkzl/hjbhxxgk/jsxmhjyxpjxx/index.html" LIST_PAGES = 5 DB_PATH = "/root/search.db" SITE_NAME = "茂名市环评审批公示" SLEEP = 1.5 HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} def get_soup(url): try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = "utf-8" return BeautifulSoup(r.text, "html.parser") except Exception as e: print(f"[WARN] 请求失败: {url} - {e}") return None def parse_list_items(soup): items = [] ul = soup.select_one("div.common-list > ul") if not ul: return items for li in ul.find_all("li", recursive=False): a = li.find("a", href=True) span = li.find("span") if not a: continue href = a.get("href", "").strip() title = a.get("title", "") or a.get_text(strip=True) date_str = span.get_text(strip=True) if span else "" if href and title: full_url = urljoin(BASE_URL, href) items.append((full_url, title, date_str)) return items def parse_detail(soup, url): title = "" meta_title = soup.find("meta", attrs={"name": "ArticleTitle"}) if meta_title: title = meta_title.get("content", "").strip() # Fallback: h1 title if not title: h1 = soup.find("h1") if h1: title = h1.get_text(strip=True) pub_date = "" meta_date = soup.find("meta", attrs={"name": "PubDate"}) if meta_date: m = re.search(r"\d{4}-\d{2}-\d{2}", meta_date.get("content", "")) if m: pub_date = m.group() # Body: div#zoomcon or div.zw content_parts = [] zoom = soup.find("div", id="zoomcon") if not zoom: zoom = soup.find("div", class_="zw") if not zoom: # Try UCAP content ucap = soup.find("UCAPCONTENT") if ucap: zoom = ucap if not zoom: return title, pub_date, "", "[]" for child in zoom.find_all(["p", "table", "img"], recursive=True): if child.name == "p": if child.find_parent("table"): continue text = child.get_text(strip=True) if text: content_parts.append(text) elif child.name == "table": md = html_table_to_html(child) if md: content_parts.append(md) elif child.name == "img": src = child.get("src", "").strip() alt = child.get("alt", "").strip() if src and not src.endswith(".gif") and "fileTypeImages" not in src: full_src = urljoin(BASE_URL, src) content_parts.append(f"![{alt}]({full_src})" if alt else f"![]({full_src})") content = "\n\n".join(content_parts) # Attachments attachments = [] for a_tag in zoom.find_all("a", class_="nfw-cms-attachment"): href = a_tag.get("href", "").strip() title_att = a_tag.get_text(strip=True) or href.split("/")[-1] if href: full_url = urljoin(BASE_URL, href) attachments.append({"url": full_url, "title": title_att}) for a_tag in zoom.find_all("a", href=re.compile(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|txt)$", re.I)): href = a_tag.get("href", "").strip() if any(a.get("href", "") == href for a in zoom.find_all("a", class_="nfw-cms-attachment")): continue title_att = a_tag.get_text(strip=True) or href.split("/")[-1] if href: full_url = urljoin(BASE_URL, href) attachments.append({"url": full_url, "title": title_att}) return title, pub_date, content, str(attachments) def html_table_to_html(table, base_url=""): """保留 HTML 表格结构，仅将相对链接/图片转绝对 URL""" from bs4 import BeautifulSoup tbl = BeautifulSoup(str(table), 'html.parser') for a in tbl.find_all('a'): href = a.get('href', '') if href and not href.startswith(('http', 'javascript', '#')): a['href'] = urllib.parse.urljoin(base_url, href) if base_url else href for img in tbl.find_all('img'): src = img.get('src', '') if src and not src.startswith(('http', '//', 'data:')): img['src'] = urllib.parse.urljoin(base_url, src) if base_url else src return str(tbl) def get_page_url(page): if page == 1: return LIST_URL return f"{BASE_URL}/zwgk/zwzl/zdlyxxgkzl/hjbhxxgk/jsxmhjyxpjxx/index_{page}.html" def main(): pages = LIST_PAGES if len(sys.argv) > 2 and sys.argv[1] == "--pages": pages = int(sys.argv[2]) total_new = 0 for page in range(1, pages + 1): url = get_page_url(page) print(f"[列表页] 第{page}页: {url}") soup = get_soup(url) if not soup: continue items = parse_list_items(soup) if not items: print(" 无数据，可能已到末页") break print(f" 找到 {len(items)} 条") for item_url, list_title, list_date in items: print(f" → {list_title[:50]}...") conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute("SELECT id FROM gov_raw WHERE page_url=? AND site_name=?", (item_url, SITE_NAME)) if c.fetchone(): conn.close() print(" 已存在，跳过") continue conn.close() time.sleep(SLEEP) detail_soup = get_soup(item_url) if not detail_soup: continue detail_title, pub_date, content, attachments = parse_detail(detail_soup, item_url) date = pub_date or list_date use_title = detail_title or list_title if not content: print(" 正文为空，跳过") continue conn = sqlite3.connect(DB_PATH) c = conn.cursor() try: summary = content[:200] has_table = 1 if "| ---" in content else 0 c.execute("""INSERT OR REPLACE INTO gov_raw (page_url, title, publish_date, site_name, content, attachments, summary, has_table) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", (item_url, use_title, date, SITE_NAME, content, attachments, summary, has_table)) conn.commit() print(f" 入库成功 (表格={has_table})") total_new += 1 except Exception as e: print(f" 入库失败: {e}") finally: conn.close() print(f"\n✅ {SITE_NAME} 爬取完成！共入库 {total_new} 条") if __name__ == "__main__": main()