#!/usr/bin/env python3 """ 都昌县人民政府 - 环评信息 爬虫脚本 抓取 https://www.duchang.gov.cn/zwgk/zfxxgkzl/bmxxgk/sthjj/zdly/03/ 所有文章 """ import os import sys import re import sqlite3 import requests from bs4 import BeautifulSoup BASE_LIST_URL = "https://www.duchang.gov.cn/zwgk/zfxxgkzl/bmxxgk/sthjj/zdly/03/" SITE_NAME = "都昌县-环评信息" PER_PAGE = 15 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" } def get_db_path(): """Get database path from SEARCH_DB env var or use default.""" return os.environ.get("SEARCH_DB", "/root/search.db") def init_db(db_path): """Ensure the table exists.""" conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS gov_raw ( id INTEGER PRIMARY KEY AUTOINCREMENT, site_name TEXT NOT NULL, source_url TEXT NOT NULL, page_url TEXT NOT NULL UNIQUE, title TEXT, publish_date TEXT, content TEXT ) """) conn.commit() return conn def parse_list_page(url): """ Parse a list page and extract (title, page_url, date) tuples. Returns a list of dicts, or empty list on failure. """ try: r = requests.get(url, headers=HEADERS, timeout=30) r.encoding = "utf-8" except Exception as e: print(f" [ERROR] Failed to fetch list page {url}: {e}", file=sys.stderr) return [] if r.status_code != 200: print(f" [ERROR] HTTP {r.status_code} for {url}", file=sys.stderr) return [] soup = BeautifulSoup(r.text, "html.parser") rows = soup.find_all("tr") items = [] for tr in rows: tds = tr.find_all("td") if len(tds) < 3: continue # td[0] = serial number, td[1] = title + link, td[2] = date a_tag = tds[1].find("a") if not a_tag: continue href = a_tag.get("href", "").strip() title = a_tag.get_text(strip=True) date_str = tds[2].get_text(strip=True) if not href or not title: continue items.append({ "title": title, "page_url": href, "publish_date": date_str, }) return items def get_page_count(): """ Fetch first page to determine total page count. Parses the JS createPage() call embedded in the page. Returns total page count (int). """ try: r = requests.get(BASE_LIST_URL, headers=HEADERS, timeout=30) r.encoding = "utf-8" except Exception as e: print(f"[ERROR] Failed to fetch list page: {e}", file=sys.stderr) return 1 if r.status_code != 200: print(f"[ERROR] HTTP {r.status_code} for list page", file=sys.stderr) return 1 soup = BeautifulSoup(r.text, "html.parser") scripts = soup.find_all("script") for script in scripts: if script.string and "createPage" in script.string: # Pattern: createPage(16, 0, "index", "html"); m = re.search(r"createPage\s*\(\s*(\d+)", script.string) if m: total_pages = int(m.group(1)) print(f" Total pages detected: {total_pages}") return total_pages # Fallback: count from the text in page body_text = soup.get_text() m = re.search(r"共\s*(\d+)\s*条", body_text) if m: total_records = int(m.group(1)) total_pages = (total_records + PER_PAGE - 1) // PER_PAGE print(f" Total records: {total_records}, pages: {total_pages}") return total_pages print(" [WARN] Could not detect page count, defaulting to 1", file=sys.stderr) return 1 def build_page_url(page_index): """Build URL for a given page index (0-based).""" if page_index == 0: return BASE_LIST_URL return BASE_LIST_URL + f"index_{page_index}.html" def parse_detail_page(page_url): """ Parse a detail page and extract title, publish_date, content. Returns dict or None on failure. """ try: r = requests.get(page_url, headers=HEADERS, timeout=30) r.encoding = "utf-8" except Exception as e: print(f" [ERROR] Failed to fetch detail {page_url}: {e}", file=sys.stderr) return None if r.status_code != 200: print(f" [ERROR] HTTP {r.status_code} for {page_url}", file=sys.stderr) return None soup = BeautifulSoup(r.text, "html.parser") # ---- Title ---- title = None title_div = soup.find("div", id="NewsArticleTitle") if title_div: title = title_div.get_text(strip=True) if not title: # Fallback: content-bg has title before "日期：" content_bg = soup.select_one("div.content-bg") if content_bg: txt = content_bg.get_text(strip=True) if "日期" in txt: title = txt[:txt.index("日期")].strip() if not title: title_tag = soup.find("title") if title_tag: title = title_tag.get_text(strip=True).split(",")[0].strip() # ---- Date ---- publish_date = None date_div = soup.find("div", id="NewsArticlePubDay") if date_div: publish_date = date_div.get_text(strip=True) if not publish_date: # Fallback: content-bg has "日期：YYYY-MM-DD" content_bg = soup.select_one("div.content-bg") if content_bg: m = re.search(r"日期[：:]\s*(\d{4}-\d{2}-\d{2})", content_bg.get_text()) if m: publish_date = m.group(1) # ---- Content ---- content = None for selector in [ "div.content.qjcontent", "div.view.TRS_UEDITOR", "div.trs_editor_view.TRS_UEDITOR", ]: content_div = soup.select_one(selector) if content_div: content = content_div.get_text(strip=True) break if not content: content_bg = soup.select_one("div.content-bg") if content_bg: # Remove title and date prefix from content-bg to get pure content txt = content_bg.get_text(strip=True) # Find "分享：" which separates the header from the content if "分享：" in txt: content = txt[txt.index("分享：") + 3:].strip() elif "来源：" in txt: idx = txt.index("来源：") # Find the end of the source/date line after_source = txt[idx:] if "】" in after_source: content = txt[txt.index("】") + 1:].strip() result = { "title": title or "", "publish_date": publish_date or "", "content": content or "", } return result def crawl(): """Main crawl function.""" db_path = get_db_path() print(f"Database: {db_path}") conn = init_db(db_path) cursor = conn.cursor() total_pages = get_page_count() print(f"Total pages to crawl: {total_pages}") total_items = 0 new_items = 0 for page_idx in range(total_pages): list_url = build_page_url(page_idx) print(f"\n--- Page {page_idx + 1}/{total_pages}: {list_url}") items = parse_list_page(list_url) print(f" Found {len(items)} items on page") for item in items: total_items += 1 page_url = item["page_url"] title = item["title"] date_str = item["publish_date"] print(f" [{total_items}] {title[:50]}...") # Fetch detail page detail = parse_detail_page(page_url) if detail is None: print(f" [SKIP] Failed to parse detail") continue # Use detail page title/date if available (more accurate) final_title = detail["title"] or title final_date = detail["publish_date"] or date_str content = detail["content"] try: cursor.execute( """INSERT OR IGNORE INTO gov_raw (site_name, source_url, page_url, title, publish_date, content) VALUES (?, ?, ?, ?, ?, ?)""", (SITE_NAME, BASE_LIST_URL, page_url, final_title, final_date, content) ) if cursor.rowcount > 0: new_items += 1 print(f" [INSERTED] {final_date} | {final_title[:50]}") else: print(f" [SKIP] already exists") conn.commit() except sqlite3.Error as e: print(f" [DB ERROR] {e}", file=sys.stderr) conn.rollback() conn.close() print(f"\n=== Done! Total: {total_items}, New: {new_items} ===") if __name__ == "__main__": crawl()