([^<]+)', html) if m: title = re.sub(r'[-_—|].*', '', m.group(1)).strip() content = "" # Extract content from ls-article-info container from bs4 import BeautifulSoup soup = BeautifulSoup(html, 'html.parser') info_div = soup.find('div', class_='ls-article-info') if info_div: raw = str(info_div) raw = re.sub(r'<script[^>]*>.*?</script>', '', raw, flags=re.DOTALL|re.I) raw = re.sub(r'<style[^>]*>.*?</style>', '', raw, flags=re.DOTALL|re.I) raw = re.sub(r' style="[^"]*"', '', raw) content = raw.strip() if not content: # Fallback: look for government document content m = re.search(r'(<p[^>]*class="MsoNormal"[^>]*>.*?</p>)', html, re.DOTALL) if m: content = m.group(1) # date date = "" m = re.search(r'PubDate["\'][^>]*content=["\']([^"\']+)', html, re.I) if m: date = m.group(1)[:10] return title, content, date def run(args=None): incremental = False if args and '1' in args: incremental = True print(f"爬取: {SITE_NAME}") html = fetch(LIST_URL) if not html: print(" ❌ 无法获取列表页") return items = parse_list(html) print(f" 列表: {len(items)} 条") results = [] for i, (title, url, date) in enumerate(items): if not url.startswith('http'): url = BASE_URL + url print(f" [{i+1}/{len(items)}] {title[:40]}...", end=" ", flush=True) dt, content, d2 = fetch_detail(url) if not content or len(content) < 30: print("内容过短") continue results.append({ "site_name": SITE_NAME, "title": dt or title, "url": url, "content": content, "pub_date": d2 or date, "category": "环评公示", "tags": "新乡", }) print(f"✅ ({len(content)}字)") time.sleep(0.3) if results: print(f"\n 入库 {len(results)} 条") push_to_searchdb(results, "xx_sthj") print("完成") if __name__ == "__main__": run(sys.argv[1] if len(sys.argv)>1 else None)

#!/usr/bin/env python3 """ 新乡市生态环境局 - 环评公示 (sthjj.xinxiang.gov.cn) 列表: /zwgk/public/column/6638717?type=4&catId=6720855&action=list 详情: /zwgk/public/6638717/{contentId}.html 正文: div.content CMS: 政府信息公开平台，静态列表+JS渲染 """ import sys, os, re, time from datetime import datetime, timezone, timedelta import requests, urllib3 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) SITE_NAME = "新乡市生态环境局" BASE_URL = "https://sthjj.xinxiang.gov.cn" LIST_URL = "https://sthjj.xinxiang.gov.cn/zwgk/public/column/6638717?type=4&catId=6720855&action=list" MAX_PAGES = 5 CUTOFF = (datetime.now(timezone.utc) - timedelta(days=365*3)).strftime("%Y-%m-%d") HEADERS = {"User-Agent": "Mozilla/5.0"} def fetch(url): try: r = requests.get(url, headers=HEADERS, timeout=20, verify=False) r.encoding = 'utf-8' return r.text except: return None def parse_list(html): """解析列表： + 附近 date""" items = [] # 政府信息公开平台列表：... 附近有日期span for m in re.finditer(r']*href="((?:https?:)?//sthjj\.xinxiang\.gov\.cn[^"]*)"[^>]*title="([^"]*)"', html): href = m.group(1) title = m.group(2).strip() if not title or not href: continue # 找日期：在href前后找或 YYYY-MM-DD span_dates = re.findall(r']*class="[^"]*date[^"]*"[^>]*>(\d{4}-\d{2}-\d{2})', html) # 简单的：按出现顺序配对，先暂时用页面中的日期 items.append((title, href, span_dates[len(items)] if len(span_dates) > len(items) else "")) return items def get_total_pages(html): m = re.search(r'page_(\d+)\.html[^>]*>末页', html) if m: return int(m.group(1)) pages = re.findall(r'page_(\d+)', html) if pages: return max(int(p) for p in pages) return 1 def fetch_detail(url): html = fetch(url) if not html: return None, None, None title = "" m = re.search(r'([^<]+)', html) if m: title = re.sub(r'[-_—|].*', '', m.group(1)).strip() content = "" # Extract content from ls-article-info container from bs4 import BeautifulSoup soup = BeautifulSoup(html, 'html.parser') info_div = soup.find('div', class_='ls-article-info') if info_div: raw = str(info_div) raw = re.sub(r'<script[^>]*>.*?</script>', '', raw, flags=re.DOTALL|re.I) raw = re.sub(r'<style[^>]*>.*?</style>', '', raw, flags=re.DOTALL|re.I) raw = re.sub(r' style="[^"]*"', '', raw) content = raw.strip() if not content: # Fallback: look for government document content m = re.search(r'(<p[^>]*class="MsoNormal"[^>]*>.*?</p>)', html, re.DOTALL) if m: content = m.group(1) # date date = "" m = re.search(r'PubDate["\'][^>]*content=["\']([^"\']+)', html, re.I) if m: date = m.group(1)[:10] return title, content, date def run(args=None): incremental = False if args and '1' in args: incremental = True print(f"爬取: {SITE_NAME}") html = fetch(LIST_URL) if not html: print(" ❌ 无法获取列表页") return items = parse_list(html) print(f" 列表: {len(items)} 条") results = [] for i, (title, url, date) in enumerate(items): if not url.startswith('http'): url = BASE_URL + url print(f" [{i+1}/{len(items)}] {title[:40]}...", end=" ", flush=True) dt, content, d2 = fetch_detail(url) if not content or len(content) < 30: print("内容过短") continue results.append({ "site_name": SITE_NAME, "title": dt or title, "url": url, "content": content, "pub_date": d2 or date, "category": "环评公示", "tags": "新乡", }) print(f"✅ ({len(content)}字)") time.sleep(0.3) if results: print(f"\n 入库 {len(results)} 条") push_to_searchdb(results, "xx_sthj") print("完成") if __name__ == "__main__": run(sys.argv[1] if len(sys.argv)>1 else None)