#!/usr/bin/env python3 """随州市生态环境局-环境影响评价 (sthjj.suizhou.gov.cn) 列表: /fbjd_15/zwgk/xxgkml/ggjgxx/hjbh/cpxk/hjyxpj/index.shtml -> index_N.shtml 详情: /fbjd_15/zwgk/xxgkml/ggjgxx/hjbh/cpxk/hjyxpj/YYYYMM/t...shtml -> div.doc_content.news-detl-c 策略: full=前5页(新站初装), full all=全量, incremental=第1页(日跑) """ import sys, os, re, json, time from datetime import datetime, timedelta import requests from bs4 import BeautifulSoup from urllib.parse import urljoin BASE_URL = "http://sthjj.suizhou.gov.cn" LIST_PATH = "/fbjd_15/zwgk/xxgkml/ggjgxx/hjbh/cpxk/hjyxpj" SITE = "随州市生态环境局" COLUMN = "环境影响评价" PROVINCE = "湖北" TOTAL_PAGES = 34 DEFAULT_FULL_PAGES = 5 HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} session = requests.Session() session.headers.update(HEADERS) MARKER = "\x00P\x00" # Attachment file extensions ATTACH_EXTS = ('.doc', '.docx', '.pdf', '.xls', '.xlsx', '.ppt', '.pptx', '.zip', '.rar', '.7z', '.tar', '.gz', '.txt', 'downfile.jsp', 'download.jsp', 'downfile', 'download') def log(msg): sys.stderr.write(msg + "\n") sys.stderr.flush() def _is_attachment(href): if not href: return False low = href.lower() return any(ext in low for ext in ATTACH_EXTS) def fetch_list(page): if page == 1: url = f"{BASE_URL}{LIST_PATH}/index.shtml" else: url = f"{BASE_URL}{LIST_PATH}/index_{page}.shtml" try: r = session.get(url, timeout=20) r.encoding = "utf-8" return r.text except Exception as e: log(f" [ERROR] list page {page}: {e}") return None def parse_list(html): soup = BeautifulSoup(html, "html.parser") items = [] for ul in soup.select("div.docs div.documents ul"): for li in ul.find_all("li"): a = li.find("a", href=True) i = li.find("i") if a and i: title = a.get("title", "") or a.get_text(strip=True) href = a["href"] if not href.startswith("http"): href = BASE_URL + href date_raw = i.get_text(strip=True) date_str = "" m = re.match(r"(\d{2})-(\d{2})-(\d{2})", date_raw) if m: date_str = f"20{m.group(1)}-{m.group(2)}-{m.group(3)}" if title and href: items.append({"title": title, "url": href, "date": date_str}) return items def extract_content(detail_url, title): try: r = session.get(detail_url, timeout=20) r.encoding = "utf-8" except Exception as e: log(f" [ERROR] detail: {detail_url} - {e}") return f"[{title}]({detail_url})", title, "", [] soup = BeautifulSoup(r.text, "html.parser") meta_title = soup.select_one("meta[name=ArticleTitle]") meta_date = soup.select_one("meta[name=PubDate]") if meta_title and meta_title.get("content"): title = meta_title["content"].strip() pub_date = "" if meta_date and meta_date.get("content"): pub_date = meta_date["content"].strip()[:10] content_el = soup.select_one("div.doc_content.news-detl-c") if not content_el: return f"[{title}]({detail_url})", title, pub_date, [] attachments = [] # 0) Extract attachment links to markdown for a_tag in content_el.find_all("a", href=True): href = a_tag["href"] text = a_tag.get_text(strip=True) if not href or href.startswith("#") or href.startswith("javascript"): continue full_url = urljoin(BASE_URL, href) if _is_attachment(href): md_link = f"[{text}]({full_url})" if text else f"[附件]({full_url})" a_tag.replace_with(md_link) attachments.append({"name": text or "附件", "url": full_url}) elif text: # External link - keep as markdown md_link = f"[{text}]({full_url})" a_tag.replace_with(md_link) # 1) Save tables tables_html = [] for table in content_el.find_all("table"): tables_html.append(str(table)) table.decompose() # 2) Paragraph markers for tag in content_el.find_all(["p", "h1", "h2", "h3", "h4", "h5", "h6"]): tag.insert(0, MARKER) tag.append(MARKER) for br in content_el.find_all("br"): br.replace_with(MARKER) # 3) Unwrap inline tags for tag in content_el.find_all(["span", "b", "strong", "font", "em", "i", "u", "s"]): tag.unwrap() # 4) Extract text text = content_el.get_text(separator="", strip=True) text = re.sub(r"\x00P\x00(\s*\x00P\x00)+", "\x00P\x00", text) text = text.replace("\x00P\x00", "\n\n") text = re.sub(r"\n{3,}", "\n\n", text) text = text.strip() for tbl_html in tables_html: text += f"\n\n{tbl_html}" if not text.strip() or len(text.strip()) < 20: text = f"[{title}]({detail_url})" return text, title, pub_date, attachments def crawl_pages(start_page, end_page, cutoff, label): """Crawl pages from start_page to end_page (inclusive).""" all_items = [] seen_urls = set() for page in range(start_page, end_page + 1): html = fetch_list(page) if not html: break items = parse_list(html) if not items: break new_count = 0 for item in items: if item["url"] not in seen_urls: if cutoff and item["date"]: try: d = datetime.strptime(item["date"], "%Y-%m-%d") if d < cutoff: continue except ValueError: pass seen_urls.add(item["url"]) all_items.append(item) new_count += 1 log(f" [{label} PAGE {page}/{end_page}] {len(items)} items, +{new_count} new") if new_count == 0 and page > 1: log(" [STOP] no new items") break time.sleep(1) return all_items def crawl_all(months_back=36): """Full crawl: first DEFAULT_FULL_PAGES pages for new site initial import.""" page_count = DEFAULT_FULL_PAGES cutoff = datetime.now() - timedelta(days=months_back * 30) if months_back else None log(f"策略: 前{page_count}页初始入库 (全量请用 'full all')") all_items = crawl_pages(1, page_count, cutoff, f"FULL 1/{page_count}") _process_details(all_items) def crawl_all_pages(months_back=36): """Crawl ALL 34 pages (full rebuilt).""" cutoff = datetime.now() - timedelta(days=months_back * 30) if months_back else None log("全量模式: 爬取所有页面...") all_items = crawl_pages(1, TOTAL_PAGES, cutoff, "FULL ALL") _process_details(all_items) def _process_details(all_items): log(f"\n共 {len(all_items)} 条待爬详情") for idx, item in enumerate(all_items, 1): content, title, pub_date, attachments = extract_content(item["url"], item["title"]) if pub_date: item["date"] = pub_date if title: item["title"] = title item["content"] = content item["attachments"] = attachments output_item(item, idx) if idx % 10 == 0: log(f" [PROGRESS] {idx}/{len(all_items)}") log(f"\n[DONE] 共爬取 {len(all_items)} 条") def crawl_incremental(): """Incremental: only first page (daily).""" html = fetch_list(1) if not html: return items = parse_list(html) log(f"[INCREMENTAL] page 1 -> {len(items)} 条") for idx, item in enumerate(items, 1): content, title, pub_date, attachments = extract_content(item["url"], item["title"]) if pub_date: item["date"] = pub_date if title: item["title"] = title item["content"] = content item["attachments"] = attachments output_item(item, idx) log(f"[DONE] 增量爬取 {len(items)} 条") def output_item(item, idx): record = { "title": item.get("title", ""), "page_url": item.get("url", ""), "publish_date": item.get("date", ""), "content": item.get("content", ""), "attachments": item.get("attachments", []), "site_name": f"{SITE}-{COLUMN}", "column": COLUMN, "province": PROVINCE, } print(json.dumps(record, ensure_ascii=False)) if __name__ == "__main__": mode = sys.argv[1] if len(sys.argv) > 1 else "incremental" if mode == "incremental": crawl_incremental() elif mode == "full": arg2 = sys.argv[2] if len(sys.argv) > 2 else "" if arg2 == "all": crawl_all_pages() else: months = int(arg2) if arg2 and arg2.isdigit() else 36 crawl_all(months) elif mode == "list": html = fetch_list(1) items = parse_list(html) log(f"共 {len(items)} 条") for it in items[:5]: log(f" {it['date']} | {it['title'][:50]} | {it['url']}") else: log(f"Usage: {sys.argv[0]} [incremental|full [months|all]|list]")