#!/usr/bin/env python3 """寻乌县-公示公告爬虫 https://www.xunwu.gov.cn/xwxrmzf/c103707/list.shtml Lonsun CMS, 12条/页, AJAX分页 list_N.shtml 需Session+Referer绕过WAF """ import requests import sqlite3 import json import os import re import argparse from urllib.parse import urljoin from bs4 import BeautifulSoup BASE_URL = "https://www.xunwu.gov.cn" LIST_URL = "https://www.xunwu.gov.cn/xwxrmzf/c103707/list.shtml" DB_PATH = os.environ.get("DB_PATH", "/root/search.db") SITE_NAME = "寻乌县-公示公告" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", } def get_soup(url, session, referer=None): hdrs = dict(HEADERS) if referer: hdrs["Referer"] = referer resp = session.get(url, headers=hdrs, timeout=30) resp.encoding = "utf-8" return BeautifulSoup(resp.text, "html.parser") def parse_list_page(soup): """div.pageList > ul > li > a + span.time""" items = [] pl = soup.find("div", class_="pageList") if not pl: return items for li in pl.find_all("li"): a = li.find("a") if not a: continue href = a.get("href", "") if not href or href.startswith("#") or href.startswith("javascript"): continue full_url = urljoin(LIST_URL, href) # Skip external links (not xunwu.gov.cn) if "xunwu.gov.cn" not in full_url: continue title = a.get("title", "") or a.get_text(strip=True) span = li.find("span", class_="time") pub_date = span.get_text(strip=True) if span else "" items.append((full_url, title, pub_date)) return items def extract_detail(soup, url): title = "" pub_date = "" source = "" content = "" attachments = [] # Title - h1 h1 = soup.find("h1") if h1: title = h1.get_text(strip=True) # Date - meta PubDate meta_pd = soup.find("meta", attrs={"name": "PubDate"}) if meta_pd and meta_pd.get("content"): pub_date = meta_pd["content"].strip()[:10] # Source - span for span in soup.find_all("span"): txt = span.get_text(strip=True) m = re.search(r"来源[：:]\s*(.+?)(?:\s|$)", txt) if m: source = m.group(1).strip() # Content - div.article-content.article-content-body article = soup.select_one("div.article-content.article-content-body") if not article: article = soup.select_one("div.cont") if article: # Attachments for a_tag in article.find_all("a"): a_href = a_tag.get("href", "") if a_href.endswith((".pdf", ".doc", ".docx", ".xls", ".xlsx", ".zip", ".rar")): if not a_href.startswith("http"): a_href = urljoin(BASE_URL, a_href) attachments.append({ "name": a_tag.get_text(strip=True) or os.path.basename(a_href), "url": a_href, }) # Build content preserving tables and paragraphs text_parts = [] for child in article.children: if not hasattr(child, "name") or not child.name: continue if child.name in ("script", "style"): continue if child.name == "table": text_parts.append(str(child)) text_parts.append("") elif child.name in ("p", "div"): inner_table = child.find("table") if inner_table: text_parts.append(str(inner_table)) text_parts.append("") continue txt = child.get_text(strip=True) if txt: text_parts.append(txt) elif child.name in ("br",): text_parts.append("") content = "\n\n".join(text_parts) return title, pub_date, source, content, attachments def crawl(max_pages=None): sess = requests.Session() sess.headers.update(HEADERS) conn = sqlite3.connect(DB_PATH, timeout=60) cur = conn.cursor() # Determine pages to crawl pages_to_crawl = [] pages_to_crawl.append(LIST_URL) # page 1 if max_pages is None or max_pages > 1: limit = max_pages if max_pages else 99 for n in range(2, limit + 1): page_url = "https://www.xunwu.gov.cn/xwxrmzf/c103707/list_%d.shtml" % n pages_to_crawl.append(page_url) all_items = [] for page_url in pages_to_crawl: try: soup = get_soup(page_url, sess, referer=LIST_URL) items = parse_list_page(soup) if not items: print(f"[列表] {page_url} -> 0 条（可能已无更多页）") break print(f"[列表] {page_url} -> {len(items)} 条") all_items.extend(items) except Exception as e: print(f"[列表] {page_url} -> 异常: {e}") break print(f"\n列表共 {len(all_items)} 条，开始抓详情...") new_count = 0 error_count = 0 skip_count = 0 for idx, (page_url, list_title, list_date) in enumerate(all_items, 1): cur.execute( "SELECT id FROM gov_raw WHERE page_url=? AND site_name=?", (page_url, SITE_NAME), ) if cur.fetchone(): skip_count += 1 print(f" [{idx}/{len(all_items)}] 跳过: {list_title[:30]}...") continue print(f" [{idx}/{len(all_items)}] {list_title[:40]}...") try: soup = get_soup(page_url, sess, referer=LIST_URL) title, pub_date, source, content, attachments = extract_detail(soup, page_url) if not title: title = list_title if not pub_date: pub_date = list_date att_json = json.dumps(attachments, ensure_ascii=False) if attachments else "" cur.execute( """INSERT OR REPLACE INTO gov_raw (page_url, site_name, title, publish_date, source_url, content, attachments, script_name) VALUES (?, ?, ?, ?, ?, ?, ?, 'crawl_xunwu.py')""", (page_url, SITE_NAME, title, pub_date, page_url, content, att_json), ) conn.commit() new_count += 1 print(f" -> 新增 ({len(content)} chars)") except Exception as e: error_count += 1 print(f" -> 异常: {e}") conn.close() return new_count, skip_count, len(all_items), error_count if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--max-pages", type=int, default=None) args = parser.parse_args() new, skip, total, errors = crawl(args.max_pages) print(f"\n完成: 新增 {new}, 跳过 {skip}, 列表 {total}, 异常 {errors}")