#!/usr/bin/env python3 """cnzhlx.com - 环评公示 (第三方环评公示平台) 列表: /hpgs/index.shtml -> /hpgs/index/p/N.shtml (23页) 详情: /hpgs/show/id/ID.shtml -> div.m-con (.news-dy) """ import sys, os, re, json, time from datetime import datetime, timedelta import requests from bs4 import BeautifulSoup BASE = "http://cnzhlx.com" SITE = "cnzhlx.com-环评公示" COLUMN = "环评公示" PROVINCE = "企业" TOTAL_PAGES = 23 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" def log(msg): sys.stderr.write(msg + "\n") sys.stderr.flush() def fetch_list(page): if page == 1: url = f"{BASE}/hpgs/index.shtml" else: url = f"{BASE}/hpgs/index/p/{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 li in soup.select("ul li"): time_div = li.select_one("div.time") txt_div = li.select_one("div.txt") if not time_div or not txt_div: continue a = txt_div.find("a", href=True) if not a: continue href = a["href"] if not href.startswith("http"): href = BASE + href title = a.get_text(strip=True) # Extract date day_el = time_div.find("h1") ym_el = time_div.find("span") date_str = "" if day_el and ym_el: day = day_el.get_text(strip=True) ym = ym_el.get_text(strip=True) if day and ym: date_str = f"{ym}-{day.zfill(2)}" if not title or not href: continue items.append({"title": title, "url": href, "date": date_str, "summary": txt_div.find("p").get_text(strip=True) if txt_div.find("p") else ""}) return items def extract_content(detail_url, title): try: r = session.get(detail_url, timeout=20) r.encoding = "utf-8" except Exception as e: return f"[{title}]({detail_url})", title, "" soup = BeautifulSoup(r.text, "html.parser") # Title from h1 h1 = soup.select_one("div.news-dy div.title3 h1") if h1: title = h1.get_text(strip=True) content_el = soup.select_one("div.m-con") if not content_el: return f"[{title}]({detail_url})", title, "" # Save attachment links attach_links = [] for a_tag in content_el.find_all("a"): href = a_tag.get("href", "") if href and not href.startswith("#") and not href.startswith("javascript"): if href.startswith("/"): href = BASE + href txt = a_tag.get_text(strip=True) if txt and href: attach_links.append(f"[{txt}]({href})") # Save tables tables_html = [] for table in content_el.find_all("table"): tables_html.append(str(table)) table.decompose() # 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) # Unwrap inline for tag in content_el.find_all(["span", "b", "strong", "font", "em", "i", "u", "s"]): tag.unwrap() 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}" for link in attach_links: if link not in text: text += f"\n\n{link}" if not text.strip() or len(text.strip()) < 20: text = f"[{title}]({detail_url})" if attach_links: text += "\n\n" + "\n".join(attach_links) return text, title, "" def crawl_all(months_back=36): cutoff = datetime.now() - timedelta(days=months_back * 30) if months_back else None all_items = [] seen_urls = set() for page in range(1, TOTAL_PAGES + 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" [PAGE {page}/{TOTAL_PAGES}] {len(items)} items, +{new_count} new") if new_count == 0 and page > 1: break time.sleep(1) log(f"\n共 {len(all_items)} 条待爬详情") for idx, item in enumerate(all_items, 1): content, title, _ = extract_content(item["url"], item["title"]) if title: item["title"] = title item["content"] = content 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(): html = fetch_list(1) if not html: return items = parse_list(html) log(f"[LIST] page 1 -> {len(items)} 条") for idx, item in enumerate(items, 1): content, title, _ = extract_content(item["url"], item["title"]) if title: item["title"] = title item["content"] = content 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", ""), "site_name": SITE, "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": months = int(sys.argv[2]) if len(sys.argv) > 2 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]]")