#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 镇江经济技术开发区 - 项目环评（审批）爬虫 栏目: https://www.zjna.gov.cn/zjna/fdzdgknr/xxgk.shtml -> 公共监管 > 生态环境 > 项目环评（审批） 列表: https://www.zjna.gov.cn/zjna/xxgkxmhpjgys/xxgk_list.shtml 注意: 必须 https://www.zjna.gov.cn (带www+https), 裸域名 502 静态分页: xxgk_list_N.shtml (createPageHTML 36页 706条, 20条/页) 详情: /zjna/xxgkxmhpjgys/YYYYMM/UUID.shtml 标题: (完整) 正文: div#zoomcon (UCAP CMS, 包裹), 保留

/ HTML 附件: 正文表格内嵌 附件名 (urljoin 绝对化) 用法: python3 crawl_zjna_xmhpjgys.py [--pages=N] [--limit=N] """ import os, sys, re, time, json, html as html_mod import requests from urllib.parse import urljoin sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from crawler_lib import push_to_searchdb BASE_URL = "https://www.zjna.gov.cn" LIST_URL = BASE_URL + "/zjna/xxgkxmhpjgys/xxgk_list.shtml" SITE_NAME = "镇江经济技术开发区-项目环评（审批）" CATEGORY = "项目环评（审批）" MAX_PAGES = 36 # createPageHTML 声明 36 页 706 条, 20条/页 UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36" HEADERS = {"User-Agent": UA, "Referer": LIST_URL} CUTOFF = "2020-01-01" # 增量日跑窗口 (首次全量可 --pages=36 抓全) def parse_args(): pages, limit = 0, 0 for a in sys.argv[1:]: if a.startswith("--pages="): pages = int(a.split("=")[1]) elif a.startswith("--limit="): limit = int(a.split("=")[1]) return pages, limit def clean_title(t): t = html_mod.unescape(t or "") t = re.sub(r"[\u200b\u200e\u200f\ufeff\xa0]", "", t) t = re.sub(r"\s+", " ", t) t = re.sub(r"^[•·\-—]\s*", "", t) t = re.sub(r"^·?\s* ?\s*", "", t) return t.strip() def fetch_list(session, page): """静态分页: xxgk_list.shtml (p1) / xxgk_list_N.shtml""" if page == 1: url = LIST_URL else: url = f"{BASE_URL}/zjna/xxgkxmhpjgys/xxgk_list_{page}.shtml" try: r = session.get(url, timeout=30) if r.status_code != 200: return None, None r.encoding = "utf-8" return r.text, url except Exception as e: print(f" [WARN] fetch_list p{page} failed: {e}", file=sys.stderr) return None, None def parse_list(html_text): """解析 ul.pageList.newsList > li > a[title] + span.time""" items = [] # 逐 li 提取 (正则, 避免 bs4 依赖) for m in re.finditer(r"
  • \s*]*title=\"([^\"]*)\"[^>]*>.*?\s*\s*([^<]+?)\s*", html_text, re.S): href, title, date = m.group(1), m.group(2), m.group(3).strip() if not href.endswith(".shtml"): continue url = urljoin(BASE_URL, href) title = clean_title(title) date = re.search(r"(\d{4}-\d{2}-\d{2})", date) date = date.group(1) if date else "" if not title or not date: continue items.append({"url": url, "title": title, "date": date}) return items def clean_content(raw_html, page_url): """正文清洗: 保留

    /

    • HTML, 附件绝对化""" # 剥 UCAPCONTENT 包裹 raw_html = re.sub(r"", "", raw_html) # 附件/图片相对路径绝对化 (以详情页 URL 为基址) def _abs(m): href = m.group(1) if href.startswith(("http://", "https://", "javascript:", "mailto:", "tel:")): return m.group(0) abs_url = urljoin(page_url, href) return f']*)>', _abs, raw_html) raw_html = re.sub(r"", "", raw_html) raw_html = re.sub(r"", "", raw_html) # 规范化

    标签 (去属性, 保证闭合

    存在走 HTML 渲染) raw_html = re.sub(r"]*>", "

    ", raw_html) # o:p 标签 (Word 内嵌) 剥除 raw_html = re.sub(r"", "", raw_html) raw_html = re.sub(r"", "", raw_html) # 其余标签去属性但保留结构 (table/tr/td/th 保留) raw_html = re.sub(r"<(table|tbody|thead|tr|td|th|div|span|br)([^>]*)>", lambda m: f"<{m.group(1)}>", raw_html) raw_html = re.sub(r"", lambda m: f"", raw_html) # 剥残留标签 (⚠️ 不能用 , o 会误伤 ) raw_html = re.sub(r"]*>", "", raw_html, flags=re.I) # 段落内空白清理 raw_html = re.sub(r"[\u3000\xa0]", " ", raw_html) # 空段落删除 raw_html = re.sub(r"

    \s*(?: )?\s*

    ", "", raw_html) # 段落间空行 raw_html = re.sub(r"\r\n", "\n", raw_html) return raw_html.strip() def fetch_detail(session, url): """抓详情页: 标题/日期/正文/附件""" try: r = session.get(url, timeout=30) if r.status_code != 200: return None r.encoding = "utf-8" html_text = r.text except Exception as e: print(f" [WARN] fetch_detail failed {url}: {e}", file=sys.stderr) return None # 标题: meta ArticleTitle (完整) m = re.search(r'ArticleTitle"\s+content="([^"]*)"', html_text) title = clean_title(m.group(1)) if m else "" if not title: m = re.search(r"([^<]*)", html_text) title = clean_title(m.group(1)) if m else "" # 日期: meta PubDate pub_date = "" m = re.search(r'PubDate"\s+content="([^"]*)"', html_text) if m: pm = re.search(r"(\d{4}-\d{2}-\d{2})", m.group(1)) pub_date = pm.group(1) if pm else "" # 正文: div#zoomcon body = "" m = re.search(r']*id="zoomcon"[^>]*>(.*?)\s*', html_text, re.S) if m: body = clean_content(m.group(1), url) if not body: m = re.search(r']*class="article-content[^"]*"[^>]*>(.*?)
    提取 (⚠️ 源站 标签可能无 > 闭合, 用 BeautifulSoup 容错) attachments = [] from bs4 import BeautifulSoup as _BS _soup = _BS(f"
    {body}
    ", "html.parser") for a in _soup.find_all("a", href=True): ahref = a["href"] atext = a.get_text(strip=True) if re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)(\?|$)", ahref, re.I) or "download" in ahref.lower(): attachments.append({"name": atext or ahref.split("/")[-1], "url": ahref}) return { "title": title, "pub_date": pub_date, "content": body, "attachments": attachments, "url": url, } def main(): pages, limit = parse_args() if not pages: pages = 5 # 默认前5页 (增量日跑) if pages > MAX_PAGES: pages = MAX_PAGES session = requests.Session() session.headers.update(HEADERS) all_items = [] seen = set() for pg in range(1, pages + 1): html_text, list_url = fetch_list(session, pg) if not html_text: print(f"[PAGE {pg}] fetch failed, stopping") break items = parse_list(html_text) if not items: print(f"[PAGE {pg}] 0 items, stopping") break new_items = [i for i in items if i["url"] not in seen] seen.update(i["url"] for i in items) all_items.extend(new_items) print(f"[PAGE {pg}] +{len(new_items)} items (total {len(all_items)})") if limit and len(all_items) >= limit: all_items = all_items[:limit] break time.sleep(1.0) print(f"\n列表共 {len(all_items)} 条, 开始抓详情...") valid = [] for i, item in enumerate(all_items, 1): d = fetch_detail(session, item["url"]) if not d or not d["content"]: print(f" [{i}/{len(all_items)}] SKIP 空正文: {item['title'][:40]}") continue # 标题回退: 详情取到的完整标题优先 title = d["title"] or item["title"] pub_date = d["pub_date"] or item["date"] content = d["content"] att = d["attachments"] # 附件嵌入正文尾部 (若正文未含附件链接) att_html = "" if att: parts = [] for _a in att: parts.append(f'

    {_a["name"]}

    ') att_html = "\n".join(parts) if att_html: has_att_link = any( re.search(r"\.(pdf|doc|docx|xls|xlsx|zip|rar|wps|7z)(\?|$)", _m.group(1), re.I) for _m in re.finditer(r'