#!/usr/bin/env python3 """ crawl_sthj_xz.py — 徐州市生态环境局 政府信息公开爬虫 ==================================================== 基于 base_crawler (GovCrawler) + API 模式。 站点: https://sthj.xz.gov.cn/dynamic/zwgk/govInfoPub.html?categorynum=003015 分类: 003015 = 法定主动公开内容（含公示公告 = 环评/环保相关） 数据: - 列表: POST /EWB-FRONT/rest/lightfrontaction/getgovinfolist (分页, 共400条) - 详情: GET /govxxgk/...html (静态页, 含索引号/发布机构/成文日期/信息名称/附件) 用法: python3 crawl_sthj_xz.py # 增量爬 (第1页) python3 crawl_sthj_xz.py --full # 全量爬所有页 python3 crawl_sthj_xz.py --sync # 仅同步 """ import os, re, sys, time, json, urllib.request, ssl from base_crawler import GovCrawler, extract_detail BASE_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, BASE_DIR) SITE_NAME = "徐州市生态环境局" DOMAIN = "sthj.xz.gov.cn" LIST_API = "https://sthj.xz.gov.cn/EWB-FRONT/rest/lightfrontaction/getgovinfolist" SITE_GUID = "fb2ed5f8-902e-48bd-998f-a622a60c0951" DEPT_CODE = "001" CATEGORY_NUM = "003015" CTX = ssl._create_unverified_context() HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/125.0.0.0 Safari/537.36", "Content-Type": "application/json;charset=UTF-8", } class XzGovCrawler(GovCrawler): def fetch_list_page(self, page_index, page_size=20): """调用列表API获取单页""" body = json.dumps({ "deptcode": DEPT_CODE, "categorynum": CATEGORY_NUM, "pageIndex": page_index, "pageSize": page_size, "siteGuid": SITE_GUID, }).encode("utf-8") req = urllib.request.Request(LIST_API, data=body, headers=HEADERS) try: resp = urllib.request.urlopen(req, timeout=25, context=CTX) result = json.loads(resp.read().decode("utf-8")) data = result.get("custom", {}).get("data", []) total = result.get("custom", {}).get("total", 0) return data, total except Exception as e: print(f" ⚠️ API请求失败(page={page_index}): {e}") return [], 0 def fetch_detail(self, infourl): """抓取详情页静态HTML，提取内容""" full_url = f"https://{DOMAIN}{infourl}" if infourl.startswith("/") else infourl req = urllib.request.Request(full_url, headers=HEADERS) try: resp = urllib.request.urlopen(req, timeout=25, context=CTX) html = resp.read().decode("utf-8", errors="replace") except Exception as e: print(f" ⚠️ 详情页加载失败: {e}") return None # 提取标题 title = "" tm = re.search(r']*id="ivs_title"[^>]*>(.*?)', html, re.DOTALL) if tm: title = re.sub(r'<[^>]+>', '', tm.group(1)).strip() if not title: tm = re.search(r'(.*?)', html, re.DOTALL) if tm: title = tm.group(1).replace("_徐州市生态环境局", "").strip() # 提取元数据表 → 组装成结构化正文 meta_rows = {} meta_patterns = [ (r'索引号[^:]*:\s*]*>(.*?)', "索引号"), (r'发布机构[^:]*:\s*]*>(.*?)', "发布机构"), (r'成文日期[^:]*:\s*]*>(.*?)', "成文日期"), (r'信息名称[^:]*:\s*]*>(.*?)', "信息名称"), (r'文号[^:]*:\s*]*>(.*?)', "文号"), ] for pat, key in meta_patterns: m = re.search(pat, html, re.DOTALL) if m: meta_rows[key] = re.sub(r'<[^>]+>', '', m.group(1)).strip() # 提取附件 attachments = [] for am in re.finditer( r']*href="([^"]+)"[^>]*>\s*\s*(.*?)\s*', html, re.DOTALL ): href = am.group(1).strip() name = am.group(2).strip() if href and name: full_href = f"https://{DOMAIN}{href}" if href.startswith("/") else href attachments.append(f"📎 {name}: {full_href}") # 提取 id="ivs_content" 正文（含表格）并清洗内联样式 body_html = "" body_match = re.search( r']*class="cont-p"[^>]*id="ivs_content"[^>]*>(.*?)\s*]*class="cont-load', html, re.DOTALL ) if body_match: body_html = body_match.group(1).strip() # fallback: 政民互动页面正文（div.ewb-article-info） if not body_html: body_match2 = re.search( r']*class="ewb-article-info"[^>]*>(.*?)\s*]*class="(?:share|ewb-attach|cont-load)', html, re.DOTALL ) if body_match2: body_html = body_match2.group(1).strip() # 清洗 script/style body_html = re.sub(r']*>.*?', '', body_html, flags=re.DOTALL|re.I) body_html = re.sub(r']*>.*?', '', body_html, flags=re.DOTALL|re.I) body_html = re.sub(r'[^<]*', '', body_html, flags=re.I) # 猛洗表格内联样式（font-family, font-size, line-height 等撑爆HTML的元凶） body_html = re.sub(r'\s+(?:style|class|height|width|valign|align|bordercolor|cellpadding|cellspacing)="[^"]*"', '', body_html) body_html = re.sub(r'\s+align="[^"]*"', '', body_html) # 构建正文内容（结构化HTML，附件URL可点击） content_parts = [] for key in ["索引号", "发布机构", "成文日期", "信息名称", "文号"]: if key in meta_rows and meta_rows[key]: content_parts.append(f"

{key}：{meta_rows[key]}

") if body_html: content_parts.append(body_html) if attachments: content_parts.append("

附件：

") for a in attachments: if ": " in a and a.startswith("📎"): name, url = a[2:].split(": ", 1) content_parts.append(f'

📎 {name}

') else: content_parts.append(f"

{a}

") content = "\n".join(content_parts) # 提取日期 date = meta_rows.get("成文日期", "") # 摘要（正文+附件URL内嵌） summary_lines = [] for key in ["索引号", "发布机构", "成文日期", "信息名称"]: if key in meta_rows and meta_rows[key]: summary_lines.append(f"{key}: {meta_rows[key]}") if body_html: body_text = re.sub(r'<[^>]+>', '', body_html) body_text = re.sub(r'\s+', ' ', body_text).strip()[:300] if body_text: summary_lines.append(body_text) if attachments: summary_lines.append("附件:") summary_lines.extend(attachments) summary = "\n".join(summary_lines) return { "title": title or meta_rows.get("信息名称", ""), "content": content, "date": date, "attachments": attachments, "summary": summary, } def do_crawl(self): import argparse parser = argparse.ArgumentParser() parser.add_argument("--full", action="store_true", help="全量爬取") parser.add_argument("--pages", type=int, default=1, help="爬取页数") parser.add_argument("--sync", action="store_true", help="仅同步") args, _ = parser.parse_known_args() if args.sync: self.sync_to_server() return max_pages = 9999 if args.full else args.pages print(f"\n📋 分类: {CATEGORY_NUM} (法定主动公开内容)") print(f"📄 最大页数: {'无限' if args.full else max_pages}") for page_index in range(max_pages): items, total = self.fetch_list_page(page_index) if not items: print(f" ⏹ 第{page_index+1}页无数据，结束") break print(f"\n📃 第{page_index+1}页 ({len(items)}条, 共{total}条)") for item in items: title = item.get("title", "") or item.get("realtitle", "") infourl = item.get("infourl", "") infodate = item.get("infodate", "") infoid = item.get("infoid", "") category = item.get("categoryname", "") if not title or not infourl: continue # 去重检查 full_url = f"https://{DOMAIN}{infourl}" if infourl.startswith("/") else infourl conn = self._get_db() exists = conn.execute( "SELECT 1 FROM crawl_results WHERE url=? AND site_id=?", (full_url, self.site_id), ).fetchone() conn.close() if exists: self._stats["skip"] += 1 continue # 抓取详情 detail = self.fetch_detail(infourl) if detail is None: self._stats["errors"] += 1 # 没有详情也用列表数据兜底 detail = { "title": title, "content": f"

信息名称：{title}

发布日期：{infodate}

", "date": infodate, "attachments": [], "summary": f"信息名称: {title}\n发布日期: {infodate}", } self.store_item( detail["title"] or title, full_url, detail["content"], detail.get("date") or infodate, detail.get("summary", ""), ) attach_info = f" +{len(detail.get('attachments',[]))}附件" if detail.get("attachments") else "" print(f" ✅ {title[:35]} | {len(detail.get('content',''))}B{attach_info}") time.sleep(self.sync_delay) # 如果这一页不满20条，说明是最后一页 if len(items) < 20: break print(f" 📊 累计: +{self._stats['new']} 新 | 跳过{self._stats['skip']} | 错误{self._stats['errors']}") if __name__ == "__main__": crawler = XzGovCrawler( site_name=SITE_NAME, domain=DOMAIN, url="https://sthj.xz.gov.cn/dynamic/zwgk/govInfoPub.html?categorynum=003015", db_name="sthj_xz_results.db", sync_delay=0.3, ) crawler.run()