#!/usr/bin/env python3 import os """ 鹤壁市生态环境局 - 环境影响评价 (sthjj.hebi.gov.cn) 列表: /zfxxgk/fdzdgknr/qtzdgkxx/sdgjz/hjyxpj/index.html (动态加载，需Playwright) 详情: /zfxxgk/.../hjyxpj/art/202X/art_XXXX.html (静态HTML) 正文: div.main.content > h1.title + 正文段落 CMS: JPaaS (动态列表) """ import sys, os, re, time from datetime import datetime, timezone, timedelta import requests, urllib3 from bs4 import BeautifulSoup from playwright.sync_api import sync_playwright 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.hebi.gov.cn" LIST_URL = "https://sthjj.hebi.gov.cn/zfxxgk/fdzdgknr/qtzdgkxx/sdgjz/hjyxpj/index.html" HEADERS = {"User-Agent": "Mozilla/5.0"} MAX_PAGES = 5 CUTOFF = (datetime.now(timezone.utc) - timedelta(days=365*3)).strftime("%Y-%m-%d") 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 extract_list_items(page): """用 Playwright 获取当前页列表项""" items = [] links = page.query_selector_all('a[href*="art_"]') for link in links: href = link.get_attribute('href') title = link.get_attribute('title') or link.inner_text().strip() if href and title: if not href.startswith('http'): href = BASE_URL + href items.append((title, href)) return items def parse_items_from_page(page_html): """从页面上已渲染的HTML中提取列表项""" items = [] soup = BeautifulSoup(page_html, 'html.parser') for a in soup.select('a[href*="art_"]'): href = a.get('href', '') title = a.get('title', '').strip() or a.get_text(strip=True) if href and title: if not href.startswith('http'): href = BASE_URL + href items.append((title, href)) return items def fetch_detail(url): """提取详情页正文""" html = fetch(url) if not html: return None, None, None soup = BeautifulSoup(html, 'html.parser') # 标题 title = "" h1 = soup.find('h1', class_='title') if h1: title = h1.get_text(strip=True) # 主体内容 content = "" main_div = soup.find('div', class_='main', attrs={'ergodic': 'article'}) if not main_div: # 尝试其他选择器 main_div = soup.find('div', class_='content') if main_div: # 清理 script/style for tag in main_div(['script', 'style']): tag.decompose() # 清理 style 属性 for tag in main_div.find_all(style=True): del tag['style'] content = str(main_div).strip() else: # 回退：找正文区域 article = soup.find('div', class_='article-content') or soup.find(id='zoom') if article: for tag in article(['script', 'style']): tag.decompose() for tag in article.find_all(style=True): del tag['style'] content = str(article).strip() # 日期 date = "" sub = soup.find('div', class_='sub-title') if sub: m = re.search(r'(\d{4})年(\d{1,2})月(\d{1,2})日', sub.get_text()) if m: date = f"{m.group(1)}-{m.group(2).zfill(2)}-{m.group(3).zfill(2)}" if not date: m = re.search(r'PubDate["\'][^>]*content=["\']([^"\']+)', html, re.I) if m: date = m.group(1)[:10] if not date: m = re.search(r'发布时间[：:]\s*(\d{4}[-/年]\d{1,2}[-/月]\d{1,2})', html) if m: date = m.group(1).replace('年', '-').replace('月', '-').replace('/', '-') return title, content, date def run(args=None): incremental = False if args and '1' in args: incremental = True print(f"爬取: {SITE_NAME}") all_items = [] with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context( user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' ) page = context.new_page() for pg in range(1, MAX_PAGES + 1): if pg == 1: page.goto(LIST_URL, wait_until='domcontentloaded', timeout=120000) page.wait_for_timeout(3000) # 等待动态列表渲染 else: # 点击下一页 next_link = page.query_selector(f'a:has-text("{pg}")') if not next_link: print(f" 第{pg}页: 找不到链接") break next_link.click() page.wait_for_timeout(2000) page.wait_for_timeout(1000) # 等待列表渲染 html = page.content() items = parse_items_from_page(html) # 过滤掉非目标URL的链接 items = [(t, h) for t, h in items if '/hjyxpj/art/' in h] # 去重 seen = set() unique_items = [] for t, h in items: if h not in seen: seen.add(h) unique_items.append((t, h)) items = unique_items if not items: print(f" 第{pg}页: 0 条") if pg > 1: break continue print(f" 第{pg}页: {len(items)} 条") all_items.extend(items) if incremental: break browser.close() print(f" 共 {len(all_items)} 条列表项") # 爬取详情 results = [] for i, (title, url) in enumerate(all_items): print(f" [{i+1}/{len(all_items)}] {title[:40]}...", end=" ", flush=True) dt, content, date = fetch_detail(url) if not content or len(content) < 50: print("⚠ 无正文") continue if date and date < CUTOFF: print("⏭ 超时范围") continue summary = re.sub(r'<[^>]+>', '', content)[:200].strip() results.append({ "site_name": SITE_NAME, "title": dt or title, "url": url, "content": content, "summary": summary, "pub_date": date or "", "category": "环评公示", "tags": "鹤壁", }) print(f"✅ ({len(content)}字)") time.sleep(0.3) if results: print(f"\n 入库 {len(results)} 条") # 判断是否在服务器上（直接入库 vs 远程推送） if os.path.exists('/root/search.db'): # 直接本地入库 import sqlite3 conn = sqlite3.connect(os.getenv("SEARCH_DB", "/root/search.db")) cur = conn.cursor() new_count = 0 for r in results: r['_type'] = 'gov' cur.execute(""" INSERT OR IGNORE INTO gov_raw (site_name, source_url, page_url, title, publish_date, summary, content, category, tags) VALUES (?,?,?,?,?,?,?,?,?) """, ( r['site_name'], r.get('url',''), r.get('url',''), r['title'], r.get('pub_date',''), r.get('summary',''), r['content'], r.get('category',''), r.get('tags','') )) if cur.lastrowid: new_count += 1 rowid = cur.lastrowid cur.execute(""" INSERT OR IGNORE INTO gov_search (rowid, title, site_name, summary) VALUES (?,?,?,?) """, (rowid, r['title'], r['site_name'], r.get('summary',''))) conn.commit() conn.close() print(f" ✅ 本地入库 {new_count} 条 (FTS已同步)") else: push_to_searchdb(results, "hebi_hjyxpj") print("完成") if __name__ == "__main__": run(sys.argv[1] if len(sys.argv) > 1 else None)