#!/usr/bin/env python3 """ crawl_hf_gov.py -- 合肥市人民政府-环评公示 Site: https://www.hefei.gov.cn/public/column/19081?type=4&catId=6999451&action=list CMS: Custom, with CloudFlare JS challenge """ import sys, re, time, os, json, subprocess from bs4 import BeautifulSoup from datetime import datetime, timedelta DB_PATH = os.getenv("SEARCH_DB", "/root/search.db") SITE_NAME = '合肥市人民政府-环评公示' DOMAIN = "www.hefei.gov.cn" BASE_URL = "https://www.hefei.gov.cn" LIST_URL = "/public/column/19081?type=4&catId=6999451&action=list" JSL_SCRIPT = "/tmp/solve_hf_jsl.js" NODE_BIN = "node" THREE_YEARS_AGO = (datetime.now() - timedelta(days=3*365)).strftime("%Y-%m-%d") stats = {"new": 0, "skip": 0, "errors": 0} def solve_jsl(): """Run Node.js to solve JS challenge, returns cookies dict""" try: result = subprocess.run([NODE_BIN, JSL_SCRIPT], capture_output=True, text=True, timeout=30) out = result.stdout.strip() data = json.loads(out) if 'error' in data: print(f" [ERROR] JSL solve: {data['error']}") return None return { '__jsluid_s': data['jsluid'], '__jsl_clearance_s': data['clearance'] } except Exception as e: print(f" [ERROR] JSL: {e}") return None def solve_jsl_page(page_url): """Solve JS challenge for a specific page URL""" try: result = subprocess.run([NODE_BIN, JSL_SCRIPT, page_url], capture_output=True, text=True, timeout=30) out = result.stdout.strip() data = json.loads(out) if 'error' in data: return None return { '__jsluid_s': data['jsluid'], '__jsl_clearance_s': data['clearance'] } except Exception as e: return None def fetch_with_cookies(path, cookies=None): """Fetch a URL with optional cookies""" import requests as req 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', } try: r = req.get(BASE_URL + path, headers=headers, cookies=cookies or {}, timeout=30, verify=False) r.encoding = 'utf-8' return r except Exception as e: return None def get_list_page(page=1): """Get list page. catId=6999451 for 环评, type=4 for list""" if page == 1: url = LIST_URL else: url = f"/public/column/19081?type=4&catId=6999451&action=list&page={page}" # First solve JS challenge cookies = solve_jsl() if not cookies: return [] r = fetch_with_cookies(url, cookies) if not r or r.status_code != 200: return [] soup = BeautifulSoup(r.text, 'html.parser') # Try different list patterns for Hefei government CMS items = [] # Skip generic/navigation items (not real announcements) SKIP_TITLES = {'政府机构网站','首页','时事新闻','组织机构','机构设置','联系我们','网站地图','领导信息','机构职能'} def is_valid_item(href, title, date_str): """Filter out navigation/placeholder items""" if not href or not title or href.startswith('javascript:'): return False if title.strip() in SKIP_TITLES: return False if not date_str or date_str.strip() in SKIP_TITLES or len(date_str.strip()) < 8: return False if len(title.strip()) < 5: return False return True # Pattern 1: li in ul with class for li in soup.select('ul.list li, ul.news_list li, ul.right-list li, ul li'): a = li.find('a') span = li.find('span') if a and span: href = a.get('href', '') title = a.get('title', '') or a.get_text(strip=True) date_str = span.get_text(strip=True) if is_valid_item(href, title, date_str): if not href.startswith('http'): href = BASE_URL + href items.append({'title': title, 'url': href, 'date': date_str[:10]}) # Pattern 2: table rows if not items: for tr in soup.select('table tr'): tds = tr.find_all('td') if len(tds) >= 2: a = tds[0].find('a') if a: href = a.get('href', '') title = a.get('title', '') or a.get_text(strip=True) date_str = tds[-1].get_text(strip=True) if is_valid_item(href, title, date_str): if not href.startswith('http'): href = BASE_URL + href items.append({'title': title, 'url': href, 'date': date_str[:10]}) # Pattern 3: div with content if not items: for item_div in soup.select('.content-list li, .list-item, .xxgk-list li'): a = item_div.find('a') spans = item_div.find_all('span') if a and spans: href = a.get('href', '') title = a.get('title', '') or a.get_text(strip=True) date_str = spans[-1].get_text(strip=True) if is_valid_item(href, title, date_str): if not href.startswith('http'): href = BASE_URL + href items.append({'title': title, 'url': href, 'date': date_str[:10]}) return items def fetch_detail(url): """Fetch detail page content""" cookies = solve_jsl() if not cookies: return "", "" r = fetch_with_cookies(url.replace(BASE_URL, ''), cookies) if not r or r.status_code != 200: return "", "" soup = BeautifulSoup(r.text, 'html.parser') # Date from meta meta_date = soup.find('meta', attrs={"name": "PubDate"}) date_str = '' if meta_date and meta_date.get('content'): date_str = meta_date['content'][:10] # Try various content selectors content_div = None for selector in ['div#zoom', 'div.content', 'div.main-content', 'div.article-content', 'div.bt-content', 'div.TRS_UEDITOR', 'div.trs_editor_view', 'div.text-content', 'div.content-body', 'div#content', '.article-content', '.news-content']: content_div = soup.select_one(selector) if content_div: break content_html = "" if content_div: content_html = str(content_div) return content_html, date_str def store_item(title, url, content, date_str): try: import sqlite3 conn = sqlite3.connect(DB_PATH, timeout=60) c = conn.cursor() summary = BeautifulSoup(content, 'html.parser').get_text(separator=' ', strip=True)[:500] if content else '' c.execute("INSERT OR IGNORE INTO gov_raw (site_name, title, page_url, publish_date, content, source_url, summary, category) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (SITE_NAME, title, url, date_str, content, url, summary, "hpgs")) if c.rowcount > 0: stats["new"] += 1 else: stats["skip"] += 1 conn.commit() conn.close() except Exception as e: print(f" [ERROR] DB: {e}") stats["errors"] += 1 def crawl(full=False, test_n=None): print("=" * 60) print("Hefei Gov EIA Announcements crawler") print("=" * 60) for page in range(1, 51): print(f"\n--- Page {page} ---") items = get_list_page(page) if not items: print(" No items found") break print(f" Got {len(items)} items") if items: print(f" Range: {items[-1]['date']} ~ {items[0]['date']}") seen = 0 for idx, it in enumerate(items): title = it['title'] url = it['url'] date_str = it['date'] if date_str and date_str < THREE_YEARS_AGO: print(f" [SKIP] Too old: {date_str} {title[:30]}") stats["skip"] += 1 continue print(f" [{idx+1}] {title[:40]} | {date_str}") if test_n and seen >= test_n: print(f"\n Test mode: {test_n} items done") return # Need to solve JSL for each detail page too content, detail_date = fetch_detail(url) final_date = detail_date or date_str store_item(title, url, content, final_date) seen += 1 time.sleep(0.5) if not full: break print(f"\nDone! New: {stats['new']}, Skip: {stats['skip']}, Errors: {stats['errors']}") if __name__ == '__main__': full = '--full' in sys.argv test_n = None for arg in sys.argv: if arg.startswith('--test='): test_n = int(arg.split('=')[1]) crawl(full=full, test_n=test_n)