#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 泸州市生态环境局 - 环评审批受理 (sthjj.luzhou.gov.cn) CMS: PowerCMS-like 列表: ul.newsListPic > li > div.con > div.title a.tit (title attr + href) + span.date 分页: /hbyw/hzsp/hpspsl (P1), /hbyw/hzsp/hpspsl_N (P2~290, 10条/页, 2895条合计) 详情: /hbyw/hzsp/hpspsl/content_NNNNNNN, 正文 div.conTxt 注意: CloudWAF封禁境外IP, 需从国内服务器运行 """ import sys, os, re, time, json from datetime import datetime, timezone, timedelta sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb import requests from bs4 import BeautifulSoup _MAX_PG = None for i, a in enumerate(sys.argv): if a == '--pages' and i + 1 < len(sys.argv): _MAX_PG = int(sys.argv[i + 1]) break if _MAX_PG is not None: print("[AutoPg] max_pages=" + str(_MAX_PG)) SITE_NAME = "泸州市生态环境局-环评审批受理" DOMAIN = "sthjj.luzhou.gov.cn" BASE_URL = f"https://{DOMAIN}" LIST_PATH = "/hbyw/hzsp/hpspsl" TOTAL_PAGES = 290 CUTOFF = (datetime.now(timezone.utc) - timedelta(days=365*3)).strftime("%Y-%m-%d") HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", "Referer": f"https://{DOMAIN}/", } def html_table_to_html(table, base_url=""): """保留 HTML 表格结构，仅将相对链接/图片转绝对 URL""" import urllib.parse tbl = BeautifulSoup(str(table), 'html.parser') for a in tbl.find_all('a'): href = a.get('href', '') if href and not href.startswith(('http', 'javascript', '#')): a['href'] = urllib.parse.urljoin(base_url, href) if base_url else href for img in tbl.find_all('img'): src = img.get('src', '') if src and not src.startswith(('http', '//', 'data:')): img['src'] = urllib.parse.urljoin(base_url, src) if base_url else src return str(tbl) def fetch(url, timeout=20): for attempt in range(3): try: r = requests.get(url, headers=HEADERS, timeout=timeout) r.encoding = "utf-8" return r.text except Exception as e: if attempt < 2: time.sleep(2) else: print(f" [WARN] {url}: {e}") return None def make_list_url(page): if page == 1: return BASE_URL + LIST_PATH return BASE_URL + LIST_PATH + f"_{page}" def parse_list_page(html): items = [] soup = BeautifulSoup(html, "html.parser") for li in soup.select("ul.newsListPic > li"): a_tag = li.find("a", class_="tit", href=True) if not a_tag: continue href = a_tag["href"].strip() title = a_tag.get("title", "").strip() or a_tag.get_text(" ", strip=True) if href.startswith("/"): href = BASE_URL + href elif not href.startswith("http"): href = BASE_URL + "/" + href # Date date_span = li.find("span", class_="date") date = date_span.get_text(strip=True) if date_span else "" if title and href: items.append((title, href, date)) return items def _clean_span_spaces(text): text = re.sub(r'(?<=[\u4e00-\u9fff])[ \t\u3000]+(?=[\u4e00-\u9fff])', '', text) text = re.sub(r'(?<=[\u4e00-\u9fff])[ \t\u3000]+(?=\d)', '', text) text = re.sub(r'(?<=\d)[ \t\u3000]+(?=[\u4e00-\u9fff])', '', text) text = re.sub(r'\s+([，。、；：？！)】」》])', r'\1', text) text = re.sub(r'([（【「《])\s+', r'\1', text) text = re.sub(r'(?<=\d)\s+(?=\d)', '', text) return re.sub(r'\s+', ' ', text).strip() def extract_content(conTxt, page_url): parts = [] for child in conTxt.children: name = getattr(child, 'name', None) if not name: continue if name == 'table': tbl_html = html_table_to_html(child, page_url) if tbl_html: parts.append(tbl_html) elif name == 'p': if child.find_parent("table"): continue txt = child.get_text(" ", strip=True) txt = _clean_span_spaces(txt) txt = re.sub(r'&#\d+;', '', txt) if txt and len(txt) > 3: parts.append(txt) parts.append("\n\n") result = ''.join(parts).strip() result = re.sub(r'\n{4,}', '\n\n', result) return result def parse_detail(url): html = fetch(url) if not html: return None, None, None soup = BeautifulSoup(html, "html.parser") # Title title = "" meta = soup.find("meta", attrs={"name": "ArticleTitle"}) if meta: title = meta.get("content", "").strip() if not title: h2 = soup.find("h2", class_="title") if h2: title = h2.get_text(strip=True) # Date date = "" meta_d = soup.find("meta", attrs={"name": "PubDate"}) if meta_d: d = meta_d.get("content", "") m = re.search(r'(\d{4}-\d{2}-\d{2})', d) if m: date = m.group(1) if not date: prop = soup.find("div", class_="property") if prop: m = re.search(r'(\d{4}-\d{2}-\d{2})', prop.get_text()) if m: date = m.group(1) # Content content = "" conTxt = soup.find("div", class_="conTxt") if conTxt: content = extract_content(conTxt, url) return title, content, date def main(): pages_to_fetch = min(TOTAL_PAGES, _MAX_PG) if _MAX_PG else TOTAL_PAGES print(f"=== {SITE_NAME} ===") print(f" 总页数: {TOTAL_PAGES}, 本次: {pages_to_fetch}") all_items = [] for page in range(1, pages_to_fetch + 1): list_url = make_list_url(page) print(f" 第 {page}/{pages_to_fetch} 页: {list_url}") html = fetch(list_url) if not html: print(f" 无法获取, 停止") break items = parse_list_page(html) if not items: print(f" 空列表, 停止") break print(f" 找到 {len(items)} 条") all_items.extend(items) time.sleep(0.3) print(f"\n 列表总计: {len(all_items)} 条\n") if not all_items: return results = [] for idx, (title, url, date) in enumerate(all_items): print(f" [{idx+1}/{len(all_items)}] {title[:40]}...", end=" ", flush=True) dt, content, dd = parse_detail(url) final_title = dt or title final_date = date or dd if not content or len(content.strip()) < 10: print("⚠ 空") continue summary = re.sub(r'<[^>]+>', '', content)[:200].strip() if not summary: summary = final_title[:200] results.append({ "site_name": SITE_NAME, "title": final_title, "url": url, "content": content, "summary": summary, "pub_date": final_date, }) print(f"✅ ({len(content)}字)") if (idx + 1) % 20 == 0: print(f" 进度: {idx+1}/{len(all_items)}, 已入库: {len(results)}") if results: print(f"\n 入库 {len(results)} 条") push_to_searchdb(results, "luzhou_hpspsl") print("完成") if __name__ == "__main__": main()