#!/usr/bin/env python3 """ 孝义市-环境保护爬虫 http://www.xiaoyi.gov.cn/xxgk/fdzdgknr/zdlyxxgk/shgysyjs/hjbh_52409/ 共6页，前5页20条/页，第6页12条，总计112条 分页: index.shtml(第1页), index_2.shtml ... index_6.shtml """ import requests import sqlite3 import json import os import argparse import re from urllib.parse import urljoin from bs4 import BeautifulSoup BASE_URL = "http://www.xiaoyi.gov.cn" LIST_PATH = "/xxgk/fdzdgknr/zdlyxxgk/shgysyjs/hjbh_52409" TOTAL_PAGES = 6 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, timeout=20): resp = requests.get(url, headers=HEADERS, timeout=timeout) resp.encoding = "utf-8" return BeautifulSoup(resp.text, "html.parser") def parse_list_page(soup, page_url): """提取列表页: table.right_cont_table > tr > td > a[title] + span.pubtime""" items = [] table = soup.select_one("table.right_cont_table") if not table: return items for tr in table.find_all("tr"): a = tr.find("a") if not a: continue href = a.get("href", "") if href.startswith("javascript") or not href: continue full_url = urljoin(page_url, href) title = a.get("title", "") or a.get_text(strip=True) span = tr.find("span", class_="pubtime") pub_date = span.get_text(strip=True) if span else "" items.append((full_url, title, pub_date)) return items def extract_date_from_info(info_text): """从 '发布时间：2026-05-06来源：XX' 提取日期""" m = re.search(r"发布时间：(\d{4}-\d{2}-\d{2})", info_text) if m: return m.group(1) return "" def extract_source_from_info(info_text): """从 '发布时间：2026-05-06来源：XX' 提取来源""" m = re.search(r"来源[：:]\s*(.*?)$", info_text) if m: src = m.group(1).strip() return src return "" def extract_detail(soup, url): """提取详情页""" title = "" pub_date = "" source = "" content = "" attachments = [] # Title - from before "_" title_tag = soup.find("title") if title_tag: raw = title_tag.get_text(strip=True) if "_" in raw: title = raw.split("_")[0].strip() else: title = raw # PubDate + Source from div.article-info info = soup.select_one("div.article-info") if info: info_text = info.get_text(strip=True) pd = extract_date_from_info(info_text) if pd: pub_date = pd src = extract_source_from_info(info_text) if src: source = src # Content - div.article-content，优先取内部的 TRS_UEDITOR 容器 article = soup.select_one("div.article-content") if article: # 实际正文在 div.TRS_UEDITOR 或 div.view 内 real_content = article.select_one("div.TRS_UEDITOR, div.view") if not real_content: real_content = article # 提取附件（从整个 article-content 中找） 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, }) # 提取正文，保留 table 原始 HTML，同时保持段落分节 text_parts = [] for child in real_content.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", "span"): 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 == "br": text_parts.append("") content = "\n\n".join(text_parts) return title, pub_date, source, content, attachments def crawl(max_pages=None): conn = sqlite3.connect(DB_PATH, timeout=60) cur = conn.cursor() all_items = [] pages_to_crawl = TOTAL_PAGES if max_pages and max_pages < pages_to_crawl: pages_to_crawl = max_pages for page in range(1, pages_to_crawl + 1): if page == 1: url = f"{BASE_URL}{LIST_PATH}/index.shtml" else: url = f"{BASE_URL}{LIST_PATH}/index_{page}.shtml" print(f"[列表] 第{page}/{pages_to_crawl}页: {url}") try: soup = get_soup(url) items = parse_list_page(soup, url) if not items: print(" -> 空页，停止") break print(f" -> {len(items)} 条") all_items.extend(items) except Exception as e: print(f" -> 失败: {e}") break print(f"\n列表共 {len(all_items)} 条，开始抓详情...") new_count = 0 error_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(): 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) 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_xiaoyi.py')""", (page_url, SITE_NAME, title, pub_date, page_url, content, att_json), ) conn.commit() new_count += 1 print(f" -> 新增") except Exception as e: error_count += 1 print(f" -> 异常: {e}") conn.close() return new_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, total, errors = crawl(args.max_pages) print(f"\n完成: 新增 {new}, 列表 {total}, 异常 {errors}")